Handle computed and distinct SUM outputs

ila · ila · commit 9f9832456a63 · 2026-07-23T16:41:57.000+02:00
diff --git a/src/core/ivm_view_classifier.cpp b/src/core/ivm_view_classifier.cpp
@@ -353,7 +353,7 @@ static void BuildModelFeatures(DeltaViewModel &model, const PlanAnalysis &analys
 		AddUnique(model.features, DeltaModelFeature::WINDOW_AFFECTED_PARTITION);
 	}
 	if (!model.HasFeature(DeltaModelFeature::FULL_ONLY)) {
-		if (input.distinct_aux_candidate) {
+		if (input.distinct_aux_candidate && !input.has_top_level_redundant_scalar_distinct) {
 			AddUnique(model.features, DeltaModelFeature::DISTINCT_STATEFUL);
 		}
 		if (input.semi_anti_aux_candidate) {
@@ -559,10 +559,19 @@ static void SelectRefreshType(DeltaViewModel &model, const PlanAnalysis &analysi
 		model.type = RefreshType::GROUP_RECOMPUTE;
 	} else if (analysis.found_filtered_list) {
 		model.type = RefreshType::FULL_REFRESH;
+	} else if (input.has_computed_sum_aggregate_projection) {
+		model.type = model.group_columns.empty() ? RefreshType::FULL_REFRESH : RefreshType::GROUP_RECOMPUTE;
+	} else if (analysis.found_distinct && !analysis.found_union_distinct && model.distinct_at_top &&
+	           analysis.found_aggregation) {
+		// DISTINCT over multiple aggregate result rows is a second non-linear
+		// aggregation level. The outer DISTINCT key is an aggregate value, not
+		// a source group key that affected-group recompute can recover.
+		model.type = RefreshType::FULL_REFRESH;
 	} else if (analysis.found_count_distinct && !model.group_columns.empty()) {
 		model.type =
 		    input.count_distinct_aux_candidate ? RefreshType::COUNT_DISTINCT_INCREMENTAL : RefreshType::GROUP_RECOMPUTE;
-	} else if (analysis.found_distinct && !model.distinct_at_top && analysis.found_aggregation) {
+	} else if (analysis.found_distinct && !input.has_top_level_redundant_scalar_distinct && !model.distinct_at_top &&
+	           analysis.found_aggregation) {
 		model.type = model.HasFeature(DeltaModelFeature::DISTINCT_STATEFUL) ? RefreshType::DISTINCT_INCREMENTAL
 		                                                                    : RefreshType::GROUP_RECOMPUTE;
 	} else if (model.union_distinct_over_agg && !model.group_columns.empty()) {
@@ -878,6 +887,9 @@ DeltaViewModel BuildDeltaViewModel(const DeltaViewModelInput &input) {
 
 	model.has_minmax_metadata = analysis.found_minmax || analysis.found_count_distinct || analysis.found_list;
 	model.distinct_at_top = IsDistinctAtTop(analysis, output_names);
+	if (input.has_top_level_redundant_scalar_distinct) {
+		model.distinct_at_top = false;
+	}
 	BuildGroupColumns(model, facts, output_names, input.visible_output_count);
 
 	if ((analysis.found_left_join || analysis.found_full_outer) && analysis.found_aggregation &&
diff --git a/src/core/parser.cpp b/src/core/parser.cpp
@@ -371,6 +371,7 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
 	bool pre_rewrite_has_aggregate_filter = false;
 	bool has_hidden_minmax_having = false;
 	bool has_computed_minmax_aggregate_projection = false;
+	bool has_computed_sum_aggregate_projection = false;
 	{
 		auto select_parse_plan_start = create_profile_now();
 		Parser select_parser;
@@ -411,6 +412,7 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
 		stored_query_has_aggregate_filter = post_rewrite_facts.has_filter_above_aggregate;
 		has_hidden_minmax_having = post_rewrite_facts.has_hidden_minmax_having_column;
 		has_computed_minmax_aggregate_projection = post_rewrite_facts.has_computed_minmax_aggregate_projection;
+		has_computed_sum_aggregate_projection = post_rewrite_facts.has_computed_sum_aggregate_projection;
 
 		// Keep data tables unlimited/unordered; apply ORDER BY/LIMIT in the user-facing view.
 		{
@@ -568,6 +570,8 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
 	model_input.stored_query_has_top_k = stored_query_retains_top_k;
 	model_input.has_hidden_minmax_having = has_hidden_minmax_having;
 	model_input.has_computed_minmax_aggregate_projection = has_computed_minmax_aggregate_projection;
+	model_input.has_computed_sum_aggregate_projection = has_computed_sum_aggregate_projection;
+	model_input.has_top_level_redundant_scalar_distinct = facts.has_top_level_redundant_scalar_distinct;
 	model_input.has_ducklake_source = HasDuckLakeSourceForModel(facts, table_names, target_is_ducklake);
 	const bool distinct_at_top = IsDistinctAtTop(analysis, output_names);
 
diff --git a/src/core/parser_plan_helpers.cpp b/src/core/parser_plan_helpers.cpp
@@ -431,6 +431,49 @@ static bool ExpressionReferencesMinMaxAggregate(Expression &expr,
 	return found;
 }
 
+static bool ExpressionReferencesUserSumAggregate(Expression &expr, const CreateMVPlanFacts &facts,
+                                                 const unordered_map<idx_t, LogicalAggregate *> &aggregates,
+                                                 idx_t depth = 0) {
+	if (depth > 16) {
+		return false;
+	}
+	if (expr.expression_class == ExpressionClass::BOUND_COLUMN_REF) {
+		auto &column_ref = expr.Cast<BoundColumnRefExpression>();
+		auto aggregate_it = aggregates.find(column_ref.binding.table_index);
+		if (aggregate_it != aggregates.end()) {
+			auto &aggregate = *aggregate_it->second;
+			if (column_ref.binding.column_index >= aggregate.expressions.size()) {
+				return false;
+			}
+			auto &aggregate_expr = aggregate.expressions[column_ref.binding.column_index];
+			if (aggregate_expr->expression_class != ExpressionClass::BOUND_AGGREGATE) {
+				return false;
+			}
+			auto &bound_aggregate = aggregate_expr->Cast<BoundAggregateExpression>();
+			return bound_aggregate.function.name == "sum" &&
+			       !IncrementalTableNames::IsInternalColumn(bound_aggregate.alias);
+		}
+		auto projection_it = facts.projections_by_index.find(column_ref.binding.table_index);
+		if (projection_it == facts.projections_by_index.end()) {
+			return false;
+		}
+		auto &projection = *projection_it->second;
+		if (column_ref.binding.column_index >= projection.expressions.size()) {
+			return false;
+		}
+		return ExpressionReferencesUserSumAggregate(*projection.expressions[column_ref.binding.column_index], facts,
+		                                            aggregates, depth + 1);
+	}
+
+	bool found = false;
+	ExpressionIterator::EnumerateChildren(expr, [&](Expression &child) {
+		if (!found && ExpressionReferencesUserSumAggregate(child, facts, aggregates, depth + 1)) {
+			found = true;
+		}
+	});
+	return found;
+}
+
 static void FinalizeCreateMVPlanFacts(CreateMVPlanFacts &facts) {
 	unordered_map<idx_t, LogicalAggregate *> aggregates;
 	for (auto *aggregate : facts.aggregates) {
@@ -441,6 +484,14 @@ static void FinalizeCreateMVPlanFacts(CreateMVPlanFacts &facts) {
 	}
 	for (auto *projection : facts.projections) {
 		for (auto &expr : projection->expressions) {
+			Expression *unwrapped = expr.get();
+			while (unwrapped->expression_class == ExpressionClass::BOUND_CAST) {
+				unwrapped = unwrapped->Cast<BoundCastExpression>().child.get();
+			}
+			if (unwrapped->expression_class != ExpressionClass::BOUND_COLUMN_REF &&
+			    ExpressionReferencesUserSumAggregate(*unwrapped, facts, aggregates)) {
+				facts.has_computed_sum_aggregate_projection = true;
+			}
 			if (expr->expression_class == ExpressionClass::BOUND_COLUMN_REF && IsHiddenHavingColumn(expr->alias)) {
 				auto &column_ref = expr->Cast<BoundColumnRefExpression>();
 				if (IsMinMaxAggregateColumn(column_ref, aggregates)) {
@@ -901,10 +952,34 @@ static void AddJoinEdgesFromFacts(CreateMVPlanFacts &facts) {
 	}
 }
 
+bool ProducesAtMostOneRow(LogicalOperator &node) {
+	LogicalOperator *current = &node;
+	while ((current->type == LogicalOperatorType::LOGICAL_PROJECTION ||
+	        current->type == LogicalOperatorType::LOGICAL_FILTER) &&
+	       !current->children.empty()) {
+		current = current->children[0].get();
+	}
+	if (current->type != LogicalOperatorType::LOGICAL_AGGREGATE_AND_GROUP_BY) {
+		return false;
+	}
+	auto &aggregate = current->Cast<LogicalAggregate>();
+	return aggregate.groups.empty() && aggregate.grouping_sets.size() <= 1;
+}
+
 CreateMVPlanFacts BuildCreateMVPlanFacts(LogicalOperator *plan, const string &current_catalog) {
 	CreateMVPlanFacts facts;
 	facts.root = plan;
 	facts.analysis = AnalyzePlan(plan);
+	LogicalOperator *top = plan;
+	while (top && top->children.size() == 1 &&
+	       (top->type == LogicalOperatorType::LOGICAL_CREATE_TABLE ||
+	        top->type == LogicalOperatorType::LOGICAL_PROJECTION || top->type == LogicalOperatorType::LOGICAL_FILTER ||
+	        top->type == LogicalOperatorType::LOGICAL_ORDER_BY || top->type == LogicalOperatorType::LOGICAL_LIMIT ||
+	        top->type == LogicalOperatorType::LOGICAL_TOP_N)) {
+		top = top->children[0].get();
+	}
+	facts.has_top_level_redundant_scalar_distinct = top && top->type == LogicalOperatorType::LOGICAL_DISTINCT &&
+	                                                !top->children.empty() && ProducesAtMostOneRow(*top->children[0]);
 	unordered_map<string, idx_t> next_occurrence;
 	CollectCreateMVPlanFacts(plan, current_catalog, facts, next_occurrence, false, false);
 	FinalizeCreateMVPlanFacts(facts);
diff --git a/src/core/plan_rewrite.cpp b/src/core/plan_rewrite.cpp
@@ -2,6 +2,7 @@
 
 #include "core/openivm_constants.hpp"
 #include "core/openivm_debug.hpp"
+#include "core/parser_plan_helpers.hpp"
 #include "core/plan_rewrite_internal.hpp"
 #include "duckdb/catalog/catalog.hpp"
 #include "duckdb/catalog/catalog_entry/aggregate_function_catalog_entry.hpp"
@@ -39,18 +40,30 @@
 namespace duckdb {
 
 /// Replace LOGICAL_DISTINCT with LOGICAL_AGGREGATE + COUNT(*).
-static void RewriteDistinct(ClientContext &context, Binder &binder, unique_ptr<LogicalOperator> &node) {
+/// Returns whether a DISTINCT state aggregate was added at the top level.
+static bool RewriteDistinct(ClientContext &context, Binder &binder, unique_ptr<LogicalOperator> &node,
+                            bool is_top_level) {
 	if (node->type != LogicalOperatorType::LOGICAL_DISTINCT) {
-		for (auto &child : node->children) {
-			RewriteDistinct(context, binder, child);
+		bool added_top_level_state = false;
+		for (idx_t child_idx = 0; child_idx < node->children.size(); child_idx++) {
+			bool child_is_top_level =
+			    is_top_level && node->type == LogicalOperatorType::LOGICAL_PROJECTION && child_idx == 0;
+			added_top_level_state = RewriteDistinct(context, binder, node->children[child_idx], child_is_top_level) ||
+			                        added_top_level_state;
 		}
-		return;
+		return added_top_level_state;
 	}
 
 	auto &distinct = node->Cast<LogicalDistinct>();
 	if (node->children.empty()) {
 		OPENIVM_DEBUG_PRINT("[PlanRewrite] DISTINCT has no children — skipping\n");
-		return;
+		return false;
+	}
+	if (ProducesAtMostOneRow(*node->children[0])) {
+		OPENIVM_DEBUG_PRINT("[PlanRewrite] Removed redundant DISTINCT over scalar aggregate\n");
+		node = std::move(node->children[0]);
+		RewriteDistinct(context, binder, node, false);
+		return false;
 	}
 	auto &child = node->children[0];
 	child->ResolveOperatorTypes();
@@ -86,6 +99,7 @@ static void RewriteDistinct(ClientContext &context, Binder &binder, unique_ptr<L
 
 	OPENIVM_DEBUG_PRINT("[PlanRewrite] DISTINCT → AGGREGATE + COUNT(*), %zu groups\n", agg_node->groups.size());
 	node = std::move(agg_node);
+	return is_top_level;
 }
 
 static bool IsSemiAntiJoinType(JoinType join_type) {
@@ -1347,9 +1361,9 @@ static void RewritePassAggregateFilters(PlanRewriteContext &rewrite_context) {
 }
 
 static void RewritePassDistinct(PlanRewriteContext &rewrite_context) {
-	bool had_distinct = HasTopLevelDistinct(rewrite_context.plan);
-	RewriteDistinct(rewrite_context.context, rewrite_context.binder, rewrite_context.plan);
-	if (had_distinct) {
+	bool added_top_level_state = RewriteDistinct(rewrite_context.context, rewrite_context.binder, rewrite_context.plan,
+	                                             HasTopLevelDistinct(rewrite_context.plan));
+	if (added_top_level_state) {
 		rewrite_context.planner_names.push_back(openivm::DISTINCT_COUNT_COL);
 	}
 }
diff --git a/src/include/core/ivm_view_classifier.hpp b/src/include/core/ivm_view_classifier.hpp
@@ -163,6 +163,8 @@ struct DeltaViewModelInput {
 	bool stored_query_has_top_k = false;
 	bool has_hidden_minmax_having = false;
 	bool has_computed_minmax_aggregate_projection = false;
+	bool has_computed_sum_aggregate_projection = false;
+	bool has_top_level_redundant_scalar_distinct = false;
 	bool has_ducklake_source = false;
 };
 
diff --git a/src/include/core/parser_plan_helpers.hpp b/src/include/core/parser_plan_helpers.hpp
@@ -85,6 +85,8 @@ struct CreateMVPlanFacts {
 	bool has_bound_aggregate_filter = false;
 	bool has_hidden_minmax_having_column = false;
 	bool has_computed_minmax_aggregate_projection = false;
+	bool has_computed_sum_aggregate_projection = false;
+	bool has_top_level_redundant_scalar_distinct = false;
 };
 
 string BuildTopKSuffix(const vector<BoundOrderByNode> &orders, idx_t limit_val, idx_t offset_val,
@@ -94,6 +96,7 @@ string QualifyCreateSourceTable(const string &table_name, const string &current_
                                 const string &default_db);
 string ExplainInitialLoadQuery(Connection &con, const string &label, const string &query);
 CreateMVPlanFacts BuildCreateMVPlanFacts(LogicalOperator *plan, const string &current_catalog);
+bool ProducesAtMostOneRow(LogicalOperator &node);
 void AddJoinKeyColumn(const unique_ptr<Expression> &expr, unordered_map<idx_t, unordered_set<idx_t>> &join_key_cols);
 bool OuterJoinAggregateNeedsRecompute(const CreateMVPlanFacts &facts, idx_t group_index);
 bool RelationExists(Connection &con, const string &qualified_name);
diff --git a/test/sql/aggregate.test b/test/sql/aggregate.test
diff --git a/test/sql/scalar_subquery_fold.test b/test/sql/scalar_subquery_fold.test
