[dataflow] Fuel the arrangement to stream operator #7171
Conversation
|
@philip-stoev when next we meet, it would be great to discuss how to test this. Specifically, it has behavior that should only be different on collections that have more than 1M records. That number can be dialed around, but it seems likely to be baked in somewhere (roughly: after how many records should the operator yield control). I noticed at least one bug (pre-fixed) that would have resulted in repeated data being output, but I could also imagine that I have others. |
|
@aljoscha this could potentially reduce some of the sink build up you were seeing. I suspect it was at least in some part due to the source of data (an arrangement) spilling out all of the data at once, rather than as downstream operators handle it. Might be a bit before this lands (wants testing) but I wanted to point it out in case you wanted to drive it around and see what it does. |
|
I've tested this against |
| .unwrap() | ||
| .do_work(&key, &mut logic, &mut fuel, output); | ||
| if fuel > 0 { | ||
| todo.pop_front(); |
There was a problem hiding this comment.
I'd consider asserting here that the PendingWork removed is indeed completed. It looks like it is from the do_work implementation, but it's always good to avoid surprises.
There was a problem hiding this comment.
Like, some sort of assert that if you try and read from or step the cursor you get nothing back?
|
@frankmcsherry Because you suggested this as a potential fix for #7043 a while back: it doesn't fix it. The problem there is that the sink machinery that sorts by timestamp and stashes messages until the frontier passes their timestamp still needs to wait for the frontier to advance. In the extreme case of all data having one and the same timestamp, that means waiting for all data and sorting all data in one go. Fueling the import doesn't change that problem because the imported data still all have the same timestamp. What would help is if the import could compact the timestamp forward as the import is happening, thus yielding a smoother distribution of timestamps. Is that at all possible with how an arrangement import currently works? |
|
Afaict, you are asking if the import could "lie" about the timestamp so that the progress tracking machinery could report "complete" times? I think in any situation that this is safe, there is no consolidation happening, and the code can instead just rely on the fact that arrangements can produce compacted data that can just be emitted once seen. I feel like there is some random mapping and such that lives between the arrangement and the sink. I have no clue what it is, but if we can nail that down (and constrain it appropriately) then the data could just be streamed out. Sinks could also read directly out of arrangements, per some issue from like a year ago. Them wanting to sort by time is neat, but it comes with this performance penalty. |
I' not 100% sure. Also, I think what I am asking for is not possible with how arrangements work. Imagine you're sinking an arrangement that has 100m entries, compaction window is set to 1ms. Importing it takes 20 seconds, but all records have the timestamp that they had when importing started. What I was asking for is that (timestamp) compaction is reflected in the records as they are emitted over 20 seconds. But I think that's not possible, because at the start of the import, you basically get a cursor over the records as of the start time and holding that cursor keeps them from being compacted, right?
Unfortunately, we can't just stream them out because of the exactly-once business that the sink does. Or at least not with the current implementation. I had some thoughts written down about that here: https://gist.github.com/aljoscha/9d134bbec1b5fceacf8a2da791a44aef
Yes! I'm aware of #2808 and would love to do it, but I think that also wouldn't easily work to provide the guarantees that the sink currently provides. There's a short comment on that in https://gist.github.com/aljoscha/9d134bbec1b5fceacf8a2da791a44aef as well. |
|
@frankmcsherry many apologies, I did not see this PR previously. Will work on it today. EDIT: RE the 1M limit , I am going to patch this locally to be 10 records or some other very small value and this should bring more of the existing tests to bear. I will also use the opportunity to see how we can start creating a "1M rows" body of tests that would be exercising the product across the board with more than a few records per source/table. |
|
Item No 1. The following query never completes with refuel =1 and refuel=10 (I gave it 20 minutes on 4 cores) but is instantaneous with refuel =1000 and higher values. CPU is pegged at 400%, mz_message_counts reports hundreds of millions of messages are being exchanged. The record counters on the SVG graph show no progress. As the participating tables have hundreds of records at most, the query should be possible to evaluate even at refuel = 10. Therefore, I think some endless loop situation has developed. Perf reports the following: The query is: To reproduce, modify refuel to be 10 and then run this SQL file through the psql client: gh7171-item1.sql.zip |
|
Looking into this, it looks like the differential dataflow cursor implementation will reset the value position when you seek a key, rather than leaving it intact. That's not something that has been spec'd out (a bug on my part), but it means that in this PR with indexed access, if you don't completely enumerate the values for a key you'll return and start over on them. In particular, I think the problematic part of the query above is the last constraint I'll work on a fix for DD, or for here, once I sort out what the right cursor behavior should be. |
|
The above commit defends against the uncertain cursor semantics, and corrects the issue for |
|
The failures seem .. idk maybe spurious, or maybe just wrong: |
|
Yes, this one is fixed on main.
…On Wed, Aug 4, 2021, 13:50 Frank McSherry ***@***.***> wrote:
The failures seem .. idk maybe spurious, or maybe just wrong:
pg-cdc.td:85:1: error: expected error containing 'error connecting to server: failed to lookup address information: Name or service not known', but got 'error connecting to server: failed to lookup address information: Temporary failure in name resolution: failed to lookup address information: Temporary failure in name resolution'
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
<#7171 (comment)>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/AAJYN2QSS35HSCO36CBWBTDT3ELPRANCNFSM47I7LKCA>
.
Triage notifications on the go with GitHub Mobile for iOS
<https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675>
or Android
<https://play.google.com/store/apps/details?id=com.github.android&utm_campaign=notification-email>
.
|
frankmcsherry
force-pushed
the
fuel_arrangement
branch
from
b0df756
to
a5c56ef
Compare
There was a problem hiding this comment.
Ok so:
- buildkite is green with refuel=1 , refuel=10
- chbench runs without crashes / loops with refuel=10
- RQG reports no wrong results with refuel=1, refuel=10
- I have pushed the endless loop test case from this PR into the branch, please squash it and take it along for the ride to main.
There are multiple hurdles to be overcome in order to push a dedicated test for this, so it can not be done right now:
- the "refuel" constant should be made configurable for testing purposes using a general mechanism for such constants going forward
- once implemented, ideally we should not need a dedicated
testsubdirectory with amzcompose.ymlfile in it in order to run such a test -- that would be too much of a boilerplate in order to test one small aspect of the system. Maybe themzworkflows.pyapproach that is currently under discussion will allow such tests to be specified in a more concise manner (and hopefully somehow run faster as well).
|
Thanks very much! |
This PR introduces fuel to the primary operator that converts arrangements to streams of updates. This should increase the responsiveness of the system, as queries deriving from arrangements needn't lock up the dataflow with the operators that turn entire arrangements into a stream of updates. At the same time, it may also improve performance, as this provides the ability of the rest of the system to respond as outputs are produced, potentially accumulating them or otherwise retiring them.
This hasn't been tested other than with
sqllogictest, and in particular isn't certain to fix the issues it is pointed at. The default number of records is set at 1M which may be too large, but should be something that we can at least distinguish from not yielding at all (on sufficiently large inputs).The next step is to test this out, determine if it alleviates previously observed responsiveness problems (it will not fix them all, but should be a start).
Fixes #7031