Update vendored DuckDB sources to 971fd3f

Author: duckdblabs-bot

src/duckdb/extension/parquet/decoder/dictionary_decoder.cpp

@@ -186,6 +186,11 @@ void DictionaryDecoder::Filter(uint8_t *defines, const idx_t read_count, Vector
 	D_ASSERT(filter_count > 0);
 	// read the dictionary values
 	const auto valid_count = Read(defines, read_count, result, 0);
+	if (valid_count == 0) {
+		// all values are NULL
+		approved_tuple_count = 0;
+		return;
+	}
 
 	// apply the filter by checking the dictionary offsets directly
 	uint32_t *offsets;

src/duckdb/extension/parquet/include/parquet_writer.hpp

@@ -13,6 +13,7 @@
 #include "duckdb/common/encryption_state.hpp"
 #include "duckdb/common/exception.hpp"
 #include "duckdb/common/mutex.hpp"
+#include "duckdb/common/atomic.hpp"
 #include "duckdb/common/serializer/buffered_file_writer.hpp"
 #include "duckdb/common/types/column/column_data_collection.hpp"
 #include "duckdb/function/copy_function.hpp"
@@ -113,8 +114,7 @@ class ParquetWriter {
 		return *writer;
 	}
 	idx_t FileSize() {
-		lock_guard<mutex> glock(lock);
-		return writer->total_written;
+		return total_written;
 	}
 	idx_t DictionarySizeLimit() const {
 		return dictionary_size_limit;
@@ -129,8 +129,7 @@ class ParquetWriter {
 		return compression_level;
 	}
 	idx_t NumberOfRowGroups() {
-		lock_guard<mutex> glock(lock);
-		return file_meta_data.row_groups.size();
+		return num_row_groups;
 	}
 	ParquetVersion GetParquetVersion() const {
 		return parquet_version;
@@ -170,6 +169,9 @@ class ParquetWriter {
 	vector<ParquetColumnSchema> column_schemas;
 
 	unique_ptr<BufferedFileWriter> writer;
+	//! Atomics to reduce contention when rotating writes to multiple Parquet files
+	atomic<idx_t> total_written;
+	atomic<idx_t> num_row_groups;
 	std::shared_ptr<duckdb_apache::thrift::protocol::TProtocol> protocol;
 	duckdb_parquet::FileMetaData file_meta_data;
 	std::mutex lock;

src/duckdb/extension/parquet/parquet_writer.cpp

@@ -352,11 +352,12 @@ ParquetWriter::ParquetWriter(ClientContext &context, FileSystem &fs, string file
       encryption_config(std::move(encryption_config_p)), dictionary_size_limit(dictionary_size_limit_p),
       string_dictionary_page_size_limit(string_dictionary_page_size_limit_p),
       bloom_filter_false_positive_ratio(bloom_filter_false_positive_ratio_p), compression_level(compression_level_p),
-      debug_use_openssl(debug_use_openssl_p), parquet_version(parquet_version) {
+      debug_use_openssl(debug_use_openssl_p), parquet_version(parquet_version), total_written(0), num_row_groups(0) {
 
 	// initialize the file writer
 	writer = make_uniq<BufferedFileWriter>(fs, file_name.c_str(),
 	                                       FileFlags::FILE_FLAGS_WRITE | FileFlags::FILE_FLAGS_FILE_CREATE_NEW);
+
 	if (encryption_config) {
 		auto &config = DBConfig::GetConfig(context);
 		if (config.encryption_util && debug_use_openssl) {
@@ -373,6 +374,7 @@ ParquetWriter::ParquetWriter(ClientContext &context, FileSystem &fs, string file
 		// parquet files start with the string "PAR1"
 		writer->WriteData(const_data_ptr_cast("PAR1"), 4);
 	}
+
 	TCompactProtocolFactoryT<MyTransport> tproto_factory;
 	protocol = tproto_factory.getProtocol(std::make_shared<MyTransport>(*writer));
 
@@ -542,6 +544,9 @@ void ParquetWriter::FlushRowGroup(PreparedRowGroup &prepared) {
 	// append the row group to the file meta data
 	file_meta_data.row_groups.push_back(row_group);
 	file_meta_data.num_rows += row_group.num_rows;
+
+	total_written = writer->GetTotalWritten();
+	num_row_groups++;
 }
 
 void ParquetWriter::Flush(ColumnDataCollection &buffer) {

src/duckdb/extension/parquet/reader/list_column_reader.cpp

@@ -184,6 +184,7 @@ ListColumnReader::ListColumnReader(ParquetReader &reader, const ParquetColumnSch
 
 void ListColumnReader::ApplyPendingSkips(data_ptr_t define_out, data_ptr_t repeat_out) {
 	ReadInternal<TemplatedListSkipper>(pending_skips, nullptr, nullptr, nullptr);
+	pending_skips = 0;
 }
 
 } // namespace duckdb

src/duckdb/src/execution/operator/csv_scanner/scanner/string_value_scanner.cpp

@@ -1894,6 +1894,7 @@ void StringValueScanner::FinalizeChunkProcess() {
 		}
 		if (states.IsQuotedCurrent() && !found_error &&
 		    state_machine->dialect_options.state_machine_options.strict_mode.GetValue()) {
+			type = UNTERMINATED_QUOTES;
 			// If we finish the execution of a buffer, and we end in a quoted state, it means we have unterminated
 			// quotes
 			result.current_errors.Insert(type, result.cur_col_id, result.chunk_col_id, result.last_position);

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

@@ -391,7 +391,8 @@ void CSVSniffer::AnalyzeDialectCandidate(unique_ptr<ColumnCountScanner> scanner,
 				return;
 			} else {
 				// Give preference to one that got escaped
-				if (!scanner->ever_escaped && candidates.front()->ever_escaped) {
+				if (!scanner->ever_escaped && candidates.front()->ever_escaped &&
+				    sniffing_state_machine.dialect_options.state_machine_options.strict_mode.GetValue()) {
 					return;
 				}
 				if (best_consistent_rows == consistent_rows && num_cols >= max_columns_found) {
@@ -413,9 +414,19 @@ void CSVSniffer::AnalyzeDialectCandidate(unique_ptr<ColumnCountScanner> scanner,
 			return;
 		}
 		if (quoted && num_cols < max_columns_found) {
-			for (auto &candidate : candidates) {
-				if (candidate->ever_quoted) {
-					return;
+			if (scanner->ever_escaped &&
+			    sniffing_state_machine.dialect_options.state_machine_options.strict_mode.GetValue()) {
+				for (auto &candidate : candidates) {
+					if (candidate->ever_quoted && candidate->ever_escaped) {
+						return;
+					}
+				}
+
+			} else {
+				for (auto &candidate : candidates) {
+					if (candidate->ever_quoted) {
+						return;
+					}
 				}
 			}
 		}
@@ -434,7 +445,6 @@ void CSVSniffer::AnalyzeDialectCandidate(unique_ptr<ColumnCountScanner> scanner,
 		} else if (!options.null_padding) {
 			sniffing_state_machine.dialect_options.skip_rows = dirty_notes;
 		}
-
 		candidates.clear();
 		sniffing_state_machine.dialect_options.num_cols = num_cols;
 		lines_sniffed = sniffed_column_counts.result_position;

src/duckdb/src/execution/operator/persistent/physical_copy_to_file.cpp

@@ -48,7 +48,8 @@ using vector_of_value_map_t = unordered_map<vector<Value>, T, VectorOfValuesHash
 class CopyToFunctionGlobalState : public GlobalSinkState {
 public:
 	explicit CopyToFunctionGlobalState(ClientContext &context)
-	    : initialized(false), rows_copied(0), last_file_offset(0) {
+	    : initialized(false), rows_copied(0), last_file_offset(0),
+	      file_write_lock_if_rotating(make_uniq<StorageLock>()) {
 		max_open_files = ClientConfig::GetConfig(context).partitioned_write_max_open_files;
 	}
 	StorageLock lock;
@@ -64,6 +65,8 @@ class CopyToFunctionGlobalState : public GlobalSinkState {
 	vector<unique_ptr<CopyToFileInfo>> written_files;
 	//! Max open files
 	idx_t max_open_files;
+	//! If rotate is true, this lock is used
+	unique_ptr<StorageLock> file_write_lock_if_rotating;
 
 	void Initialize(ClientContext &context, const PhysicalCopyToFile &op) {
 		if (initialized) {
@@ -472,6 +475,43 @@ PhysicalCopyToFile::PhysicalCopyToFile(vector<LogicalType> types, CopyFunction f
       function(std::move(function_p)), bind_data(std::move(bind_data)), parallel(false) {
 }
 
+void PhysicalCopyToFile::WriteRotateInternal(ExecutionContext &context, GlobalSinkState &global_state,
+                                             const std::function<void(GlobalFunctionData &)> &fun) const {
+	auto &g = global_state.Cast<CopyToFunctionGlobalState>();
+
+	// Loop until we can write (synchronize using locks when using parallel writes to the same files and "rotate")
+	while (true) {
+		// Grab global lock and dereference the current file state (and corresponding lock)
+		auto global_guard = g.lock.GetExclusiveLock();
+		auto &file_state = *g.global_state;
+		auto &file_lock = *g.file_write_lock_if_rotating;
+		if (rotate && function.rotate_next_file(file_state, *bind_data, file_size_bytes)) {
+			// Global state must be rotated. Move to local scope, create an new one, and immediately release global lock
+			auto owned_gstate = std::move(g.global_state);
+			g.global_state = CreateFileState(context.client, *sink_state, *global_guard);
+			auto owned_lock = std::move(g.file_write_lock_if_rotating);
+			g.file_write_lock_if_rotating = make_uniq<StorageLock>();
+			global_guard.reset();
+
+			// This thread now waits for the exclusive lock on this file while other threads complete their writes
+			// Note that new writes can still start, as there is already a new global state
+			auto file_guard = owned_lock->GetExclusiveLock();
+			function.copy_to_finalize(context.client, *bind_data, *owned_gstate);
+		} else {
+			// Get shared file write lock while holding global lock,
+			// so file can't be rotated before we get the write lock
+			auto file_guard = file_lock.GetSharedLock();
+
+			// Because we got the shared lock on the file, we're sure that it will keep existing until we release it
+			global_guard.reset();
+
+			// Sink/Combine!
+			fun(file_state);
+			break;
+		}
+	}
+}
+
 SinkResultType PhysicalCopyToFile::Sink(ExecutionContext &context, DataChunk &chunk, OperatorSinkInput &input) const {
 	auto &g = input.global_state.Cast<CopyToFunctionGlobalState>();
 	auto &l = input.local_state.Cast<CopyToFunctionLocalState>();
@@ -507,20 +547,9 @@ SinkResultType PhysicalCopyToFile::Sink(ExecutionContext &context, DataChunk &ch
 		return SinkResultType::NEED_MORE_INPUT;
 	}
 
-	// FILE_SIZE_BYTES/rotate is set, but threads write to the same file, synchronize using lock
-	auto &gstate = g.global_state;
-	auto global_lock = g.lock.GetExclusiveLock();
-	if (rotate && function.rotate_next_file(*gstate, *bind_data, file_size_bytes)) {
-		auto owned_gstate = std::move(gstate);
-		gstate = CreateFileState(context.client, *sink_state, *global_lock);
-		global_lock.reset();
-		function.copy_to_finalize(context.client, *bind_data, *owned_gstate);
-	} else {
-		global_lock.reset();
-	}
-
-	global_lock = g.lock.GetSharedLock();
-	function.copy_to_sink(context, *bind_data, *gstate, *l.local_state, chunk);
+	WriteRotateInternal(context, input.global_state, [&](GlobalFunctionData &gstate) {
+		function.copy_to_sink(context, *bind_data, gstate, *l.local_state, chunk);
+	});
 
 	return SinkResultType::NEED_MORE_INPUT;
 }
@@ -540,9 +569,9 @@ SinkCombineResultType PhysicalCopyToFile::Combine(ExecutionContext &context, Ope
 				function.copy_to_finalize(context.client, *bind_data, *l.global_state);
 			}
 		} else if (rotate) {
-			// 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);
+			WriteRotateInternal(context, input.global_state, [&](GlobalFunctionData &gstate) {
+				function.copy_to_combine(context, *bind_data, gstate, *l.local_state);
+			});
 		} else if (g.global_state) {
 			function.copy_to_combine(context, *bind_data, *g.global_state, *l.local_state);
 		}

src/duckdb/src/execution/operator/projection/physical_unnest.cpp

@@ -186,7 +186,8 @@ OperatorResultType PhysicalUnnest::ExecuteInternal(ExecutionContext &context, Da
 						list_offset = list_entry.offset;
 					}
 					// unnest any entries we can
-					idx_t unnest_length = MinValue<idx_t>(list_length - state.list_position, current_row_length);
+					idx_t unnest_length = MinValue<idx_t>(
+					    list_length - MinValue<idx_t>(list_length, state.list_position), current_row_length);
 					auto &unnest_sel = state.unnest_sels[col_idx];
 					for (idx_t r = 0; r < unnest_length; r++) {
 						unnest_sel.set_index(result_length + r, list_offset + state.list_position + r);
@@ -228,14 +229,12 @@ OperatorResultType PhysicalUnnest::ExecuteInternal(ExecutionContext &context, Da
 			col_offset = input.ColumnCount();
 		}
 		for (idx_t col_idx = 0; col_idx < state.list_data.ColumnCount(); col_idx++) {
-			if (state.list_data.data[col_idx].GetType() == LogicalType::SQLNULL) {
-				// UNNEST(NULL)
-				chunk.SetCardinality(0);
-				break;
-			}
 			auto &list_vector = state.list_data.data[col_idx];
 			auto &result_vector = chunk.data[col_offset + col_idx];
-			if (ListVector::GetListSize(list_vector) == 0) {
+			if (state.list_data.data[col_idx].GetType() == LogicalType::SQLNULL ||
+			    ListType::GetChildType(state.list_data.data[col_idx].GetType()) == LogicalType::SQLNULL ||
+			    ListVector::GetListSize(list_vector) == 0) {
+				// UNNEST(NULL) or UNNEST([])
 				// we cannot slice empty lists - but if our child list is empty we can only return NULL anyway
 				result_vector.SetVectorType(VectorType::CONSTANT_VECTOR);
 				ConstantVector::SetNull(result_vector, true);

src/duckdb/src/function/scalar/string/concat.cpp

@@ -335,9 +335,9 @@ static unique_ptr<BaseStatistics> ListConcatStats(ClientContext &context, Functi
 
 ScalarFunction ListConcatFun::GetFunction() {
 	// The arguments and return types are set in the binder function.
-	auto fun = ScalarFunction({LogicalType::LIST(LogicalType::ANY), LogicalType::LIST(LogicalType::ANY)},
-	                          LogicalType::LIST(LogicalType::ANY), ConcatFunction, BindConcatFunction, nullptr,
+	auto fun = ScalarFunction({}, LogicalType::LIST(LogicalType::ANY), ConcatFunction, BindConcatFunction, nullptr,
 	                          ListConcatStats);
+	fun.varargs = LogicalType::LIST(LogicalType::ANY);
 	fun.null_handling = FunctionNullHandling::SPECIAL_HANDLING;
 	return fun;
 }

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

@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "0-dev2049"
+#define DUCKDB_PATCH_VERSION "0-dev2110"
 #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-dev2049"
+#define DUCKDB_VERSION "v1.3.0-dev2110"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "daf2fa3674"
+#define DUCKDB_SOURCE_ID "7912713493"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"