vendor: Update vendored sources to duckdb/duckdb@7a9e96d

Merge V1.2 histrionicus into main (duckdb/duckdb#16324)
Bump Julia FixedPointDecimals dependency version (duckdb/duckdb#16323)

Author: krlmlr

src/duckdb/extension/core_functions/aggregate/distributive/string_agg.cpp

@@ -44,41 +44,33 @@ struct StringAggFunction {
 		if (!state.dataptr) {
 			finalize_data.ReturnNull();
 		} else {
-			target = StringVector::AddString(finalize_data.result, state.dataptr, state.size);
-		}
-	}
-
-	template <class STATE>
-	static void Destroy(STATE &state, AggregateInputData &aggr_input_data) {
-		if (state.dataptr) {
-			delete[] state.dataptr;
+			target = string_t(state.dataptr, state.size);
 		}
 	}
 
 	static bool IgnoreNull() {
 		return true;
 	}
 
-	static inline void PerformOperation(StringAggState &state, const char *str, const char *sep, idx_t str_size,
-	                                    idx_t sep_size) {
+	static inline void PerformOperation(StringAggState &state, ArenaAllocator &allocator, const char *str,
+	                                    const char *sep, idx_t str_size, idx_t sep_size) {
 		if (!state.dataptr) {
 			// first iteration: allocate space for the string and copy it into the state
 			state.alloc_size = MaxValue<idx_t>(8, NextPowerOfTwo(str_size));
-			state.dataptr = new char[state.alloc_size];
+			state.dataptr = char_ptr_cast(allocator.Allocate(state.alloc_size));
 			state.size = str_size;
 			memcpy(state.dataptr, str, str_size);
 		} else {
 			// subsequent iteration: first check if we have space to place the string and separator
 			idx_t required_size = state.size + str_size + sep_size;
 			if (required_size > state.alloc_size) {
 				// no space! allocate extra space
+				const auto old_size = state.alloc_size;
 				while (state.alloc_size < required_size) {
 					state.alloc_size *= 2;
 				}
-				auto new_data = new char[state.alloc_size];
-				memcpy(new_data, state.dataptr, state.size);
-				delete[] state.dataptr;
-				state.dataptr = new_data;
+				state.dataptr =
+				    char_ptr_cast(allocator.Reallocate(data_ptr_cast(state.dataptr), old_size, state.alloc_size));
 			}
 			// copy the separator
 			memcpy(state.dataptr + state.size, sep, sep_size);
@@ -89,14 +81,15 @@ struct StringAggFunction {
 		}
 	}
 
-	static inline void PerformOperation(StringAggState &state, string_t str, optional_ptr<FunctionData> data_p) {
+	static inline void PerformOperation(StringAggState &state, ArenaAllocator &allocator, string_t str,
+	                                    optional_ptr<FunctionData> data_p) {
 		auto &data = data_p->Cast<StringAggBindData>();
-		PerformOperation(state, str.GetData(), data.sep.c_str(), str.GetSize(), data.sep.size());
+		PerformOperation(state, allocator, str.GetData(), data.sep.c_str(), str.GetSize(), data.sep.size());
 	}
 
 	template <class INPUT_TYPE, class STATE, class OP>
 	static void Operation(STATE &state, const INPUT_TYPE &input, AggregateUnaryInput &unary_input) {
-		PerformOperation(state, input, unary_input.input.bind_data);
+		PerformOperation(state, unary_input.input.allocator, input, unary_input.input.bind_data);
 	}
 
 	template <class INPUT_TYPE, class STATE, class OP>
@@ -113,8 +106,8 @@ struct StringAggFunction {
 			// source is not set: skip combining
 			return;
 		}
-		PerformOperation(target, string_t(source.dataptr, UnsafeNumericCast<uint32_t>(source.size)),
-		                 aggr_input_data.bind_data);
+		PerformOperation(target, aggr_input_data.allocator,
+		                 string_t(source.dataptr, UnsafeNumericCast<uint32_t>(source.size)), aggr_input_data.bind_data);
 	}
 };
 
@@ -162,8 +155,7 @@ AggregateFunctionSet StringAggFun::GetFunctions() {
 	    AggregateFunction::UnaryScatterUpdate<StringAggState, string_t, StringAggFunction>,
 	    AggregateFunction::StateCombine<StringAggState, StringAggFunction>,
 	    AggregateFunction::StateFinalize<StringAggState, string_t, StringAggFunction>,
-	    AggregateFunction::UnaryUpdate<StringAggState, string_t, StringAggFunction>, StringAggBind,
-	    AggregateFunction::StateDestroy<StringAggState, StringAggFunction>);
+	    AggregateFunction::UnaryUpdate<StringAggState, string_t, StringAggFunction>, StringAggBind);
 	string_agg_param.serialize = StringAggSerialize;
 	string_agg_param.deserialize = StringAggDeserialize;
 	string_agg.AddFunction(string_agg_param);

src/duckdb/extension/core_functions/aggregate/nested/list.cpp

@@ -116,7 +116,6 @@ static void ListFinalize(Vector &states_vector, AggregateInputData &aggr_input_d
 
 	// first iterate over all entries and set up the list entries, and get the newly required total length
 	for (idx_t i = 0; i < count; i++) {
-
 		auto &state = *states[states_data.sel->get_index(i)];
 		const auto rid = i + offset;
 		result_data[rid].offset = total_len;

src/duckdb/extension/core_functions/scalar/list/list_aggregates.cpp

@@ -15,7 +15,17 @@
 
 namespace duckdb {
 
-// FIXME: use a local state for each thread to increase performance?
+struct ListAggregatesLocalState : public FunctionLocalState {
+	explicit ListAggregatesLocalState(Allocator &allocator) : arena_allocator(allocator) {
+	}
+
+	ArenaAllocator arena_allocator;
+};
+
+unique_ptr<FunctionLocalState> ListAggregatesInitLocalState(ExpressionState &state, const BoundFunctionExpression &expr,
+                                                            FunctionData *bind_data) {
+	return make_uniq<ListAggregatesLocalState>(BufferAllocator::Get(state.GetContext()));
+}
 // FIXME: benchmark the use of simple_update against using update (if applicable)
 
 static unique_ptr<FunctionData> ListAggregatesBindFailure(ScalarFunction &bound_function) {
@@ -207,7 +217,8 @@ static void ListAggregatesFunction(DataChunk &args, ExpressionState &state, Vect
 	auto &func_expr = state.expr.Cast<BoundFunctionExpression>();
 	auto &info = func_expr.bind_info->Cast<ListAggregatesBindData>();
 	auto &aggr = info.aggr_expr->Cast<BoundAggregateExpression>();
-	ArenaAllocator allocator(Allocator::DefaultAllocator());
+	auto &allocator = ExecuteFunctionState::GetFunctionState(state)->Cast<ListAggregatesLocalState>().arena_allocator;
+	allocator.Reset();
 	AggregateInputData aggr_input_data(aggr.bind_info.get(), allocator);
 
 	D_ASSERT(aggr.function.update);
@@ -511,8 +522,9 @@ static unique_ptr<FunctionData> ListUniqueBind(ClientContext &context, ScalarFun
 }
 
 ScalarFunction ListAggregateFun::GetFunction() {
-	auto result = ScalarFunction({LogicalType::LIST(LogicalType::ANY), LogicalType::VARCHAR}, LogicalType::ANY,
-	                             ListAggregateFunction, ListAggregateBind);
+	auto result =
+	    ScalarFunction({LogicalType::LIST(LogicalType::ANY), LogicalType::VARCHAR}, LogicalType::ANY,
+	                   ListAggregateFunction, ListAggregateBind, nullptr, nullptr, ListAggregatesInitLocalState);
 	BaseScalarFunction::SetReturnsError(result);
 	result.null_handling = FunctionNullHandling::SPECIAL_HANDLING;
 	result.varargs = LogicalType::ANY;
@@ -523,12 +535,12 @@ ScalarFunction ListAggregateFun::GetFunction() {
 
 ScalarFunction ListDistinctFun::GetFunction() {
 	return ScalarFunction({LogicalType::LIST(LogicalType::ANY)}, LogicalType::LIST(LogicalType::ANY),
-	                      ListDistinctFunction, ListDistinctBind);
+	                      ListDistinctFunction, ListDistinctBind, nullptr, nullptr, ListAggregatesInitLocalState);
 }
 
 ScalarFunction ListUniqueFun::GetFunction() {
 	return ScalarFunction({LogicalType::LIST(LogicalType::ANY)}, LogicalType::UBIGINT, ListUniqueFunction,
-	                      ListUniqueBind);
+	                      ListUniqueBind, nullptr, nullptr, ListAggregatesInitLocalState);
 }
 
 } // namespace duckdb

src/duckdb/src/common/compressed_file_system.cpp

@@ -31,6 +31,8 @@ void CompressedFile::Initialize(bool write) {
 	stream_data.out_buff_start = stream_data.out_buff.get();
 	stream_data.out_buff_end = stream_data.out_buff.get();
 
+	current_position = 0;
+
 	stream_wrapper = compressed_fs.CreateStream();
 	stream_wrapper->Initialize(*this, write);
 }

src/duckdb/src/execution/operator/aggregate/physical_window.cpp

@@ -575,6 +575,11 @@ class WindowLocalSourceState : public LocalSourceState {
 
 	explicit WindowLocalSourceState(WindowGlobalSourceState &gsource);
 
+	void ReleaseLocalStates() {
+		auto &local_states = window_hash_group->thread_states.at(task->thread_idx);
+		local_states.clear();
+	}
+
 	//! Does the task have more work to do?
 	bool TaskFinished() const {
 		return !task || task->begin_idx == task->end_idx;
@@ -792,6 +797,12 @@ void WindowGlobalSourceState::FinishTask(TaskPtr task) {
 }
 
 bool WindowLocalSourceState::TryAssignTask() {
+	D_ASSERT(TaskFinished());
+	if (task && task->stage == WindowGroupStage::GETDATA) {
+		// If this state completed the last block in the previous iteration,
+		// release out local state memory.
+		ReleaseLocalStates();
+	}
 	// Because downstream operators may be using our internal buffers,
 	// we can't "finish" a task until we are about to get the next one.
 
@@ -888,10 +899,6 @@ void WindowLocalSourceState::GetData(DataChunk &result) {
 		++task->begin_idx;
 	}
 
-	// If that was the last block, release out local state memory.
-	if (TaskFinished()) {
-		local_states.clear();
-	}
 	result.Verify();
 }
 

src/duckdb/src/execution/operator/csv_scanner/sniffer/dialect_detection.cpp

@@ -491,7 +491,7 @@ void CSVSniffer::RefineCandidates() {
 
 	for (idx_t i = 1; i <= options.sample_size_chunks; i++) {
 		vector<unique_ptr<ColumnCountScanner>> successful_candidates;
-		bool done = false;
+		bool done = candidates.empty();
 		for (auto &cur_candidate : candidates) {
 			const bool finished_file = cur_candidate->FinishedFile();
 			if (successful_candidates.empty()) {

src/duckdb/src/execution/operator/helper/physical_reservoir_sample.cpp

@@ -86,6 +86,7 @@ SourceResultType PhysicalReservoirSample::GetData(ExecutionContext &context, Dat
 		return SourceResultType::FINISHED;
 	}
 	auto sample_chunk = sink.sample->GetChunk();
+
 	if (!sample_chunk) {
 		return SourceResultType::FINISHED;
 	}

src/duckdb/src/execution/operator/schema/physical_create_art_index.cpp

@@ -34,6 +34,7 @@ PhysicalCreateARTIndex::PhysicalCreateARTIndex(LogicalOperator &op, TableCatalog
 
 class CreateARTIndexGlobalSinkState : public GlobalSinkState {
 public:
+	//! We merge the local indexes into one global index.
 	unique_ptr<BoundIndex> global_index;
 };
 
@@ -53,8 +54,10 @@ class CreateARTIndexLocalSinkState : public LocalSinkState {
 };
 
 unique_ptr<GlobalSinkState> PhysicalCreateARTIndex::GetGlobalSinkState(ClientContext &context) const {
-	// Create the global sink state and add the global index.
+	// Create the global sink state.
 	auto state = make_uniq<CreateARTIndexGlobalSinkState>();
+
+	// Create the global index.
 	auto &storage = table.GetStorage();
 	state->global_index = make_uniq<ART>(info->index_name, info->constraint_type, storage_ids,
 	                                     TableIOManager::Get(storage), unbound_expressions, storage.db);
@@ -123,7 +126,6 @@ SinkResultType PhysicalCreateARTIndex::SinkSorted(OperatorSinkInput &input) cons
 
 SinkResultType PhysicalCreateARTIndex::Sink(ExecutionContext &context, DataChunk &chunk,
                                             OperatorSinkInput &input) const {
-
 	D_ASSERT(chunk.ColumnCount() >= 2);
 	auto &l_state = input.local_state.Cast<CreateARTIndexLocalSinkState>();
 	l_state.arena_allocator.Reset();
@@ -151,11 +153,10 @@ SinkResultType PhysicalCreateARTIndex::Sink(ExecutionContext &context, DataChunk
 
 SinkCombineResultType PhysicalCreateARTIndex::Combine(ExecutionContext &context,
                                                       OperatorSinkCombineInput &input) const {
-
 	auto &g_state = input.global_state.Cast<CreateARTIndexGlobalSinkState>();
-	auto &l_state = input.local_state.Cast<CreateARTIndexLocalSinkState>();
 
 	// Merge the local index into the global index.
+	auto &l_state = input.local_state.Cast<CreateARTIndexLocalSinkState>();
 	if (!g_state.global_index->MergeIndexes(*l_state.local_index)) {
 		throw ConstraintException("Data contains duplicates on indexed column(s)");
 	}
@@ -165,8 +166,6 @@ SinkCombineResultType PhysicalCreateARTIndex::Combine(ExecutionContext &context,
 
 SinkFinalizeType PhysicalCreateARTIndex::Finalize(Pipeline &pipeline, Event &event, ClientContext &context,
                                                   OperatorSinkFinalizeInput &input) const {
-
-	// Here, we set the resulting global index as the newly created index of the table.
 	auto &state = input.global_state.Cast<CreateARTIndexGlobalSinkState>();
 
 	// Vacuum excess memory and verify.
@@ -182,7 +181,6 @@ SinkFinalizeType PhysicalCreateARTIndex::Finalize(Pipeline &pipeline, Event &eve
 	auto &schema = table.schema;
 	info->column_ids = storage_ids;
 
-	// FIXME: We should check for catalog exceptions prior to index creation, and later double-check.
 	if (!alter_table_info) {
 		// Ensure that the index does not yet exist in the catalog.
 		auto entry = schema.GetEntry(schema.GetCatalogTransaction(context), CatalogType::INDEX_ENTRY, info->index_name);

src/duckdb/src/execution/physical_plan/plan_create_index.cpp

@@ -6,10 +6,21 @@
 #include "duckdb/planner/expression/bound_reference_expression.hpp"
 #include "duckdb/planner/operator/logical_create_index.hpp"
 #include "duckdb/planner/operator/logical_get.hpp"
+#include "duckdb/execution/operator/scan/physical_dummy_scan.hpp"
 
 namespace duckdb {
 
 unique_ptr<PhysicalOperator> PhysicalPlanGenerator::CreatePlan(LogicalCreateIndex &op) {
+	// Early-out, if the index already exists.
+	auto &schema = op.table.schema;
+	auto entry = schema.GetEntry(schema.GetCatalogTransaction(context), CatalogType::INDEX_ENTRY, op.info->index_name);
+	if (entry) {
+		if (op.info->on_conflict != OnCreateConflict::IGNORE_ON_CONFLICT) {
+			throw CatalogException("Index with name \"%s\" already exists!", op.info->index_name);
+		}
+		return make_uniq<PhysicalDummyScan>(op.types, op.estimated_cardinality);
+	}
+
 	// Ensure that all expressions contain valid scalar functions.
 	// E.g., get_current_timestamp(), random(), and sequence values cannot be index keys.
 	for (idx_t i = 0; i < op.unbound_expressions.size(); i++) {