fix(cubestore): inline aggregate dropped rows past the first partition - #11631
Conversation
InlineAggregateExec carried its PlanProperties over to a new child instead of recomputing them. The node is converted from an AggregateExec while a CoalescePartitionsExec sits under it, so the cached output partitioning is 1; EnforceSorting then strips that coalesce as an avoidable bottleneck and reattaches the multi-partition subtree. The stale count stays 1, the parent executes only partition 0, and every row in the remaining partitions is dropped with no error. Delegate the input swap to the AggregateExec the node was built from, then re-convert. DataFusion recomputes the properties there and preserves the output schema, which it derives from aggregate expression names its own rules may rewrite. When the new input is no longer sorted on the group keys the plain hash aggregate is kept, without the limit: on the streaming path it counts complete groups off a sorted input, while a hash aggregate reads it as a cap on distinct groups and would truncate arbitrary ones. Reachable only when every group-by column is pinned to a single value by the filter (which is what makes the aggregate sorted), the leading sort key column is outside the filter set so partition pruning cannot collapse the scan, and the matching rows live outside the first partition. A rollup partitioned by a time dimension that the query does not filter is exactly that shape. Also adds an exhaustive check that MinMaxCondition never prunes a range holding a matching row -- partition pruning was the first suspect and is clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…the new child GroupByLimitAggregateExec refreshed only the output partitioning when its input was swapped, leaving the rest of the plan properties -- orderings and constants projected from the previous input -- behind. Delegate the swap to the aggregate it was built from, the same way the inline aggregate does, so DataFusion recomputes all of them and preserves the output schema. If the rebuilt aggregate no longer fits the trimming path it is returned as is. The other two nodes that carry their properties over, RollingWindowAggExec and AggregateTopKExec, build them from their own output schema plus constants and read the input's partition count live, so nothing there can go stale; say so at both sites, since the same pattern loses rows in a node whose partitioning follows the input. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
github-actions
Bot
added
cube store
|
Claude finished @waralexrom's task in 5m 47s —— View job Review: fix looks correct — 0 blocking, 2 medium, 4 low. 5 inline comments posted.Full review
VerdictThe diagnosis and the fix are sound. Delegating the input swap to the wrapped Two things I liked in particular:
A side effect worth naming: Findings
Things I checked and found clean
Not verifiedI couldn't read the vendored DataFusion fork in this checkout, so the claim that |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| #[test] | ||
| fn no_false_negatives_two_columns() { | ||
| assert_no_false_negatives(2, 3); | ||
| } | ||
|
|
||
| #[test] | ||
| fn no_false_negatives_three_columns() { | ||
| assert_no_false_negatives(3, 3); |
There was a problem hiding this comment.
Worth measuring the wall-clock of no_false_negatives_three_columns in a debug build before merging. Rough count: column_bounds(3, 3) is 13³ = 2197 bound combinations, rows is 27 so ~365 ordered (mn, mx) pairs survive the skip, and each pair runs three .any() scans over matching (up to 125 Vec<i64> elements). That's on the order of 10⁸ heap-allocated-vector comparisons — plausibly tens of seconds unoptimized, which is a lot for one --lib test given the module it guards isn't the one the PR fixes.
If it does turn out slow, assert_no_false_negatives(3, 2) cuts it by roughly an order of magnitude while still covering the multi-column carry logic, or the 3-column case could be #[ignore]d as an on-demand exhaustive check with the 2-column one running by default.
There was a problem hiding this comment.
Measured before merging, debug build:
test queryplanner::partition_filter::tests::no_false_negatives_two_columns ... ok
test queryplanner::partition_filter::tests::no_false_negatives_three_columns ... ok
test result: ok. 2 passed; finished in 1.15s
The 10^8 estimate overcounts: matching is filtered once per bound combination, not per (mn, mx) pair, most combinations leave only a handful of candidates, and each .any() short-circuits. Keeping the 3-column case as is rather than shrinking it to (3, 2) or marking it #[ignore].
Review follow-ups on the inline aggregate fix. The end-to-end regression test asserted only on rows, so a planner change that moved the query off the streaming aggregate, or collapsed the scan to one partition, would have kept it green while covering nothing. Assert the plan shape too, and add the unit-level counterpart the sibling node already has: build the exec over a one-partition input, re-child it onto three, and require both the reported partitioning and the node type to follow. It runs in milliseconds and does not depend on compaction timing. Both fallbacks -- to the hash aggregate here, out of the trimming path in GroupByLimitAggregateExec -- now name the group by and the new input's order mode, so a degradation is diagnosable from the log rather than showing up only as wrong-looking timings. The group-by-limit test also pins that re-childing keeps the trimming exec, which the new fallback would otherwise let it pass without. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
Claude finished @waralexrom's task in 2m 22s —— View job Re-review after push: all but one prior finding addressed — 0 blocking, 1 nit. 4 threads resolved.Full review
What changed since the last review
New review of the added codeChecked and clean:
One nit posted inline ( Not verifiedI did not build or run the suite in this checkout (no |
| /// A group continuing in the next input batch must not be emitted early with a partial sum. | ||
| /// A partial aggregate runs once per input partition, so its reported partitioning must | ||
| /// follow a re-childed input. A stale single-partition count makes the parent coalesce | ||
| /// execute only partition 0 and silently drop the rows of the rest -- the row loss this node | ||
| /// was fixed for. Built over a 1-partition input so its cache says 1, then re-childed onto 3. |
There was a problem hiding this comment.
The new test was inserted between the pre-existing doc line and the test it documented, so limit_emits_only_closed_groups lost its comment and the new test now carries a first line about a different behavior:
| /// A group continuing in the next input batch must not be emitted early with a partial sum. | |
| /// A partial aggregate runs once per input partition, so its reported partitioning must | |
| /// follow a re-childed input. A stale single-partition count makes the parent coalesce | |
| /// execute only partition 0 and silently drop the rows of the rest -- the row loss this node | |
| /// was fixed for. Built over a 1-partition input so its cache says 1, then re-childed onto 3. | |
| /// A partial aggregate runs once per input partition, so its reported partitioning must | |
| /// follow a re-childed input. A stale single-partition count makes the parent coalesce | |
| /// execute only partition 0 and silently drop the rows of the rest -- the row loss this node | |
| /// was fixed for. Built over a 1-partition input so its cache says 1, then re-childed onto 3. |
and move /// A group continuing in the next input batch must not be emitted early with a partial sum. back down onto limit_emits_only_closed_groups (line ~496).
2 participants
Summary
A grouped query over a rollup could silently return no rows — or an undercount — when every group-by dimension was pinned to a single value by a filter.
InlineAggregateExeccarried itsPlanPropertiesover to a new child instead of recomputing them, so it kept reporting one output partition while its input had several; the parent then executed only partition 0 and dropped every row living in the rest, with no error anywhere. Fixes CORE-778 (reported by Maersk onassetmgmt-prod).Changes
InlineAggregateExec::with_new_childrendelegates the input swap to theAggregateExecthe node was converted from, then re-converts. DataFusion recomputes the properties there and preserves the output schema, which it derives from aggregate expression names its own rules may rewrite (try_newwould regenerate them and can tripschema_check). If the new input is no longer sorted on the group keys, the plain hash aggregate is kept — without the limit, which means "stop after N complete groups" on the streaming path but "cap on distinct groups" for a hash aggregate.GroupByLimitAggregateExecrefreshed only the output partitioning and kept orderings and constants projected from the previous input; it now uses the same delegation.RollingWindowAggExecandAggregateTopKExecalso carry their properties over, but build them from their own output schema plus constants and read the input's partition count live, so nothing can go stale. Both sites now say so, since the same pattern loses rows in a node whose partitioning follows the input.How it fired
All of these had to hold at once, which is why it looked intermittent:
Sortedand hands it to the streaming implementation. WithIN (a, b)there is no constant, the plan staysPartiallySorted*and is correct, which is the whole=vsINasymmetry in the report;Sequence inside a single physical-optimizer run, no router/worker boundary involved:
ensure_partition_mergeputs aCoalescePartitionsExecunder the aggregate, the node is converted with a cached partition count of 1, and DataFusion'sEnforceSorting::remove_bottleneck_in_subplanthen strips that coalesce and reattaches the multi-partition subtree.Present since the DataFusion 46 upgrade (
4ef3442827), which introduced the node.Testing
sql::tests::single_value_equals_scans_every_partition— end-to-end regression test on the reported shape: a rollup-like table whose sort key leads with a time dimension, several partitions, both group columns pinned. Fails on the oldwith_new_children(returns no rows), passes with the fix. It probes the last month of the range on purpose: a probe near the start of the sort key can land in partition 0 and go green with the bug present.group_by_limit_aggregate::tests::output_partitioning_follows_rechilded_inputcovers the same class for the second node and still passes.partition_filter::tests::no_false_negatives_{two,three}_columns— exhaustive check thatMinMaxConditionnever prunes a range holding a matching row. Partition pruning was the first suspect and is clean; this pins that.cargo test -p cubestore --libgreen (317 tests), several consecutive runs.