Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
38 changes: 19 additions & 19 deletions src/upsert/refresh_sql.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1210,7 +1210,7 @@
OPENIVM_DEBUG_PRINT("[UPSERT] DISTINCT_INCREMENTAL view has no aux meta — "
"falling through to "
"GROUP_RECOMPUTE\n");
[[fallthrough]];

Check warning on line 1213 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 1213 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 1213 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 1213 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 +1244,7 @@
OPENIVM_DEBUG_PRINT("[UPSERT] SEMI_ANTI_RECOMPUTE view has no aux meta — "
"falling through to "
"GROUP_RECOMPUTE\n");
[[fallthrough]];

Check warning on line 1247 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 1247 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 1247 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 1247 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 +1296,7 @@
break;
}
case RefreshType::TOP_K:
[[fallthrough]];

Check warning on line 1299 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 1299 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 Expand Up @@ -1565,6 +1565,11 @@
string raw_refresh_sql;
if (IsEmptyDeltaPlan(plan.get())) {
raw_refresh_sql = BuildEmptyDeltaInsert(view_name, column_names, column_types, active_facts.target_dialect);
// The generated statement is ours (no user text), so its prefix is known exactly.
string empty_prefix = "INSERT INTO " + SqlUtils::DeltaName(view_name) + " SELECT ";
D_ASSERT(raw_refresh_sql.compare(0, empty_prefix.size(), empty_prefix) == 0);
raw_refresh_sql = "INSERT INTO " + delta_view_name + " (" + SqlUtils::JoinQuotedColumns(column_names) +
") SELECT " + raw_refresh_sql.substr(empty_prefix.size());
OPENIVM_DEBUG_PRINT("[UPSERT] Delta plan is empty; generated no-op delta insert for '%s'\n",
view_name.c_str());
} else {
Expand All @@ -1577,14 +1582,20 @@
bool emit_spark_hints = dialect == SqlDialect::SPARK &&
(active_facts.emit_spark_hints ||
SqlUtils::GetBoolSetting(con_ctx, "openivm_emit_spark_hints", false));
auto cte_list = AstToCteList(*ast, dialect, emit_spark_hints);
raw_refresh_sql = cte_list->ToQuery(false);
if (use_transient_mv_delta && ast->NodeType() == "Insert" && ast->children.size() == 1) {
auto inline_cte_list = AstToCteList(*ast->children[0], dialect, emit_spark_hints);
inline_delta_select_query = inline_cte_list->ToQuery(false, column_names);
if (!inline_delta_select_query.empty() && inline_delta_select_query.back() == ';') {
inline_delta_select_query.pop_back();
}
if (ast->NodeType() != "Insert" || ast->children.size() != 1) {
throw InternalException("delta plan root is not a single-child INSERT");
}
// Build the INSERT target and column list from the AST's Insert node instead of searching the
// serialized text (a string literal could contain the target name).
auto select_cte_list = AstToCteList(*ast->children[0], dialect, emit_spark_hints);
string delta_select = select_cte_list->ToQuery(false, column_names);
if (!delta_select.empty() && delta_select.back() == ';') {
delta_select.pop_back();
}
raw_refresh_sql = "INSERT INTO " + delta_view_name + " (" + SqlUtils::JoinQuotedColumns(column_names) +
") " + delta_select + ";";
if (use_transient_mv_delta) {
inline_delta_select_query = delta_select;
}
add_profile_step("generate_refresh_sql.lpts", lpts_start,
"delta_sql_bytes=" + to_string(raw_refresh_sql.size()));
Expand All @@ -1596,17 +1607,6 @@
view_name + "': " + e.what());
}
}
string insert_target_bare = "INSERT INTO " + SqlUtils::DeltaName(view_name);
auto insert_pos = raw_refresh_sql.find(insert_target_bare);
if (insert_pos != string::npos) {
if (!internal_catalog_prefix.empty()) {
raw_refresh_sql.replace(insert_pos, insert_target_bare.size(), "INSERT INTO " + delta_view_name);
insert_pos = raw_refresh_sql.find("INSERT INTO " + delta_view_name);
}
string col_list = "(" + SqlUtils::JoinQuotedColumns(column_names) + ") ";
string full_insert = "INSERT INTO " + delta_view_name;
raw_refresh_sql.insert(insert_pos + full_insert.size(), " " + col_list);
}
delta_query += raw_refresh_sql;

if (has_downstream && view_query_type == RefreshType::SIMPLE_AGGREGATE) {
Expand Down
42 changes: 42 additions & 0 deletions test/sql/insert_into_literal_delta_name.test
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
# name: test/sql/insert_into_literal_delta_name.test
# description: a string literal containing the internal delta INSERT target must not hijack the refresh INSERT column list
# group: [sql]

require openivm

statement ok
SET openivm_files_path='__TEST_DIR__';

statement ok
CREATE TABLE lit_t (grp INT, v INT, note VARCHAR);

statement ok
INSERT INTO lit_t VALUES (1, 10, 'k');

statement ok
CREATE MATERIALIZED VIEW mv_lit AS SELECT grp, SUM(v) AS s FROM lit_t
WHERE note <> 'INSERT INTO openivm_delta___openivm_mv_6d656d6f7279_6d61696e_6d765f6c6974_' GROUP BY grp;

statement ok
INSERT INTO lit_t VALUES (1, 5, 'k'), (2, 7, 'k'), (3, 1, 'INSERT INTO openivm_delta___openivm_mv_6d656d6f7279_6d61696e_6d765f6c6974_');

statement ok
PRAGMA refresh('mv_lit');

query II
SELECT grp, s FROM mv_lit ORDER BY grp;
----
1 15
2 7

query I
SELECT count(*) FROM (
(SELECT * FROM mv_lit EXCEPT ALL SELECT grp, SUM(v) FROM lit_t
WHERE note <> 'INSERT INTO openivm_delta___openivm_mv_6d656d6f7279_6d61696e_6d765f6c6974_' GROUP BY grp)
UNION ALL
(SELECT grp, SUM(v) FROM lit_t
WHERE note <> 'INSERT INTO openivm_delta___openivm_mv_6d656d6f7279_6d61696e_6d765f6c6974_' GROUP BY grp
EXCEPT ALL SELECT * FROM mv_lit)
);
----
0
Loading