Skip to content

Commit 21fb899

Browse files
ilaila
committed
Optimize insert-only derived aggregate refresh
1 parent 2cb752d commit 21fb899

12 files changed

Lines changed: 1304 additions & 207 deletions

‎src/core/parser.cpp‎

Lines changed: 31 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -368,6 +368,7 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
368368
bool pre_rewrite_has_aggregate_filter = false;
369369
bool has_hidden_minmax_having = false;
370370
bool has_computed_minmax_aggregate_projection = false;
371+
DerivedAggregateOutputInfo derived_aggregate_outputs;
371372
{
372373
auto select_parse_plan_start = create_profile_now();
373374
Parser select_parser;
@@ -448,6 +449,9 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
448449
select_plan = std::move(select_plan->children[0]);
449450
OPENIVM_DEBUG_PRINT("[CREATE MV] Stripped standalone ORDER_BY, suffix='%s'\n", top_k_suffix.c_str());
450451
}
452+
if (select_plan) {
453+
derived_aggregate_outputs = ExtractDerivedAggregateOutputs(*select_plan, output_names);
454+
}
451455
add_create_profile_step("create_compile_select_rewrite", select_rewrite_start,
452456
"output_cols=" + to_string(output_names.size()));
453457

@@ -471,13 +475,15 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
471475
} catch (...) {
472476
view_query = original_view_query;
473477
lpts_fallback = true;
474-
OPENIVM_DEBUG_PRINT("[CREATE MV] LPTS fallback (unknown exception) to original query: %s\n",
478+
OPENIVM_DEBUG_PRINT("[CREATE MV] LPTS fallback (unknown exception) to "
479+
"original query: %s\n",
475480
view_query.c_str());
476481
}
477482
if (PlanNeedsOriginalSqlForLpts(select_plan.get())) {
478483
view_query = original_view_query;
479484
lpts_fallback = true;
480-
OPENIVM_DEBUG_PRINT("[CREATE MV] LPTS can't round-trip this construct — using original SQL: %s\n",
485+
OPENIVM_DEBUG_PRINT("[CREATE MV] LPTS can't round-trip this construct — "
486+
"using original SQL: %s\n",
481487
view_query.c_str());
482488
}
483489
add_create_profile_step("create_compile_lpts", lpts_start,
@@ -521,7 +527,8 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
521527
if (analysis.found_filtered_list) {
522528
view_query = original_view_query;
523529
lpts_fallback = true;
524-
OPENIVM_DEBUG_PRINT("[CREATE MV] LIST FILTER requires original SQL for group-recompute: %s\n",
530+
OPENIVM_DEBUG_PRINT("[CREATE MV] LIST FILTER requires original SQL for "
531+
"group-recompute: %s\n",
525532
view_query.c_str());
526533
}
527534
auto classification_start = create_profile_now();
@@ -590,10 +597,10 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
590597
if (analysis.found_semi_anti_join && !analysis.found_aggregation) {
591598
if (ExtractSemiAntiQuery(original_view_query, semi_anti_extract)) {
592599
string left_table_name = SqlUtils::LastIdentifierPart(semi_anti_extract.left_table);
593-
auto col_result =
594-
con.Query("SELECT column_name FROM information_schema.columns WHERE lower(table_name) = lower('" +
595-
SqlUtils::EscapeSingleQuotes(left_table_name) + "') AND table_schema = '" +
596-
SqlUtils::EscapeSingleQuotes(current_schema) + "' ORDER BY ordinal_position");
600+
auto col_result = con.Query("SELECT column_name FROM information_schema.columns WHERE "
601+
"lower(table_name) = lower('" +
602+
SqlUtils::EscapeSingleQuotes(left_table_name) + "') AND table_schema = '" +
603+
SqlUtils::EscapeSingleQuotes(current_schema) + "' ORDER BY ordinal_position");
597604
auto add_semi_anti_left_col = [&](const string &col_name) {
598605
if (!ContainsColumnCI(semi_anti_left_cols, col_name)) {
599606
semi_anti_left_cols.push_back(col_name);
@@ -760,7 +767,8 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
760767
if (!outer_agg || unsupported_agg || sum_count != 1) {
761768
distinct_sum_arg.clear();
762769
distinct_sum_out.clear();
763-
OPENIVM_DEBUG_PRINT("[CREATE MV] DISTINCT_INCREMENTAL outer-agg not single-SUM — demoting "
770+
OPENIVM_DEBUG_PRINT("[CREATE MV] DISTINCT_INCREMENTAL outer-agg not "
771+
"single-SUM — demoting "
764772
"to GROUP_RECOMPUTE\n");
765773
} else if (distinct_sum_out.empty()) {
766774
// `SUM(c) AS s` puts the alias `s` on the SELECT-list BCR above the
@@ -889,12 +897,14 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
889897
if (ducklake_window_partition && !lpts_fallback) {
890898
view_query = original_view_query;
891899
lpts_fallback = true;
892-
OPENIVM_DEBUG_PRINT("[CREATE MV] DuckLake window MV uses original SQL for initial data table: %s\n",
900+
OPENIVM_DEBUG_PRINT("[CREATE MV] DuckLake window MV uses original SQL for "
901+
"initial data table: %s\n",
893902
view_query.c_str());
894903
}
895904
add_create_profile_step("create_compile_classification", classification_start, model_profile_detail);
896905

897-
OPENIVM_DEBUG_PRINT("[CREATE MV] Detected IVM type: %s (aggregation=%d, projection=%d, group_cols=%zu)\n",
906+
OPENIVM_DEBUG_PRINT("[CREATE MV] Detected IVM type: %s (aggregation=%d, "
907+
"projection=%d, group_cols=%zu)\n",
898908
RefreshTypeName(refresh_type), (int)analysis.found_aggregation, (int)analysis.found_projection,
899909
aggregate_columns.size());
900910
for (auto reason : view_model.strategy_reasons) {
@@ -916,7 +926,8 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
916926
domain.source_tables.size(), (int)domain.delta_local, (int)domain.needs_base_lookup);
917927
}
918928
for (auto &node : view_model.nodes) {
919-
OPENIVM_DEBUG_PRINT("[CREATE MV] Delta node %llu: %s rule=%s children=%zu sources=%zu keys=%zu "
929+
OPENIVM_DEBUG_PRINT("[CREATE MV] Delta node %llu: %s rule=%s children=%zu sources=%zu "
930+
"keys=%zu "
920931
"occurrence=%llu domains=%zu lineage=%zu semantics=%zu unsupported=%zu "
921932
"maintenance=%s state=%s\n",
922933
static_cast<unsigned long long>(node.id), DeltaModelNodeKindName(node.kind),
@@ -996,7 +1007,8 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
9961007
" (view_name, sql_string, type, has_minmax, has_left_join, has_join, last_update, "
9971008
"refresh_interval, refresh_in_progress, group_columns, aggregate_types, "
9981009
"having_predicate, group_recompute_affected_mode, "
999-
"group_recompute_source_occurrences_json, has_full_outer, full_outer_join_cols) values ('" +
1010+
"group_recompute_source_occurrences_json, has_full_outer, "
1011+
"full_outer_join_cols) values ('" +
10001012
view_name + "', '" + SqlUtils::EscapeSingleQuotes(view_query) + "', " +
10011013
to_string((int)refresh_type) + ", " + (has_minmax_metadata ? "true" : "false") + ", " +
10021014
(analysis.found_left_join ? "true" : "false") + ", " +
@@ -1008,6 +1020,9 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
10081020
if (!lineage_json.empty()) {
10091021
aux_metadata_ddl.push_back(BuildUpdateViewJsonSQL("lineage_json", lineage_json, view_name));
10101022
}
1023+
aux_metadata_ddl.push_back(
1024+
BuildUpdateViewJsonSQL("derived_aggregate_outputs_json",
1025+
RefreshMetadata::DerivedAggregateOutputsToJson(derived_aggregate_outputs), view_name));
10111026
add_profile_record("create_compile_lineage", lineage_duration_ms, "entries=" + to_string(lineage_entry_count));
10121027

10131028
Value match_flag_val;
@@ -1180,7 +1195,8 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
11801195
if (!source_metadata_values.empty()) {
11811196
source_metadata_ddl.push_back("insert or replace into " + string(openivm::DELTA_TABLES_TABLE) +
11821197
" (view_name, table_name, last_update, catalog_type, last_snapshot_id, "
1183-
"last_refresh_ts, source_catalog, source_schema, source_table_id) values " +
1198+
"last_refresh_ts, source_catalog, source_schema, source_table_id) "
1199+
"values " +
11841200
StringUtil::Join(source_metadata_values, ", "));
11851201
}
11861202

@@ -1429,7 +1445,8 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
14291445
if (visible_ddl_idx < 3) {
14301446
system_tables_sql += ddl[i] + ";\n\n";
14311447
} else if (StringUtil::StartsWith(ddl[i], OPENIVM_DDL_CREATE_DELTA_FROM_DATA_PREFIX)) {
1432-
compiled_sql += "-- OpenIVM derives the MV delta-table schema from the physical data table "
1448+
compiled_sql += "-- OpenIVM derives the MV delta-table schema from the "
1449+
"physical data table "
14331450
"at DDL execution time.\n\n";
14341451
} else {
14351452
compiled_sql += ddl[i] + ";\n\n";

‎src/core/parser_create_mv_helpers.cpp‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ void AppendCreateMVSystemTablesDDL(vector<string> &ddl, const string &view_name,
4141
" refresh_in_progress boolean default false,"
4242
" group_columns varchar default null,"
4343
" aggregate_types varchar default null,"
44+
" derived_aggregate_outputs_json varchar default null,"
4445
" having_predicate varchar default null,"
4546
" group_recompute_affected_mode varchar default null,"
4647
" group_recompute_source_occurrences_json varchar default null,"
@@ -67,13 +68,16 @@ void AppendCreateMVSystemTablesDDL(vector<string> &ddl, const string &view_name,
6768
AddColumnIfNotExists(ddl, openivm::VIEWS_TABLE, "has_join boolean default null");
6869
AddColumnIfNotExists(ddl, openivm::VIEWS_TABLE, "group_recompute_affected_mode varchar default null");
6970
AddColumnIfNotExists(ddl, openivm::VIEWS_TABLE, "group_recompute_source_occurrences_json varchar default null");
71+
AddColumnIfNotExists(ddl, openivm::VIEWS_TABLE, "derived_aggregate_outputs_json varchar default null");
7072
if (!is_replace) {
7173
string escaped_view_name = SqlUtils::EscapeSingleQuotes(view_name);
7274
string escaped_data_table = SqlUtils::EscapeSingleQuotes(IncrementalTableNames::DataTableName(view_name));
7375
string stale_mv_condition = "view_name = '" + escaped_view_name +
74-
"' AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = '" +
76+
"' AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE "
77+
"table_name = '" +
7578
escaped_view_name +
76-
"') AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = '" +
79+
"') AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE "
80+
"table_name = '" +
7781
escaped_data_table + "')";
7882
// CREATE MV executes as multiple catalog statements. If a process dies or loses
7983
// a DuckDB file lock after writing metadata but before creating the physical
@@ -94,7 +98,8 @@ void AppendCreateMVSystemTablesDDL(vector<string> &ddl, const string &view_name,
9498

9599
ddl.push_back("create table if not exists " + string(openivm::DELTA_TABLES_TABLE) +
96100
" (view_name varchar, table_name varchar, last_update timestamp,"
97-
" catalog_type varchar default 'duckdb', last_snapshot_id bigint default null,"
101+
" catalog_type varchar default 'duckdb', last_snapshot_id bigint default "
102+
"null,"
98103
" last_refresh_ts timestamp default null,"
99104
" pending_row_estimate bigint default null,"
100105
" pending_estimate_ts timestamp default null,"
@@ -120,8 +125,10 @@ void AppendCreateMVSystemTablesDDL(vector<string> &ddl, const string &view_name,
120125
// Refresh history: stores execution stats for learned cost model calibration.
121126
// Stage A.5 adds `strategy` (default 'incremental') for per-strategy regression.
122127
ddl.push_back("create table if not exists " + string(openivm::HISTORY_TABLE) +
123-
" (view_name varchar, refresh_timestamp timestamp default current_timestamp,"
124-
" method varchar, incremental_compute_est double, incremental_upsert_est double,"
128+
" (view_name varchar, refresh_timestamp timestamp default "
129+
"current_timestamp,"
130+
" method varchar, incremental_compute_est double, "
131+
"incremental_upsert_est double,"
125132
" recompute_compute_est double, recompute_replace_est double,"
126133
" actual_duration_ms bigint,"
127134
" strategy varchar default 'incremental',"
@@ -130,7 +137,8 @@ void AppendCreateMVSystemTablesDDL(vector<string> &ddl, const string &view_name,
130137
ddl.push_back("create table if not exists " + string(openivm::PROFILE_TABLE) +
131138
" (refresh_id varchar, view_name varchar,"
132139
" profile_timestamp timestamp default current_timestamp,"
133-
" step_order integer, step_name varchar, duration_ms bigint, detail varchar,"
140+
" step_order integer, step_name varchar, duration_ms bigint, "
141+
"detail varchar,"
134142
" primary key(refresh_id, step_order))");
135143
}
136144

0 commit comments

Comments
 (0)