Skip to content

Commit 16b0098

Browse files
ilaila
committed
Use NULL-safe window partition refresh
1 parent fa69ce0 commit 16b0098

7 files changed

Lines changed: 251 additions & 101 deletions

File tree

‎benchmark/queries/tpcc/ducklake_0318.sql‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
-- {"operators": "DUCKLAKE,WINDOW", "complexity": "low", "is_incremental": true, "has_nulls": true, "has_cast": false, "has_case": false, "tables": "ORDER_LINE", "ducklake": true, "openivm_verified": true}
2+
SELECT
3+
OL_W_ID AS warehouse_id,
4+
OL_D_ID AS district_id,
5+
OL_O_ID AS order_id,
6+
OL_NUMBER AS line_number,
7+
OL_DELIVERY_D AS delivery_at,
8+
ROW_NUMBER() OVER (
9+
PARTITION BY OL_DELIVERY_D
10+
ORDER BY OL_W_ID, OL_D_ID, OL_O_ID, OL_NUMBER
11+
) AS delivery_row_number
12+
FROM dl.ORDER_LINE;

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,10 @@ maintenance path. A bug must not be hidden by weakening a test or silently chang
2828
- [x] **Continue cascade traversal when the current node has no source deltas.** Empty-delta detection now skips only the
2929
current node; it cannot terminate traversal of the downstream DAG. Regression coverage exercises both a pending parent
3030
delta left by a cascade-off refresh and independent child-source DML, including the hook-aware empty-node path.
31-
- [ ] **Use NULL-safe affected-window partition matching.** `IN` does not select a NULL partition even though SQL window
32-
partitioning groups NULL values. Join the affected-key relation with `IS NOT DISTINCT FROM` for every partition key.
31+
- [x] **Use NULL-safe affected-window partition matching.** Affected-key refresh now uses hash-semi-joinable
32+
`IS NOT DISTINCT FROM` predicates for standard, cascade, running-window, and DuckLake paths. Regression coverage includes
33+
batched conflicting DML on single and composite NULL partitions plus a DuckLake TPCC window whose delivery timestamp moves
34+
from the NULL partition to a non-NULL partition.
3335
- [ ] **Preserve `SUM` NULL semantics.** Weighted SUM alone cannot distinguish numeric zero from no non-NULL inputs. Persist
3436
a hidden `COUNT(sum_argument)` and render NULL when that count reaches zero.
3537
- [ ] **Use an unconditional row count for group existence.** `COUNT(nullable_expression)=0` does not mean a group is empty.

‎src/include/upsert/refresh_compiler.hpp‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,6 @@ string CompileWindowRecompute(const string &view_name, const string &view_query_
5050
const string &catalog_prefix = "", const vector<string> &partition_columns = {},
5151
const vector<WindowPartitionDeltaSpec> &partition_delta_specs = {},
5252
bool emit_cascade_delta = false, const string &affected_keys_sql = "",
53-
const string &affected_key_cols = "", const string &affected_key_tuple = "",
5453
const vector<string> &column_names = {}, bool running_window_incremental = false);
5554
string CompileFullRecompute(const string &view_name, const string &view_query_sql, const string &catalog_prefix = "");
5655

‎src/upsert/refresh_compiler_aux.cpp‎

Lines changed: 84 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,41 @@ static string PartitionOutputColumn(const string &input) {
5151
return StripIdentifierQuotes(pos == string::npos ? input : input.substr(0, pos));
5252
}
5353

