perf(merge_insert): stream missing partial-schema columns instead of collecting them in the join#7630
Conversation
Benchmark: peak memory before vs afterMemory experiment on a synthetic dataset shaped like the failing workload (wide binary payload column, partial-schema source). Setup:
The key structural difference: before, peak memory is ~3× the total payload size of the whole table regardless of how few rows are updated (the CollectLeft build side materializes the payload column for every target row, then join/write copies stack on top). After, peak memory tracks the number of matched rows (4× fewer updated rows → ~3× lower peak) and no longer grows with the payload width of the target table at all. Extrapolated to the motivating workload (hundreds of GB of payload columns, updating 30–40% of rows via a partial-schema source): before the fix the build side alone must hold the table's full payload in RAM, which OOMs any reasonable machine; after the fix the merge streams. |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
…collecting them in the join
A partial-schema merge_insert filled the dataset columns missing from
the source by copying them out of the target side of the join
(with_column(col("target.x"))). The target scan feeds the CollectLeft
build side of the hash join, so every payload column of every target
row was collected into memory before the source streamed through, and
the plan executes with an unbounded memory pool. On wide tables (rows
carrying large binary columns, datasets in the hundreds of GB) this
reliably OOM-kills the process even when the merge only touches a
small fraction of rows.
The join now carries only the join keys, _rowid/_rowaddr, the source
columns, and the action column. The write exec fetches the missing
columns per output batch with a take by row address
(TargetColumnFiller): updated rows read the matched target row's
values, inserted rows fill NULL. Peak memory is bounded by the batch
size, and only the rows actually rewritten are read, instead of the
full column for every target row.
264a161 to
1e49757
Compare
wjones127
left a comment
wjones127
left a comment
There was a problem hiding this comment.
I appreciate you taking an interest in this work. However, this isn't aligned with the direction we are going for.
One fix we plan to do immediately:
- Get statistics from the input data, so we can make the input data the build side instead of the target table, when the input data is smaller. #7368
- Support filter pushdown from hash joins: #7562
- Consider using TakeExec to defer read of large columns using an optimizer rule: #7363
That third one is closer to what you are trying to do here, except it does it via an optimizer rule so it can be shared with more query type. In general, the direction we are trying to go for is write less bespoke merge-insert code, and write more general query optimization primitives.
The problem I found with deferring the take is that it's not always beneficial, particularly for narrow columns. So more work is needed to design a better optimizer rule.
On your benchmark, one thing that isn't clear to me is how much of the same benefit could be gotten by applying PR #7368 to flip the join order. Perhaps worth trying that same situation out on that PR branch to compare to this solution. Perhaps it can do just as well.
|
Thank you for pointing out. Closing in favor of the optimizer-primitive direction — #7363 covers the failure mode this |
2 participants
A partial-schema merge_insert filled the dataset columns missing from the source by copying them out of the target side of the join (with_column(col("target.x"))). The target scan feeds the CollectLeft build side of the hash join, so every payload column of every target row was collected into memory before the source streamed through, and the plan executes with an unbounded memory pool. On wide tables (rows carrying large binary columns, datasets in the hundreds of GB) this reliably OOM-kills the process even when the merge only touches a small fraction of rows.
The join now carries only the join keys, _rowid/_rowaddr, the source columns, and the action column. The write exec fetches the missing columns per output batch with a take by row address (TargetColumnFiller): updated rows read the matched target row's values, inserted rows fill NULL. Peak memory is bounded by the batch size, and only the rows actually rewritten are read, instead of the full column for every target row.
Related to #1983 — fixes the partial-schema OOM; a configurable memory pool for the remaining paths is follow-up work.