Update vendored DuckDB sources to 9507d49

Author: duckdblabs-bot

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

@@ -132,6 +132,10 @@ struct ParquetUnionData : public BaseUnionData {
 };
 
 class ParquetReader : public BaseFileReader {
+public:
+	// Reserved field id used for the "ord" field according to the iceberg spec (used for file_row_number)
+	static constexpr int32_t ORDINAL_FIELD_ID = 2147483645;
+
 public:
 	ParquetReader(ClientContext &context, string file_name, ParquetOptions parquet_options,
 	              shared_ptr<ParquetFileMetadataCache> metadata = nullptr);
@@ -174,6 +178,8 @@ class ParquetReader : public BaseFileReader {
 		return "Parquet";
 	}
 
+	void AddVirtualColumn(column_t virtual_column_id) override;
+
 private:
 	//! Construct a parquet reader but **do not** open a file, used in ReadStatistics only
 	ParquetReader(ClientContext &context, ParquetOptions parquet_options,

src/duckdb/extension/parquet/parquet_extension.cpp

@@ -134,7 +134,6 @@ static void ParseFileRowNumberOption(MultiFileReaderBindData &bind_data, Parquet
 			    "Using file_row_number option on file with column named file_row_number is not supported");
 		}
 
-		bind_data.file_row_number_idx = names.size();
 		return_types.emplace_back(LogicalType::BIGINT);
 		names.emplace_back("file_row_number");
 	}
@@ -186,6 +185,14 @@ static void BindSchema(ClientContext &context, vector<LogicalType> &return_types
 		res.default_expression = make_uniq<ConstantExpression>(column.default_value);
 		reader_bind.schema.emplace_back(std::move(res));
 	}
