duckdb · Mytherin · Mar 20, 2025 · Mar 19, 2025 · Mar 19, 2025
diff --git a/extension/core_functions/function_list.cpp b/extension/core_functions/function_list.cpp
@@ -385,8 +385,8 @@ static const StaticFunctionDefinition core_functions[] = {
 	DUCKDB_SCALAR_FUNCTION(UrlDecodeFun),
 	DUCKDB_SCALAR_FUNCTION(UrlEncodeFun),
 	DUCKDB_SCALAR_FUNCTION(UUIDFun),
-	DUCKDB_SCALAR_FUNCTION_SET(ExtractUuidTimestampFun),
-	DUCKDB_SCALAR_FUNCTION_SET(ExtractUuidVerisonFun),
+	DUCKDB_SCALAR_FUNCTION(UUIDExtractTimestampFun),
+	DUCKDB_SCALAR_FUNCTION(UUIDExtractVersionFun),
 	DUCKDB_SCALAR_FUNCTION(UUIDv4Fun),
 	DUCKDB_SCALAR_FUNCTION(UUIDv7Fun),
 	DUCKDB_AGGREGATE_FUNCTION(VarPopFun),

diff --git a/extension/core_functions/include/core_functions/scalar/random_functions.hpp b/extension/core_functions/include/core_functions/scalar/random_functions.hpp
@@ -66,22 +66,22 @@ struct UUIDv7Fun {
 	static ScalarFunction GetFunction();
 };
 
-struct ExtractUuidVerisonFun {
+struct UUIDExtractVersionFun {
 	static constexpr const char *Name = "uuid_extract_version";
 	static constexpr const char *Parameters = "uuid";
 	static constexpr const char *Description = "Extract a version for the given UUID.";
 	static constexpr const char *Example = "uuid_extract_version('019482e4-1441-7aad-8127-eec99573b0a0')";
 
-	static ScalarFunctionSet GetFunctions();
+	static ScalarFunction GetFunction();
 };
 
-struct ExtractUuidTimestampFun {
+struct UUIDExtractTimestampFun {
 	static constexpr const char *Name = "uuid_extract_timestamp";
 	static constexpr const char *Parameters = "uuid";
 	static constexpr const char *Description = "Extract the timestamp for the given UUID v7.";
 	static constexpr const char *Example = "uuid_extract_timestamp('019482e4-1441-7aad-8127-eec99573b0a0')";
 
-	static ScalarFunctionSet GetFunctions();
+	static ScalarFunction GetFunction();
 };
 
 } // namespace duckdb
diff --git a/extension/core_functions/scalar/random/functions.json b/extension/core_functions/scalar/random/functions.json
@@ -38,15 +38,15 @@
         "parameters": "uuid",
         "description": "Extract a version for the given UUID.",
         "example": "uuid_extract_version('019482e4-1441-7aad-8127-eec99573b0a0')",
-        "type": "scalar_function_set",
-        "struct": "ExtractUuidVerisonFun"
+        "type": "scalar_function",
+        "struct": "UUIDExtractVersionFun"
     },
     {
         "name": "uuid_extract_timestamp",
         "parameters": "uuid",
         "description": "Extract the timestamp for the given UUID v7.",
         "example": "uuid_extract_timestamp('019482e4-1441-7aad-8127-eec99573b0a0')",
-        "type": "scalar_function_set",
-        "struct": "ExtractUuidTimestampFun"
+        "type": "scalar_function",
+        "struct": "UUIDExtractTimestampFun"
     }
 ]
