Skip to content

Commit 13c4cd5

Browse files
ilaila
committed
Fix table-function left join refreshes
1 parent 9f98324 commit 13c4cd5

7 files changed

Lines changed: 274 additions & 15 deletions

File tree

‎src/core/ivm_view_classifier.cpp‎

Lines changed: 30 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -612,7 +612,8 @@ static void SelectGroupRecomputeAffectedMode(DeltaViewModel &model, const DeltaV
612612
if ((input.facts && input.facts->analysis.found_asof_join) || input.stored_query_has_top_k ||
613613
aggregate_filter_join || (input.stored_query_has_aggregate_filter && input.has_ducklake_source)) {
614614
model.group_recompute_affected_mode = GroupRecomputeAffectedMode::CURRENT_DIFF;
615-
} else if (HasStrategyReason(model, DeltaStrategyReason::JOIN_AGGREGATE_PROJECTION_FALLBACK)) {
615+
} else if (HasStrategyReason(model, DeltaStrategyReason::JOIN_AGGREGATE_PROJECTION_FALLBACK) ||
616+
HasStrategyReason(model, DeltaStrategyReason::OUTER_JOIN_PRESERVED_TABLE_FUNCTION_RECOMPUTE)) {
616617
model.group_recompute_affected_mode = GroupRecomputeAffectedMode::CURRENT_DIFF;
617618
} else if (input.stored_query_has_aggregate_filter) {
618619
model.group_recompute_affected_mode = GroupRecomputeAffectedMode::SOURCE_DELTA_RELAX_AGGREGATE_FILTER;
@@ -663,6 +664,8 @@ const char *DeltaStrategyReasonName(DeltaStrategyReason reason) {
663664
return "SEMI_ANTI_AGGREGATE_GROUP_FALLBACK";
664665
case DeltaStrategyReason::OUTER_JOIN_AGGREGATE_RECOMPUTE:
665666
return "OUTER_JOIN_AGGREGATE_RECOMPUTE";
667+
case DeltaStrategyReason::OUTER_JOIN_PRESERVED_TABLE_FUNCTION_RECOMPUTE:
668+
return "OUTER_JOIN_PRESERVED_TABLE_FUNCTION_RECOMPUTE";
666669
default:
667670
return "UNKNOWN";
668671
}
@@ -852,22 +855,31 @@ idx_t DeltaViewModel::LineageEntryCount() const {
852855
return count;
853856
}
854857

855-
bool IsDistinctAtTop(const PlanAnalysis &analysis, const vector<string> &output_names) {
856-
if (!analysis.found_distinct || analysis.aggregate_columns.empty() || output_names.empty()) {
858+
bool IsDistinctAtTop(const CreateMVPlanFacts &facts, const vector<string> &output_names) {
859+
if (!facts.analysis.found_distinct) {
857860
return false;
858861
}
859862

860-
unordered_set<string> output_lc;
861-
for (auto &name : output_names) {
862-
output_lc.insert(StringUtil::Lower(name));
863+
auto *node = facts.root;
864+
while (node && node->children.size() == 1 &&
865+
(node->type == LogicalOperatorType::LOGICAL_CREATE_TABLE ||
866+
node->type == LogicalOperatorType::LOGICAL_PROJECTION ||
867+
node->type == LogicalOperatorType::LOGICAL_FILTER || node->type == LogicalOperatorType::LOGICAL_ORDER_BY ||
868+
node->type == LogicalOperatorType::LOGICAL_LIMIT || node->type == LogicalOperatorType::LOGICAL_TOP_N)) {
869+
node = node->children[0].get();
863870
}
864-
865-
for (auto &target : analysis.aggregate_columns) {
866-
if (!output_lc.count(StringUtil::Lower(target))) {
867-
return false;
871+
if (node && node->type == LogicalOperatorType::LOGICAL_DISTINCT) {
872+
return true;
873+
}
874+
if (!node || node->type != LogicalOperatorType::LOGICAL_AGGREGATE_AND_GROUP_BY) {
875+
return false;
876+
}
877+
for (auto &name : output_names) {
878+
if (name == openivm::DISTINCT_COUNT_COL) {
879+
return true;
868880
}
869881
}
870-
return true;
882+
return false;
871883
}
872884

873885
DeltaViewModel BuildDeltaViewModel(const DeltaViewModelInput &input) {
@@ -886,7 +898,7 @@ DeltaViewModel BuildDeltaViewModel(const DeltaViewModelInput &input) {
886898
}
887899

888900
model.has_minmax_metadata = analysis.found_minmax || analysis.found_count_distinct || analysis.found_list;
889-
model.distinct_at_top = IsDistinctAtTop(analysis, output_names);
901+
model.distinct_at_top = IsDistinctAtTop(facts, output_names);
890902
if (input.has_top_level_redundant_scalar_distinct) {
891903
model.distinct_at_top = false;
892904
}
@@ -899,6 +911,12 @@ DeltaViewModel BuildDeltaViewModel(const DeltaViewModelInput &input) {
899911
OPENIVM_DEBUG_PRINT("[CREATE MV] LEFT/OUTER JOIN aggregate with computed aggregate or projection wrapper -- "
900912
"using group-recompute metadata\n");
901913
}
914+
if (analysis.found_aggregation && OuterJoinPreservedSideHasTableFunction(facts)) {
915+
model.has_minmax_metadata = true;
916+
AddUnique(model.strategy_reasons, DeltaStrategyReason::OUTER_JOIN_PRESERVED_TABLE_FUNCTION_RECOMPUTE);
917+
OPENIVM_DEBUG_PRINT("[CREATE MV] LEFT/RIGHT JOIN aggregate with a table function on the preserved side -- "
918+
"using current-diff group recompute\n");
919+
}
902920

903921
if (analysis.found_full_outer) {
904922
model.full_outer_join_cols = ExtractFullOuterJoinMetadata(facts);

‎src/core/parser.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -573,7 +573,7 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
573573
model_input.has_computed_sum_aggregate_projection = has_computed_sum_aggregate_projection;
574574
model_input.has_top_level_redundant_scalar_distinct = facts.has_top_level_redundant_scalar_distinct;
575575
model_input.has_ducklake_source = HasDuckLakeSourceForModel(facts, table_names, target_is_ducklake);
576-
const bool distinct_at_top = IsDistinctAtTop(analysis, output_names);
576+
const bool distinct_at_top = IsDistinctAtTop(facts, output_names);
577577

578578
// Populated by ExtractInnerDistinct when classified as DISTINCT_INCREMENTAL.
579579
vector<string> distinct_extracted_cols;

‎src/core/parser_plan_helpers.cpp‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,6 +230,35 @@ bool OuterJoinAggregateNeedsRecompute(const CreateMVPlanFacts &facts, idx_t grou
230230
return false;
231231
}
232232

233+
static bool ContainsTableFunction(LogicalOperator &op) {
234+
if (op.type == LogicalOperatorType::LOGICAL_GET && !op.Cast<LogicalGet>().GetTable().get()) {
235+
return true;
236+
}
237+
for (auto &child : op.children) {
238+
if (ContainsTableFunction(*child)) {
239+
return true;
240+
}
241+
}
242+
return false;
243+
}
244+
245+
bool OuterJoinPreservedSideHasTableFunction(const CreateMVPlanFacts &facts) {
246+
for (auto *join : facts.comparison_joins) {
247+
idx_t preserved_child;
248+
if (join->join_type == JoinType::LEFT) {
249+
preserved_child = 0;
250+
} else if (join->join_type == JoinType::RIGHT) {
251+
preserved_child = 1;
252+
} else {
253+
continue;
254+
}
255+
if (join->children.size() > preserved_child && ContainsTableFunction(*join->children[preserved_child])) {
256+
return true;
257+
}
258+
}
259+
return false;
260+
}
261+
233262
static void AddGetFacts(LogicalGet &get, const string &current_catalog, CreateMVPlanFacts &facts,
234263
unordered_map<string, idx_t> &next_occurrence) {
235264
facts.gets_by_index[get.table_index] = &get;

‎src/include/core/ivm_view_classifier.hpp‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,8 @@ enum class DeltaStrategyReason {
1717
NESTED_AGGREGATE_GROUP_FALLBACK,
1818
REPEATED_CTE_AGGREGATE_GROUP_FALLBACK,
1919
SEMI_ANTI_AGGREGATE_GROUP_FALLBACK,
20-
OUTER_JOIN_AGGREGATE_RECOMPUTE
20+
OUTER_JOIN_AGGREGATE_RECOMPUTE,
21+
OUTER_JOIN_PRESERVED_TABLE_FUNCTION_RECOMPUTE
2122
};
2223

2324
enum class DeltaModelFeature {
@@ -221,7 +222,7 @@ const char *DeltaRuleKindName(DeltaRuleKind kind);
221222
const char *DeltaUnsupportedReasonName(DeltaUnsupportedReason reason);
222223
const char *DeltaUpdateSemanticsName(DeltaUpdateSemantics semantics);
223224
const char *DeltaAffectedDomainKindName(DeltaAffectedDomainKind kind);
224-
bool IsDistinctAtTop(const PlanAnalysis &analysis, const vector<string> &output_names);
225+
bool IsDistinctAtTop(const CreateMVPlanFacts &facts, const vector<string> &output_names);
225226
DeltaViewModel BuildDeltaViewModel(const DeltaViewModelInput &input);
226227

227228
} // namespace duckdb

‎src/include/core/parser_plan_helpers.hpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,7 @@ CreateMVPlanFacts BuildCreateMVPlanFacts(LogicalOperator *plan, const string &cu
9999
bool ProducesAtMostOneRow(LogicalOperator &node);
100100
void AddJoinKeyColumn(const unique_ptr<Expression> &expr, unordered_map<idx_t, unordered_set<idx_t>> &join_key_cols);
101101
bool OuterJoinAggregateNeedsRecompute(const CreateMVPlanFacts &facts, idx_t group_index);
102+
bool OuterJoinPreservedSideHasTableFunction(const CreateMVPlanFacts &facts);
102103
bool RelationExists(Connection &con, const string &qualified_name);
103104
vector<string> DeriveGroupColumnNames(const CreateMVPlanFacts &facts, idx_t group_index, size_t group_count,
104105
const vector<string> &output_names);

‎test/sql/lateral.test‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -610,6 +610,87 @@ SELECT COUNT(*) FROM (
610610
----
611611
0
612612

613+
# DISTINCT inside the left input of a LATERAL aggregate is not a top-level
614+
# DISTINCT. It remains incrementally maintainable via affected-group recompute.
615+
statement ok
616+
CREATE TABLE lat_distinct_customer (id INT, warehouse_id INT, state VARCHAR);
617+
618+
statement ok
619+
INSERT INTO lat_distinct_customer VALUES
620+
(1, 1, 'CA'), (2, 1, 'CA'), (3, 1, 'NY'), (4, 2, 'CA');
621+
622+
statement ok
623+
CREATE MATERIALIZED VIEW mv_lat_distinct_count AS
624+
SELECT c.warehouse_id, c.state, agg.customer_count
625+
FROM (
626+
SELECT DISTINCT warehouse_id, state
627+
FROM lat_distinct_customer
628+
) c
629+
JOIN LATERAL (
630+
SELECT COUNT(*) AS customer_count
631+
FROM lat_distinct_customer
632+
WHERE warehouse_id = c.warehouse_id AND state = c.state
633+
) agg ON TRUE;
634+
635+
query I
636+
SELECT type FROM openivm_views WHERE view_name = 'mv_lat_distinct_count';
637+
----
638+
6
639+
640+
statement ok
641+
UPDATE lat_distinct_customer SET state = 'TX' WHERE id = 2;
642+
643+
statement ok
644+
DELETE FROM lat_distinct_customer WHERE id = 3;
645+
646+
statement ok
647+
INSERT INTO lat_distinct_customer VALUES (5, 2, 'WA'), (6, 3, 'CA');
648+
649+
statement ok
650+
UPDATE lat_distinct_customer SET warehouse_id = 3 WHERE id = 6;
651+
652+
statement ok
653+
DELETE FROM lat_distinct_customer WHERE id = 5;
654+
655+
statement ok
656+
PRAGMA refresh('mv_lat_distinct_count');
657+
658+
query I
659+
SELECT COUNT(*) FROM (
660+
SELECT * FROM mv_lat_distinct_count
661+
EXCEPT ALL
662+
SELECT c.warehouse_id, c.state, agg.customer_count
663+
FROM (
664+
SELECT DISTINCT warehouse_id, state
665+
FROM lat_distinct_customer
666+
) c
667+
JOIN LATERAL (
668+
SELECT COUNT(*) AS customer_count
669+
FROM lat_distinct_customer
670+
WHERE warehouse_id = c.warehouse_id AND state = c.state
671+
) agg ON TRUE
672+
);
673+
----
674+
0
675+
676+
query I
677+
SELECT COUNT(*) FROM (
678+
SELECT c.warehouse_id, c.state, agg.customer_count
679+
FROM (
680+
SELECT DISTINCT warehouse_id, state
681+
FROM lat_distinct_customer
682+
) c
683+
JOIN LATERAL (
684+
SELECT COUNT(*) AS customer_count
685+
FROM lat_distinct_customer
686+
WHERE warehouse_id = c.warehouse_id AND state = c.state
687+
) agg ON TRUE
688+
EXCEPT ALL
689+
SELECT * FROM mv_lat_distinct_count
690+
);
691+
----
692+
0
693+
613694
statement ok
614695
INSERT INTO lat_history VALUES (1, 1, 7);
615696

‎test/sql/left_join.test‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1020,6 +1020,135 @@ SELECT COUNT(*) FROM (
10201020
----
10211021
0
10221022

1023+
# A table function on the preserved side has no source delta of its own.
1024+
# LEFT JOIN MERGE cannot infer which zero-match generated groups must remain,
1025+
# so these aggregates use current-vs-stored affected-group recomputation.
1026+
statement ok
1027+
CREATE TABLE tf_customer (id INT, warehouse_id INT, balance INT);
1028+
1029+
statement ok
1030+
INSERT INTO tf_customer VALUES (1, 1, 10), (2, 1, 20), (3, 2, 30);
1031+
1032+
statement ok
1033+
CREATE MATERIALIZED VIEW mv_tf_customer_count AS
1034+
SELECT generated_id AS warehouse_id, COUNT(c.id) AS customer_count
1035+
FROM generate_series(1, 5) generated(generated_id)
1036+
LEFT JOIN tf_customer c ON c.warehouse_id = generated.generated_id
1037+
GROUP BY generated_id;
1038+
1039+
query I
1040+
SELECT type FROM openivm_views WHERE view_name = 'mv_tf_customer_count';
1041+
----
1042+
6
1043+
1044+
# Conflicting insert/delete/update operations are deliberately batched before
1045+
# one refresh. In particular, groups 3 and 5 must retain their generated
1046+
# zero-count rows.
1047+
statement ok
1048+
UPDATE tf_customer SET balance = 11 WHERE id = 1;
1049+
1050+
statement ok
1051+
UPDATE tf_customer SET warehouse_id = 2 WHERE id = 2;
1052+
1053+
statement ok
1054+
DELETE FROM tf_customer WHERE id = 3;
1055+
1056+
statement ok
1057+
INSERT INTO tf_customer VALUES (4, 1, 40), (5, 4, 50);
1058+
1059+
statement ok
1060+
UPDATE tf_customer SET balance = 41 WHERE id = 4;
1061+
1062+
statement ok
1063+
DELETE FROM tf_customer WHERE id = 5;
1064+
1065+
statement ok
1066+
PRAGMA refresh('mv_tf_customer_count');
1067+
1068+
query I
1069+
SELECT COUNT(*) FROM (
1070+
SELECT * FROM mv_tf_customer_count
1071+
EXCEPT ALL
1072+
SELECT generated_id, COUNT(c.id)
1073+
FROM generate_series(1, 5) generated(generated_id)
1074+
LEFT JOIN tf_customer c ON c.warehouse_id = generated.generated_id
1075+
GROUP BY generated_id
1076+
);
1077+
----
1078+
0
1079+
1080+
query I
1081+
SELECT COUNT(*) FROM (
1082+
SELECT generated_id, COUNT(c.id)
1083+
FROM generate_series(1, 5) generated(generated_id)
1084+
LEFT JOIN tf_customer c ON c.warehouse_id = generated.generated_id
1085+
GROUP BY generated_id
1086+
EXCEPT ALL
1087+
SELECT * FROM mv_tf_customer_count
1088+
);
1089+
----
1090+
0
1091+
1092+
statement ok
1093+
CREATE TABLE tf_stock (id INT, quantity INT);
1094+
1095+
statement ok
1096+
INSERT INTO tf_stock VALUES (1, 3), (2, 12), (3, 27);
1097+
1098+
statement ok
1099+
CREATE MATERIALIZED VIEW mv_tf_stock_thresholds AS
1100+
SELECT threshold, COUNT(s.id) AS low_stock_items
1101+
FROM range(5, 30, 5) generated(threshold)
1102+
LEFT JOIN tf_stock s ON s.quantity < generated.threshold
1103+
GROUP BY threshold;
1104+
1105+
query I
1106+
SELECT type FROM openivm_views WHERE view_name = 'mv_tf_stock_thresholds';
1107+
----
1108+
6
1109+
1110+
statement ok
1111+
UPDATE tf_stock SET quantity = 18 WHERE id = 1;
1112+
1113+
statement ok
1114+
DELETE FROM tf_stock WHERE id = 2;
1115+
1116+
statement ok
1117+
INSERT INTO tf_stock VALUES (4, 7), (5, 22);
1118+
1119+
statement ok
1120+
UPDATE tf_stock SET quantity = 2 WHERE id = 5;
1121+
1122+
statement ok
1123+
DELETE FROM tf_stock WHERE id = 4;
1124+
1125+
statement ok
1126+
PRAGMA refresh('mv_tf_stock_thresholds');
1127+
1128+
query I
1129+
SELECT COUNT(*) FROM (
1130+
SELECT * FROM mv_tf_stock_thresholds
1131+
EXCEPT ALL
1132+
SELECT threshold, COUNT(s.id)
1133+
FROM range(5, 30, 5) generated(threshold)
1134+
LEFT JOIN tf_stock s ON s.quantity < generated.threshold
1135+
GROUP BY threshold
1136+
);
1137+
----
1138+
0
1139+
1140+
query I
1141+
SELECT COUNT(*) FROM (
1142+
SELECT threshold, COUNT(s.id)
1143+
FROM range(5, 30, 5) generated(threshold)
1144+
LEFT JOIN tf_stock s ON s.quantity < generated.threshold
1145+
GROUP BY threshold
1146+
EXCEPT ALL
1147+
SELECT * FROM mv_tf_stock_thresholds
1148+
);
1149+
----
1150+
0
1151+
10231152
# Delete from a deeper-right table: the matched MV row should
10241153
# revert to NULL on that side.
10251154
statement ok

0 commit comments

Comments
 (0)