diff --git a/src/core/parser.cpp b/src/core/parser.cpp index 0d01a53..21912c6 100644 --- a/src/core/parser.cpp +++ b/src/core/parser.cpp @@ -1082,9 +1082,13 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC add_profile_record(step.step_name, step.duration_ms, step.detail); } - add_profile_marker("create_mv_system_tables", "refresh_type=" + string(RefreshTypeName(refresh_type)) + - "; lpts_fallback=" + string(lpts_fallback ? "true" : "false")); - AppendCreateMVSystemTablesDDL(ddl, view_name, parse_data_ref.is_replace); + if (MVSystemTablesNeedMigration(con)) { + add_profile_marker("create_mv_system_tables", + "refresh_type=" + string(RefreshTypeName(refresh_type)) + + "; lpts_fallback=" + string(lpts_fallback ? "true" : "false")); + AppendCreateMVSystemTablesDDL(ddl); + } + AppendCreateMVPreflightDDL(ddl, view_name, parse_data_ref.is_replace); if (parse_data_ref.is_replace && !staged_cross_catalog_replace) { add_profile_marker("create_mv_replace_cleanup"); diff --git a/src/core/parser_create_mv_helpers.cpp b/src/core/parser_create_mv_helpers.cpp index 5ffc919..949e942 100644 --- a/src/core/parser_create_mv_helpers.cpp +++ b/src/core/parser_create_mv_helpers.cpp @@ -30,7 +30,25 @@ string BuildUpdateViewJsonSQL(const string &column_name, const string &json, con SqlUtils::EscapeSingleQuotes(json) + "' WHERE view_name = '" + SqlUtils::EscapeSingleQuotes(view_name) + "'"; } -void AppendCreateMVSystemTablesDDL(vector &ddl, const string &view_name, bool is_replace) { +bool MVSystemTablesNeedMigration(Connection &con) { + // Binding this zero-row query checks the complete current schema without + // rerunning DuckLake DDL for every materialized view. + auto schema = con.Query("SELECT v.view_catalog, v.view_schema, v.distinct_aux_meta_json, " + "v.count_distinct_aux_meta_json, v.semi_anti_aux_meta_json, v.lineage_json, " + "v.leftjoin_secondary_meta_json, v.has_join, v.group_recompute_affected_mode, " + "v.group_recompute_source_occurrences_json, v.derived_aggregate_outputs_json, " + "d.last_refresh_ts, d.pending_row_estimate, d.pending_estimate_ts, " + "d.source_catalog, d.source_schema, d.source_table_id, h.mode, r.strategy, " + "p.detail FROM openivm_views v, openivm_delta_tables d, openivm_refresh_hooks h, " + "openivm_refresh_history r, openivm_refresh_profile p LIMIT 0"); + if (schema->HasError()) { + return true; + } + auto legacy_rows = con.Query("SELECT 1 FROM openivm_views WHERE has_join IS NULL LIMIT 1"); + return legacy_rows->HasError() || legacy_rows->RowCount() != 0; +} + +void AppendCreateMVSystemTablesDDL(vector &ddl) { // Matcher metadata columns (signature_hash..nullified_columns_json) stay // NULL unless openivm_enable_view_matching=true; populated by Stage I wiring. ddl.push_back("create table if not exists " + string(openivm::VIEWS_TABLE) + @@ -73,26 +91,6 @@ void AppendCreateMVSystemTablesDDL(vector &ddl, const string &view_name, AddColumnIfNotExists(ddl, openivm::VIEWS_TABLE, "derived_aggregate_outputs_json varchar default null"); AddColumnIfNotExists(ddl, openivm::VIEWS_TABLE, "view_catalog varchar default null"); AddColumnIfNotExists(ddl, openivm::VIEWS_TABLE, "view_schema varchar default null"); - if (!is_replace) { - string escaped_view_name = SqlUtils::EscapeSingleQuotes(view_name); - string escaped_data_table = SqlUtils::EscapeSingleQuotes(IncrementalTableNames::DataTableName(view_name)); - string stale_mv_condition = "view_name = '" + escaped_view_name + - "' AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE " - "table_name = '" + - escaped_view_name + - "') AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE " - "table_name = '" + - escaped_data_table + "')"; - // CREATE MV executes as multiple catalog statements. If a process dies or loses - // a DuckDB file lock after writing metadata but before creating the physical - // DuckLake/default-catalog objects, a retry should clean that stale row rather - // than report a misleading duplicate MV. - ddl.push_back("DELETE FROM " + string(openivm::VIEWS_TABLE) + " WHERE " + stale_mv_condition); - ddl.push_back("SELECT CASE WHEN EXISTS (SELECT 1 FROM " + string(openivm::VIEWS_TABLE) + - " WHERE view_name = '" + escaped_view_name + - "') THEN error('Duplicate key: materialized view \"" + escaped_view_name + - "\" already exists') ELSE NULL END"); - } // Refresh hooks: extensions can register custom SQL to run on MV refresh // mode: 'replace' (instead of ivm), 'before' (before ivm), 'after' (after ivm) @@ -146,4 +144,27 @@ void AppendCreateMVSystemTablesDDL(vector &ddl, const string &view_name, " primary key(refresh_id, step_order))"); } +void AppendCreateMVPreflightDDL(vector &ddl, const string &view_name, bool is_replace) { + if (!is_replace) { + string escaped_view_name = SqlUtils::EscapeSingleQuotes(view_name); + string escaped_data_table = SqlUtils::EscapeSingleQuotes(IncrementalTableNames::DataTableName(view_name)); + string stale_mv_condition = "view_name = '" + escaped_view_name + + "' AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE " + "table_name = '" + + escaped_view_name + + "') AND NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE " + "table_name = '" + + escaped_data_table + "')"; + // CREATE MV executes as multiple catalog statements. If a process dies or loses + // a DuckDB file lock after writing metadata but before creating the physical + // DuckLake/default-catalog objects, a retry should clean that stale row rather + // than report a misleading duplicate MV. + ddl.push_back("DELETE FROM " + string(openivm::VIEWS_TABLE) + " WHERE " + stale_mv_condition); + ddl.push_back("SELECT CASE WHEN EXISTS (SELECT 1 FROM " + string(openivm::VIEWS_TABLE) + + " WHERE view_name = '" + escaped_view_name + + "') THEN error('Duplicate key: materialized view \"" + escaped_view_name + + "\" already exists') ELSE NULL END"); + } +} + } // namespace duckdb diff --git a/src/core/parser_ddl.cpp b/src/core/parser_ddl.cpp index 7efe44a..adbcc73 100644 --- a/src/core/parser_ddl.cpp +++ b/src/core/parser_ddl.cpp @@ -2,6 +2,7 @@ #include "core/openivm_constants.hpp" #include "core/openivm_debug.hpp" +#include "core/parser_create_mv_helpers.hpp" #include "core/refresh_locks.hpp" #include "core/sql_utils.hpp" #include "duckdb/catalog/catalog.hpp" @@ -93,6 +94,21 @@ static bool IsMetadataSchemaStatement(const string &statement) { return StringUtil::StartsWith(lower, "create table") || StringUtil::StartsWith(lower, "alter table"); } +static void EnsureTransactionalMetadataSchemas(vector &statements) { + for (auto &statement : statements) { + if (IsMetadataSchemaStatement(statement)) { + return; + } + } + vector schema; + AppendCreateMVSystemTablesDDL(schema); + for (auto &statement : schema) { + if (IsMetadataSchemaStatement(statement)) { + statements.push_back(std::move(statement)); + } + } +} + static bool MatchesNowCall(const string &statement, idx_t offset) { static const string now_call = "now()"; if (offset + now_call.size() > statement.size()) { @@ -553,6 +569,7 @@ void TransactionalMVMetadataState::Register(ClientContext &context, const vector statements.push_back(StabilizeTransactionalMetadata(statement, timestamp_sql)); } } + EnsureTransactionalMetadataSchemas(statements); } void TransactionalMVMetadataState::RegisterSQL(const string &sql, const string &view_name) { @@ -562,6 +579,7 @@ void TransactionalMVMetadataState::RegisterSQL(const string &sql, const string & statements.push_back(std::move(statement)); } } + EnsureTransactionalMetadataSchemas(statements); } void TransactionalMVMetadataState::IncludeView(const string &view_name) { diff --git a/src/include/core/parser_create_mv_helpers.hpp b/src/include/core/parser_create_mv_helpers.hpp index 31cbc01..6c11369 100644 --- a/src/include/core/parser_create_mv_helpers.hpp +++ b/src/include/core/parser_create_mv_helpers.hpp @@ -6,7 +6,9 @@ namespace duckdb { string SqlCsvLiteralOrNull(const vector &values); -void AppendCreateMVSystemTablesDDL(vector &ddl, const string &view_name, bool is_replace); +bool MVSystemTablesNeedMigration(Connection &con); +void AppendCreateMVSystemTablesDDL(vector &ddl); +void AppendCreateMVPreflightDDL(vector &ddl, const string &view_name, bool is_replace); string BuildUpdateViewJsonSQL(const string &column_name, const string &json, const string &view_name); } // namespace duckdb diff --git a/test/sql/metadata.test b/test/sql/metadata.test index fd86998..47b5fa4 100644 --- a/test/sql/metadata.test +++ b/test/sql/metadata.test @@ -133,6 +133,9 @@ statement ok INSERT INTO openivm_delta_tables(view_name, table_name, last_update) VALUES ('mv_legacy_self_join', 'openivm_delta_legacy_self_join_src', now()); +statement ok +SET openivm_profile_refresh = true; + statement ok CREATE MATERIALIZED VIEW mv_trigger_join_backfill AS SELECT id FROM legacy_self_join_src; @@ -141,6 +144,24 @@ SELECT has_join FROM openivm_views WHERE view_name = 'mv_legacy_self_join'; ---- true +query I +SELECT count(*) FROM openivm_refresh_profile +WHERE view_name = 'mv_trigger_join_backfill' AND step_name = 'create_mv_system_tables'; +---- +1 + +statement ok +CREATE MATERIALIZED VIEW mv_after_join_backfill AS SELECT id FROM legacy_self_join_src; + +query I +SELECT count(*) FROM openivm_refresh_profile +WHERE view_name = 'mv_after_join_backfill' AND step_name = 'create_mv_system_tables'; +---- +0 + +statement ok +SET openivm_profile_refresh = false; + statement ok DELETE FROM openivm_delta_tables WHERE view_name = 'mv_legacy_self_join'; @@ -150,6 +171,9 @@ DELETE FROM openivm_views WHERE view_name = 'mv_legacy_self_join'; statement ok DROP VIEW mv_trigger_join_backfill; +statement ok +DROP VIEW mv_after_join_backfill; + statement ok DROP TABLE legacy_self_join_src;