diff --git a/extension/core_functions/scalar/random/random.cpp b/extension/core_functions/scalar/random/random.cpp
@@ -9,18 +9,6 @@
 
 namespace duckdb {
 
-struct ExtractVersionStrOperator {
-	template <typename INPUT_TYPE, typename RESULT_TYPE>
-	static RESULT_TYPE Operation(INPUT_TYPE input, Vector &result) {
-		const idx_t len = input.GetSize();
-		if (len != 36) {
-			throw InvalidInputException("Given string '%s' is invalid UUID.", input.GetString());
-		}
-		// UUIDv4 and UUIDv7 stores version as the 15-th uint8_t.
-		return input.GetPointer()[14] - '0';
-	}
-};
-
 struct ExtractVersionUuidOperator {
 	template <typename INPUT_TYPE, typename RESULT_TYPE>
 	static RESULT_TYPE Operation(INPUT_TYPE input, Vector &result) {
@@ -51,19 +39,6 @@ struct ExtractTimestampUuidOperator {
 	}
 };
 
-struct ExtractTimestampStrOperator {
-	template <typename INPUT_TYPE, typename RESULT_TYPE>
-	static RESULT_TYPE Operation(INPUT_TYPE input, Vector &result) {
-		// Validate whether the give input is a valid UUID.
-		hugeint_t uuid_hugeint;
-		if (!BaseUUID::FromCString(input.GetData(), input.GetSize(), uuid_hugeint)) {
-			throw InvalidInputException("Given string '%s' is invalid UUID.", input.GetString());
-		}
-
-		return ExtractTimestampUuidOperator::Operation<hugeint_t, RESULT_TYPE>(uuid_hugeint, result);
-	}
-};
-
 template <typename INPUT, typename OP>
 static void ExtractVersionFunction(DataChunk &args, ExpressionState &state, Vector &result) {
 	D_ASSERT(args.ColumnCount() == 1);
@@ -157,22 +132,14 @@ ScalarFunction UUIDv7Fun::GetFunction() {
 	return uuid_v7_function;
 }
 
-ScalarFunctionSet ExtractUuidVerisonFun::GetFunctions() {
-	ScalarFunctionSet version_extraction;
-	version_extraction.AddFunction(ScalarFunction({LogicalType::VARCHAR}, LogicalType::UINTEGER,
-	                                              ExtractVersionFunction<string_t, ExtractVersionStrOperator>));
-	version_extraction.AddFunction(ScalarFunction({LogicalType::UUID}, LogicalType::UINTEGER,
-	                                              ExtractVersionFunction<hugeint_t, ExtractVersionUuidOperator>));
-	return version_extraction;
+ScalarFunction UUIDExtractVersionFun::GetFunction() {
+	return ScalarFunction({LogicalType::UUID}, LogicalType::UINTEGER,
+	                      ExtractVersionFunction<hugeint_t, ExtractVersionUuidOperator>);
 }
 
-ScalarFunctionSet ExtractUuidTimestampFun::GetFunctions() {
-	ScalarFunctionSet timestamp_extraction;
-	timestamp_extraction.AddFunction(ScalarFunction({LogicalType::VARCHAR}, LogicalType::TIMESTAMP_TZ,
-	                                                ExtractTimestampFunction<string_t, ExtractTimestampStrOperator>));
-	timestamp_extraction.AddFunction(ScalarFunction({LogicalType::UUID}, LogicalType::TIMESTAMP_TZ,
-	                                                ExtractTimestampFunction<hugeint_t, ExtractTimestampUuidOperator>));
-	return timestamp_extraction;
+ScalarFunction UUIDExtractTimestampFun::GetFunction() {
+	return ScalarFunction({LogicalType::UUID}, LogicalType::TIMESTAMP_TZ,
+	                      ExtractTimestampFunction<hugeint_t, ExtractTimestampUuidOperator>);
 }
 
 } // namespace duckdb
diff --git a/src/execution/operator/persistent/physical_batch_copy_to_file.cpp b/src/execution/operator/persistent/physical_batch_copy_to_file.cpp
@@ -67,17 +67,18 @@ class FixedBatchCopyGlobalState : public GlobalSinkState {
 	static constexpr const idx_t MINIMUM_MEMORY_PER_COLUMN_PER_THREAD = 4ULL * 1024ULL * 1024ULL;
 
 public:
-	explicit FixedBatchCopyGlobalState(ClientContext &context_p, unique_ptr<GlobalFunctionData> global_state,
-	                                   idx_t minimum_memory_per_thread)
-	    : memory_manager(context_p, minimum_memory_per_thread), rows_copied(0), global_state(std::move(global_state)),
-	      batch_size(0), scheduled_batch_index(0), flushed_batch_index(0), any_flushing(false), any_finished(false),
+	explicit FixedBatchCopyGlobalState(ClientContext &context_p, idx_t minimum_memory_per_thread)
+	    : memory_manager(context_p, minimum_memory_per_thread), initialized(false), rows_copied(0), batch_size(0),
+	      scheduled_batch_index(0), flushed_batch_index(0), any_flushing(false), any_finished(false),
 	      minimum_memory_per_thread(minimum_memory_per_thread) {
 	}
 
 	BatchMemoryManager memory_manager;
 	BatchTaskManager<BatchCopyTask> task_manager;
 	mutex lock;
 	mutex flush_lock;
+	//! Whether or not the copy has been initialized
+	atomic<bool> initialized;
 	//! The total number of rows copied to the file
 	atomic<idx_t> rows_copied;
 	//! Global copy state
@@ -101,6 +102,24 @@ class FixedBatchCopyGlobalState : public GlobalSinkState {
 	//! Written file info (for RETURN_STATS)
 	unique_ptr<CopyToFileInfo> written_file_info;
 
+	void Initialize(ClientContext &context, const PhysicalBatchCopyToFile &op) {
+		if (initialized) {
+			return;
+		}
+		lock_guard<mutex> guard(lock);
+		if (initialized) {
+			return;
+		}
+		// initialize writing to the file
+		global_state = op.function.copy_to_initialize_global(context, *op.bind_data, op.file_path);
+		if (op.return_type == CopyFunctionReturnType::WRITTEN_FILE_STATISTICS) {
+			written_file_info = make_uniq<CopyToFileInfo>(op.file_path);
+			written_file_info->file_stats = make_uniq<CopyFunctionFileStatistics>();
+			op.function.copy_to_get_written_statistics(context, *op.bind_data, *global_state,
+			                                           *written_file_info->file_stats);
+		}
+	}
+
 	void AddBatchData(idx_t batch_index, unique_ptr<PreparedBatchData> new_batch, idx_t memory_usage) {
 		// move the batch data to the set of prepared batch data
 		lock_guard<mutex> l(lock);
@@ -186,6 +205,10 @@ SinkResultType PhysicalBatchCopyToFile::Sink(ExecutionContext &context, DataChun
 			return Sink(context, chunk, input);
 		}
 	}
+	if (!write_empty_file) {
+		// if we are not writing empty files - initialize after we have received rows
+		gstate.Initialize(context.client, *this);
+	}
 	if (!state.collection) {
 		state.InitializeCollection(context.client, *this);
 		state.batch_index = batch_index;
@@ -288,7 +311,7 @@ SinkFinalizeType PhysicalBatchCopyToFile::FinalFlush(ClientContext &context, Glo
 	if (gstate.scheduled_batch_index != gstate.flushed_batch_index) {
 		throw InternalException("Not all batches were flushed to disk - incomplete file?");
 	}
-	if (function.copy_to_finalize) {
+	if (function.copy_to_finalize && gstate.global_state) {
 		function.copy_to_finalize(context, *bind_data, *gstate.global_state);
 
 		if (use_tmp_file) {
@@ -599,13 +622,10 @@ unique_ptr<GlobalSinkState> PhysicalBatchCopyToFile::GetGlobalSinkState(ClientCo
 	// request memory based on the minimum amount of memory per column
 	auto minimum_memory_per_thread =
 	    FixedBatchCopyGlobalState::MINIMUM_MEMORY_PER_COLUMN_PER_THREAD * children[0].get().GetTypes().size();
-	auto result = make_uniq<FixedBatchCopyGlobalState>(
-	    context, function.copy_to_initialize_global(context, *bind_data, file_path), minimum_memory_per_thread);
-	if (return_type == CopyFunctionReturnType::WRITTEN_FILE_STATISTICS) {
-		result->written_file_info = make_uniq<CopyToFileInfo>(file_path);
-		result->written_file_info->file_stats = make_uniq<CopyFunctionFileStatistics>();
-		function.copy_to_get_written_statistics(context, *bind_data, *result->global_state,
-		                                        *result->written_file_info->file_stats);
+	auto result = make_uniq<FixedBatchCopyGlobalState>(context, minimum_memory_per_thread);
+	if (write_empty_file) {
+		// if we are writing the file also if it is empty - initialize now
+		result->Initialize(context, *this);
 	}
 	result->batch_size = function.desired_batch_size ? function.desired_batch_size(context, *bind_data) : 0;
 	return std::move(result);
@@ -617,19 +637,28 @@ unique_ptr<GlobalSinkState> PhysicalBatchCopyToFile::GetGlobalSinkState(ClientCo
 SourceResultType PhysicalBatchCopyToFile::GetData(ExecutionContext &context, DataChunk &chunk,
                                                   OperatorSourceInput &input) const {
 	auto &g = sink_state->Cast<FixedBatchCopyGlobalState>();
-	chunk.SetCardinality(1);
 	auto fp = use_tmp_file ? PhysicalCopyToFile::GetNonTmpFile(context.client, file_path) : file_path;
 	switch (return_type) {
 	case CopyFunctionReturnType::CHANGED_ROWS:
 		chunk.SetValue(0, 0, Value::BIGINT(NumericCast<int64_t>(g.rows_copied.load())));
+		chunk.SetCardinality(1);
 		break;
 	case CopyFunctionReturnType::CHANGED_ROWS_AND_FILE_LIST: {
+		vector<Value> file_list;
+		if (g.global_state) {
+			file_list.emplace_back(std::move(fp));
+		}
 		chunk.SetValue(0, 0, Value::BIGINT(NumericCast<int64_t>(g.rows_copied.load())));
-		chunk.SetValue(1, 0, Value::LIST(LogicalType::VARCHAR, {fp}));
+		chunk.SetValue(1, 0, Value::LIST(LogicalType::VARCHAR, std::move(file_list)));
+		chunk.SetCardinality(1);
 		break;
 	}
 	case CopyFunctionReturnType::WRITTEN_FILE_STATISTICS: {
-		PhysicalCopyToFile::ReturnStatistics(chunk, 0, *g.written_file_info);
+		if (g.written_file_info) {
+			g.written_file_info->file_path = std::move(fp);
+			PhysicalCopyToFile::ReturnStatistics(chunk, 0, *g.written_file_info);
+			chunk.SetCardinality(1);
+		}
 		break;
 	}
 	default:

diff --git a/src/execution/operator/persistent/physical_copy_to_file.cpp b/src/execution/operator/persistent/physical_copy_to_file.cpp
@@ -47,11 +47,12 @@ using vector_of_value_map_t = unordered_map<vector<Value>, T, VectorOfValuesHash
 
 class CopyToFunctionGlobalState : public GlobalSinkState {
 public:
-	explicit CopyToFunctionGlobalState(ClientContext &context, unique_ptr<GlobalFunctionData> global_state)
-	    : rows_copied(0), last_file_offset(0), global_state(std::move(global_state)) {
+	explicit CopyToFunctionGlobalState(ClientContext &context)
+	    : initialized(false), rows_copied(0), last_file_offset(0) {
 		max_open_files = ClientConfig::GetConfig(context).partitioned_write_max_open_files;
 	}
 	StorageLock lock;
+	atomic<bool> initialized;
 	atomic<idx_t> rows_copied;
 	atomic<idx_t> last_file_offset;
 	unique_ptr<GlobalFunctionData> global_state;
@@ -64,6 +65,24 @@ class CopyToFunctionGlobalState : public GlobalSinkState {
 	//! Max open files
 	idx_t max_open_files;
 
+	void Initialize(ClientContext &context, const PhysicalCopyToFile &op) {
+		if (initialized) {
+			return;
+		}
+		auto write_lock = lock.GetExclusiveLock();
+		if (initialized) {
+			return;
+		}
+		// initialize writing to the file
+		global_state = op.function.copy_to_initialize_global(context, *op.bind_data, op.file_path);
+		auto written_file_info = AddFile(*write_lock, op.file_path, op.return_type);
+		if (written_file_info) {
+			op.function.copy_to_get_written_statistics(context, *op.bind_data, *global_state,
+			                                           *written_file_info->file_stats);
+		}
+		initialized = true;
+	}
+
 	void CreateDir(const string &dir_path, FileSystem &fs) {
 		if (created_directories.find(dir_path) != created_directories.end()) {
 			// already attempted to create this directory
@@ -400,7 +419,7 @@ unique_ptr<GlobalSinkState> PhysicalCopyToFile::GetGlobalSinkState(ClientContext
 			CheckDirectory(fs, file_path, overwrite_mode);
 		}
 
-		auto state = make_uniq<CopyToFunctionGlobalState>(context, nullptr);
+		auto state = make_uniq<CopyToFunctionGlobalState>(context);
 		if (!per_thread_output && rotate) {
 			auto global_lock = state->lock.GetExclusiveLock();
 			state->global_state = CreateFileState(context, *state, *global_lock);
@@ -413,13 +432,10 @@ unique_ptr<GlobalSinkState> PhysicalCopyToFile::GetGlobalSinkState(ClientContext
 		return std::move(state);
 	}
 
-	auto state = make_uniq<CopyToFunctionGlobalState>(
-	    context, function.copy_to_initialize_global(context, *bind_data, file_path));
-	auto global_lock = state->lock.GetExclusiveLock();
-	auto written_file_info = state->AddFile(*global_lock, file_path, return_type);
-	if (written_file_info) {
-		function.copy_to_get_written_statistics(context, *bind_data, *state->global_state,
-		                                        *written_file_info->file_stats);
+	auto state = make_uniq<CopyToFunctionGlobalState>(context);
+	if (write_empty_file) {
+		// if we are writing the file also if it is empty - initialize now
+		state->Initialize(context, *this);
 	}
 	return std::move(state);
 }
@@ -460,6 +476,10 @@ SinkResultType PhysicalCopyToFile::Sink(ExecutionContext &context, DataChunk &ch
 	auto &g = input.global_state.Cast<CopyToFunctionGlobalState>();
 	auto &l = input.local_state.Cast<CopyToFunctionLocalState>();
 
+	if (!write_empty_file) {
+		// if we are only writing the file when there are rows to write we need to initialize here
+		g.Initialize(context.client, *this);
+	}
 	g.rows_copied += chunk.size();
 
 	if (partition_output) {
@@ -523,7 +543,7 @@ SinkCombineResultType PhysicalCopyToFile::Combine(ExecutionContext &context, Ope
 			// File in global state may change with FILE_SIZE_BYTES/rotate, need to grab lock
 			auto lock = g.lock.GetSharedLock();
 			function.copy_to_combine(context, *bind_data, *g.global_state, *l.local_state);
-		} else {
+		} else if (g.global_state) {
 			function.copy_to_combine(context, *bind_data, *g.global_state, *l.local_state);
 		}
 	}
@@ -549,7 +569,7 @@ SinkFinalizeType PhysicalCopyToFile::Finalize(Pipeline &pipeline, Event &event,
 		}
 		return SinkFinalizeType::READY;
 	}
-	if (function.copy_to_finalize) {
+	if (function.copy_to_finalize && gstate.global_state) {
 		function.copy_to_finalize(context, *bind_data, *gstate.global_state);
 
 		if (use_tmp_file) {
@@ -633,6 +653,9 @@ SourceResultType PhysicalCopyToFile::GetData(ExecutionContext &context, DataChun
 		idx_t count = next_end - source_state.offset;
 		for (idx_t i = 0; i < count; i++) {
 			auto &file_entry = *g.written_files[source_state.offset + i];
+			if (use_tmp_file) {
+				file_entry.file_path = GetNonTmpFile(context.client, file_entry.file_path);
+			}
 			ReturnStatistics(chunk, i, file_entry);
 		}
 		chunk.SetCardinality(count);
@@ -650,7 +673,11 @@ SourceResultType PhysicalCopyToFile::GetData(ExecutionContext &context, DataChun
 		chunk.SetValue(0, 0, Value::BIGINT(NumericCast<int64_t>(g.rows_copied.load())));
 		vector<Value> file_name_list;
 		for (auto &file_info : g.written_files) {
-			file_name_list.emplace_back(file_info->file_path);
+			if (use_tmp_file) {
+				file_name_list.emplace_back(GetNonTmpFile(context.client, file_info->file_path));
+			} else {
+				file_name_list.emplace_back(file_info->file_path);
+			}
 		}
 		chunk.SetValue(1, 0, Value::LIST(LogicalType::VARCHAR, std::move(file_name_list)));
 		break;

diff --git a/src/execution/physical_plan/plan_copy_to_file.cpp b/src/execution/physical_plan/plan_copy_to_file.cpp
@@ -41,6 +41,7 @@ PhysicalOperator &PhysicalPlanGenerator::CreatePlan(LogicalCopyToFile &op) {
 		cast_copy.use_tmp_file = op.use_tmp_file;
 		cast_copy.children.push_back(plan);
 		cast_copy.return_type = op.return_type;
+		cast_copy.write_empty_file = op.write_empty_file;
 		return copy;
 	}
 
@@ -67,6 +68,7 @@ PhysicalOperator &PhysicalPlanGenerator::CreatePlan(LogicalCopyToFile &op) {
 	cast_copy.names = op.names;
 	cast_copy.expected_types = op.expected_types;
 	cast_copy.parallel = mode == CopyFunctionExecutionMode::PARALLEL_COPY_TO_FILE;
+	cast_copy.write_empty_file = op.write_empty_file;
 
 	cast_copy.children.push_back(plan);
 	return copy;

diff --git a/src/include/duckdb/execution/operator/persistent/physical_batch_copy_to_file.hpp b/src/include/duckdb/execution/operator/persistent/physical_batch_copy_to_file.hpp
@@ -30,6 +30,7 @@ class PhysicalBatchCopyToFile : public PhysicalOperator {
 	string file_path;
 	bool use_tmp_file;
 	CopyFunctionReturnType return_type;
+	bool write_empty_file;
 
 public:
 	// Source interface

diff --git a/src/include/duckdb/execution/operator/persistent/physical_copy_to_file.hpp b/src/include/duckdb/execution/operator/persistent/physical_copy_to_file.hpp
@@ -51,6 +51,7 @@ class PhysicalCopyToFile : public PhysicalOperator {
 
 	bool partition_output;
 	bool write_partition_columns;
+	bool write_empty_file;
 	vector<idx_t> partition_columns;
 	vector<string> names;
 	vector<LogicalType> expected_types;