54+
static vector<string> PartitionOutputColumns(const vector<string> &partition_columns) {
55+
vector<string> output_columns;
56+
output_columns.reserve(partition_columns.size());
57+
for (auto &partition_column : partition_columns) {
58+
output_columns.push_back(PartitionOutputColumn(partition_column));
59+
}
60+
return output_columns;
61+
}
62+
63+
static string BuildAffectedTableFilter(const vector<string> &columns, const string &outer_alias,
64+
const string &affected_table) {
65+
string match = SqlUtils::BuildNullSafeMatch(columns, "openivm_aff", outer_alias);
66+
return "EXISTS (SELECT 1 FROM " + affected_table + " openivm_aff WHERE " + match + ")";
67+
}
68+
69+
static string BuildDeltaAffectedFilter(const vector<WindowPartitionDeltaSpec> &partition_delta_specs,
70+
const string &delta_where, const string &outer_alias) {
71+
string filter;
72+
for (idx_t i = 0; i < partition_delta_specs.size(); i++) {
73+
if (i > 0) {
74+
filter += " OR ";
75+
}
76+
const auto &spec = partition_delta_specs[i];
77+
string output_col = SqlUtils::QuoteIdentifier(spec.output_column);
78+
string source_col = SqlUtils::QuoteIdentifier(spec.source_column);
79+
string delta_table =
80+
spec.delta_table_sql.empty() ? SqlUtils::QuoteIdentifier(spec.delta_table) : spec.delta_table_sql;
81+
string affected_keys =
82+
"SELECT DISTINCT " + source_col + " AS " + output_col + " FROM " + delta_table + delta_where;
83+
filter += "EXISTS (SELECT 1 FROM (" + affected_keys + ") openivm_aff WHERE " + outer_alias + "." + output_col +
84+
" IS NOT DISTINCT FROM openivm_aff." + output_col + ")";
85+
}
86+
return filter;
87+
}
88+
5489
static vector<string> SplitTopLevelComma(const string &input) {
5590
vector<string> parts;
5691
idx_t start = 0;
@@ -710,14 +745,16 @@ static string BuildRunningWindowSuffixRefreshSQL(const string &view_name, const
710745
string key_match_dt_fk = SqlUtils::BuildNullSafeMatch(vector<string> {plan.partition_column}, "dt", "fk");
711746
string key_match_b_m = "b." + part_q + " IS NOT DISTINCT FROM m." + part_q;
712747
string key_match_d_fk = SqlUtils::BuildNullSafeMatch(vector<string> {plan.partition_column}, "d", "fk");
748+
string affected_data_filter =
749+
BuildAffectedTableFilter(vector<string> {plan.partition_column}, "dt", affected_table);
713750

714751
string sql;
715752
sql += "CREATE OR REPLACE TEMP TABLE " + affected_table + " AS\nSELECT DISTINCT " +
716753
QualifiedColumn("d", plan.partition_column) + " AS " + part_q + "\nFROM " + delta_q + " d\nWHERE " +
717754
delta_positive + ";\n\n";
718-
sql += "CREATE OR REPLACE TEMP TABLE " + bounds_table + " AS\nWITH old_max AS (\n SELECT " + part_q + ", MAX(" +
719-
order_q + ") AS openivm_old_max_order FROM " + data_table + " WHERE " + part_q + " IN (SELECT " + part_q +
720-
" FROM " + affected_table + ") GROUP BY " + part_q + "\n), delta_min AS (\n SELECT " +
755+
sql += "CREATE OR REPLACE TEMP TABLE " + bounds_table + " AS\nWITH old_max AS (\n SELECT dt." + part_q +
756+
", MAX(dt." + order_q + ") AS openivm_old_max_order FROM " + data_table + " dt WHERE " +
757+
affected_data_filter + " GROUP BY dt." + part_q + "\n), delta_min AS (\n SELECT " +
721758
QualifiedColumn("d", plan.partition_column) + " AS " + part_q + ", MIN(" +
722759
QualifiedColumn("d", plan.order_column) + ") AS openivm_delta_min_order\n FROM " + delta_q +
723760
" d\n WHERE " + delta_positive + "\n GROUP BY " + QualifiedColumn("d", plan.partition_column) +
@@ -754,18 +791,29 @@ static string BuildRunningWindowSuffixRefreshSQL(const string &view_name, const
754791
" ORDER BY " + QualifiedColumn("dt", plan.order_column) + " DESC) AS openivm_rn\n FROM " + data_table +
755792
" dt\n JOIN " + fast_table + " fk ON " + key_match_dt_fk +
756793
"\n) openivm_state_ranked\nWHERE openivm_rn = 1;\n\n";
757-
string fallback_filter = part_q + " IN (SELECT " + part_q + " FROM " + fallback_table + ")";
794+
string fallback_keys = "SELECT " + part_q + " FROM " + fallback_table;
795+
string fallback_target_match =
796+
SqlUtils::BuildNullSafeMatch(vector<string> {plan.partition_column}, "openivm_aff", "openivm_target");
797+
string fallback_recompute_match =
798+
SqlUtils::BuildNullSafeMatch(vector<string> {plan.partition_column}, "openivm_aff", "openivm_recompute");
758799
if (emit_cascade_delta) {
759-
sql += "CREATE OR REPLACE TEMP TABLE " + old_temp_table + " AS\nSELECT * FROM " + data_table + "\nWHERE " +
760-
fallback_filter + ";\n\n";
800+
string fallback_old_filter =
801+
BuildAffectedTableFilter(vector<string> {plan.partition_column}, "openivm_old", fallback_table);
802+
string fallback_new_filter =
803+
BuildAffectedTableFilter(vector<string> {plan.partition_column}, "openivm_recompute", fallback_table);
804+
string fallback_delete_filter =
805+
BuildAffectedTableFilter(vector<string> {plan.partition_column}, "openivm_target", fallback_table);
806+
sql += "CREATE OR REPLACE TEMP TABLE " + old_temp_table + " AS\nSELECT * FROM " + data_table +
807+
" openivm_old\nWHERE " + fallback_old_filter + ";\n\n";
761808
sql += "CREATE OR REPLACE TEMP TABLE " + new_temp_table + " AS\nSELECT * FROM (" + view_query_sql +
762-
") openivm_recompute\nWHERE " + fallback_filter + ";\n\n";
763-
sql += "DELETE FROM " + data_table + " WHERE " + fallback_filter + ";\n";
809+
") openivm_recompute\nWHERE " + fallback_new_filter + ";\n\n";
810+
sql += "DELETE FROM " + data_table + " AS openivm_target WHERE " + fallback_delete_filter + ";\n";
764811
sql += "INSERT INTO " + data_table + "\nSELECT * FROM " + new_temp_table + ";\n";
765812
sql += "\n" + BuildSignedMultisetDeltaInsertSQL(delta_table, old_temp_table, new_temp_table);
766813
} else {
767-
sql += BuildDeleteInsertRefreshSQL(data_table, view_query_sql, "openivm_recompute", fallback_filter,
768-
fallback_filter);
814+
sql +=
815+
BuildAffectedKeyRefreshSQL(data_table, view_query_sql, fallback_keys, "openivm_target", "openivm_recompute",
816+
"openivm_aff", fallback_target_match, fallback_recompute_match);
769817
}
770818

771819
auto emit_column_names = plan.output_columns.empty() ? visible_column_names : plan.output_columns;
@@ -1548,10 +1596,9 @@ string CompileFilteredGroupCount(const string &view_name, const string &aux_tabl
15481596
string CompileWindowRecompute(const string &view_name, const string &view_query_sql, const string &delta_ts_filter,
15491597
const string &catalog_prefix, const vector<string> &partition_columns,
15501598
const vector<WindowPartitionDeltaSpec> &partition_delta_specs, bool emit_cascade_delta,
1551-
const string &affected_keys_sql, const string &affected_key_cols,
1552-
const string &affected_key_tuple, const vector<string> &column_names,
1599+
const string &affected_keys_sql, const vector<string> &column_names,
15531600
bool running_window_incremental) {
1554-
bool have_affected_keys = !affected_keys_sql.empty() && !affected_key_cols.empty() && !affected_key_tuple.empty();
1601+
bool have_affected_keys = !affected_keys_sql.empty();
15551602
if (!have_affected_keys && (partition_columns.empty() || partition_delta_specs.empty())) {
15561603
return CompileFullRecompute(view_name, view_query_sql, catalog_prefix);
15571604
}
@@ -1566,41 +1613,25 @@ string CompileWindowRecompute(const string &view_name, const string &view_query_
15661613
string data_table = catalog_prefix + SqlUtils::QuoteIdentifier(IncrementalTableNames::DataTableName(view_name));
15671614
string delta_where = delta_ts_filter.empty() ? "" : " WHERE " + delta_ts_filter;
15681615
string affected_temp_table = SqlUtils::QuoteIdentifier("openivm_affected_" + view_name);
1569-
1570-
string affected_filter;
1571-
if (have_affected_keys) {
1572-
affected_filter =
1573-
affected_key_tuple + " IN (SELECT " + affected_key_cols + " FROM " + affected_temp_table + ")";
1574-
} else {
1575-
for (size_t i = 0; i < partition_delta_specs.size(); i++) {
1576-
if (i > 0) {
1577-
affected_filter += " OR ";
1578-
}
1579-
const auto &spec = partition_delta_specs[i];
1580-
string output_col = SqlUtils::QuoteIdentifier(spec.output_column);
1581-
string source_col = SqlUtils::QuoteIdentifier(spec.source_column);
1582-
string delta_table =
1583-
spec.delta_table_sql.empty() ? SqlUtils::QuoteIdentifier(spec.delta_table) : spec.delta_table_sql;
1584-
affected_filter +=
1585-
output_col + " IN (SELECT DISTINCT " + source_col + " FROM " + delta_table + delta_where + ")";
1586-
}
1587-
}
1616+
auto output_columns = PartitionOutputColumns(partition_columns);
15881617

15891618
OPENIVM_DEBUG_PRINT(
15901619
"[CompileWindowRecompute] Partition columns: %zu, delta specs: %zu, lineage keys: %s, cascade delta: %s\n",
15911620
partition_columns.size(), partition_delta_specs.size(), have_affected_keys ? "yes" : "no",
15921621
emit_cascade_delta ? "enabled" : "disabled");
15931622
if (!emit_cascade_delta) {
15941623
if (!have_affected_keys) {
1595-
return BuildDeleteInsertRefreshSQL(data_table, view_query_sql, "openivm_recompute", affected_filter,
1596-
affected_filter);
1624+
string target_filter = BuildDeltaAffectedFilter(partition_delta_specs, delta_where, "openivm_target");
1625+
string recompute_filter = BuildDeltaAffectedFilter(partition_delta_specs, delta_where, "openivm_recompute");
1626+
return "DELETE FROM " + data_table + " AS openivm_target WHERE " + target_filter + ";\n" + "INSERT INTO " +
1627+
data_table + "\nSELECT * FROM (" + view_query_sql + ") openivm_recompute\nWHERE " +
1628+
recompute_filter + ";\n";
15971629
}
1598-
string sql;
1599-
sql += "CREATE OR REPLACE TEMP TABLE " + affected_temp_table + " AS\n" + affected_keys_sql + ";\n\n";
1600-
sql += BuildDeleteInsertRefreshSQL(data_table, view_query_sql, "openivm_recompute", affected_filter,
1601-
affected_filter);
1602-
sql += "DROP TABLE IF EXISTS " + affected_temp_table + ";\n";
1603-
return sql;
1630+
string target_match = SqlUtils::BuildNullSafeMatch(output_columns, "openivm_aff", "openivm_target");
1631+
string recompute_match = SqlUtils::BuildNullSafeMatch(output_columns, "openivm_aff", "openivm_recompute");
1632+
return BuildAffectedKeyRefreshSQL(data_table, view_query_sql, affected_keys_sql, "openivm_target",
1633+
"openivm_recompute", "openivm_aff", target_match, recompute_match,
1634+
affected_temp_table);
16041635
}
16051636

16061637
string delta_table = catalog_prefix + SqlUtils::QuoteIdentifier(SqlUtils::DeltaName(view_name));
@@ -1610,11 +1641,20 @@ string CompileWindowRecompute(const string &view_name, const string &view_query_
16101641
if (have_affected_keys) {
16111642
sql += "CREATE OR REPLACE TEMP TABLE " + affected_temp_table + " AS\n" + affected_keys_sql + ";\n\n";
16121643
}
1613-
sql += "CREATE OR REPLACE TEMP TABLE " + old_temp_table + " AS\nSELECT * FROM " + data_table + "\nWHERE " +
1614-
affected_filter + ";\n\n";
1644+
string old_filter = have_affected_keys
1645+
? BuildAffectedTableFilter(output_columns, "openivm_old", affected_temp_table)
1646+
: BuildDeltaAffectedFilter(partition_delta_specs, delta_where, "openivm_old");
1647+
string recompute_filter = have_affected_keys
1648+
? BuildAffectedTableFilter(output_columns, "openivm_recompute", affected_temp_table)
1649+
: BuildDeltaAffectedFilter(partition_delta_specs, delta_where, "openivm_recompute");
1650+
string target_filter = have_affected_keys
1651+
? BuildAffectedTableFilter(output_columns, "openivm_target", affected_temp_table)
1652+
: BuildDeltaAffectedFilter(partition_delta_specs, delta_where, "openivm_target");
1653+
sql += "CREATE OR REPLACE TEMP TABLE " + old_temp_table + " AS\nSELECT * FROM " + data_table +
1654+
" openivm_old\nWHERE " + old_filter + ";\n\n";
16151655
sql += "CREATE OR REPLACE TEMP TABLE " + new_temp_table + " AS\nSELECT * FROM (" + view_query_sql +
1616-
") openivm_recompute\nWHERE " + affected_filter + ";\n\n";
1617-
sql += "DELETE FROM " + data_table + " WHERE " + affected_filter + ";\n";
1656+
") openivm_recompute\nWHERE " + recompute_filter + ";\n\n";
1657+
sql += "DELETE FROM " + data_table + " AS openivm_target WHERE " + target_filter + ";\n";
16181658
sql += "INSERT INTO " + data_table + "\nSELECT * FROM " + new_temp_table + ";\n";
16191659
sql += "\n" + BuildSignedMultisetDeltaInsertSQL(delta_table, old_temp_table, new_temp_table);
16201660
if (have_affected_keys) {

0 commit comments

Comments
 (0)