Finish transactional MV lifecycle

Author: ila

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

@@ -19,10 +19,12 @@ maintenance path. A bug must not be hidden by weakening a test or silently chang
   decorator now captures only the INSERT, UPDATE, and DELETE rows selected by DuckDB after match and action predicates are
   resolved. Updates emit the required delete-old/insert-new pair, volatile defaults are evaluated once, `RETURNING` is
   preserved, and all writes remain in the base DML transaction. The same boundary also covers native `MERGE` actions.
-- [ ] **Make MV lifecycle and native refresh transactional.** CREATE, REPLACE, DROP, ALTER, and `PRAGMA refresh` currently
-  execute material work through helper autocommit connections. Caller rollback cannot restore a consistent collection of
-  user view, backing table, metadata, and deltas; failed replacement can destroy the old MV. Use the caller transaction for
-  native-catalog work and stage cross-catalog replacement before publication.
+- [x] **Make MV lifecycle and native refresh transactional.** Native CREATE, REPLACE, DROP, ALTER, and refresh inside an
+  explicit caller transaction now mutate the user view, backing state, metadata, and deltas atomically and retain their
+  OpenIVM locks through commit or rollback. Autocommit refresh temporarily retains the established locked executor because
+  DuckDB query-pragma preprocessing ends a transaction before its returned program completes; `refresh.cpp` records the
+  native-operator follow-up. DuckLake replacement materializes data and auxiliary state under unpublished staging names
+  before publishing them.
 - [ ] **Continue cascade traversal when the current node has no source deltas.** The root empty-delta fast path returns
   before visiting downstream nodes. A child can have independent source deltas or pending parent-delta rows after an earlier
   failed cascade. Model `node skipped` separately from `graph traversal complete` and checkpoint each node independently.

src/core/parser.cpp

