Skip to content

Element-wise LIST MERGE path in the refresh compiler is unreachable #103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ila wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Element-wise LIST MERGE path in the refresh compiler is unreachable #103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3dedcdd
Element-wise LIST MERGE path in the refresh compiler is unreachable
ila Oct 5, 2026
be6a514
Element-wise LIST MERGE path in the refresh compiler is unreachable
ila Oct 5, 2026
514df6d
Element-wise LIST MERGE path in the refresh compiler is unreachable
ila Oct 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 11 additions & 14 deletions src/include/upsert/refresh_compiler.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,21 +32,18 @@ struct WindowPartitionDeltaSpec {
string source_column;
};

string CompileAggregateGroups(const string &view_name, optional_ptr<CatalogEntry> index_delta_view_catalog_entry,
vector<string> column_names, const string &view_query_sql = "", bool has_minmax = false,
bool list_mode = false, const string &delta_ts_filter = "",
const vector<string> &group_column_names = {}, const string &catalog_prefix = "",
bool insert_only = false, const vector<string> &aggregate_types = {},
const vector<LogicalType> &column_types = {}, bool use_current_diff_affected_keys = false,
const vector<GroupRecomputeDeltaSpec> *cascade_delta_specs = nullptr,
const string &cascade_lpts_table_prefix = "", bool emit_cascade_delta = false,
bool inline_cascade_delta = false, bool *out_handled_cascade_delta = nullptr,
const unordered_map<string, string> &derived_output_expressions = {},
bool derived_output_expressions_complete = false,
const vector<string> &preserved_side_cols = {}, bool *out_used_group_recompute = nullptr,
bool force_group_recompute = false);
string CompileAggregateGroups(
const string &view_name, optional_ptr<CatalogEntry> index_delta_view_catalog_entry, vector<string> column_names,
const string &view_query_sql = "", bool has_minmax = false, const string &delta_ts_filter = "",
const vector<string> &group_column_names = {}, const string &catalog_prefix = "", bool insert_only = false,
const vector<string> &aggregate_types = {}, const vector<LogicalType> &column_types = {},
bool use_current_diff_affected_keys = false, const vector<GroupRecomputeDeltaSpec> *cascade_delta_specs = nullptr,
const string &cascade_lpts_table_prefix = "", bool emit_cascade_delta = false, bool inline_cascade_delta = false,
bool *out_handled_cascade_delta = nullptr, const unordered_map<string, string> &derived_output_expressions = {},
bool derived_output_expressions_complete = false, const vector<string> &preserved_side_cols = {},
bool *out_used_group_recompute = nullptr, bool force_group_recompute = false);
string CompileSimpleAggregates(const string &view_name, const vector<string> &column_names,
const string &view_query_sql = "", bool has_minmax = false, bool list_mode = false,
const string &view_query_sql = "", bool has_minmax = false,
const string &delta_ts_filter = "", const string &catalog_prefix = "",
bool insert_only = false, const vector<LogicalType> &column_types = {},
bool *out_full_recompute = nullptr);
Expand Down
55 changes: 15 additions & 40 deletions src/upsert/refresh_compiler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,6 @@

