feat(vortex): read DuckLake data files in the vortex format (per-file dispatch)

.gitignore

@@ -9,3 +9,4 @@ test/python/__pycache__/
 .Rhistory
 scripts/data/
 duckdb_benchmark_data/
+.cache/

.gitmodules

@@ -5,4 +5,7 @@
 [submodule "extension-ci-tools"]
 	path = extension-ci-tools
 	url = https://github.com/duckdb/extension-ci-tools
-	branch = main
+	branch = main
+[submodule "vortex"]
+	path = vortex
+	url = https://github.com/vortex-data/duckdb-vortex.git

extension_config.cmake

@@ -11,6 +11,14 @@ if(NOT DEFINED ENV{DISABLE_EXTENSIONS_FOR_TEST})
     duckdb_extension_load(tpch)
 endif()
 
+# Vortex format support (./vortex submodule). Opt-in: needs a Rust toolchain and DuckDB pinned to
+# vortex's fork (see .gitmodules), so it is off in the standard distribution CI. Enable with ENABLE_VORTEX=1.
+if(DEFINED ENV{ENABLE_VORTEX})
+    duckdb_extension_load(vortex
+            SOURCE_DIR ${CMAKE_CURRENT_LIST_DIR}/vortex
+    )
+endif()
+
 set(EXTENSION_CONFIG_BASE_DIR "${CMAKE_CURRENT_LIST_DIR}/.github/config/extensions/")
 if($ENV{ENABLE_SQLITE_SCANNER})
     include("${EXTENSION_CONFIG_BASE_DIR}/sqlite_scanner.cmake")

src/common/ducklake_data_file.cpp

