Handle 0-list maps (erroring out) and add test-case

.github/config/uncovered_files.csv

@@ -72,7 +72,6 @@ common/types/batched_data_collection.cpp	11
 common/types/bit.cpp	9
 common/types/blob.cpp	3
 common/types/cast_helpers.cpp	2
-common/types/chunk_collection.cpp	94
 common/types/column/column_data_allocator.cpp	13
 common/types/column/column_data_collection.cpp	55
 common/types/column/partitioned_column_data.cpp	8
@@ -385,7 +384,6 @@ include/duckdb/common/sort/sorted_block.hpp	1
 include/duckdb/common/string_util.hpp	9
 include/duckdb/common/type_util.hpp	2
 include/duckdb/common/types.hpp	5
-include/duckdb/common/types/chunk_collection.hpp	6
 include/duckdb/common/types/column/column_data_allocator.hpp	3
 include/duckdb/common/types/column/partitioned_column_data.hpp	2
 include/duckdb/common/types/datetime.hpp	13

.github/patches/extensions/substrait/disable_tpch.patch

@@ -0,0 +1,19 @@
+diff --git a/test/sql/test_substrait_tpch.test b/test/sql/test_substrait_tpch.test
+index ffa2666..10f815b 100644
+--- a/test/sql/test_substrait_tpch.test
++++ b/test/sql/test_substrait_tpch.test
+@@ -2,6 +2,14 @@
+ # description: Test get_substrait with TPC-H queries
+ # group: [sql]
+
++# test skipped since PR https://github.com/duckdb/duckdb/pull/9993
++# the PR re-introduces DelimJoins in TPC-H again for performance reasons
++# if there is a selection in the duplicate-eliminated side, we keep the DelimJoin
++# this is checked in Deliminator::HasSelection
++# if this function returns false, all DelimJoins are removed from TPC-H
++
++mode skip
++
+ require substrait
+
+ require tpch

benchmark/micro/aggregate/ordered_first.benchmark

@@ -5,13 +5,13 @@
 name Ordered First (Grouped)
 group aggregate
 
-#load
-#PRAGMA ordered_aggregate_threshold=262144
+load
+CREATE TABLE t AS FROM range(10000000) tbl(i);
 
 run
 SELECT SUM(agg) FROM (
 	SELECT i // 2048 AS grp, FIRST(i ORDER BY i DESC) AS agg
-	FROM range(10000000) tbl(i)
+	FROM t
 	GROUP BY ALL
 )
 

extension/icu/icu-timezone.cpp