namespace duckdb {

// Zero-initialized 64-element float list, used as COALESCE default for NULL list aggregates.
static constexpr const char *ZEROS_LIST = "[0.0::FLOAT FOR x IN generate_series(1, 64)]";
static constexpr idx_t GROUP_RECOMPUTE_DIFF_OCCURRENCE_THRESHOLD = 32;
static constexpr idx_t GROUP_RECOMPUTE_DIFF_SQL_BYTES_THRESHOLD = 16ULL * 1024ULL * 1024ULL;

Expand Down Expand Up @@ -529,7 +527,7 @@ static unordered_map<string, string> DetectSumNullCountColumns(const vector<stri

string CompileAggregateGroups(const string &view_name, optional_ptr<CatalogEntry> index_delta_view_catalog_entry,
vector<string> column_names, const string &view_query_sql, bool has_minmax,
bool list_mode, const string &delta_ts_filter, const vector<string> &group_column_names,
const string &delta_ts_filter, const vector<string> &group_column_names,
const string &catalog_prefix, bool insert_only, const vector<string> &aggregate_types,
const vector<LogicalType> &column_types, bool use_current_diff_affected_keys,
const vector<GroupRecomputeDeltaSpec> *cascade_delta_specs,
Expand Down Expand Up @@ -904,14 +902,6 @@ string CompileAggregateGroups(const string &view_name, optional_ptr<CatalogEntry
} else if (insert_only && agg_type == "max") {
// Insert-only MAX: consolidate with MAX (new max can only be >= current)
cte_select_string += "\n\tmax(" + column + ") as " + column + ", ";
} else if (list_mode) {
// Z-set bag-aware list sum: per-row scale every element by the row's weight,
// then list_reduce-add across the group.
cte_select_string += "\n\tlist_reduce(list(list_transform(" + column +
", lambda x: " + string(openivm::MULTIPLICITY_COL) +
" * x)), lambda a, b: list_transform(list_zip(a, "
"b), lambda x: x[1] + x[2])) AS " +
column + ", ";
} else {
// Z-set bag-aware sum: weight w∈ℤ scales the column value before SUM.
cte_select_string = cte_select_string + "\n\tsum(" + string(openivm::MULTIPLICITY_COL) + " * " + column +
Expand Down Expand Up @@ -982,9 +972,6 @@ string CompileAggregateGroups(const string &view_name, optional_ptr<CatalogEntry
updated_column_expressions[raw_column] = BuildNullSafeExtremumUpdate(column, "LEAST");
} else if (insert_only && agg_type == "max") {
updated_column_expressions[raw_column] = BuildNullSafeExtremumUpdate(column, "GREATEST");
} else if (list_mode) {
updated_column_expressions[raw_column] =
"list_transform(list_zip(v." + column + ", d." + column + "), lambda x: x[1] + x[2])";
} else {
updated_column_expressions[raw_column] = BuildUpdatedAggregateColumn(column);
}
Expand Down Expand Up @@ -1262,8 +1249,8 @@ string CompileAggregateGroups(const string &view_name, optional_ptr<CatalogEntry
}

string CompileSimpleAggregates(const string &view_name, const vector<string> &column_names,
const string &view_query_sql, bool has_minmax, bool list_mode,
const string &delta_ts_filter, const string &catalog_prefix, bool /*insert_only*/,
const string &view_query_sql, bool has_minmax, const string &delta_ts_filter,
const string &catalog_prefix, bool /*insert_only*/,
const vector<LogicalType> &column_types, bool *out_full_recompute) {
string data_table = catalog_prefix + SqlUtils::QuoteIdentifier(IncrementalTableNames::DataTableName(view_name));

Expand Down Expand Up @@ -1318,31 +1305,19 @@ string CompileSimpleAggregates(const string &view_name, const vector<string> &co
update_set += ",\n ";
}
first = false;
if (list_mode) {
// Z-set bag-aware list sum: per-row scale every element by w∈ℤ, then
// list_reduce-add. NULL-list COALESCE preserves the previous semantics
// for empty groups.
cte += "COALESCE(list_reduce(list(list_transform(" + column + ", lambda x: " + mul +
" * x)), lambda a, b: list_transform(list_zip(a, b), lambda x: "
"x[1] + x[2])), " +
string(ZEROS_LIST) + ") AS d_" + column;
update_set += column + " = list_transform(list_zip(" + column + ", (SELECT d_" + column +
" FROM openivm_delta)), lambda x: x[1] + x[2])";
// Z-set bag-aware sum: weight w∈ℤ scales the column value before SUM.
cte += "SUM(" + mul + " * " + column + ") AS d_" + column;
auto sum_count = sum_null_count_cols.find(column);
if (sum_count == sum_null_count_cols.end()) {
update_set +=
column + " = COALESCE(" + column + ", 0) + COALESCE((SELECT d_" + column + " FROM openivm_delta), 0)";
} else {
// Z-set bag-aware sum: weight w∈ℤ scales the column value before SUM.
cte += "SUM(" + mul + " * " + column + ") AS d_" + column;
auto sum_count = sum_null_count_cols.find(column);
if (sum_count == sum_null_count_cols.end()) {
update_set += column + " = COALESCE(" + column + ", 0) + COALESCE((SELECT d_" + column +
" FROM openivm_delta), 0)";
} else {
string count_col = sum_count->second;
string updated_sum =
"COALESCE(" + column + ", 0) + COALESCE((SELECT d_" + column + " FROM openivm_delta), 0)";
string updated_count =
"COALESCE(" + count_col + ", 0) + COALESCE((SELECT d_" + count_col + " FROM openivm_delta), 0)";
update_set += column + " = " + BuildNullableSum(updated_sum, updated_count);
}
string count_col = sum_count->second;
string updated_sum =
"COALESCE(" + column + ", 0) + COALESCE((SELECT d_" + column + " FROM openivm_delta), 0)";
string updated_count =
"COALESCE(" + count_col + ", 0) + COALESCE((SELECT d_" + count_col + " FROM openivm_delta), 0)";
update_set += column + " = " + BuildNullableSum(updated_sum, updated_count);
}
}
cte += "\n FROM " + delta_view + ts_where + "\n)\n";
Expand Down
28 changes: 11 additions & 17 deletions src/upsert/refresh_sql.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -780,15 +780,11 @@
auto column_metadata_start = profile_now();
vector<string> column_names;
vector<LogicalType> column_types;
bool list_mode = false;
auto read_delta_columns = [&](TableCatalogEntry &entry) {
const auto &columns = entry.GetColumns();
column_names = columns.GetColumnNames();
for (auto &column : columns.Logical()) {
column_types.push_back(column.GetType());
if (column.GetName() != openivm::MULTIPLICITY_COL && column.GetType().id() == LogicalTypeId::LIST) {
list_mode = true;
}
}
};
if (delta_view_catalog_entry) {
Expand All @@ -812,10 +808,8 @@
column_types.erase(column_types.begin() + offset);
}
}
OPENIVM_DEBUG_PRINT("[UPSERT] List mode: %s\n", list_mode ? "true" : "false");
add_profile_step("generate_refresh_sql.column_metadata", column_metadata_start,
"columns=" + to_string(column_names.size()) +
"; list_mode=" + string(list_mode ? "true" : "false"));
"columns=" + to_string(column_names.size()));

string upsert_query;
string appended_projection_rows;
Expand Down Expand Up @@ -929,13 +923,13 @@
has_argminmax ? false : (has_minmax ? (insert_only && minmax_incremental) : skip_agg_delete);
upsert_query =
CompileAggregateGroups(view_name, index_delta_view_catalog_entry.get(), column_names, view_query_sql,
has_minmax, list_mode, delta_ts_filter, group_cols, internal_catalog_prefix,
has_minmax, delta_ts_filter, group_cols, internal_catalog_prefix,
effective_insert_only, agg_types, column_types, has_unstripped_having);
} else {
upsert_query = CompileAggregateGroups(
view_name, index_delta_view_catalog_entry.get(), column_names, view_query_sql,
/*has_minmax=*/true, list_mode, delta_ts_filter, group_cols, internal_catalog_prefix,
/*insert_only=*/false, agg_types, column_types, has_unstripped_having);
upsert_query =
CompileAggregateGroups(view_name, index_delta_view_catalog_entry.get(), column_names, view_query_sql,
/*has_minmax=*/true, delta_ts_filter, group_cols, internal_catalog_prefix,
/*insert_only=*/false, agg_types, column_types, has_unstripped_having);
}
break;
}
Expand All @@ -948,7 +942,7 @@
bool effective_insert_only = has_argminmax ? false : skip_agg_delete;
upsert_query =
CompileAggregateGroups(view_name, index_delta_view_catalog_entry.get(), column_names, view_query_sql,
/*has_minmax=*/has_argminmax, list_mode, delta_ts_filter, group_cols,
/*has_minmax=*/has_argminmax, delta_ts_filter, group_cols,
internal_catalog_prefix, effective_insert_only, agg_types, column_types);
upsert_query += BuildFullOuterAffectedGroupRefresh(metadata, view_name, delta_table_names, group_cols,
data_table, view_query_sql, delta_ts_filter,
Expand Down Expand Up @@ -1072,7 +1066,7 @@
}
bool ljsec_used_group_recompute = false;
upsert_query = CompileAggregateGroups(
view_name, index_delta_view_catalog_entry.get(), column_names, view_query_sql, has_minmax, list_mode,
view_name, index_delta_view_catalog_entry.get(), column_names, view_query_sql, has_minmax,
delta_ts_filter, group_cols, internal_catalog_prefix, effective_insert_only, agg_types, column_types,
/*use_current_diff_affected_keys=*/false, aggregate_cascade_specs_ptr, aggregate_recompute_lpts_prefix,
/*emit_cascade_delta=*/aggregate_cascade_specs_ptr != nullptr,
Expand Down Expand Up @@ -1133,9 +1127,9 @@
aux_meta.group_col.c_str(), aux_meta.sum_col.c_str(), aux_meta.comparison_op.c_str(),
aux_meta.threshold_sql.c_str());
} else {
upsert_query = CompileSimpleAggregates(view_name, column_names, view_query_sql, has_minmax, list_mode,
delta_ts_filter, internal_catalog_prefix, sa_insert_only,
column_types, &simple_aggregate_full_recompute);
upsert_query = CompileSimpleAggregates(view_name, column_names, view_query_sql, has_minmax, delta_ts_filter,
internal_catalog_prefix, sa_insert_only, column_types,
&simple_aggregate_full_recompute);
}
if (!has_minmax && aux_meta.aux_table.empty() && !simple_aggregate_full_recompute) {
AppendSimpleAggregateEmptySourceNulling(metadata, upsert_query, view_name, column_names, data_table,
Expand Down Expand Up @@ -1210,7 +1204,7 @@
OPENIVM_DEBUG_PRINT("[UPSERT] DISTINCT_INCREMENTAL view has no aux meta — "
"falling through to "
"GROUP_RECOMPUTE\n");
[[fallthrough]];

Check warning on line 1207 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored

Check warning on line 1207 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored

Check warning on line 1207 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored

Check warning on line 1207 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored
}
case RefreshType::SEMI_ANTI_RECOMPUTE: {
RefreshMetadata::SemiAntiAuxMeta aux_meta;
Expand Down Expand Up @@ -1244,7 +1238,7 @@
OPENIVM_DEBUG_PRINT("[UPSERT] SEMI_ANTI_RECOMPUTE view has no aux meta — "
"falling through to "
"GROUP_RECOMPUTE\n");
[[fallthrough]];

Check warning on line 1241 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored

Check warning on line 1241 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored

Check warning on line 1241 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored

Check warning on line 1241 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored
}
case RefreshType::GROUP_RECOMPUTE: {
auto group_columns = metadata.GetGroupColumns(view_name);
Expand Down Expand Up @@ -1296,7 +1290,7 @@
break;
}
case RefreshType::TOP_K:
[[fallthrough]];

Check warning on line 1293 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored

Check warning on line 1293 in src/upsert/refresh_sql.cpp

View workflow job for this annotation

GitHub Actions / Build extension binaries / Windows (windows_amd64, windows-latest, x64-windows-static-release, x64-windows-static-release, t... 

 attribute [[fallthrough]] requires at least '/std:c++17'; ignored
case RefreshType::FULL_REFRESH: {
string full_recompute_query = view_query_sql;
if (active_facts.target_dialect == SqlDialect::SPARK) {
Expand Down
Loading
Loading