Harden transactional delta capture

Author: ila

src/core/refresh_locks.cpp

@@ -5,7 +5,41 @@ namespace duckdb {
 std::mutex RefreshLocks::map_mutex_;
 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<string, unique_ptr<std::mutex>> RefreshLocks::delta_catalog_mutexes_;
+std::unordered_map<const Catalog *, unique_ptr<DeltaCatalogPhaseGate>> RefreshLocks::delta_catalog_gates_;
+
+void DeltaCatalogPhaseGate::EnterWrite() {
+	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++;
+}
+
+void DeltaCatalogPhaseGate::ExitWrite() {
+	lock_guard<mutex> guard(lock);
+	D_ASSERT(active_writers > 0);
+	active_writers--;
+	if (active_writers == 0) {
+		condition.notify_all();
+	}
+}
+
+void DeltaCatalogPhaseGate::EnterRefresh() {
+	std::unique_lock<mutex> guard(lock);
+	waiting_refreshes++;
+	condition.wait(guard, [&]() { return active_writers == 0; });
+	waiting_refreshes--;
+	active_refreshes++;
+}
+
+void DeltaCatalogPhaseGate::ExitRefresh() {
+	lock_guard<mutex> guard(lock);
+	D_ASSERT(active_refreshes > 0);
+	active_refreshes--;
+	if (active_refreshes == 0) {
+		condition.notify_all();
+	}
+}
 
 std::mutex &RefreshLocks::GetViewMutex(const string &view_name) {
 	std::lock_guard<std::mutex> guard(map_mutex_);
@@ -25,11 +59,11 @@ std::mutex &RefreshLocks::GetDeltaMutex(const string &delta_table_name) {
 	return *entry;
 }
 
-std::mutex &RefreshLocks::GetDeltaCatalogMutex(const string &catalog_name) {
+DeltaCatalogPhaseGate &RefreshLocks::GetDeltaCatalogGate(Catalog &catalog) {
 	std::lock_guard<std::mutex> guard(map_mutex_);
-	auto &entry = delta_catalog_mutexes_[catalog_name];
+	auto &entry = delta_catalog_gates_[&catalog];
 	if (!entry) {
-		entry = duckdb::unique_ptr<std::mutex>(new std::mutex());
+		entry = make_uniq<DeltaCatalogPhaseGate>();
 	}
 	return *entry;
 }
@@ -54,12 +88,20 @@ void RefreshLocks::UnlockDelta(const string &delta_table_name) {
 	GetDeltaMutex(delta_table_name).unlock();
 }
 
-void RefreshLocks::LockDeltaCatalog(const string &catalog_name) {
-	GetDeltaCatalogMutex(catalog_name).lock();
+void RefreshLocks::EnterDeltaWrite(Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).EnterWrite();
+}
+
+void RefreshLocks::ExitDeltaWrite(Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).ExitWrite();
+}
+
+void RefreshLocks::EnterDeltaRefresh(Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).EnterRefresh();
 }
 
-void RefreshLocks::UnlockDeltaCatalog(const string &catalog_name) {
-	GetDeltaCatalogMutex(catalog_name).unlock();
+void RefreshLocks::ExitDeltaRefresh(Catalog &catalog) {
+	GetDeltaCatalogGate(catalog).ExitRefresh();
 }
 
 } // namespace duckdb

src/core/refresh_metadata.cpp

@@ -122,6 +122,28 @@ RefreshMetadata::SourceLocation RefreshMetadata::GetSourceLocation(const string
 	return loc;
 }
 
+vector<RefreshMetadata::DeltaSource> RefreshMetadata::GetDeltaSources(const string &view_name,
+                                                                      const string &fallback_catalog,
+                                                                      const string &fallback_schema) {
+	auto result = con.Query("SELECT table_name, catalog_type, source_catalog, source_schema FROM " +
+	                        string(openivm::DELTA_TABLES_TABLE) + " WHERE view_name = '" +
+	                        SqlUtils::EscapeValue(view_name) + "'");
+	vector<DeltaSource> sources;
+	if (result->HasError()) {
+		return sources;
+	}
+	for (idx_t row = 0; row < result->RowCount(); row++) {
+		DeltaSource source;
+		source.table_name = result->GetValue(0, row).ToString();
+		source.catalog_type = result->GetValue(1, row).IsNull() ? "duckdb" : result->GetValue(1, row).ToString();
+		source.catalog_name =
+		    result->GetValue(2, row).IsNull() ? fallback_catalog : result->GetValue(2, row).ToString();
+		source.schema_name = result->GetValue(3, row).IsNull() ? fallback_schema : result->GetValue(3, row).ToString();
+		sources.push_back(std::move(source));
+	}
+	return sources;
+}
+
 string RefreshMetadata::ResolveDeltaQualifiedName(const string &view_name, const string &delta_table_name,
                                                   const string &fallback_catalog, const string &fallback_schema) {
 	auto loc = GetSourceLocation(view_name, delta_table_name, fallback_catalog, fallback_schema);
@@ -259,6 +281,12 @@ vector<string> RefreshMetadata::GetDownstreamViews(const string &view_name) {
 	return result; // topological order: downstream parents before fan-in children
 }
 
+bool RefreshMetadata::HasDownstreamViews(const string &view_name) {
+	auto result = con.Query("SELECT 1 FROM " + string(openivm::DELTA_TABLES_TABLE) + " WHERE table_name = '" +
+	                        SqlUtils::EscapeValue(SqlUtils::DeltaName(view_name)) + "' LIMIT 1");
+	return !result->HasError() && result->RowCount() > 0;
+}
+
 vector<string> RefreshMetadata::GetGroupColumns(const string &view_name) {
 	auto result = con.Query("SELECT group_columns FROM " + string(openivm::VIEWS_TABLE) + " WHERE view_name = '" +
 	                        SqlUtils::EscapeValue(view_name) + "'");

src/include/core/refresh_locks.hpp

@@ -3,6 +3,7 @@
 
 #include "duckdb.hpp"
 
+#include <condition_variable>
 #include <mutex>
 #include <unordered_map>
 
@@ -13,9 +14,26 @@ namespace duckdb {
 // Per-view mutex: prevents two concurrent refreshes of the same MV (which would
 // cause write-write conflicts on the MV table).
 //
-// Per-delta-table mutex: serializes the refresh's "read deltas + set last_update"
-// critical section with the insert rule's delta row writes. This closes the window
-// where concurrent DML deltas could be permanently skipped.
+// Per-delta-table mutex: serializes refreshes that consume the same delta table.
+//
+// Per-catalog phase gate: allows concurrent DML writers and concurrent refreshes, but
+// prevents those two phases from overlapping. This closes the window where refresh
+// could advance a watermark past a delta row that commits afterward.
+class DeltaCatalogPhaseGate {
+public:
+	void EnterWrite();
+	void ExitWrite();
+	void EnterRefresh();
+	void ExitRefresh();
+
+private:
+	mutex lock;
+	std::condition_variable condition;
+	idx_t active_writers = 0;
+	idx_t active_refreshes = 0;
+	idx_t waiting_refreshes = 0;
+};
+
 class RefreshLocks {
 public:
 	// --- View-level locks (prevent concurrent refresh of same MV) ---
@@ -29,47 +47,61 @@ class RefreshLocks {
 
 	static void UnlockView(const string &view_name);
 
-	// --- Delta-table-level locks (serialize delta reads/writes) ---
+	// --- Delta-table-level locks (serialize overlapping refreshes) ---
 
-	// Blocking lock — held briefly by both refresh (read + timestamp update)
-	// and insert rule (delta row write).
+	// Blocking lock held by refresh while consuming and checkpointing one delta table.
 	static void LockDelta(const string &delta_table_name);
 
 	static void UnlockDelta(const string &delta_table_name);
 
-	// Serializes refresh with transactions that write native delta tables in the same catalog.
-	// The writer holds this lock until transaction commit/rollback so a refresh cannot advance
-	// its timestamp cursor past an uncommitted delta row.
-	static void LockDeltaCatalog(const string &catalog_name);
-
-	static void UnlockDeltaCatalog(const string &catalog_name);
+	static void EnterDeltaWrite(Catalog &catalog);
+	static void ExitDeltaWrite(Catalog &catalog);
+	static void EnterDeltaRefresh(Catalog &catalog);
+	static void ExitDeltaRefresh(Catalog &catalog);
 
 private:
 	static std::mutex &GetViewMutex(const string &view_name);
 	static std::mutex &GetDeltaMutex(const string &delta_table_name);
-	static std::mutex &GetDeltaCatalogMutex(const string &catalog_name);
+	static DeltaCatalogPhaseGate &GetDeltaCatalogGate(Catalog &catalog);
 
 	static std::mutex map_mutex_;
 	static std::unordered_map<string, unique_ptr<std::mutex>> view_mutexes_;
 	static std::unordered_map<string, unique_ptr<std::mutex>> delta_mutexes_;
-	static std::unordered_map<string, unique_ptr<std::mutex>> delta_catalog_mutexes_;
+	static std::unordered_map<const Catalog *, unique_ptr<DeltaCatalogPhaseGate>> delta_catalog_gates_;
 };
 
-// RAII guard for catalog-level delta transaction locks.
-class DeltaCatalogLockGuard {
-	string name_;
+// Write guards can be retained by transaction state and released from the commit thread;
+// unlike std::mutex ownership, phase-gate membership is not tied to one OS thread.
+class DeltaCatalogWriteGuard {
+	Catalog *catalog;
+
+public:
+	explicit DeltaCatalogWriteGuard(Catalog &catalog_p) : catalog(&catalog_p) {
+		RefreshLocks::EnterDeltaWrite(*catalog);
+	}
+	~DeltaCatalogWriteGuard() {
+		RefreshLocks::ExitDeltaWrite(*catalog);
+	}
+	DeltaCatalogWriteGuard(const DeltaCatalogWriteGuard &) = delete;
+	DeltaCatalogWriteGuard &operator=(const DeltaCatalogWriteGuard &) = delete;
+	DeltaCatalogWriteGuard(DeltaCatalogWriteGuard &&) = delete;
+	DeltaCatalogWriteGuard &operator=(DeltaCatalogWriteGuard &&) = delete;
+};
+
+class DeltaCatalogRefreshGuard {
+	Catalog *catalog;
 
 public:
-	explicit DeltaCatalogLockGuard(const string &catalog_name) : name_(catalog_name) {
-		RefreshLocks::LockDeltaCatalog(name_);
+	explicit DeltaCatalogRefreshGuard(Catalog &catalog_p) : catalog(&catalog_p) {
+		RefreshLocks::EnterDeltaRefresh(*catalog);
 	}
-	~DeltaCatalogLockGuard() {
-		RefreshLocks::UnlockDeltaCatalog(name_);
+	~DeltaCatalogRefreshGuard() {
+		RefreshLocks::ExitDeltaRefresh(*catalog);
 	}
-	DeltaCatalogLockGuard(const DeltaCatalogLockGuard &) = delete;
-	DeltaCatalogLockGuard &operator=(const DeltaCatalogLockGuard &) = delete;
-	DeltaCatalogLockGuard(DeltaCatalogLockGuard &&) = delete;
-	DeltaCatalogLockGuard &operator=(DeltaCatalogLockGuard &&) = delete;
+	DeltaCatalogRefreshGuard(const DeltaCatalogRefreshGuard &) = delete;
+	DeltaCatalogRefreshGuard &operator=(const DeltaCatalogRefreshGuard &) = delete;
+	DeltaCatalogRefreshGuard(DeltaCatalogRefreshGuard &&) = delete;
+	DeltaCatalogRefreshGuard &operator=(DeltaCatalogRefreshGuard &&) = delete;
 };
 
 // RAII guard for delta-table locks. Automatically unlocks on scope exit (including exceptions).

src/include/core/refresh_metadata.hpp

@@ -54,9 +54,17 @@ class RefreshMetadata {
 		string schema_name;
 		string table_name;
 	};
+	struct DeltaSource {
+		string table_name;
+		string catalog_type;
+		string catalog_name;
+		string schema_name;
+	};
 
 	SourceLocation GetSourceLocation(const string &view_name, const string &table_name,
 	                                 const string &fallback_catalog = "", const string &fallback_schema = "");
+	vector<DeltaSource> GetDeltaSources(const string &view_name, const string &fallback_catalog = "",
+	                                    const string &fallback_schema = "");
 	string ResolveDeltaQualifiedName(const string &view_name, const string &delta_table_name,
 	                                 const string &fallback_catalog = "", const string &fallback_schema = "");
 
@@ -79,6 +87,7 @@ class RefreshMetadata {
 	// Get all downstream MV dependents in topological order (closest first).
 	// For table→mv1→mv2→mv3, GetDownstreamViews("mv1") returns ["mv2", "mv3"].
 	vector<string> GetDownstreamViews(const string &view_name);
+	bool HasDownstreamViews(const string &view_name);
 
 	// Get refresh_interval in seconds for a view. Returns -1 if not set (manual only).
 	int64_t GetRefreshInterval(const string &view_name);