+	ParseFileRowNumberOption(reader_bind, options, return_types, names);
+	if (options.file_row_number) {
+		MultiFileColumnDefinition res("file_row_number", LogicalType::BIGINT);
+		res.identifier = Value::INTEGER(ParquetReader::ORDINAL_FIELD_ID);
+		schema_col_names.push_back(res.name);
+		schema_col_types.push_back(res.type);
+		reader_bind.schema.emplace_back(std::move(res));
+	}
 
 	if (match_by_field_id) {
 		reader_bind.mapping = MultiFileColumnMappingMode::BY_FIELD_ID;
@@ -200,8 +207,6 @@ static void BindSchema(ClientContext &context, vector<LogicalType> &return_types
 	names = schema_col_names;
 	return_types = schema_col_types;
 	D_ASSERT(names.size() == return_types.size());
-
-	ParseFileRowNumberOption(reader_bind, options, return_types, names);
 }
 
 void ParquetMultiFileInfo::BindReader(ClientContext &context, vector<LogicalType> &return_types, vector<string> &names,
@@ -465,10 +470,6 @@ optional_idx ParquetMultiFileInfo::MaxThreads(const MultiFileBindData &bind_data
 
 void ParquetMultiFileInfo::FinalizeBindData(MultiFileBindData &multi_file_data) {
 	auto &bind_data = multi_file_data.bind_data->Cast<ParquetReadBindData>();
-	// Enable the parquet file_row_number on the parquet options if the file_row_number_idx was set
-	if (multi_file_data.reader_bind.file_row_number_idx != DConstants::INVALID_INDEX) {
-		bind_data.parquet_options.file_row_number = true;
-	}
 	if (multi_file_data.initial_reader) {
 		auto &initial_reader = multi_file_data.initial_reader->Cast<ParquetReader>();
 		bind_data.initial_file_cardinality = initial_reader.NumRows();
@@ -498,7 +499,9 @@ double ParquetMultiFileInfo::GetProgressInFile(ClientContext &context, const Bas
 	return 100.0 * (static_cast<double>(read_rows) / static_cast<double>(parquet_reader.NumRows()));
 }
 
-void ParquetMultiFileInfo::GetVirtualColumns(ClientContext &, MultiFileBindData &, virtual_column_map_t &) {
+void ParquetMultiFileInfo::GetVirtualColumns(ClientContext &, MultiFileBindData &, virtual_column_map_t &result) {
+	result.insert(make_pair(MultiFileReader::COLUMN_IDENTIFIER_FILE_ROW_NUMBER,
+	                        TableColumn("file_row_number", LogicalType::BIGINT)));
 }
 
 shared_ptr<BaseFileReader> ParquetMultiFileInfo::CreateReader(ClientContext &context, GlobalTableFunctionState &,

src/duckdb/extension/parquet/parquet_reader.cpp

@@ -27,6 +27,7 @@
 #include "duckdb/storage/object_cache.hpp"
 #include "duckdb/optimizer/statistics_propagator.hpp"
 #include "duckdb/planner/table_filter_state.hpp"
+#include "duckdb/common/multi_file/multi_file_reader.hpp"
 
 #include <cassert>
 #include <chrono>
@@ -35,6 +36,8 @@
 
 namespace duckdb {
 
+constexpr int32_t ParquetReader::ORDINAL_FIELD_ID;
+
 using duckdb_parquet::ColumnChunk;
 using duckdb_parquet::ConvertedType;
 using duckdb_parquet::FieldRepetitionType;
@@ -573,6 +576,11 @@ ParquetColumnSchema ParquetReader::ParseSchemaRecursive(idx_t depth, idx_t max_d
 	}
 }
 
+ParquetColumnSchema FileRowNumberSchema() {
+	return ParquetColumnSchema("file_row_number", LogicalType::BIGINT, 0, 0, 0, 0,
+	                           ParquetColumnSchemaType::FILE_ROW_NUMBER);
+}
+
 unique_ptr<ParquetColumnSchema> ParquetReader::ParseSchema() {
 	auto file_meta_data = GetFileMetadata();
 	idx_t next_schema_idx = 0;
@@ -598,16 +606,18 @@ unique_ptr<ParquetColumnSchema> ParquetReader::ParseSchema() {
 				    "Using file_row_number option on file with column named file_row_number is not supported");
 			}
 		}
-		ParquetColumnSchema file_row_number("file_row_number", LogicalType::BIGINT, 0, 0, 0, 0,
-		                                    ParquetColumnSchemaType::FILE_ROW_NUMBER);
-		root.children.push_back(std::move(file_row_number));
+		root.children.push_back(FileRowNumberSchema());
 	}
 	return make_uniq<ParquetColumnSchema>(root);
 }
 
 MultiFileColumnDefinition ParquetReader::ParseColumnDefinition(const FileMetaData &file_meta_data,
                                                                ParquetColumnSchema &element) {
 	MultiFileColumnDefinition result(element.name, element.type);
+	if (element.schema_type == ParquetColumnSchemaType::FILE_ROW_NUMBER) {
+		result.identifier = Value::INTEGER(ORDINAL_FIELD_ID);
+		return result;
+	}
 	auto &column_schema = file_meta_data.schema[element.schema_index];
 
 	if (column_schema.__isset.field_id) {
@@ -645,6 +655,14 @@ void ParquetReader::InitializeSchema(ClientContext &context) {
 	}
 }
 
+void ParquetReader::AddVirtualColumn(column_t virtual_column_id) {
+	if (virtual_column_id == MultiFileReader::COLUMN_IDENTIFIER_FILE_ROW_NUMBER) {
+		root_schema->children.push_back(FileRowNumberSchema());
+	} else {
+		throw InternalException("Unsupported virtual column id %d for parquet reader", virtual_column_id);
+	}
+}
+
 ParquetOptions::ParquetOptions(ClientContext &context) {
 	Value binary_as_string_val;
 	if (context.TryGetCurrentSetting("binary_as_string", binary_as_string_val)) {

src/duckdb/src/common/multi_file/multi_file_column_mapper.cpp

@@ -365,6 +365,14 @@ unique_ptr<Expression> ConstructMapExpression(ClientContext &context, idx_t loca
 	                                          std::move(bind_data));
 }
 
+bool VirtualColumnIsConstant(column_t column_id) {
+	if (column_id == COLUMN_IDENTIFIER_EMPTY ||
+	    COLUMN_IDENTIFIER_EMPTY == MultiFileReader::COLUMN_IDENTIFIER_FILENAME) {
+		return true;
+	}
+	return false;
+}
+
 ResultColumnMapping MultiFileColumnMapper::CreateColumnMappingByMapper(const ColumnMapper &mapper) {
 	auto &reader = *reader_data.reader;
 	auto &local_columns = reader.GetColumns();
@@ -397,28 +405,36 @@ ResultColumnMapping MultiFileColumnMapper::CreateColumnMappingByMapper(const Col
 		auto &global_id = global_column_ids[i];
 		auto global_column_id = global_id.GetPrimaryIndex();
 
+		auto local_idx = MultiFileLocalIndex(reader.column_ids.size());
 		if (IsVirtualColumn(global_column_id)) {
-			// virtual column - these are emitted for every file
+			// virtual column - look it up in the virtual column entry map
 			auto virtual_entry = virtual_columns.find(global_column_id);
 			if (virtual_entry == virtual_columns.end()) {
 				throw InternalException("Virtual column id %d not found in virtual columns map", global_column_id);
 			}
-			expressions.push_back(make_uniq<BoundConstantExpression>(Value(virtual_entry->second.type)));
-			continue;
-		}
-
-		auto local_idx = MultiFileLocalIndex(reader.column_ids.size());
-		if (global_column_id >= global_columns.size()) {
-			if (bind_data.file_row_number_idx == global_column_id) {
-				// FIXME: this needs a more extensible solution
-				auto new_column_id = MultiFileLocalColumnId(mapper.MapCount());
-				reader.column_ids.push_back(new_column_id);
-				reader.column_indexes.emplace_back(mapper.MapCount());
-				//! FIXME: what to do here???
-				expressions.push_back(make_uniq<BoundReferenceExpression>(LogicalType::BIGINT, local_idx));
-			} else {
-				throw InternalException("Unexpected generated column");
+			auto &virtual_column_type = virtual_entry->second.type;
+			// check if this column is constant for the file
+			if (VirtualColumnIsConstant(global_column_id)) {
+				// the column is constant (e.g. filename or empty) - no need to project from the file
+				expressions.push_back(make_uniq<BoundConstantExpression>(Value(virtual_column_type)));
+				continue;
 			}
+			// the column is not constant for the file - need to push into the projection list
+			auto expr = make_uniq<BoundReferenceExpression>(virtual_column_type, local_idx.GetIndex());
+			reader_data.expressions.push_back(std::move(expr));
+
+			MultiFileLocalColumnId local_id(reader.columns.size());
+			ColumnIndex local_index(local_id.GetId());
+
+			// add the virtual column to the reader
+			reader.columns.emplace_back(virtual_entry->second.name, virtual_column_type);
+			reader.AddVirtualColumn(global_column_id);
+
+			// set it as being projected in this spot
+			MultiFileColumnMap index_mapping(local_idx, virtual_column_type, virtual_column_type);
+			result.global_to_local.insert(make_pair(global_idx.GetIndex(), std::move(index_mapping)));
+			reader.column_ids.push_back(local_id);
+			reader.column_indexes.push_back(std::move(local_index));
 			continue;
 		}
 
@@ -433,8 +449,7 @@ ResultColumnMapping MultiFileColumnMapper::CreateColumnMappingByMapper(const Col
 			ColumnIndex local_index(local_id.GetId());
 			auto &local_type = local_columns[local_id.GetId()].type;
 			auto &global_type = global_column.type;
-			auto local_idx = reader.column_ids.size();
-			auto expr = make_uniq<BoundReferenceExpression>(global_type, local_idx);
+			auto expr = make_uniq<BoundReferenceExpression>(global_type, local_idx.GetIndex());
 			if (global_type != local_type) {
 				reader.cast_map[local_id.GetId()] = global_type;
 			} else {

src/duckdb/src/common/multi_file/multi_file_reader.cpp

@@ -16,6 +16,7 @@
 namespace duckdb {
 
 constexpr column_t MultiFileReader::COLUMN_IDENTIFIER_FILENAME;
+constexpr column_t MultiFileReader::COLUMN_IDENTIFIER_FILE_ROW_NUMBER;
 
 MultiFileReaderGlobalState::~MultiFileReaderGlobalState() {
 }

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

@@ -1,5 +1,5 @@
 #ifndef DUCKDB_PATCH_VERSION
-#define DUCKDB_PATCH_VERSION "0-dev2195"
+#define DUCKDB_PATCH_VERSION "0-dev2212"
 #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-dev2195"
+#define DUCKDB_VERSION "v1.3.0-dev2212"
 #endif
 #ifndef DUCKDB_SOURCE_ID
-#define DUCKDB_SOURCE_ID "cf02bffeb0"
+#define DUCKDB_SOURCE_ID "946a2bd1bf"
 #endif
 #include "duckdb/function/table/system_functions.hpp"
 #include "duckdb/main/database.hpp"

src/duckdb/src/include/duckdb/common/multi_file/base_file_reader.hpp

@@ -56,6 +56,10 @@ class BaseFileReader {
 		//! Whether or not to push casts into the cast map
 		return false;
 	}
+	//! Adds a virtual column to be projected at the end
+	virtual void AddVirtualColumn(column_t virtual_column_id) {
+		throw InternalException("Reader %s does not support AddVirtualColumn", GetReaderType());
+	}
 
 public:
 	template <class TARGET>

src/duckdb/src/include/duckdb/common/multi_file/multi_file_reader.hpp

@@ -33,6 +33,7 @@ enum class ReaderInitializeType { INITIALIZED, SKIP_READING_FILE };
 struct MultiFileReader {
 public:
 	static constexpr column_t COLUMN_IDENTIFIER_FILENAME = UINT64_C(9223372036854775808);
+	static constexpr column_t COLUMN_IDENTIFIER_FILE_ROW_NUMBER = UINT64_C(9223372036854775809);
 
 public:
 	virtual ~MultiFileReader();

src/duckdb/src/include/duckdb/common/multi_file/multi_file_states.hpp

@@ -21,8 +21,6 @@ struct MultiFileReaderBindData {
 	column_t filename_idx = DConstants::INVALID_INDEX;
 	//! The set of hive partitioning indexes (if any)
 	vector<HivePartitioningIndex> hive_partitioning_indexes;
-	//! The (global) column id of the file_row_number column (if any)
-	column_t file_row_number_idx = DConstants::INVALID_INDEX;
 	//! (optional) The schema set by the multi file reader
 	vector<MultiFileColumnDefinition> schema;
 	//! The method used to map local -> global columns

src/duckdb/src/include/duckdb/main/config.hpp

@@ -290,6 +290,12 @@ struct DBConfigOptions {
 	set<string> allowed_directories;
 	//! The log configuration
 	LogConfig log_config = LogConfig();
+	//! Partially process tasks before rescheduling - allows for more scheduler fairness between separate queries
+#ifdef DUCKDB_ALTERNATIVE_VERIFY
+	bool scheduler_process_partial = true;
+#else
+	bool scheduler_process_partial = false;
+#endif
 
 	bool operator==(const DBConfigOptions &other) const;
 };

src/duckdb/src/include/duckdb/main/settings.hpp

@@ -1122,6 +1122,17 @@ struct ScalarSubqueryErrorOnMultipleRowsSetting {
 	static Value GetSetting(const ClientContext &context);
 };
 
+struct SchedulerProcessPartialSetting {
+	using RETURN_TYPE = bool;
+	static constexpr const char *Name = "scheduler_process_partial";
+	static constexpr const char *Description =
+	    "Partially process tasks before rescheduling - allows for more scheduler fairness between separate queries";
+	static constexpr const char *InputType = "BOOLEAN";
+	static void SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &parameter);
+	static void ResetGlobal(DatabaseInstance *db, DBConfig &config);
+	static Value GetSetting(const ClientContext &context);
+};
+
 struct SchemaSetting {
 	using RETURN_TYPE = string;
 	static constexpr const char *Name = "schema";

src/duckdb/src/include/duckdb/parallel/task.hpp

@@ -9,6 +9,7 @@
 #pragma once
 
 #include "duckdb/common/common.hpp"
+#include "duckdb/common/optional_ptr.hpp"
 
 namespace duckdb {
 class ClientContext;
@@ -50,6 +51,9 @@ class Task : public enable_shared_from_this<Task> {
 	virtual bool TaskBlockedOnResult() const {
 		return false;
 	}
+
+public:
+	optional_ptr<ProducerToken> token;
 };
 
 } // namespace duckdb

src/duckdb/src/main/config.cpp

@@ -161,6 +161,7 @@ static const ConfigurationOption internal_options[] = {
     DUCKDB_LOCAL(ProfilingModeSetting),
     DUCKDB_LOCAL(ProgressBarTimeSetting),
     DUCKDB_LOCAL(ScalarSubqueryErrorOnMultipleRowsSetting),
+    DUCKDB_GLOBAL(SchedulerProcessPartialSetting),
     DUCKDB_LOCAL(SchemaSetting),
     DUCKDB_LOCAL(SearchPathSetting),
     DUCKDB_GLOBAL(SecretDirectorySetting),

src/duckdb/src/main/settings/autogenerated_settings.cpp

@@ -1117,6 +1117,22 @@ Value ScalarSubqueryErrorOnMultipleRowsSetting::GetSetting(const ClientContext &
 	return Value::BOOLEAN(config.scalar_subquery_error_on_multiple_rows);
 }
 
+//===----------------------------------------------------------------------===//
+// Scheduler Process Partial
+//===----------------------------------------------------------------------===//
+void SchedulerProcessPartialSetting::SetGlobal(DatabaseInstance *db, DBConfig &config, const Value &input) {
+	config.options.scheduler_process_partial = input.GetValue<bool>();
+}
+
+void SchedulerProcessPartialSetting::ResetGlobal(DatabaseInstance *db, DBConfig &config) {
+	config.options.scheduler_process_partial = DBConfig().options.scheduler_process_partial;
+}
+
+Value SchedulerProcessPartialSetting::GetSetting(const ClientContext &context) {
+	auto &config = DBConfig::GetConfig(context);
+	return Value::BOOLEAN(config.options.scheduler_process_partial);
+}
+
 //===----------------------------------------------------------------------===//
 // Zstd Min String Length
 //===----------------------------------------------------------------------===//

src/duckdb/src/parallel/task_executor.cpp

@@ -63,8 +63,6 @@ BaseExecutorTask::BaseExecutorTask(TaskExecutor &executor) : executor(executor)
 }
 
 TaskExecutionResult BaseExecutorTask::Execute(TaskExecutionMode mode) {
-	(void)mode;
-	D_ASSERT(mode == TaskExecutionMode::PROCESS_ALL);
 	if (executor.HasError()) {
 		// another task encountered an error - bailout
 		executor.FinishTask();