Update vendored DuckDB sources to 1265bd0

duckdblabs-bot · duckdblabs-bot · commit 4c92b7428387 · 2025-04-10T00:36:30.000Z
diff --git a/src/duckdb/extension/parquet/column_reader.cpp b/src/duckdb/extension/parquet/column_reader.cpp
@@ -318,12 +318,14 @@ void ColumnReader::PreparePageV2(PageHeader &page_hdr) {
 
 	auto compressed_bytes = page_hdr.compressed_page_size - uncompressed_bytes;
 
-	ResizeableBuffer compressed_buffer;
-	compressed_buffer.resize(GetAllocator(), compressed_bytes);
-	reader.ReadData(*protocol, compressed_buffer.ptr, compressed_bytes);
+	if (compressed_bytes > 0) {
+		ResizeableBuffer compressed_buffer;
+		compressed_buffer.resize(GetAllocator(), compressed_bytes);
+		reader.ReadData(*protocol, compressed_buffer.ptr, compressed_bytes);
 
-	DecompressInternal(chunk->meta_data.codec, compressed_buffer.ptr, compressed_bytes, block->ptr + uncompressed_bytes,
-	                   page_hdr.uncompressed_page_size - uncompressed_bytes);
+		DecompressInternal(chunk->meta_data.codec, compressed_buffer.ptr, compressed_bytes,
+		                   block->ptr + uncompressed_bytes, page_hdr.uncompressed_page_size - uncompressed_bytes);
+	}
 }
 
 void ColumnReader::AllocateBlock(idx_t size) {
diff --git a/src/duckdb/extension/parquet/include/writer/templated_column_writer.hpp b/src/duckdb/extension/parquet/include/writer/templated_column_writer.hpp
@@ -224,6 +224,7 @@ class StandardColumnWriter : public PrimitiveColumnWriter {
 
 		auto &state = state_p.Cast<StandardColumnWriterState<SRC, TGT, OP>>();
 		if (state.dictionary.GetSize() == 0 || state.dictionary.IsFull()) {
+			state.dictionary.Reset();
 			if (writer.GetParquetVersion() == ParquetVersion::V1) {
 				// Can't do the cool stuff for V1
 				state.encoding = duckdb_parquet::Encoding::PLAIN;
diff --git a/src/duckdb/extension/parquet/parquet_extension.cpp b/src/duckdb/extension/parquet/parquet_extension.cpp
@@ -781,6 +781,9 @@ struct ParquetWriteBindData : public TableFunctionData {
 
 struct ParquetWriteGlobalState : public GlobalFunctionData {
 	unique_ptr<ParquetWriter> writer;
+
+	mutex lock;
+	unique_ptr<ColumnDataCollection> combine_buffer;
 };
 
 struct ParquetWriteLocalState : public LocalFunctionData {
@@ -986,16 +989,45 @@ void ParquetWriteSink(ExecutionContext &context, FunctionData &bind_data_p, Glob
 	}
 }
 
-void ParquetWriteCombine(ExecutionContext &context, FunctionData &bind_data, GlobalFunctionData &gstate,
+void ParquetWriteCombine(ExecutionContext &context, FunctionData &bind_data_p, GlobalFunctionData &gstate,
                          LocalFunctionData &lstate) {
+	auto &bind_data = bind_data_p.Cast<ParquetWriteBindData>();
 	auto &global_state = gstate.Cast<ParquetWriteGlobalState>();
 	auto &local_state = lstate.Cast<ParquetWriteLocalState>();
-	// flush any data left in the local state to the file
-	global_state.writer->Flush(local_state.buffer);
+
+	if (local_state.buffer.Count() >= bind_data.row_group_size / 2 ||
+	    local_state.buffer.SizeInBytes() >= bind_data.row_group_size_bytes / 2) {
+		// local state buffer is more than half of the row_group_size(_bytes), just flush it
+		global_state.writer->Flush(local_state.buffer);
+		return;
+	}
+
+	unique_lock<mutex> guard(global_state.lock);
+	if (global_state.combine_buffer) {
+		// There is still some data, combine it
+		global_state.combine_buffer->Combine(local_state.buffer);
+		if (global_state.combine_buffer->Count() >= bind_data.row_group_size / 2 ||
+		    global_state.combine_buffer->SizeInBytes() >= bind_data.row_group_size_bytes / 2) {
+			// After combining, the combine buffer is more than half of the row_group_size(_bytes), so we flush
+			auto owned_combine_buffer = std::move(global_state.combine_buffer);
+			guard.unlock();
+			// Lock free, of course
+			global_state.writer->Flush(*owned_combine_buffer);
+		}
+		return;
+	}
+
+	global_state.combine_buffer = make_uniq<ColumnDataCollection>(context.client, local_state.buffer.Types());
+	global_state.combine_buffer->Combine(local_state.buffer);
 }
 
 void ParquetWriteFinalize(ClientContext &context, FunctionData &bind_data, GlobalFunctionData &gstate) {
 	auto &global_state = gstate.Cast<ParquetWriteGlobalState>();
+	// flush the combine buffer (if it's there)
+	if (global_state.combine_buffer) {
+		global_state.writer->Flush(*global_state.combine_buffer);
+	}
+
 	// finalize: write any additional metadata to the file here
 	global_state.writer->Finalize();
 }
diff --git a/src/duckdb/extension/parquet/writer/primitive_column_writer.cpp b/src/duckdb/extension/parquet/writer/primitive_column_writer.cpp
@@ -391,6 +391,12 @@ void PrimitiveColumnWriter::WriteDictionary(PrimitiveColumnWriterState &state, u
 	             write_info.compressed_buf);
 	hdr.compressed_page_size = UnsafeNumericCast<int32_t>(write_info.compressed_size);
 
+	if (write_info.compressed_buf) {
+		// if the data has been compressed, we no longer need the uncompressed data
+		D_ASSERT(write_info.compressed_buf.get() == write_info.compressed_data);
+		write_info.temp_writer.reset();
+	}
+
 	// insert the dictionary page as the first page to write for this column
 	state.write_info.insert(state.write_info.begin(), std::move(write_info));
 }
diff --git a/src/duckdb/src/catalog/catalog_entry/table_catalog_entry.cpp b/src/duckdb/src/catalog/catalog_entry/table_catalog_entry.cpp
@@ -343,4 +343,10 @@ virtual_column_map_t TableCatalogEntry::GetVirtualColumns() const {
 	return virtual_columns;
 }
 
+vector<column_t> TableCatalogEntry::GetRowIdColumns() const {
+	vector<column_t> result;
+	result.push_back(COLUMN_IDENTIFIER_ROW_ID);
+	return result;
+}
+
 } // namespace duckdb
diff --git a/src/duckdb/src/common/exception.cpp b/src/duckdb/src/common/exception.cpp
@@ -334,8 +334,7 @@ FatalException::FatalException(ExceptionType type, const string &msg) : Exceptio
 
 InternalException::InternalException(const string &msg) : Exception(ExceptionType::INTERNAL, msg) {
 #ifdef DUCKDB_CRASH_ON_ASSERT
-	Printer::Print("ABORT THROWN BY INTERNAL EXCEPTION: " + msg);
-	Printer::Print(StackTrace::GetStackTrace());
+	Printer::Print("ABORT THROWN BY INTERNAL EXCEPTION: " + msg + "\n" + StackTrace::GetStackTrace());
 	abort();
 #endif
 }
diff --git a/src/duckdb/src/common/file_system.cpp b/src/duckdb/src/common/file_system.cpp
@@ -399,6 +399,11 @@ time_t FileSystem::GetLastModifiedTime(FileHandle &handle) {
 	throw NotImplementedException("%s: GetLastModifiedTime is not implemented!", GetName());
 }
 
+string FileSystem::GetVersionTag(FileHandle &handle) {
+	// Used to check cache invalidation for httpfs files with an ETag in CachingFileSystem
+	return "";
+}
+
 FileType FileSystem::GetFileType(FileHandle &handle) {
 	return FileType::FILE_TYPE_INVALID;
 }
diff --git a/src/duckdb/src/common/multi_file/multi_file_reader.cpp b/src/duckdb/src/common/multi_file/multi_file_reader.cpp
@@ -275,7 +275,8 @@ void MultiFileReader::FinalizeBind(MultiFileReaderData &reader_data, const Multi
 		auto global_idx = MultiFileGlobalIndex(i);
 		auto &col_id = global_column_ids[i];
 		auto column_id = col_id.GetPrimaryIndex();
-		if (options.filename_idx.IsValid() && column_id == options.filename_idx.GetIndex()) {
+		if ((options.filename_idx.IsValid() && column_id == options.filename_idx.GetIndex()) ||
+		    column_id == MultiFileReader::COLUMN_IDENTIFIER_FILENAME) {
 			// filename
 			reader_data.constant_map.Add(global_idx, Value(filename));
 			continue;
diff --git a/src/duckdb/src/common/printer.cpp b/src/duckdb/src/common/printer.cpp
@@ -30,11 +30,17 @@ void Printer::RawPrint(OutputStream stream, const string &str) {
 #endif
 }
 
-// LCOV_EXCL_START
-void Printer::Print(OutputStream stream, const string &str) {
+void Printer::DefaultLinePrint(OutputStream stream, const string &str) {
 	Printer::RawPrint(stream, str);
 	Printer::RawPrint(stream, "\n");
 }
+
+line_printer_f Printer::line_printer = Printer::DefaultLinePrint;
+
+// LCOV_EXCL_START
+void Printer::Print(OutputStream stream, const string &str) {
+	Printer::line_printer(stream, str);
+}
 void Printer::Flush(OutputStream stream) {
 #ifndef DUCKDB_DISABLE_PRINT
 	fflush(stream == OutputStream::STREAM_STDERR ? stderr : stdout);
diff --git a/src/duckdb/src/execution/operator/persistent/physical_copy_to_file.cpp b/src/duckdb/src/execution/operator/persistent/physical_copy_to_file.cpp
@@ -580,9 +580,8 @@ SinkCombineResultType PhysicalCopyToFile::Combine(ExecutionContext &context, Ope
 	return SinkCombineResultType::FINISHED;
 }
 
-SinkFinalizeType PhysicalCopyToFile::Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
-                                              OperatorSinkFinalizeInput &input) const {
-	auto &gstate = input.global_state.Cast<CopyToFunctionGlobalState>();
+SinkFinalizeType PhysicalCopyToFile::FinalizeInternal(ClientContext &context, GlobalSinkState &global_state) const {
+	auto &gstate = global_state.Cast<CopyToFunctionGlobalState>();
 	if (partition_output) {
 		// finalize any outstanding partitions
 		gstate.FinalizePartitions(context, *this);
@@ -612,6 +611,11 @@ SinkFinalizeType PhysicalCopyToFile::Finalize(Pipeline &pipeline, Event &event,
 	return SinkFinalizeType::READY;
 }
 
+SinkFinalizeType PhysicalCopyToFile::Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
+                                              OperatorSinkFinalizeInput &input) const {
+	return FinalizeInternal(context, input.global_state);
+}
+
 //===--------------------------------------------------------------------===//
 // Source
 //===--------------------------------------------------------------------===//
diff --git a/src/duckdb/src/execution/operator/persistent/physical_insert.cpp b/src/duckdb/src/execution/operator/persistent/physical_insert.cpp
@@ -276,7 +276,7 @@ static idx_t PerformOnConflictAction(InsertLocalState &lstate, InsertGlobalState
 		}
 		auto &local_storage = LocalStorage::Get(context.client, data_table.db);
 		if (gstate.initialized) {
-			// Flush the data first, it might be referenced by the Update
+			// Flush any local appends that could be referenced by the UPDATE.
 			data_table.FinalizeLocalAppend(gstate.append_state);
 			gstate.initialized = false;
 		}
@@ -289,6 +289,11 @@ 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());
 	}
 
diff --git a/src/duckdb/src/execution/physical_plan/plan_delete.cpp b/src/duckdb/src/execution/physical_plan/plan_delete.cpp
@@ -24,8 +24,6 @@ PhysicalOperator &Catalog::PlanDelete(ClientContext &context, PhysicalPlanGenera
 
 PhysicalOperator &PhysicalPlanGenerator::CreatePlan(LogicalDelete &op) {
 	D_ASSERT(op.children.size() == 1);
-	D_ASSERT(op.expressions.size() == 1);
-	D_ASSERT(op.expressions[0]->GetExpressionType() == ExpressionType::BOUND_REF);
 
 	dependencies.AddDependency(op.table);
 	return op.table.catalog.PlanDelete(context, *this, op);
diff --git a/src/duckdb/src/function/table/version/pragma_version.cpp b/src/duckdb/src/function/table/version/pragma_version.cpp
@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "0-dev2262"
+#define DUCKDB_PATCH_VERSION "0-dev2339"
 #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-dev2262"
+#define DUCKDB_VERSION "v1.3.0-dev2339"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "589e10ae54"
+#define DUCKDB_SOURCE_ID "e1dcf13ed5"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"
diff --git a/src/duckdb/src/include/duckdb/catalog/catalog_entry/table_catalog_entry.hpp b/src/duckdb/src/include/duckdb/catalog/catalog_entry/table_catalog_entry.hpp
@@ -127,6 +127,8 @@ class TableCatalogEntry : public StandardEntry {
 	//! Returns the virtual columns for this table
 	virtual virtual_column_map_t GetVirtualColumns() const;
 
+	virtual vector<column_t> GetRowIdColumns() const;
+
 protected:
 	//! A list of columns that are part of this table
 	ColumnList columns;
diff --git a/src/duckdb/src/include/duckdb/common/file_system.hpp b/src/duckdb/src/include/duckdb/common/file_system.hpp
@@ -140,6 +140,9 @@ class FileSystem {
 	DUCKDB_API virtual int64_t GetFileSize(FileHandle &handle);
 	//! Returns the file last modified time of a file handle, returns timespec with zero on all attributes on error
 	DUCKDB_API virtual time_t GetLastModifiedTime(FileHandle &handle);
+	//! Returns a tag that uniquely identifies the version of the file,
+	//! used for checking cache invalidation for CachingFileSystem httpfs files
+	DUCKDB_API virtual string GetVersionTag(FileHandle &handle);
 	//! Returns the file type of the attached handle
 	DUCKDB_API virtual FileType GetFileType(FileHandle &handle);
 	//! Truncate a file to a maximum size of new_size, new_size should be smaller than or equal to the current size of
diff --git a/src/duckdb/src/include/duckdb/common/primitive_dictionary.hpp b/src/duckdb/src/include/duckdb/common/primitive_dictionary.hpp
@@ -120,6 +120,11 @@ class PrimitiveDictionary {
 		return result;
 	}
 
+	void Reset() {
+		allocated_dictionary.Reset();
+		allocated_target.Reset();
+	}
+
 private:
 	//! Look up a value in the dictionary using linear probing
 	primitive_dictionary_entry_t &Lookup(const SRC &value) const {
diff --git a/src/duckdb/src/include/duckdb/common/printer.hpp b/src/duckdb/src/include/duckdb/common/printer.hpp
@@ -15,6 +15,8 @@ namespace duckdb {
 
 enum class OutputStream : uint8_t { STREAM_STDOUT = 1, STREAM_STDERR = 2 };
 
+typedef void (*line_printer_f)(OutputStream stream, const string &str);
+
 //! Printer is a static class that allows printing to logs or stdout/stderr
 class Printer {
 public:
@@ -40,5 +42,11 @@ class Printer {
 	DUCKDB_API static bool IsTerminal(OutputStream stream);
 	//! The terminal width
 	DUCKDB_API static idx_t TerminalWidth();
+
+	// hook to allow capturing the output and routing it somewhere else / reformat it};
+	static line_printer_f line_printer;
+
+private:
+	static void DefaultLinePrint(OutputStream stream, const string &str);
 };
 } // namespace duckdb
diff --git a/src/duckdb/src/include/duckdb/execution/operator/persistent/physical_copy_to_file.hpp b/src/duckdb/src/include/duckdb/execution/operator/persistent/physical_copy_to_file.hpp
@@ -71,6 +71,7 @@ class PhysicalCopyToFile : public PhysicalOperator {
 	SinkCombineResultType Combine(ExecutionContext &context, OperatorSinkCombineInput &input) const override;
 	SinkFinalizeType Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
 	                          OperatorSinkFinalizeInput &input) const override;
+	SinkFinalizeType FinalizeInternal(ClientContext &context, GlobalSinkState &gstate) const;
 	unique_ptr<LocalSinkState> GetLocalSinkState(ExecutionContext &context) const override;
 	unique_ptr<GlobalSinkState> GetGlobalSinkState(ClientContext &context) const override;
 
diff --git a/src/duckdb/src/include/duckdb/main/client_context_state.hpp b/src/duckdb/src/include/duckdb/main/client_context_state.hpp
@@ -13,6 +13,7 @@
 #include "duckdb/common/optional_ptr.hpp"
 #include "duckdb/main/config.hpp"
 #include "duckdb/main/valid_checker.hpp"
+#include "duckdb/planner/expression/bound_parameter_data.hpp"
 #include "duckdb/transaction/meta_transaction.hpp"
 #include "duckdb/transaction/transaction_manager.hpp"
 #include "duckdb/main/database_manager.hpp"
@@ -39,6 +40,11 @@ struct PreparedStatementCallbackInfo {
 	const PendingQueryParameters &parameters;
 };
 
+struct BindPreparedStatementCallbackInfo {
+	PreparedStatementData &prepared_statement;
+	optional_ptr<case_insensitive_map_t<BoundParameterData>> parameters;
+};
+
 //! ClientContextState is virtual base class for ClientContext-local (or Query-Local, using QueryEnd callback) state
 //! e.g. caches that need to live as long as a ClientContext or Query.
 class ClientContextState {
@@ -78,6 +84,10 @@ class ClientContextState {
 	                                          RebindQueryInfo current_rebind) {
 		return RebindQueryInfo::DO_NOT_REBIND;
 	}
+	virtual RebindQueryInfo OnRebindPreparedStatement(ClientContext &context, BindPreparedStatementCallbackInfo &info,
+	                                                  RebindQueryInfo current_rebind) {
+		return RebindQueryInfo::DO_NOT_REBIND;
+	}
 	virtual void WriteProfilingInformation(std::ostream &ss) {
 	}
 	virtual void OnTaskStart(ClientContext &context) {
diff --git a/src/duckdb/src/include/duckdb/main/extension_entries.hpp b/src/duckdb/src/include/duckdb/main/extension_entries.hpp
diff --git a/src/duckdb/src/include/duckdb/parser/constraints/unique_constraint.hpp b/src/duckdb/src/include/duckdb/parser/constraints/unique_constraint.hpp
diff --git a/src/duckdb/src/include/duckdb/planner/binder.hpp b/src/duckdb/src/include/duckdb/planner/binder.hpp
diff --git a/src/duckdb/src/parser/constraints/unique_constraint.cpp b/src/duckdb/src/parser/constraints/unique_constraint.cpp
diff --git a/src/duckdb/src/parser/transform/constraint/transform_constraint.cpp b/src/duckdb/src/parser/transform/constraint/transform_constraint.cpp
diff --git a/src/duckdb/src/planner/binder/statement/bind_create_table.cpp b/src/duckdb/src/planner/binder/statement/bind_create_table.cpp
diff --git a/src/duckdb/src/planner/binder/statement/bind_delete.cpp b/src/duckdb/src/planner/binder/statement/bind_delete.cpp
diff --git a/src/duckdb/src/planner/binder/statement/bind_execute.cpp b/src/duckdb/src/planner/binder/statement/bind_execute.cpp
diff --git a/src/duckdb/src/planner/binder/statement/bind_update.cpp b/src/duckdb/src/planner/binder/statement/bind_update.cpp

Original file line number	Diff line number	Diff line change
`@@ -334,8 +334,7 @@ FatalException::FatalException(ExceptionType type, const string &msg) : Exceptio`
`334`	`334`
`335`	`335`	`InternalException::InternalException(const string &msg) : Exception(ExceptionType::INTERNAL, msg) {`
`336`	`336`	`#ifdef DUCKDB_CRASH_ON_ASSERT`
`337`		`- Printer::Print("ABORT THROWN BY INTERNAL EXCEPTION: " + msg);`
`338`		`- Printer::Print(StackTrace::GetStackTrace());`
	`337`	`+ Printer::Print("ABORT THROWN BY INTERNAL EXCEPTION: " + msg + "\n" + StackTrace::GetStackTrace());`
`339`	`338`	`abort();`
`340`	`339`	`#endif`
`341`	`340`	`}`
Original file line number	Diff line number	Diff line change
`@@ -399,6 +399,11 @@ time_t FileSystem::GetLastModifiedTime(FileHandle &handle) {`
`399`	`399`	`throw NotImplementedException("%s: GetLastModifiedTime is not implemented!", GetName());`
`400`	`400`	`}`
`401`	`401`
	`402`	`+string FileSystem::GetVersionTag(FileHandle &handle) {`
	`403`	`+ // Used to check cache invalidation for httpfs files with an ETag in CachingFileSystem`
	`404`	`+ return "";`
	`405`	`+}`
	`406`	`+`
`402`	`407`	`FileType FileSystem::GetFileType(FileHandle &handle) {`
`403`	`408`	`return FileType::FILE_TYPE_INVALID;`
`404`	`409`	`}`