Materialize: Spike: explore time-sources

[awang]

While time operators are possible in queries, it's difficult to do time-related comparisons in a materialized view, as it's difficult to map.

As a thought experiment, we should try out the dead-simple prototype of pinging an input each 1s with an update, and trying to use that in a query. It will have defective performance, but we should be able to determine if that level of defect works for customers before building it.

The immediate next step task is a time-bounded (1/2-1 day exercise) around exploring Frank's suggested workaround (above), running some hypothetical queries, and better understanding the limits on that approach. The goal of this is to identify just how well our workaround might work before recommending it to users, and to explore potential ways forward for how to interact with time

cc @frankmcsherry
Related to https://github.com/MaterializeInc/materialize/issues/2439 and https://github.com/MaterializeInc/product/issues/7 and

[umanwizard]

@frankmcsherry are you able to comment on why the perf might be defective?

[benesch]

If you join via SELECT * FROM t WHERE ts > now() - '30 day', that'll get planned as a cross join, because we don't support inequality joins. So every time now() ticks you'll have a boatload of recomputation to do if t is large.

[frankmcsherry]

Additionally, even if we supported inequality joins we would need to maintain an index over all of t, whereas many users of SELECT * FROM t WHERE ts > now() - '30 day' would expect t to be transiently maintained and then retired once the 30 day interval had elapsed.

[JLDLaughlin]

@benesch why don't we support inequality joins?

[frankmcsherry]

It's a matter of implementing a differential dataflow operator to do it, or thinking up a clever way to put together a few operators. In particular, you most likely want the operator to do incremental work proportional to the change in values.

For example, if you have

select * from a, b
where a.x = b.x and a.y < b.z

then for each group, if someone changes a.y say to a.y' the goal is to figure out how to quickly get at the b.z that lie between a.y and a.y' (and figure out their sign, to know if they are now included in a result or discarded).

[frankmcsherry]

So, we currently support them in the sense that you can 100% write them, but they will be planned as an equijoin followed by a filter. In the case of time sources (where there is no other key) it would be a cross join followed by a filter, which is just painful. It would be less painful if we had an inequality join operator in DD and only had to deal with changes that were

(now(), now(), -1)
(now() + 1, now() + 1, +1)

because there is fairly little movement in the value, and in principle we could leap directly to the data whose columns lie in between the values.

[frankmcsherry]

There is an implementation you could put together if you knew that the value domain was .. integer-like. You could assemble for each i the results where you bin the values into bins of width 2^i, extract the corresponding bits out of the threshold, and cross join wherever the bits are set. This doesn't sound much better (maybe worse), but it means that if a.y changes to a.y' and only lower-order bits change, and we only need to update cross joins for narrow buckets.

[JLDLaughlin]

To test this, I've started up two EC2 instances in the scratch AWS environment. In the first instance, I've just created the source and views described in our wikipedia demo. In the second instance, I've done the following:

1. Created a dummy time source as described by @quodlibetor here -- thanks @quodlibetor!
2. Materialized the same views from the wikipedia demo, but as of the last five minutes:

CREATE MATERIALIZED VIEW fivemincounter AS 
SELECT COUNT(*) FROM recentchanges WHERE r_ts > (SELECT max - 300 FROM current_time_view);

CREATE MATERIALIZED VIEW fiveminsuseredits AS
SELECT user, COUNT(*) FROM recentchanges WHERE r_ts > (select max - 300 FROM current_time_view) 
GROUP BY user;

CREATE MATERIALIZED VIEW fiveminstop10 AS
SELECT * FROM fiveminsuseredits
ORDER BY count
DESC LIMIT 10;

I plan on letting both instances run for ~24 hours and comparing their memory usage at that time.

[JLDLaughlin]

Update: the materialized instance with the time source exited with code 137 (OOM). Here are screenshots of the Grafana dashboards to compare:

Without time source:

With time source:

[JLDLaughlin]

I'm going to restart the test with two r5ad.4xlarge EC2 instances -- will post updates here.

[JLDLaughlin]

Even using a r5ad.4xlarge EC2 instance, the materialized instance maintaining the time source exited with code 137 (OOM) after ~8 hours.

Without time source:

With time source:

At this point, I don't think we can reasonably recommend that anyone create their own per-second time source. On the other hand, a daily time source might work? (I can run a test for that).

[awang]

Sounds good! What about the querying/joining the time source? Are Nikhil & Frank's recommendations straightforward to write in practice?

[JLDLaughlin]

Yes! Querying and joining with the time source is straightforward.

[frankmcsherry]

I think neither inequality joins nor automatically retiring data are "straightforward". Inequality joins probably involve inventing something (a DD implementation or pattern) and retiring data needs a bit of help rewriting queries whose scope is currently unknown to me.

[JLDLaughlin]

Sorry - I may have misunderstood the question. I meant actually querying and joining (as in executing the queries I'd written above), not implementing an improvement.

[benesch]

Why did memory usage balloon? That is surprising to me. Yes, we need to
cross join, but we need to cross join with a single row table and that
shouldn’t require any additional memory?

I’m just waking up, so quite possible I’m missing something.

On Tue, Aug 18, 2020 at 9:21 AM Jessica Laughlin [e-mail]
wrote:

>
>

Sorry - I may have misunderstood the question. I meant actually querying
and joining (as in executing the queries I'd written above), not
implementing an improvement.

—
You are receiving this because you were mentioned.

Reply to this email directly, view it on GitHub
https://github.com/MaterializeInc/materialize/issues/3880#issuecomment-675475074,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/AAGXSICLX4MFUCHMTPETRDLSBJ553ANCNFSM4PW6XGRQ
.

[frankmcsherry]

I'm not certain the reason here, but at the least we'll need to hold the input in an arrangement for a join, with the time field preserved. Without the time source, we would be able to avoid that arrangement, and potentially project out the time field and collapse down more?

[frankmcsherry]

It's also possible, though again not certain, that MZ is just falling behind and OOMing because it has 1,000 outstanding seconds and that means 1,000x the input relation to work its way though?

[umanwizard]

@JLDLaughlin thanks for testing this out. Seems like we can mark the quick 'n dirty approach as "not feasible", which we always knew was in the range of outcomes.

@frankmcsherry would a /prof trace help diagnose why we're OOMing, or do you already have a fairly good idea?

[frankmcsherry]

I am 100% not certain! It would be great for us to nail down diagnostic procedure though. I'm thinking if it is falling behind we should see that in the timestamp lag between output and input? If it is not falling behind (i.e. is stable) then more traditional tools might explain it?

It's reasonable to think through "does this look the way we expect?" and imo for the expectation of "falls behind on CPU, because re-upping the data 1/s is hard" this does look like symptoms you would expect. That doesn't need to be the end of the diagnosis, but imo no outstanding mystery (but, 100% up for others wanting more clarity / smacking down easy hypotheses).

[JLDLaughlin]

I re-ran the test for a daily time source (meaning a time source that ticks once a day instead of once a second):

Memory didn't balloon in the same way, but is still steadily increasing.

[JLDLaughlin]

The testing around this was a failure, but since then @frankmcsherry has improved the performance of max() on a monotonic stream: https://github.com/MaterializeInc/materialize/pull/3994. This does not fix the problem with the frequent cross join, but might improve performance a bit.

[krishmanoh2]

I have a shell script that is populating a file on disk with the time updated 1 sec and am using that file as a source. I am maintaining the max() value from that.

So far, it is working fine. I am not running out of memory. I am doing a cross join with other views. However, unsure of performance implications of cross join of a single row view.