Skip to content

Commit 7697928

Browse files
ilaila
committed
Propagate group row counts through wrappers
1 parent b7503c9 commit 7697928

3 files changed

Lines changed: 79 additions & 48 deletions

File tree

‎docs/codebase-audit-2026-07-22.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,11 @@ maintenance path. A bug must not be hidden by weakening a test or silently chang
7373
- [ ] **Finish or reverse the `CURRENT_DIFF_RECOMPUTE` removal coherently.** Deterministic SAMPLE, POSITIONAL, and ASOF shapes
7474
are demoted to unconditional full refresh while stale nonlocal strategies, documentation, and benchmark paths remain.
7575
Preserve a typed affected/current-diff recompute path where correctness permits it; delete truly dead strategies.
76+
- [ ] **Strip computed top-k wrappers from stored aggregate state.** CREATE only moves a root `TOP_N` or
77+
`LIMIT -> ORDER_BY` into the user-facing view. Plans such as
78+
`PROJECTION -> TOP_N -> PROJECTION -> AGGREGATE` therefore initialize a limited backing table, while refresh strips the
79+
top-k operator and applies unbounded deltas. Resolve the bound ordering expressions through the projection path, keep the
80+
backing state unbounded, and apply the computed `ORDER BY`/`LIMIT` only in the user-facing view.
7681

7782
## P2: remove SQL-text hacks and duplicate abstractions
7883

‎src/core/plan_rewrite.cpp‎

Lines changed: 50 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -868,6 +868,27 @@ void FoldConstantScalarSubqueries(ClientContext &context, unique_ptr<LogicalOper
868868
plan->ResolveOperatorTypes();
869869
}
870870