@@ -378,6 +378,12 @@ MaterializedViewParserExtension::PlanFunction(ParserExtensionInfo *info, ClientC
 		select_planner.CreatePlan(std::move(select_parser.statements[0]));
 		auto select_plan = std::move(select_planner.plan);
 		visible_output_count = select_planner.names.size();
+		for (auto &name : select_planner.names) {
+			if (StringUtil::CIEquals(name, openivm::MULTIPLICITY_COL) ||
+			    StringUtil::CIEquals(name, openivm::TIMESTAMP_COL)) {
+				throw BinderException("Materialized-view output uses reserved OpenIVM column '%s'", name);
+			}
+		}
 		add_create_profile_step("create_compile_select_plan", select_parse_plan_start);
 
 		// Inline CTEs without running the full optimizer, which can reshape plans
@@ -1519,9 +1525,11 @@ string MaterializedViewLifecycleQuery(ClientContext &context, const FunctionPara
 	    MaterializedViewParserExtension::PlanFunction(nullptr, context, std::move(parse_result.parse_data));
 	if (plan_result.function.name == OPENIVM_TRANSACTIONAL_DDL_FUNCTION) {
 		if (!lock_view_name.empty()) {
-			TransactionalMVLockState::Get(context).Acquire({lock_view_name}, {});
+			auto &default_entry = ClientData::Get(context).catalog_search_path->GetDefault();
+			auto &catalog = Catalog::GetCatalog(context, default_entry.catalog);
+			TransactionalMVLockState::Get(context).Acquire({lock_view_name}, {}, {&catalog});
 		}
-		return RenderTransactionalDDL(plan_result.parameters);
+		return RenderTransactionalDDL(context, plan_result.parameters);
 	}
 	if (!lock_view_name.empty()) {
 		ViewLockGuard view_guard(lock_view_name);
@@ -1606,7 +1614,8 @@ string MaterializedViewDropQuery(ClientContext &context, const FunctionParameter
 	}
 	auto view_delta_table = SqlUtils::DeltaName(drop.info->name);
 	delta_tables.push_back(view_delta_table);
-	TransactionalMVLockState::Get(context).Acquire({drop.info->name}, delta_tables);
+	auto &catalog = Catalog::GetCatalog(context, catalog_name);
+	TransactionalMVLockState::Get(context).Acquire({drop.info->name}, delta_tables, {&catalog});
 	return program;
 }
 } // namespace duckdb

src/core/parser_ddl.cpp

@@ -461,9 +461,23 @@ void ExecuteDropView(ClientContext &context, TableFunctionInput &input, DataChun
 	state.finished = true;
 }
 
-string RenderTransactionalDDL(const vector<Value> &parameters) {
+string RenderTransactionalDDL(ClientContext &context, const vector<Value> &parameters) {
+	struct ProfileRow {
+		string view_name;
+		string step_name;
+		int64_t duration_ms;
+		string detail;
+	};
+	struct PendingProfileMarker {
+		string view_name;
+		string step_name;
+		string detail;
+		idx_t statement_start;
+	};
+
 	string sql;
-	auto append_statement = [&](const string &statement) {
+	idx_t logical_statement_count = 0;
+	auto append_statement = [&](const string &statement, bool logical_statement = true) {
 		if (statement.empty()) {
 			return;
 		}
@@ -472,12 +486,57 @@ string RenderTransactionalDDL(const vector<Value> &parameters) {
 			sql += ";";
 		}
 		sql += "\n";
+		if (logical_statement) {
+			logical_statement_count++;
+		}
+	};
+	Value profile_value;
+	bool profile_enabled = context.TryGetCurrentSetting("openivm_profile_refresh", profile_value) &&
+	                       !profile_value.IsNull() && BooleanValue::Get(profile_value);
+	vector<ProfileRow> profile_rows;
+	unique_ptr<PendingProfileMarker> pending_profile;
+	auto finish_profile_marker = [&]() {
+		if (!pending_profile) {
+			return;
+		}
+		auto statement_count = logical_statement_count - pending_profile->statement_start;
+		auto detail = pending_profile->detail;
+		if (!detail.empty()) {
+			detail += "; ";
+		}
+		detail +=
+		    "statements=" + to_string(statement_count) + "; transactional_program=true; duration_not_measured=true";
+		profile_rows.push_back({pending_profile->view_name, pending_profile->step_name, 0, std::move(detail)});
+		pending_profile.reset();
 	};
 	for (auto &parameter : parameters) {
 		auto statement = parameter.GetValue<string>();
-		if (statement.empty() || StringUtil::StartsWith(statement, OPENIVM_DDL_CLEANUP_PREFIX) ||
-		    StringUtil::StartsWith(statement, OPENIVM_DDL_PROFILE_PREFIX) ||
-		    StringUtil::StartsWith(statement, OPENIVM_DDL_PROFILE_RECORD_PREFIX)) {
+		if (statement.empty() || StringUtil::StartsWith(statement, OPENIVM_DDL_CLEANUP_PREFIX)) {
+			continue;
+		}
+		if (StringUtil::StartsWith(statement, OPENIVM_DDL_PROFILE_RECORD_PREFIX)) {
+			if (profile_enabled) {
+				string view_name;
+				string step_name;
+				string detail;
+				int64_t duration_ms = 0;
+				ParseCreateMVProfileRecord(statement.substr(strlen(OPENIVM_DDL_PROFILE_RECORD_PREFIX)), view_name,
+				                           step_name, duration_ms, detail);
+				profile_rows.push_back({std::move(view_name), std::move(step_name), duration_ms, std::move(detail)});
+			}
+			continue;
+		}
+		if (StringUtil::StartsWith(statement, OPENIVM_DDL_PROFILE_PREFIX)) {
+			if (profile_enabled) {
+				finish_profile_marker();
+				string view_name;
+				string step_name;
+				string detail;
+				ParseCreateMVProfileMarker(statement.substr(strlen(OPENIVM_DDL_PROFILE_PREFIX)), view_name, step_name,
+				                           detail);
+				pending_profile = make_uniq<PendingProfileMarker>(PendingProfileMarker {
+				    std::move(view_name), std::move(step_name), std::move(detail), logical_statement_count});
+			}
 			continue;
 		}
 		if (StringUtil::StartsWith(statement, OPENIVM_DDL_CREATE_DELTA_FROM_DATA_PREFIX)) {
@@ -493,10 +552,11 @@ string RenderTransactionalDDL(const vector<Value> &parameters) {
 			                 " AS SELECT *, 1::INTEGER AS " + string(openivm::MULTIPLICITY_COL) +
 			                 ", now()::TIMESTAMP AS " + string(openivm::TIMESTAMP_COL) + " FROM " + data_table +
 			                 " LIMIT 0");
-			append_statement("ALTER TABLE " + delta_table + " ALTER " + string(openivm::MULTIPLICITY_COL) +
-			                 " SET DEFAULT 1");
+			append_statement(
+			    "ALTER TABLE " + delta_table + " ALTER " + string(openivm::MULTIPLICITY_COL) + " SET DEFAULT 1", false);
 			append_statement("ALTER TABLE " + delta_table + " ALTER " + string(openivm::TIMESTAMP_COL) +
-			                 " SET DEFAULT now()");
+			                     " SET DEFAULT now()",
+			                 false);
 			continue;
 		}
 
@@ -516,6 +576,27 @@ string RenderTransactionalDDL(const vector<Value> &parameters) {
 		}
 		append_statement(statement);
 	}
+	if (profile_enabled) {
+		finish_profile_marker();
+		if (!profile_rows.empty()) {
+			auto view_name = profile_rows.front().view_name;
+			profile_rows.push_back(
+			    {view_name, "create_mv_total", 0, "transactional_program=true; duration_not_measured=true"});
+			auto now = std::chrono::steady_clock::now().time_since_epoch();
+			auto refresh_id = view_name + "_create_tx_" +
+			                  to_string(std::chrono::duration_cast<std::chrono::nanoseconds>(now).count());
+			for (idx_t step_order = 0; step_order < profile_rows.size(); step_order++) {
+				auto &row = profile_rows[step_order];
+				append_statement("INSERT OR REPLACE INTO " + string(openivm::PROFILE_TABLE) +
+				                     " (refresh_id, view_name, step_order, step_name, duration_ms, detail) VALUES ('" +
+				                     SqlUtils::EscapeValue(refresh_id) + "', '" + SqlUtils::EscapeValue(row.view_name) +
+				                     "', " + to_string(step_order) + ", '" + SqlUtils::EscapeValue(row.step_name) +
+				                     "', " + to_string(row.duration_ms) + ", '" + SqlUtils::EscapeValue(row.detail) +
+				                     "')",
+				                 false);
+			}
+		}
+	}
 	append_statement("SELECT true AS \"MATERIALIZED VIEW CREATION\"");
 	return sql;
 }

src/core/refresh_locks.cpp

@@ -7,38 +7,49 @@ std::unordered_map<string, unique_ptr<std::mutex>> RefreshLocks::view_mutexes_;
 std::unordered_map<string, unique_ptr<std::mutex>> RefreshLocks::delta_mutexes_;
 std::unordered_map<const Catalog *, unique_ptr<DeltaCatalogPhaseGate>> RefreshLocks::delta_catalog_gates_;
 
-void DeltaCatalogPhaseGate::EnterWrite() {
+static bool HasOtherOwner(const unordered_map<ClientContext *, idx_t> &owners, ClientContext &owner) {
+	return !owners.empty() && (owners.size() > 1 || owners.find(&owner) == owners.end());
+}
+
+void DeltaCatalogPhaseGate::EnterWrite(ClientContext &owner) {
 	std::unique_lock<mutex> guard(lock);
 	// Once a refresh is waiting, stop admitting new writers so the finite set of
-	// active transactions can drain. Writers otherwise run concurrently.
-	condition.wait(guard, [&]() { return active_refreshes == 0 && waiting_refreshes == 0; });
-	active_writers++;
+	// active transactions can drain. The refresh owner's own later DML remains
+	// reentrant, which lets transaction-local DML and refresh compose safely.
+	condition.wait(guard, [&]() {
+		bool owner_is_active = active_writers.find(&owner) != active_writers.end() ||
+		                       active_refreshes.find(&owner) != active_refreshes.end();
+		return !HasOtherOwner(active_refreshes, owner) && (waiting_refreshes == 0 || owner_is_active);
+	});
+	active_writers[&owner]++;
 }
 
-void DeltaCatalogPhaseGate::ExitWrite() {
+void DeltaCatalogPhaseGate::ExitWrite(ClientContext &owner) {
 	lock_guard<mutex> guard(lock);
-	D_ASSERT(active_writers > 0);
-	active_writers--;
-	if (active_writers == 0) {
-		condition.notify_all();
+	auto entry = active_writers.find(&owner);
+	D_ASSERT(entry != active_writers.end() && entry->second > 0);
+	if (--entry->second == 0) {
+		active_writers.erase(entry);
 	}
+	condition.notify_all();
 }
 
-void DeltaCatalogPhaseGate::EnterRefresh() {
+void DeltaCatalogPhaseGate::EnterRefresh(ClientContext &owner) {
 	std::unique_lock<mutex> guard(lock);
 	waiting_refreshes++;
-	condition.wait(guard, [&]() { return active_writers == 0; });
+	condition.wait(guard, [&]() { return !HasOtherOwner(active_writers, owner); });
 	waiting_refreshes--;
-	active_refreshes++;
+	active_refreshes[&owner]++;
 }
 
-void DeltaCatalogPhaseGate::ExitRefresh() {
+void DeltaCatalogPhaseGate::ExitRefresh(ClientContext &owner) {
 	lock_guard<mutex> guard(lock);
-	D_ASSERT(active_refreshes > 0);
-	active_refreshes--;
-	if (active_refreshes == 0) {
-		condition.notify_all();
+	auto entry = active_refreshes.find(&owner);
+	D_ASSERT(entry != active_refreshes.end() && entry->second > 0);
+	if (--entry->second == 0) {
+		active_refreshes.erase(entry);
 	}
+	condition.notify_all();
 }
 
 std::mutex &RefreshLocks::GetViewMutex(const string &view_name) {
@@ -88,27 +99,52 @@ void RefreshLocks::UnlockDelta(const string &delta_table_name) {
 	GetDeltaMutex(delta_table_name).unlock();
 }
 
-void RefreshLocks::EnterDeltaWrite(Catalog &catalog) {
-	GetDeltaCatalogGate(catalog).EnterWrite();
+void RefreshLocks::EnterDeltaWrite(ClientContext &owner, Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).EnterWrite(owner);
 }
 
-void RefreshLocks::ExitDeltaWrite(Catalog &catalog) {
-	GetDeltaCatalogGate(catalog).ExitWrite();
+void RefreshLocks::ExitDeltaWrite(ClientContext &owner, Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).ExitWrite(owner);
 }
 
-void RefreshLocks::EnterDeltaRefresh(Catalog &catalog) {
-	GetDeltaCatalogGate(catalog).EnterRefresh();
+void RefreshLocks::EnterDeltaRefresh(ClientContext &owner, Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).EnterRefresh(owner);
 }
 
-void RefreshLocks::ExitDeltaRefresh(Catalog &catalog) {
-	GetDeltaCatalogGate(catalog).ExitRefresh();
+void RefreshLocks::ExitDeltaRefresh(ClientContext &owner, Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).ExitRefresh(owner);
 }
 
 TransactionalMVLockState &TransactionalMVLockState::Get(ClientContext &context) {
-	return *context.registered_state->GetOrCreate<TransactionalMVLockState>("openivm_transactional_mv_locks");
+	auto state = context.registered_state->GetOrCreate<TransactionalMVLockState>("openivm_transactional_mv_locks");
+	state->owner = &context;
+	return *state;
+}
+
+optional_ptr<TransactionalMVLockState> TransactionalMVLockState::TryGet(ClientContext &context) {
+	return context.registered_state->Get<TransactionalMVLockState>("openivm_transactional_mv_locks");
+}
+
+bool TransactionalMVLockState::OwnsRefreshDelta(Catalog &catalog, const string &delta_table_name) const {
+	return locked_catalogs.find(&catalog) != locked_catalogs.end() &&
+	       locked_delta_tables.find(delta_table_name) != locked_delta_tables.end();
 }
 
-void TransactionalMVLockState::Acquire(const vector<string> &view_names, const vector<string> &delta_table_names) {
+void TransactionalMVLockState::Acquire(const vector<string> &view_names, const vector<string> &delta_table_names,
+                                       const vector<Catalog *> &source_catalogs) {
+	if (!owner) {
+		throw InternalException("OpenIVM transactional lock state has no owning client context");
+	}
+	auto sorted_catalogs = source_catalogs;
+	std::sort(sorted_catalogs.begin(), sorted_catalogs.end(),
+	          [](Catalog *left, Catalog *right) { return left->GetName() < right->GetName(); });
+	sorted_catalogs.erase(std::unique(sorted_catalogs.begin(), sorted_catalogs.end()), sorted_catalogs.end());
+	for (auto catalog : sorted_catalogs) {
+		if (locked_catalogs.insert(catalog).second) {
+			catalog_guards.push_back(make_uniq<DeltaCatalogRefreshGuard>(*owner, *catalog));
+		}
+	}
+
 	auto sorted_views = view_names;
 	std::sort(sorted_views.begin(), sorted_views.end());
 	sorted_views.erase(std::unique(sorted_views.begin(), sorted_views.end()), sorted_views.end());
@@ -139,8 +175,10 @@ void TransactionalMVLockState::TransactionRollback(MetaTransaction &transaction,
 void TransactionalMVLockState::Release() {
 	delta_guards.clear();
 	view_guards.clear();
+	catalog_guards.clear();
 	locked_delta_tables.clear();
 	locked_views.clear();
+	locked_catalogs.clear();
 }
 
 } // namespace duckdb

src/include/core/parser_ddl.hpp

@@ -19,7 +19,7 @@ enum class DDLExecutionMode : uint8_t { CALLER_TRANSACTION, STAGED_CROSS_CATALOG
 
 void ConfigureDDLExecutorResult(ParserExtensionPlanResult &result,
                                 DDLExecutionMode mode = DDLExecutionMode::STAGED_CROSS_CATALOG);
-string RenderTransactionalDDL(const vector<Value> &parameters);
+string RenderTransactionalDDL(ClientContext &context, const vector<Value> &parameters);
 void ExecuteStagedDDL(ClientContext &context, const vector<Value> &parameters);
 string BuildCreateDeltaFromDataOperation(const string &delta_table, const string &data_table, bool replace);
 string BuildDropViewStatement(const DropInfo &drop_info);