vendor: Update vendored sources to duckdb/duckdb@892a03d

Fix an issue in upserts where the local append state was not correctly flushed (duckdb/duckdb#17109)

Author: krlmlr

src/duckdb/src/execution/operator/persistent/physical_insert.cpp

@@ -75,7 +75,7 @@ void PhysicalInsert::GetInsertInfo(const BoundCreateTableInfo &info, vector<Logi
 
 InsertGlobalState::InsertGlobalState(ClientContext &context, const vector<LogicalType> &return_types,
                                      DuckTableEntry &table)
-    : table(table), insert_count(0), initialized(false), return_collection(context, return_types) {
+    : table(table), insert_count(0), return_collection(context, return_types) {
 }
 
 InsertLocalState::InsertLocalState(ClientContext &context, const vector<LogicalType> &types,
@@ -275,11 +275,6 @@ static idx_t PerformOnConflictAction(InsertLocalState &lstate, InsertGlobalState
 			return update_chunk.size();
 		}
 		auto &local_storage = LocalStorage::Get(context.client, data_table.db);
-		if (gstate.initialized) {
-			// Flush any local appends that could be referenced by the UPDATE.
-			data_table.FinalizeLocalAppend(gstate.append_state);
-			gstate.initialized = false;
-		}
 		local_storage.Update(data_table, row_ids, set_columns, update_chunk);
 		return update_chunk.size();
 	}
@@ -289,11 +284,6 @@ static idx_t PerformOnConflictAction(InsertLocalState &lstate, InsertGlobalState
 		data_table.Delete(delete_state, context.client, row_ids, update_chunk.size());
 	} else {
 		auto &local_storage = LocalStorage::Get(context.client, data_table.db);
-		if (gstate.initialized) {
-			// Flush any local appends that could be referenced by the DELETE.
-			data_table.FinalizeLocalAppend(gstate.append_state);
-			gstate.initialized = false;
-		}
 		local_storage.Delete(data_table, row_ids, update_chunk.size());
 	}
 
@@ -427,7 +417,7 @@ static void VerifyOnConflictCondition(ExecutionContext &context, DataChunk &comb
 		throw InternalException("VerifyAppendConstraints was expected to throw but didn't");
 	}
 
-	auto &indexes = local_storage.GetIndexes(data_table);
+	auto &indexes = local_storage.GetIndexes(context.client, data_table);
 	auto storage = local_storage.GetStorage(data_table);
 	DataTable::VerifyUniqueIndexes(indexes, storage, tuples, nullptr);
 	throw InternalException("VerifyUniqueIndexes was expected to throw but didn't");
@@ -451,7 +441,7 @@ static idx_t HandleInsertConflicts(TableCatalogEntry &table, ExecutionContext &c
 		auto &constraint_state = lstate.GetConstraintState(data_table, table);
 		data_table.VerifyAppendConstraints(constraint_state, context.client, tuples, storage, &conflict_manager);
 	} else {
-		auto &indexes = local_storage.GetIndexes(data_table);
+		auto &indexes = local_storage.GetIndexes(context.client, data_table);
 		DataTable::VerifyUniqueIndexes(indexes, storage, tuples, &conflict_manager);
 	}
 
@@ -529,7 +519,7 @@ idx_t PhysicalInsert::OnConflictHandling(TableCatalogEntry &table, ExecutionCont
 	ConflictInfo conflict_info(conflict_target);
 
 	auto &global_indexes = data_table.GetDataTableInfo()->GetIndexes();
-	auto &local_indexes = local_storage.GetIndexes(data_table);
+	auto &local_indexes = local_storage.GetIndexes(context.client, data_table);
 
 	unordered_set<BoundIndex *> matched_indexes;
 	if (conflict_info.column_ids.empty()) {
@@ -623,19 +613,14 @@ SinkResultType PhysicalInsert::Sink(ExecutionContext &context, DataChunk &insert
 	insert_chunk.Flatten();
 
 	if (!parallel) {
-		if (!gstate.initialized) {
-			storage.InitializeLocalAppend(gstate.append_state, table, context.client, bound_constraints);
-			gstate.initialized = true;
-		}
-
 		idx_t updated_tuples = OnConflictHandling(table, context, gstate, lstate, insert_chunk);
 
 		gstate.insert_count += insert_chunk.size();
 		gstate.insert_count += updated_tuples;
 		if (return_chunk) {
 			gstate.return_collection.Append(insert_chunk);
 		}
-		storage.LocalAppend(gstate.append_state, context.client, insert_chunk, true);
+		storage.LocalAppend(table, context.client, insert_chunk, bound_constraints);
 		if (action_type == OnConflictAction::UPDATE && lstate.update_chunk.size() != 0) {
 			(void)HandleInsertConflicts<true>(table, context, lstate, gstate, lstate.update_chunk, *this);
 			(void)HandleInsertConflicts<false>(table, context, lstate, gstate, lstate.update_chunk, *this);
@@ -701,13 +686,14 @@ SinkCombineResultType PhysicalInsert::Combine(ExecutionContext &context, Operato
 	gstate.insert_count += append_count;
 	if (append_count < row_group_size) {
 		// we have few rows - append to the local storage directly
-		storage.InitializeLocalAppend(gstate.append_state, table, context.client, bound_constraints);
+		LocalAppendState append_state;
+		storage.InitializeLocalAppend(append_state, table, context.client, bound_constraints);
 		auto &transaction = DuckTransaction::Get(context.client, table.catalog);
 		collection.Scan(transaction, [&](DataChunk &insert_chunk) {
-			storage.LocalAppend(gstate.append_state, context.client, insert_chunk, false);
+			storage.LocalAppend(append_state, context.client, insert_chunk, false);
 			return true;
 		});
-		storage.FinalizeLocalAppend(gstate.append_state);
+		storage.FinalizeLocalAppend(append_state);
 	} else {
 		// we have written rows to disk optimistically - merge directly into the transaction-local storage
 		lstate.optimistic_writer->WriteLastRowGroup(collection);
@@ -722,12 +708,6 @@ SinkCombineResultType PhysicalInsert::Combine(ExecutionContext &context, Operato
 
 SinkFinalizeType PhysicalInsert::Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
                                           OperatorSinkFinalizeInput &input) const {
-	auto &gstate = input.global_state.Cast<InsertGlobalState>();
-	if (!parallel && gstate.initialized) {
-		auto &table = gstate.table;
-		auto &storage = table.GetStorage();
-		storage.FinalizeLocalAppend(gstate.append_state);
-	}
 	return SinkFinalizeType::READY;
 }
 

src/duckdb/src/function/table/version/pragma_version.cpp

@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "0-dev2456"
+#define DUCKDB_PATCH_VERSION "0-dev2459"
 #endif
 #ifndef DUCKDB_MINOR_VERSION
 #define DUCKDB_MINOR_VERSION 3
@@ -8,10 +8,10 @@
 #define DUCKDB_MAJOR_VERSION 1
 #endif
 #ifndef DUCKDB_VERSION
-#define DUCKDB_VERSION "v1.3.0-dev2456"
+#define DUCKDB_VERSION "v1.3.0-dev2459"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "cae7a680bc"
+#define DUCKDB_SOURCE_ID "892a03d26e"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"

src/duckdb/src/include/duckdb/execution/operator/persistent/physical_insert.hpp

@@ -31,8 +31,6 @@ class InsertGlobalState : public GlobalSinkState {
 	mutex lock;
 	DuckTableEntry &table;
 	idx_t insert_count;
-	bool initialized;
-	LocalAppendState append_state;
 	ColumnDataCollection return_collection;
 };
 

src/duckdb/src/include/duckdb/storage/data_table.hpp

@@ -111,6 +111,9 @@ class DataTable : public enable_shared_from_this<DataTable> {
 	void LocalAppend(TableCatalogEntry &table, ClientContext &context, DataChunk &chunk,
 	                 const vector<unique_ptr<BoundConstraint>> &bound_constraints, Vector &row_ids,
 	                 DataChunk &delete_chunk);
+	//! Appends to the transaction-local storage of this table
+	void LocalAppend(TableCatalogEntry &table, ClientContext &context, DataChunk &chunk,
+	                 const vector<unique_ptr<BoundConstraint>> &bound_constraints);
 	//! Append a chunk to the transaction-local storage of this table.
 	void LocalWALAppend(TableCatalogEntry &table, ClientContext &context, DataChunk &chunk,
 	                    const vector<unique_ptr<BoundConstraint>> &bound_constraints);

src/duckdb/src/include/duckdb/transaction/local_storage.hpp

@@ -180,7 +180,7 @@ class LocalStorage {
 	void MoveStorage(DataTable &old_dt, DataTable &new_dt);
 	void FetchChunk(DataTable &table, Vector &row_ids, idx_t count, const vector<StorageIndex> &col_ids,
 	                DataChunk &chunk, ColumnFetchState &fetch_state);
-	TableIndexList &GetIndexes(DataTable &table);
+	TableIndexList &GetIndexes(ClientContext &context, DataTable &table);
 	optional_ptr<LocalTableStorage> GetStorage(DataTable &table);
 
 	void VerifyNewConstraint(DataTable &parent, const BoundConstraint &constraint);

src/duckdb/src/storage/data_table.cpp

@@ -580,7 +580,7 @@ void DataTable::VerifyForeignKeyConstraint(optional_ptr<LocalTableStorage> stora
 
 	// Local constraint verification.
 	if (local_verification) {
-		auto &local_indexes = local_storage.GetIndexes(data_table);
+		auto &local_indexes = local_storage.GetIndexes(context, data_table);
 		local_indexes.VerifyForeignKey(storage, dst_keys_ptr, dst_chunk, local_conflicts);
 		local_conflicts.Finalize();
 		auto &local_matches = local_conflicts.Conflicts();
@@ -601,7 +601,7 @@ void DataTable::VerifyForeignKeyConstraint(optional_ptr<LocalTableStorage> stora
 	// check whether or not the chunk can be inserted or deleted into the referenced table' storage
 	index = data_table.info->indexes.FindForeignKeyIndex(dst_keys_ptr, fk_type);
 	if (local_verification) {
-		auto &transact_index = local_storage.GetIndexes(data_table);
+		auto &transact_index = local_storage.GetIndexes(context, data_table);
 		// check whether or not the chunk can be inserted or deleted into the referenced table' storage
 		transaction_index = transact_index.FindForeignKeyIndex(dst_keys_ptr, fk_type);
 	}
@@ -878,6 +878,14 @@ void DataTable::LocalAppend(LocalAppendState &state, ClientContext &context, Dat
 	LocalStorage::Append(state, chunk);
 }
 
+void DataTable::LocalAppend(TableCatalogEntry &table, ClientContext &context, DataChunk &chunk,
+                            const vector<unique_ptr<BoundConstraint>> &bound_constraints) {
+	LocalAppendState append_state;
+	InitializeLocalAppend(append_state, table, context, bound_constraints);
+	LocalAppend(append_state, context, chunk, false);
+	FinalizeLocalAppend(append_state);
+}
+
 void DataTable::FinalizeLocalAppend(LocalAppendState &state) {
 	LocalStorage::FinalizeAppend(state);
 }

src/duckdb/src/storage/local_storage.cpp

@@ -657,12 +657,9 @@ void LocalStorage::FetchChunk(DataTable &table, Vector &row_ids, idx_t count, co
 	storage->row_groups->Fetch(transaction, chunk, col_ids, row_ids, count, fetch_state);
 }
 
-TableIndexList &LocalStorage::GetIndexes(DataTable &table) {
-	auto storage = table_manager.GetStorage(table);
-	if (!storage) {
-		throw InternalException("LocalStorage::GetIndexes - local storage not found");
-	}
-	return storage->append_indexes;
+TableIndexList &LocalStorage::GetIndexes(ClientContext &context, DataTable &table) {
+	auto &storage = table_manager.GetOrCreateStorage(context, table);
+	return storage.append_indexes;
 }
 
 optional_ptr<LocalTableStorage> LocalStorage::GetStorage(DataTable &table) {