871+
static ColumnBinding AppendProjectionPassthrough(LogicalProjection &proj, const ColumnBinding &binding,
872+
const LogicalType &type, const string &alias) {
873+
auto passthrough = make_uniq<BoundColumnRefExpression>(type, binding);
874+
passthrough->alias = alias;
875+
proj.expressions.push_back(std::move(passthrough));
876+
proj.ResolveOperatorTypes();
877+
auto bindings = proj.GetColumnBindings();
878+
if (bindings.empty()) {
879+
throw InternalException("OpenIVM: projection produced no bindings after appending hidden passthrough");
880+
}
881+
return bindings.back();
882+
}
883+
884+
static void PropagateHiddenBindingThroughProjectionPath(vector<LogicalProjection *> &projection_path,
885+
ColumnBinding binding, LogicalType type, const string &alias) {
886+
for (auto it = projection_path.rbegin(); it != projection_path.rend(); ++it) {
887+
binding = AppendProjectionPassthrough(**it, binding, type, alias);
888+
type = (*it)->types.back();
889+
}
890+
}
891+
871892
/// Inject a hidden COUNT(*) (alias `openivm_count_star`) into AGGREGATE_GROUP
872893
/// aggregates that don't already have a reliable total-row-count aggregate.
873894
///
@@ -886,24 +907,37 @@ static void InjectGroupCountStar(unique_ptr<LogicalOperator> &plan) {
886907
// Only inject into the aggregate on the top-level output path. ORDER BY/LIMIT/TOP_N
887908
// preserve the projection schema; inner aggregates under joins or set operations
888909
// are handled by other refresh strategies.
910+
vector<LogicalProjection *> projection_path;
889911
LogicalOperator *node = plan.get();
890-
while (node && node->children.size() == 1 &&
891-
(node->type == LogicalOperatorType::LOGICAL_ORDER_BY || node->type == LogicalOperatorType::LOGICAL_LIMIT ||
892-
node->type == LogicalOperatorType::LOGICAL_TOP_N)) {
893-
node = node->children[0].get();
894-
}
895-
if (!node || node->type != LogicalOperatorType::LOGICAL_PROJECTION || node->children.empty()) {
896-
return;
897-
}
898-
auto &projection = node->Cast<LogicalProjection>();
899-
auto *agg_search = node->children[0].get();
900-
if (agg_search->type == LogicalOperatorType::LOGICAL_FILTER) {
901-
if (agg_search->children.empty()) {
902-
return;
912+
LogicalOperator *agg_search = nullptr;
913+
while (node) {
914+
switch (node->type) {
915+
case LogicalOperatorType::LOGICAL_ORDER_BY:
916+
case LogicalOperatorType::LOGICAL_LIMIT:
917+
case LogicalOperatorType::LOGICAL_TOP_N:
918+
node = node->children.size() == 1 ? node->children[0].get() : nullptr;
919+
continue;
920+
case LogicalOperatorType::LOGICAL_PROJECTION:
921+
projection_path.push_back(&node->Cast<LogicalProjection>());
922+
node = node->children.size() == 1 ? node->children[0].get() : nullptr;
923+
continue;
924+
case LogicalOperatorType::LOGICAL_FILTER:
925+
if (node->children.size() == 1 &&
926+
node->children[0]->type == LogicalOperatorType::LOGICAL_AGGREGATE_AND_GROUP_BY) {
927+
agg_search = node->children[0].get();
928+
}
929+
node = nullptr;
930+
continue;
931+
case LogicalOperatorType::LOGICAL_AGGREGATE_AND_GROUP_BY:
932+
agg_search = node;
933+
node = nullptr;
934+
continue;
935+
default:
936+
node = nullptr;
937+
continue;
903938
}
904-
agg_search = agg_search->children[0].get();
905939
}
906-
if (agg_search->type != LogicalOperatorType::LOGICAL_AGGREGATE_AND_GROUP_BY) {
940+
if (!agg_search || projection_path.empty()) {
907941
return;
908942
}
909943
auto &agg = agg_search->Cast<LogicalAggregate>();
@@ -954,10 +988,7 @@ static void InjectGroupCountStar(unique_ptr<LogicalOperator> &plan) {
954988
agg.ResolveOperatorTypes();
955989

956990
ColumnBinding count_binding(agg.aggregate_index, new_agg_idx);
957-
auto count_pt = make_uniq<BoundColumnRefExpression>(count_type, count_binding);
958-
count_pt->alias = openivm::COUNT_STAR_COL;
959-
projection.expressions.push_back(std::move(count_pt));
960-
projection.ResolveOperatorTypes();
991+
PropagateHiddenBindingThroughProjectionPath(projection_path, count_binding, count_type, openivm::COUNT_STAR_COL);
961992
plan->ResolveOperatorTypes();
962993

963994
OPENIVM_DEBUG_PRINT("[PlanRewrite] Injected openivm_count_star for AGGREGATE_GROUP\n");
@@ -1028,27 +1059,6 @@ void PropagateHiddenAggregateColumns(unique_ptr<LogicalOperator> &plan) {
10281059
}
10291060
}
10301061

1031-
static ColumnBinding AppendProjectionPassthrough(LogicalProjection &proj, const ColumnBinding &binding,
1032-
const LogicalType &type, const string &alias) {
1033-
auto passthrough = make_uniq<BoundColumnRefExpression>(type, binding);
1034-
passthrough->alias = alias;
1035-
proj.expressions.push_back(std::move(passthrough));
1036-
proj.ResolveOperatorTypes();
1037-
auto bindings = proj.GetColumnBindings();
1038-
if (bindings.empty()) {
1039-
throw InternalException("OpenIVM: projection produced no bindings after appending hidden passthrough");
1040-
}
1041-
return bindings.back();
1042-
}
1043-
1044-
static void PropagateHiddenBindingThroughProjectionPath(vector<LogicalProjection *> &projection_path,
1045-
ColumnBinding binding, LogicalType type, const string &alias) {
1046-
for (auto it = projection_path.rbegin(); it != projection_path.rend(); ++it) {
1047-
binding = AppendProjectionPassthrough(**it, binding, type, alias);
1048-
type = (*it)->types.back();
1049-
}
1050-
}
1051-
10521062
struct OuterJoinBindings {
10531063
bool found = false;
10541064
bool is_full_outer = false;

‎test/sql/aggregate.test‎

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3579,9 +3579,9 @@ SELECT count(*) FROM (
35793579
----
35803580
0
35813581

3582-
# Group existence must use an unconditional row count even when ORDER BY wraps
3583-
# the aggregate projection. Removing the last non-NULL x leaves the group alive
3584-
# because another row with x=NULL remains.
3582+
# Group existence must use an unconditional row count when computed ORDER BY and
3583+
# LIMIT introduce projections around the aggregate. Removing the last non-NULL x
3584+
# leaves the group alive because another row with x=NULL remains.
35853585
statement ok
35863586
CREATE TABLE ordered_nullable_count_input (id INT, g INT, x INT);
35873587

@@ -3593,19 +3593,23 @@ CREATE MATERIALIZED VIEW ordered_nullable_count_mv AS
35933593
SELECT g, COUNT(x) AS n
35943594
FROM ordered_nullable_count_input
35953595
GROUP BY g
3596-
ORDER BY g;
3596+
ORDER BY g + 0
3597+
LIMIT 2;
35973598

35983599
statement ok
35993600
DELETE FROM ordered_nullable_count_input WHERE id = 2;
36003601

36013602
statement ok
3602-
INSERT INTO ordered_nullable_count_input VALUES (4, 3, NULL), (5, 3, 9);
3603+
INSERT INTO ordered_nullable_count_input VALUES (4, 1, 9), (5, 2, 11);
36033604

36043605
statement ok
36053606
UPDATE ordered_nullable_count_input SET x = NULL WHERE id = 5;
36063607

36073608
statement ok
3608-
DELETE FROM ordered_nullable_count_input WHERE id = 4;
3609+
UPDATE ordered_nullable_count_input SET x = NULL WHERE id = 4;
3610+
3611+
statement ok
3612+
DELETE FROM ordered_nullable_count_input WHERE id = 5;
36093613

36103614
statement ok
36113615
PRAGMA refresh('ordered_nullable_count_mv');
@@ -3614,14 +3618,26 @@ query I
36143618
SELECT COUNT(*) FROM (
36153619
SELECT * FROM ordered_nullable_count_mv
36163620
EXCEPT ALL
3617-
SELECT g, COUNT(x) FROM ordered_nullable_count_input GROUP BY g
3621+
SELECT * FROM (
3622+
SELECT g, COUNT(x)
3623+
FROM ordered_nullable_count_input
3624+
GROUP BY g
3625+
ORDER BY g + 0
3626+
LIMIT 2
3627+
)
36183628
);
36193629
----
36203630
0
36213631

36223632
query I
36233633
SELECT COUNT(*) FROM (
3624-
SELECT g, COUNT(x) FROM ordered_nullable_count_input GROUP BY g
3634+
SELECT * FROM (
3635+
SELECT g, COUNT(x)
3636+
FROM ordered_nullable_count_input
3637+
GROUP BY g
3638+
ORDER BY g + 0
3639+
LIMIT 2
3640+
)
36253641
EXCEPT ALL
36263642
SELECT * FROM ordered_nullable_count_mv
36273643
);

0 commit comments

Comments
 (0)