@@ -10,6 +10,16 @@
 
 namespace duckdb {
 
+template <typename T>
+static bool ICUIsFinite(const T &t) {
+	return true;
+}
+
+template <>
+bool ICUIsFinite(const timestamp_t &t) {
+	return Timestamp::IsFinite(t);
+}
+
 struct ICUTimeZoneData : public GlobalTableFunctionState {
 	ICUTimeZoneData() : tzs(icu::TimeZone::createEnumeration()) {
 		UErrorCode status = U_ZERO_ERROR;
@@ -93,7 +103,7 @@ static void ICUTimeZoneFunction(ClientContext &context, TableFunctionInput &data
 
 struct ICUFromNaiveTimestamp : public ICUDateFunc {
 	static inline timestamp_t Operation(icu::Calendar *calendar, timestamp_t naive) {
-		if (!Timestamp::IsFinite(naive)) {
+		if (!ICUIsFinite(naive)) {
 			return naive;
 		}
 
@@ -157,7 +167,7 @@ struct ICUFromNaiveTimestamp : public ICUDateFunc {
 
 struct ICUToNaiveTimestamp : public ICUDateFunc {
 	static inline timestamp_t Operation(icu::Calendar *calendar, timestamp_t instant) {
-		if (!Timestamp::IsFinite(instant)) {
+		if (!ICUIsFinite(instant)) {
 			return instant;
 		}
 
@@ -289,8 +299,23 @@ struct ICULocalTimeFunc : public ICUDateFunc {
 	}
 };
 
+struct ICUToTimeTZ : public ICUDateFunc {
+	static inline dtime_tz_t Operation(icu::Calendar *calendar, dtime_tz_t timetz) {
+		// Normalise to +00:00, add TZ offset, then set offset to TZ
+		auto time = Time::NormalizeTimeTZ(timetz);
+
+		auto offset = ExtractField(calendar, UCAL_ZONE_OFFSET);
+		offset += ExtractField(calendar, UCAL_DST_OFFSET);
+		offset /= Interval::MSECS_PER_SEC;
+
+		date_t date(0);
+		time = Interval::Add(time, {0, 0, offset * Interval::MICROS_PER_SEC}, date);
+		return dtime_tz_t(time, offset);
+	}
+};
+
 struct ICUTimeZoneFunc : public ICUDateFunc {
-	template <typename OP>
+	template <typename OP, typename T>
 	static void Execute(DataChunk &input, ExpressionState &state, Vector &result) {
 		auto &func_expr = state.expr.Cast<BoundFunctionExpression>();
 		auto &info = func_expr.bind_info->Cast<BindData>();
@@ -307,28 +332,29 @@ struct ICUTimeZoneFunc : public ICUDateFunc {
 				ConstantVector::SetNull(result, true);
 			} else {
 				SetTimeZone(calendar, *ConstantVector::GetData<string_t>(tz_vec));
-				UnaryExecutor::Execute<timestamp_t, timestamp_t>(
-				    ts_vec, result, input.size(), [&](timestamp_t ts) { return OP::Operation(calendar, ts); });
+				UnaryExecutor::Execute<T, T>(ts_vec, result, input.size(),
+				                             [&](T ts) { return OP::Operation(calendar, ts); });
 			}
 		} else {
-			BinaryExecutor::Execute<string_t, timestamp_t, timestamp_t>(tz_vec, ts_vec, result, input.size(),
-			                                                            [&](string_t tz_id, timestamp_t ts) {
-				                                                            if (Timestamp::IsFinite(ts)) {
-					                                                            SetTimeZone(calendar, tz_id);
-					                                                            return OP::Operation(calendar, ts);
-				                                                            } else {
-					                                                            return ts;
-				                                                            }
-			                                                            });
+			BinaryExecutor::Execute<string_t, T, T>(tz_vec, ts_vec, result, input.size(), [&](string_t tz_id, T ts) {
+				if (ICUIsFinite(ts)) {
+					SetTimeZone(calendar, tz_id);
+					return OP::Operation(calendar, ts);
+				} else {
+					return ts;
+				}
+			});
 		}
 	}
 
 	static void AddFunction(const string &name, DatabaseInstance &db) {
 		ScalarFunctionSet set(name);
 		set.AddFunction(ScalarFunction({LogicalType::VARCHAR, LogicalType::TIMESTAMP}, LogicalType::TIMESTAMP_TZ,
-		                               Execute<ICUFromNaiveTimestamp>, Bind));
+		                               Execute<ICUFromNaiveTimestamp, timestamp_t>, Bind));
 		set.AddFunction(ScalarFunction({LogicalType::VARCHAR, LogicalType::TIMESTAMP_TZ}, LogicalType::TIMESTAMP,
-		                               Execute<ICUToNaiveTimestamp>, Bind));
+		                               Execute<ICUToNaiveTimestamp, timestamp_t>, Bind));
+		set.AddFunction(ScalarFunction({LogicalType::VARCHAR, LogicalType::TIME_TZ}, LogicalType::TIME_TZ,
+		                               Execute<ICUToTimeTZ, dtime_tz_t>, Bind));
 		ExtensionUtil::AddFunctionOverload(db, set);
 	}
 };

extension/json/buffered_json_reader.cpp

@@ -14,8 +14,7 @@ JSONBufferHandle::JSONBufferHandle(idx_t buffer_index_p, idx_t readers_p, Alloca
 
 JSONFileHandle::JSONFileHandle(unique_ptr<FileHandle> file_handle_p, Allocator &allocator_p)
     : file_handle(std::move(file_handle_p)), allocator(allocator_p), can_seek(file_handle->CanSeek()),
-      plain_file_source(file_handle->OnDiskFile() && can_seek), file_size(file_handle->GetFileSize()), read_position(0),
-      requested_reads(0), actual_reads(0), cached_size(0) {
+      file_size(file_handle->GetFileSize()), read_position(0), requested_reads(0), actual_reads(0), cached_size(0) {
 }
 
 bool JSONFileHandle::IsOpen() const {
@@ -55,6 +54,10 @@ bool JSONFileHandle::CanSeek() const {
 	return can_seek;
 }
 
+FileHandle &JSONFileHandle::GetHandle() {
+	return *file_handle;
+}
+
 idx_t JSONFileHandle::GetPositionAndSize(idx_t &position, idx_t requested_size) {
 	D_ASSERT(requested_size != 0);
 
@@ -68,12 +71,15 @@ idx_t JSONFileHandle::GetPositionAndSize(idx_t &position, idx_t requested_size)
 	return actual_size;
 }
 
-void JSONFileHandle::ReadAtPosition(char *pointer, idx_t size, idx_t position, bool sample_run) {
+void JSONFileHandle::ReadAtPosition(char *pointer, idx_t size, idx_t position, bool sample_run,
+                                    optional_ptr<FileHandle> override_handle) {
 	D_ASSERT(size != 0);
-	if (plain_file_source) {
-		file_handle->Read(pointer, size, position);
+	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
-		file_handle->Read(pointer, size, position);
+		handle.Read(pointer, size, position);
 
 		cached_buffers.emplace_back(allocator.Allocate(size));
 		memcpy(cached_buffers.back().get(), pointer, size);
@@ -84,7 +90,7 @@ void JSONFileHandle::ReadAtPosition(char *pointer, idx_t size, idx_t position, b
 		}
 
 		if (size != 0) {
-			file_handle->Read(pointer, size, position);
+			handle.Read(pointer, size, position);
 		}
 	}
 	if (++actual_reads > requested_reads) {
@@ -94,7 +100,7 @@ void JSONFileHandle::ReadAtPosition(char *pointer, idx_t size, idx_t position, b
 
 idx_t JSONFileHandle::Read(char *pointer, idx_t requested_size, bool sample_run) {
 	D_ASSERT(requested_size != 0);
-	if (plain_file_source) {
+	if (can_seek) {
 		auto actual_size = ReadInternal(pointer, requested_size);
 		read_position += actual_size;
 		return actual_size;

extension/json/include/buffered_json_reader.hpp

@@ -66,9 +66,13 @@ struct JSONFileHandle {
 
 	bool CanSeek() const;
 
+	FileHandle &GetHandle();
+
 	idx_t GetPositionAndSize(idx_t &position, idx_t requested_size);
-	void ReadAtPosition(char *pointer, idx_t size, idx_t position, bool sample_run);
 	idx_t Read(char *pointer, idx_t requested_size, bool sample_run);
+	//! Read at position optionally allows passing a custom handle to read from, otherwise the default one is used
+	void ReadAtPosition(char *pointer, idx_t size, idx_t position, bool sample_run,
+	                    optional_ptr<FileHandle> override_handle = nullptr);
 
 private:
 	idx_t ReadInternal(char *pointer, const idx_t requested_size);
@@ -81,7 +85,6 @@ struct JSONFileHandle {
 
 	//! File properties
 	const bool can_seek;
-	const bool plain_file_source;
 	const idx_t file_size;
 
 	//! Read properties

extension/json/include/json_scan.hpp

@@ -250,6 +250,12 @@ struct JSONScanLocalState {
 	//! Whether this is the last batch of the file
 	bool is_last;
 
+	//! The current main filesystem
+	FileSystem &fs;
+
+	//! For some filesystems (e.g. S3), using a filehandle per thread increases performance
+	unique_ptr<FileHandle> thread_local_filehandle;
+
 	//! Current buffer read info
 	char *buffer_ptr;
 	idx_t buffer_size;

extension/json/json_functions/copy_json.cpp

@@ -12,7 +12,7 @@
 namespace duckdb {
 
 static void ThrowJSONCopyParameterException(const string &loption) {
-	throw BinderException("COPY (FORMAT JSON) parameter %s expects a single argument.");
+	throw BinderException("COPY (FORMAT JSON) parameter %s expects a single argument.", loption);
 }
 
 static BoundStatement CopyToJSONPlan(Binder &binder, CopyStatement &stmt) {
@@ -23,7 +23,8 @@ static BoundStatement CopyToJSONPlan(Binder &binder, CopyStatement &stmt) {
 	// Parse the options, creating options for the CSV writer while doing so
 	string date_format;
 	string timestamp_format;
-	case_insensitive_map_t<vector<Value>> csv_copy_options;
+	// We insert the JSON file extension here so it works properly with PER_THREAD_OUTPUT/FILE_SIZE_BYTES etc.
+	case_insensitive_map_t<vector<Value>> csv_copy_options {{"file_extension", {"json"}}};
 	for (const auto &kv : info.options) {
 		const auto &loption = StringUtil::Lower(kv.first);
 		if (loption == "dateformat" || loption == "date_format") {
@@ -36,8 +37,6 @@ static BoundStatement CopyToJSONPlan(Binder &binder, CopyStatement &stmt) {
 				ThrowJSONCopyParameterException(loption);
 			}
 			timestamp_format = StringValue::Get(kv.second.back());
-		} else if (loption == "compression") {
-			csv_copy_options.insert(kv);
 		} else if (loption == "array") {
 			if (kv.second.size() > 1) {
 				ThrowJSONCopyParameterException(loption);
@@ -47,6 +46,11 @@ static BoundStatement CopyToJSONPlan(Binder &binder, CopyStatement &stmt) {
 				csv_copy_options["suffix"] = {"\n]\n"};
 				csv_copy_options["new_line"] = {",\n\t"};
 			}
+		} else if (loption == "compression" || loption == "encoding" || loption == "per_thread_output" ||
+		           loption == "file_size_bytes" || loption == "use_tmp_file" || loption == "overwrite_or_ignore" ||
+		           loption == "filename_pattern" || loption == "file_extension") {
+			// We support these base options
+			csv_copy_options.insert(kv);
 		} else {
 			throw BinderException("Unknown option for COPY ... TO ... (FORMAT JSON): \"%s\".", loption);
 		}

extension/json/json_scan.cpp

@@ -147,7 +147,8 @@ JSONScanGlobalState::JSONScanGlobalState(ClientContext &context, const JSONScanD
 JSONScanLocalState::JSONScanLocalState(ClientContext &context, JSONScanGlobalState &gstate)
     : scan_count(0), batch_index(DConstants::INVALID_INDEX), total_read_size(0), total_tuple_count(0),
       bind_data(gstate.bind_data), allocator(BufferAllocator::Get(context)), current_reader(nullptr),
-      current_buffer_handle(nullptr), is_last(false), buffer_size(0), buffer_offset(0), prev_buffer_remainder(0) {
+      current_buffer_handle(nullptr), is_last(false), fs(FileSystem::GetFileSystem(context)),
+      thread_local_filehandle(nullptr), buffer_size(0), buffer_offset(0), prev_buffer_remainder(0) {
 
 	// Buffer to reconstruct JSON values when they cross a buffer boundary
 	reconstruct_buffer = gstate.allocator.Allocate(gstate.buffer_capacity);
@@ -718,9 +719,21 @@ void JSONScanLocalState::ReadNextBufferSeek(JSONScanGlobalState &gstate, optiona
 		return;
 	}
 
+	auto &raw_handle = file_handle.GetHandle();
+	// For non-on-disk files, we create a handle per thread: this is faster for e.g. S3Filesystem where throttling
+	// per tcp connection can occur meaning that using multiple connections is faster.
+	if (!raw_handle.OnDiskFile() && raw_handle.CanSeek()) {
+		if (!thread_local_filehandle || thread_local_filehandle->GetPath() != raw_handle.GetPath()) {
+			thread_local_filehandle =
+			    fs.OpenFile(raw_handle.GetPath(), FileFlags::FILE_FLAGS_READ | FileFlags::FILE_FLAGS_DIRECT_IO);
+		}
+	} else if (thread_local_filehandle) {
+		thread_local_filehandle = nullptr;
+	}
+
 	// Now read the file lock-free!
 	file_handle.ReadAtPosition(buffer_ptr + prev_buffer_remainder, read_size, read_position,
-	                           gstate.bind_data.type == JSONScanType::SAMPLE);
+	                           gstate.bind_data.type == JSONScanType::SAMPLE, thread_local_filehandle);
 }
 
 void JSONScanLocalState::ReadNextBufferNoSeek(JSONScanGlobalState &gstate, optional_idx &buffer_index) {

extension/parquet/column_reader.cpp

@@ -20,7 +20,6 @@
 #ifndef DUCKDB_AMALGAMATION
 #include "duckdb/common/types/bit.hpp"
 #include "duckdb/common/types/blob.hpp"
-#include "duckdb/common/types/chunk_collection.hpp"
 #endif
 
 namespace duckdb {

extension/parquet/column_writer.cpp

@@ -13,7 +13,6 @@
 #include "duckdb/common/serializer/memory_stream.hpp"
 #include "duckdb/common/serializer/write_stream.hpp"
 #include "duckdb/common/string_map_set.hpp"
-#include "duckdb/common/types/chunk_collection.hpp"
 #include "duckdb/common/types/date.hpp"
 #include "duckdb/common/types/hugeint.hpp"
 #include "duckdb/common/types/uhugeint.hpp"

extension/parquet/include/column_reader.hpp

@@ -19,7 +19,6 @@
 #ifndef DUCKDB_AMALGAMATION
 
 #include "duckdb/common/operator/cast_operators.hpp"
-#include "duckdb/common/types/chunk_collection.hpp"
 #include "duckdb/common/types/string_type.hpp"
 #include "duckdb/common/types/vector.hpp"
 #include "duckdb/common/types/vector_cache.hpp"

extension/parquet/include/parquet_writer.hpp

@@ -87,6 +87,10 @@ class ParquetWriter {
 	BufferedFileWriter &GetWriter() {
 		return *writer;
 	}
+	idx_t FileSize() {
+		lock_guard<mutex> glock(lock);
+		return writer->total_written;
+	}
 
 	static CopyTypeSupport TypeIsSupported(const LogicalType &type);