vendor: Update vendored sources to duckdb/duckdb@5141aa4

Combine small row groups in Parquet writer (duckdb/duckdb#17036)

Author: krlmlr

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;

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();
 }

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));
 }

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

@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "0-dev2266"
+#define DUCKDB_PATCH_VERSION "0-dev2271"
 #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-dev2266"
+#define DUCKDB_VERSION "v1.3.0-dev2271"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "6c176c0a3b"
+#define DUCKDB_SOURCE_ID "5141aa4560"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"

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 {