@@ -25,6 +25,7 @@ DeleteFileFormat DeleteFileFormatFromString(const string &str) {
 
 DuckLakeDataFile::DuckLakeDataFile(const DuckLakeDataFile &other) {
 	file_name = other.file_name;
+	file_format = other.file_format;
 	row_count = other.row_count;
 	file_size_bytes = other.file_size_bytes;
 	footer_size = other.footer_size;
@@ -42,6 +43,7 @@ DuckLakeDataFile::DuckLakeDataFile(const DuckLakeDataFile &other) {
 
 DuckLakeDataFile &DuckLakeDataFile::operator=(const DuckLakeDataFile &other) {
 	file_name = other.file_name;
+	file_format = other.file_format;
 	row_count = other.row_count;
 	file_size_bytes = other.file_size_bytes;
 	footer_size = other.footer_size;

src/functions/ducklake_flush_inlined_data.cpp

@@ -739,7 +739,7 @@ static unique_ptr<LogicalOperator> FlushInlinedDataBind(ClientContext &context,
 	auto projection = make_uniq<LogicalProjection>(bind_index, std::move(proj_exprs));
 	projection->children.push_back(std::move(filter));
 
-	return projection;
+	return unique_ptr<LogicalOperator>(projection.release());
 }
 
 DuckLakeFlushInlinedDataFunction::DuckLakeFlushInlinedDataFunction()

src/functions/ducklake_set_option.cpp

@@ -3,6 +3,7 @@
 #include "storage/ducklake_catalog.hpp"
 #include "storage/ducklake_table_entry.hpp"
 #include "storage/ducklake_schema_entry.hpp"
+#include "storage/ducklake_insert.hpp"
 
 namespace duckdb {
 
@@ -98,6 +99,9 @@ static unique_ptr<FunctionData> DuckLakeSetOptionBind(ClientContext &context, Ta
 		value = val.CastAs(context, LogicalType::BOOLEAN).GetValue<bool>() ? "true" : "false";
 	} else if (option == "write_deletion_vectors") {
 		value = val.CastAs(context, LogicalType::BOOLEAN).GetValue<bool>() ? "true" : "false";
+	} else if (option == "data_file_format") {
+		// Physical format new data files are written in: 'parquet' (default) or 'vortex'.
+		value = DuckLakeInsert::ValidateDataFileFormat(val.DefaultCastAs(LogicalType::VARCHAR).GetValue<string>());
 	} else {
 		throw NotImplementedException("Unsupported option %s", option);
 	}

src/include/common/ducklake_data_file.hpp

@@ -62,6 +62,8 @@ struct DuckLakeDataFile {
 	DuckLakeDataFile &operator=(const DuckLakeDataFile &);
 
 	string file_name;
+	//! physical format ("parquet", "vortex", ...); recorded in ducklake_data_file.file_format
+	string file_format = "parquet";
 	idx_t row_count;
 	idx_t file_size_bytes;
 	optional_idx footer_size;

src/include/storage/ducklake_format_reader.hpp

@@ -0,0 +1,67 @@
+//===----------------------------------------------------------------------===//
+//                         DuckDB
+//
+// storage/ducklake_format_reader.hpp
+//
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "duckdb/common/multi_file/base_file_reader.hpp"
+#include "duckdb/function/table_function.hpp"
+#include "duckdb/parallel/thread_context.hpp"
+
+namespace duckdb {
+struct DuckLakeFunctionInfo;
+
+//! How an output column is produced.
+enum class FormatReaderColumn : uint8_t {
+	FILE_COLUMN, //! read from the file (see output_source_index)
+	ORDINAL,     //! file_row_number, emitted by the reader
+	MISSING      //! not in the file -> NULL
+};
+
+//! Reads a single non-parquet DuckLake data file ("vortex", "json", ...) by driving its read_<format>
+//! table function and adapting the chunks to DuckLake's MultiFileReader contract (column mapping,
+//! delete filters, pushed-down filters, file_row_number), like the parquet reader path does.
+class DuckLakeFormatReader : public BaseFileReader {
+public:
+	DuckLakeFormatReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, string format,
+	                     vector<MultiFileColumnDefinition> columns);
+
+public:
+	bool TryInitializeScan(ClientContext &context, GlobalTableFunctionState &gstate,
+	                       LocalTableFunctionState &lstate) override;
+	AsyncResult Scan(ClientContext &context, GlobalTableFunctionState &global_state,
+	                 LocalTableFunctionState &local_state, DataChunk &chunk) override;
+	string GetReaderType() const override;
+	void AddVirtualColumn(column_t virtual_column_id) override;
+
+private:
+	static TableFunction GetReadFunction(ClientContext &context, const string &format);
+
+	mutex lock;
+	DuckLakeFunctionInfo &read_info;
+	string format; //! selects the read_<format> function
+	bool initialized_scan = false;
+	int64_t file_row_number = 0;
+
+	//! read_<format> scan bound to the file
+	TableFunction file_scan;
+	unique_ptr<FunctionData> bind_data;
+	vector<LogicalType> file_types;
+	vector<string> file_names;
+
+	unique_ptr<ThreadContext> thread_context;
+	unique_ptr<ExecutionContext> execution_context;
+	unique_ptr<GlobalTableFunctionState> global_state;
+	unique_ptr<LocalTableFunctionState> local_state;
+
+	vector<column_t> scan_column_ids;              //! columns projected out of the file
+	DataChunk scan_chunk;                          //! raw columns read from the file
+	vector<FormatReaderColumn> output_kinds;       //! how each output column is produced
+	vector<idx_t> output_source_index;             //! for FILE_COLUMN: position in scan_chunk
+};
+
+} // namespace duckdb

src/include/storage/ducklake_insert.hpp

@@ -37,6 +37,8 @@ class DuckLakeInsertGlobalState : public GlobalSinkState {
 	case_insensitive_set_t not_null_fields;
 	//! Total rows flushed (used by flush_inlined_data)
 	idx_t rows_flushed = 0;
+	//! Format written files are recorded with (resolved from the table's data_file_format option)
+	string file_format = "parquet";
 };
 
 class DuckLakeInsert : public PhysicalOperator {
@@ -78,6 +80,8 @@ class DuckLakeInsert : public PhysicalOperator {
 	static unique_ptr<LogicalOperator> InsertCasts(Binder &binder, unique_ptr<LogicalOperator> &plan);
 
 	static DuckLakeColumnStats ParseColumnStats(const LogicalType &type, const vector<Value> &stats);
+	//! Normalize and validate a configured data file format; throws for unsupported formats.
+	static string ValidateDataFileFormat(string format);
 	static DuckLakeCopyOptions GetCopyOptions(ClientContext &context, DuckLakeCopyInput &copy_input);
 	static PhysicalOperator &PlanCopyForInsert(ClientContext &context, PhysicalPlanGenerator &planner,
 	                                           DuckLakeCopyInput &copy_input, optional_ptr<PhysicalOperator> plan);

src/include/storage/ducklake_metadata_info.hpp

@@ -149,6 +149,8 @@ struct DuckLakeFileInfo {
 	DataFileIndex id;
 	TableIndex table_id;
 	string file_name;
+	//! physical format; persisted to ducklake_data_file.file_format
+	string file_format = "parquet";
 	idx_t row_count;
 	idx_t file_size_bytes;
 	optional_idx footer_size;
@@ -352,7 +354,10 @@ struct DuckLakeFileData {
 	string encryption_key;
 	idx_t file_size_bytes = 0;
 	optional_idx footer_size;
+	//! delete-file format (parquet/puffin); delete files only
 	DeleteFileFormat format = DeleteFileFormat::PARQUET;
+	//! data-file format from ducklake_data_file.file_format; selects the reader
+	string data_file_format = "parquet";
 };
 
 enum class DuckLakeDataType {

src/include/storage/ducklake_metadata_manager.hpp

@@ -295,12 +295,13 @@ class DuckLakeMetadataManager {
 	template <class T>
 	string FlushDrop(const string &metadata_table_name, const string &id_name, const set<T> &dropped_entries);
 	template <class T>
-	DuckLakeFileData ReadDataFile(DuckLakeTableEntry &table, T &row, idx_t &col_idx, bool is_encrypted);
+	DuckLakeFileData ReadDataFile(DuckLakeTableEntry &table, T &row, idx_t &col_idx, bool is_encrypted,
+	                              bool read_data_file_format = false);
 	template <class T>
 	DuckLakeFileData ReadDeleteFile(DuckLakeTableEntry &table, T &row, idx_t &col_idx, bool is_encrypted);
 
 	bool IsEncrypted() const;
-	string GetFileSelectList(const string &prefix);
+	string GetFileSelectList(const string &prefix, bool include_data_file_format = false);
 	string GetDeleteFileSelectList(const string &prefix);
 	FilterPushdownQueryComponents GenerateFilterPushdownComponents(const FilterPushdownInfo &filter_info,
 	                                                               TableIndex table_id);

src/include/storage/ducklake_multi_file_reader.hpp

@@ -80,6 +80,8 @@ struct DuckLakeMultiFileReader : public MultiFileReader {
 
 private:
 	shared_ptr<BaseFileReader> TryCreateInlinedDataReader(const OpenFileInfo &file);
+	//! Create a non-parquet format reader based on the file's file_format extended option (else nullptr).
+	shared_ptr<BaseFileReader> TryCreateFormatReader(ClientContext &context, const OpenFileInfo &file);
 	//! For deletion scans we need to get the snapshot_id values using per-row snapshot information
 	void GatherDeletionScanSnapshots(BaseFileReader &reader, const MultiFileReaderData &reader_data, DataChunk &chunk,
 	                                 optional_idx rowid_col_override = optional_idx()) const;

src/storage/CMakeLists.txt

@@ -21,6 +21,7 @@ add_library(
   ducklake_delete.cpp
   ducklake_deletion_vector.cpp
   ducklake_multi_file_reader.cpp
+  ducklake_format_reader.cpp
   ducklake_partition_data.cpp
   ducklake_secret.cpp
   ducklake_stats.cpp