Merge pull request #155 from duckdb/bump121-2

update bundled duckdb to 1.2.1

Author: hannes

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/lambda_functions.cpp

@@ -223,17 +223,6 @@ void ExecuteExpression(const idx_t elem_cnt, const LambdaFunctions::ColumnInfo &
 // ListLambdaBindData
 //===--------------------------------------------------------------------===//
 
-unique_ptr<FunctionData> ListLambdaBindData::Copy() const {
-	auto lambda_expr_copy = lambda_expr ? lambda_expr->Copy() : nullptr;
-	return make_uniq<ListLambdaBindData>(return_type, std::move(lambda_expr_copy), has_index);
-}
-
-bool ListLambdaBindData::Equals(const FunctionData &other_p) const {
-	auto &other = other_p.Cast<ListLambdaBindData>();
-	return Expression::Equals(lambda_expr, other.lambda_expr) && return_type == other.return_type &&
-	       has_index == other.has_index;
-}
-
 void ListLambdaBindData::Serialize(Serializer &serializer, const optional_ptr<FunctionData> bind_data_p,
                                    const ScalarFunction &) {
 	auto &bind_data = bind_data_p->Cast<ListLambdaBindData>();

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/extension/icu/icu-datefunc.cpp

@@ -71,13 +71,20 @@ unique_ptr<FunctionData> ICUDateFunc::Bind(ClientContext &context, ScalarFunctio
 	return make_uniq<BindData>(context);
 }
 
-void ICUDateFunc::SetTimeZone(icu::Calendar *calendar, const string_t &tz_id) {
+bool ICUDateFunc::TrySetTimeZone(icu::Calendar *calendar, const string_t &tz_id) {
 	auto tz = icu_66::TimeZone::createTimeZone(icu::UnicodeString::fromUTF8(icu::StringPiece(tz_id.GetString())));
 	if (*tz == icu::TimeZone::getUnknown()) {
 		delete tz;
-		throw NotImplementedException("Unknown TimeZone '%s'", tz_id.GetString());
+		return false;
 	}
 	calendar->adoptTimeZone(tz);
+	return true;
+}
+
+void ICUDateFunc::SetTimeZone(icu::Calendar *calendar, const string_t &tz_id) {
+	if (!TrySetTimeZone(calendar, tz_id)) {
+		throw NotImplementedException("Unknown TimeZone '%s'", tz_id.GetString());
+	}
 }
 
 timestamp_t ICUDateFunc::GetTimeUnsafe(icu::Calendar *calendar, uint64_t micros) {

src/duckdb/extension/icu/icu-strptime.cpp

@@ -11,9 +11,7 @@
 #include "duckdb/execution/expression_executor.hpp"
 #include "duckdb/function/scalar/strftime_format.hpp"
 #include "duckdb/main/client_context.hpp"
-#include "duckdb/parser/parsed_data/create_scalar_function_info.hpp"
 #include "duckdb/planner/expression/bound_function_expression.hpp"
-#include "duckdb/function/function_binder.hpp"
 #include "duckdb/function/cast/default_casts.hpp"
 #include "duckdb/main/extension_util.hpp"
 
@@ -60,14 +58,7 @@ struct ICUStrptime : public ICUDateFunc {
 	}
 
 	static uint64_t ToMicros(icu::Calendar *calendar, const ParseResult &parsed, const StrpTimeFormat &format) {
-		// Set TZ first, if any.
-		// Note that empty TZ names are not allowed,
-		// but unknown names will map to GMT.
-		if (!parsed.tz.empty()) {
-			SetTimeZone(calendar, parsed.tz);
-		}
-
-		// Now get the parts in the given time zone
+		// Get the parts in the current time zone
 		uint64_t micros = parsed.GetMicros();
 		calendar->set(UCAL_EXTENDED_YEAR, parsed.data[0]); // strptime doesn't understand eras
 		calendar->set(UCAL_MONTH, parsed.data[1] - 1);
@@ -110,6 +101,11 @@ struct ICUStrptime : public ICUDateFunc {
 						if (parsed.is_special) {
 							return parsed.ToTimestamp();
 						} else {
+							// Set TZ first, if any.
+							if (!parsed.tz.empty()) {
+								SetTimeZone(calendar, parsed.tz);
+							}
+
 							return GetTime(calendar, ToMicros(calendar, parsed, format));
 						}
 					}
@@ -143,7 +139,7 @@ struct ICUStrptime : public ICUDateFunc {
 					    if (format.Parse(input, parsed)) {
 						    if (parsed.is_special) {
 							    return parsed.ToTimestamp();
-						    } else {
+						    } else if (parsed.tz.empty() || TrySetTimeZone(calendar, parsed.tz)) {
 							    timestamp_t result;
 							    if (TryGetTime(calendar, ToMicros(calendar, parsed, format), result)) {
 								    return result;

src/duckdb/extension/icu/include/icu-datefunc.hpp

@@ -49,7 +49,9 @@ struct ICUDateFunc {
 	static duckdb::unique_ptr<FunctionData> Bind(ClientContext &context, ScalarFunction &bound_function,
 	                                             vector<duckdb::unique_ptr<Expression>> &arguments);
 
-	//! Sets the time zone for the calendar.
+	//! Tries to set the time zone for the calendar and returns false if it is not valid.
+	static bool TrySetTimeZone(icu::Calendar *calendar, const string_t &tz_id);
+	//! Sets the time zone for the calendar. Throws if it is not valid
 	static void SetTimeZone(icu::Calendar *calendar, const string_t &tz_id);
 	//! Gets the timestamp from the calendar, throwing if it is not in range.
 	static bool TryGetTime(icu::Calendar *calendar, uint64_t micros, timestamp_t &result);

src/duckdb/extension/json/buffered_json_reader.cpp

@@ -90,22 +90,16 @@ void JSONFileHandle::ReadAtPosition(char *pointer, idx_t size, idx_t position, b
                                     optional_ptr<FileHandle> override_handle) {
 	if (size != 0) {
 		auto &handle = override_handle ? *override_handle.get() : *file_handle.get();
-		if (can_seek) {
-			handle.Read(pointer, size, position);
-		} else if (sample_run) { // Cache the buffer
-			handle.Read(pointer, size, position);
 
+		if (!cached_buffers.empty() || position < cached_size) {
+			ReadFromCache(pointer, size, position);
+		}
+
+		handle.Read(pointer, size, position);
+		if (file_handle->IsPipe()) { // Cache the buffer
 			cached_buffers.emplace_back(allocator.Allocate(size));
 			memcpy(cached_buffers.back().get(), pointer, size);
 			cached_size += size;
-		} else {
-			if (!cached_buffers.empty() || position < cached_size) {
-				ReadFromCache(pointer, size, position);
-			}
-
-			if (size != 0) {
-				handle.Read(pointer, size, position);
-			}
 		}
 	}
 
@@ -121,30 +115,23 @@ void JSONFileHandle::ReadAtPosition(char *pointer, idx_t size, idx_t position, b
 
 bool JSONFileHandle::Read(char *pointer, idx_t &read_size, idx_t requested_size, bool &file_done, bool sample_run) {
 	D_ASSERT(requested_size != 0);
+	read_size = 0;
 	if (last_read_requested) {
 		return false;
 	}
 
-	if (can_seek) {
-		read_size = ReadInternal(pointer, requested_size);
-		read_position += read_size;
-	} else if (sample_run) { // Cache the buffer
-		read_size = ReadInternal(pointer, requested_size);
-		if (read_size > 0) {
-			cached_buffers.emplace_back(allocator.Allocate(read_size));
-			memcpy(cached_buffers.back().get(), pointer, read_size);
-		}
-		cached_size += read_size;
-		read_position += read_size;
-	} else {
-		read_size = 0;
-		if (!cached_buffers.empty() || read_position < cached_size) {
-			read_size += ReadFromCache(pointer, requested_size, read_position);
-		}
-		if (requested_size != 0) {
-			read_size += ReadInternal(pointer, requested_size);
-		}
+	if (!cached_buffers.empty() || read_position < cached_size) {
+		read_size += ReadFromCache(pointer, requested_size, read_position);
+	}
+
+	auto temp_read_size = ReadInternal(pointer, requested_size);
+	if (file_handle->IsPipe() && temp_read_size != 0) { // Cache the buffer
+		cached_buffers.emplace_back(allocator.Allocate(temp_read_size));
+		memcpy(cached_buffers.back().get(), pointer, temp_read_size);
 	}
+	cached_size += temp_read_size;
+	read_position += temp_read_size;
+	read_size += temp_read_size;
 
 	if (read_size == 0) {
 		last_read_requested = true;

src/duckdb/extension/json/json_extension.cpp

@@ -17,12 +17,17 @@
 namespace duckdb {
 
 static DefaultMacro json_macros[] = {
-    {DEFAULT_SCHEMA, "json_group_array", {"x", nullptr}, {{nullptr, nullptr}}, "to_json(list(x))"},
+    {DEFAULT_SCHEMA,
+     "json_group_array",
+     {"x", nullptr},
+     {{nullptr, nullptr}},
+     "CAST('[' || string_agg(CASE WHEN x IS NULL THEN 'null'::JSON ELSE to_json(x) END, ',') || ']' AS JSON)"},
     {DEFAULT_SCHEMA,
      "json_group_object",
-     {"name", "value", nullptr},
+     {"n", "v", nullptr},
      {{nullptr, nullptr}},
-     "to_json(map(list(name), list(value)))"},
+     "CAST('{' || string_agg(to_json(n::VARCHAR) || ':' || CASE WHEN v IS NULL THEN 'null'::JSON ELSE to_json(v) END, "
+     "',') || '}' AS JSON)"},
     {DEFAULT_SCHEMA,
      "json_group_structure",
      {"x", nullptr},

src/duckdb/extension/parquet/column_reader.cpp

@@ -319,7 +319,8 @@ void ColumnReader::PreparePageV2(PageHeader &page_hdr) {
 
 	auto compressed_bytes = page_hdr.compressed_page_size - uncompressed_bytes;
 
-	AllocateCompressed(compressed_bytes);
+	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,
@@ -334,10 +335,6 @@ void ColumnReader::AllocateBlock(idx_t size) {
 	}
 }
 
-void ColumnReader::AllocateCompressed(idx_t size) {
-	compressed_buffer.resize(GetAllocator(), size);
-}
-
 void ColumnReader::PreparePage(PageHeader &page_hdr) {
 	AllocateBlock(page_hdr.uncompressed_page_size + 1);
 	if (chunk->meta_data.codec == CompressionCodec::UNCOMPRESSED) {
@@ -348,7 +345,8 @@ void ColumnReader::PreparePage(PageHeader &page_hdr) {
 		return;
 	}
 
-	AllocateCompressed(page_hdr.compressed_page_size + 1);
+	ResizeableBuffer compressed_buffer;
+	compressed_buffer.resize(GetAllocator(), page_hdr.compressed_page_size + 1);
 	reader.ReadData(*protocol, compressed_buffer.ptr, page_hdr.compressed_page_size);
 
 	DecompressInternal(chunk->meta_data.codec, compressed_buffer.ptr, page_hdr.compressed_page_size, block->ptr,