From c0712039c09ea63f48c8cd9f1fe803377d597a0c Mon Sep 17 00:00:00 2001 From: Mosha Pasumansky Date: Tue, 4 Aug 2026 10:26:20 -0700 Subject: [PATCH 1/3] feat(read): dispatch DuckLake data files by file_format (parquet + vortex) Make the DuckLake scan path honor the per-file file_format column of ducklake_data_file instead of assuming parquet for every file. Files recorded as "vortex" are now read via a dedicated reader that drives the vortex extension's read_vortex table function; parquet files are unchanged. Formats can be mixed within a single table (per-file dispatch), reusing the existing delete-filter, row-id and column-mapping machinery. - thread file_format from ducklake_data_file into DuckLakeFileData and on into OpenFileInfo.extended_info (GetFileSelectList/ReadDataFile gain an opt-in flag, enabled only on the data-file scan paths) - DuckLakeMultiFileReader::CreateReader dispatches "vortex" files to the new DuckLakeVortexReader (BaseFileReader), mirroring the existing inlined-data reader dispatch - vendor vortex-data/duckdb-vortex as the ./vortex submodule and load it alongside parquet via extension_config.cmake (opt-out with DISABLE_VORTEX) Write support (recording file_format="vortex" on insert / COPY FORMAT VORTEX) is intentionally deferred to a follow-up, so end-to-end vortex round-trip tests land with the write path. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitmodules | 5 +- extension_config.cmake | 10 + .../storage/ducklake_metadata_info.hpp | 4 + .../storage/ducklake_metadata_manager.hpp | 5 +- .../storage/ducklake_multi_file_reader.hpp | 2 + .../storage/ducklake_vortex_reader.hpp | 76 +++++++ src/storage/CMakeLists.txt | 1 + src/storage/ducklake_metadata_manager.cpp | 31 ++- src/storage/ducklake_multi_file_list.cpp | 5 + src/storage/ducklake_multi_file_reader.cpp | 27 +++ src/storage/ducklake_vortex_reader.cpp | 191 ++++++++++++++++++ vortex | 1 + 12 files changed, 347 insertions(+), 11 deletions(-) create mode 100644 src/include/storage/ducklake_vortex_reader.hpp create mode 100644 src/storage/ducklake_vortex_reader.cpp create mode 160000 vortex diff --git a/.gitmodules b/.gitmodules index 01b02cb4..62348cfe 100644 --- a/.gitmodules +++ b/.gitmodules @@ -5,4 +5,7 @@ [submodule "extension-ci-tools"] path = extension-ci-tools url = https://github.com/duckdb/extension-ci-tools - branch = main \ No newline at end of file + branch = main +[submodule "vortex"] + path = vortex + url = https://github.com/vortex-data/duckdb-vortex.git diff --git a/extension_config.cmake b/extension_config.cmake index 2bb8bb13..d7823493 100644 --- a/extension_config.cmake +++ b/extension_config.cmake @@ -11,6 +11,16 @@ if(NOT DEFINED ENV{DISABLE_EXTENSIONS_FOR_TEST}) duckdb_extension_load(tpch) endif() +# Vortex file format support (vendored as the ./vortex submodule). Building it +# requires a Rust toolchain (it compiles the vortex-extension crate via Corrosion), +# so it is opt-out via DISABLE_VORTEX for toolchains without cargo. When enabled, +# DuckLake can read and write data files whose file_format is 'vortex'. +if(NOT DEFINED ENV{DISABLE_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") diff --git a/src/include/storage/ducklake_metadata_info.hpp b/src/include/storage/ducklake_metadata_info.hpp index 23ededb5..66cad9f7 100644 --- a/src/include/storage/ducklake_metadata_info.hpp +++ b/src/include/storage/ducklake_metadata_info.hpp @@ -352,7 +352,11 @@ struct DuckLakeFileData { string encryption_key; idx_t file_size_bytes = 0; optional_idx footer_size; + //! Delete-file format (parquet/puffin); only meaningful for delete files. DeleteFileFormat format = DeleteFileFormat::PARQUET; + //! Physical data-file format read from ducklake_data_file.file_format ("parquet" or "vortex"). + //! Selects which reader is used to scan this file. + string data_file_format = "parquet"; }; enum class DuckLakeDataType { diff --git a/src/include/storage/ducklake_metadata_manager.hpp b/src/include/storage/ducklake_metadata_manager.hpp index 9590aea1..c0d54666 100644 --- a/src/include/storage/ducklake_metadata_manager.hpp +++ b/src/include/storage/ducklake_metadata_manager.hpp @@ -295,12 +295,13 @@ class DuckLakeMetadataManager { template string FlushDrop(const string &metadata_table_name, const string &id_name, const set &dropped_entries); template - 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 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); diff --git a/src/include/storage/ducklake_multi_file_reader.hpp b/src/include/storage/ducklake_multi_file_reader.hpp index d89dccfd..7bd7677c 100644 --- a/src/include/storage/ducklake_multi_file_reader.hpp +++ b/src/include/storage/ducklake_multi_file_reader.hpp @@ -80,6 +80,8 @@ struct DuckLakeMultiFileReader : public MultiFileReader { private: shared_ptr TryCreateInlinedDataReader(const OpenFileInfo &file); + //! Create a non-parquet format reader based on the file's file_format extended option (else nullptr). + shared_ptr TryCreateVortexReader(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; diff --git a/src/include/storage/ducklake_vortex_reader.hpp b/src/include/storage/ducklake_vortex_reader.hpp new file mode 100644 index 00000000..55274fff --- /dev/null +++ b/src/include/storage/ducklake_vortex_reader.hpp @@ -0,0 +1,76 @@ +//===----------------------------------------------------------------------===// +// DuckDB +// +// storage/ducklake_vortex_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; + +//! Marks how a single output column of the DuckLakeVortexReader is produced. +enum class VortexReaderColumn : uint8_t { + //! Projected directly from the underlying read_vortex scan (see source_index) + FILE_COLUMN, + //! Ordinal position of the row within the file (row_id_start + file_row_number is applied by the + //! MultiFileReader on top). Emitted by the reader itself. + ORDINAL, + //! Column that is not present in the vortex file - emit its default value (usually NULL) + MISSING +}; + +//! Reads a single DuckLake data file stored in the Vortex format. It drives the vortex extension's +//! `read_vortex` table function for the physical file and adapts the produced chunks to DuckLake's +//! MultiFileReader contract (column mapping, positional delete filters, pushed-down filters and the +//! file_row_number virtual column) exactly like the parquet reader path does. +class DuckLakeVortexReader : public BaseFileReader { +public: + DuckLakeVortexReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, + vector 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: + //! Look up the vortex extension `read_vortex` table function from the catalog. + static TableFunction GetReadVortexFunction(ClientContext &context); + + mutex lock; + DuckLakeFunctionInfo &read_info; + bool initialized_scan = false; + int64_t file_row_number = 0; + + //! Underlying read_vortex scan bound to the physical file. + TableFunction vortex_scan; + unique_ptr bind_data; + vector file_types; + vector file_names; + + unique_ptr thread_context; + unique_ptr execution_context; + unique_ptr global_state; + unique_ptr local_state; + + //! Column ids projected out of the vortex file (indexes into file_names/file_types). + vector scan_column_ids; + //! Chunk holding the raw columns read from read_vortex (in scan_column_ids order). + DataChunk scan_chunk; + //! For each output column: how it is produced. + vector output_kinds; + //! For FILE_COLUMN outputs: the position within scan_chunk to reference. + vector output_source_index; +}; + +} // namespace duckdb diff --git a/src/storage/CMakeLists.txt b/src/storage/CMakeLists.txt index 41e6ac9e..a9bf2b48 100644 --- a/src/storage/CMakeLists.txt +++ b/src/storage/CMakeLists.txt @@ -21,6 +21,7 @@ add_library( ducklake_delete.cpp ducklake_deletion_vector.cpp ducklake_multi_file_reader.cpp + ducklake_vortex_reader.cpp ducklake_partition_data.cpp ducklake_secret.cpp ducklake_stats.cpp diff --git a/src/storage/ducklake_metadata_manager.cpp b/src/storage/ducklake_metadata_manager.cpp index 2b7c0e8e..5417f101 100644 --- a/src/storage/ducklake_metadata_manager.cpp +++ b/src/storage/ducklake_metadata_manager.cpp @@ -873,7 +873,7 @@ ORDER BY table_id; return TransformGlobalStats(*result); } -string DuckLakeMetadataManager::GetFileSelectList(const string &prefix) { +string DuckLakeMetadataManager::GetFileSelectList(const string &prefix, bool include_data_file_format) { static const vector column_list { "path", "path_is_relative", "file_size_bytes", "footer_size", "encryption_key", }; @@ -887,6 +887,12 @@ string DuckLakeMetadataManager::GetFileSelectList(const string &prefix) { return prefix + "." + column + " AS " + prefix + "_" + column; }); + if (include_data_file_format) { + // Appended as the trailing column so it is read right after ReadDataFile's fixed columns. + // Only valid for the ducklake_data_file table (delete files do not have a file_format column). + result += ", " + prefix + ".file_format AS " + prefix + "_file_format"; + } + return result; } @@ -896,7 +902,7 @@ string DuckLakeMetadataManager::GetDeleteFileSelectList(const string &prefix) { template DuckLakeFileData DuckLakeMetadataManager::ReadDataFile(DuckLakeTableEntry &table, T &row, idx_t &col_idx, - bool is_encrypted) { + bool is_encrypted, bool read_data_file_format) { DuckLakeFileData data; if (row.IsNull(col_idx)) { // file is not there @@ -904,6 +910,9 @@ DuckLakeFileData DuckLakeMetadataManager::ReadDataFile(DuckLakeTableEntry &table if (is_encrypted) { col_idx++; } + if (read_data_file_format) { + col_idx++; + } return data; } DuckLakePath path; @@ -923,6 +932,12 @@ DuckLakeFileData DuckLakeMetadataManager::ReadDataFile(DuckLakeTableEntry &table } data.encryption_key = Blob::FromBase64(row.template GetValue(col_idx++)); } + if (read_data_file_format) { + if (!row.IsNull(col_idx)) { + data.data_file_format = row.template GetValue(col_idx); + } + col_idx++; + } return data; } @@ -1370,7 +1385,7 @@ vector DuckLakeMetadataManager::GetFilesForTable(DuckLake } } - string select_list = "data.data_file_id, " + GetFileSelectList("data") + + string select_list = "data.data_file_id, " + GetFileSelectList("data", true) + ", data.row_id_start, data.begin_snapshot, data.partial_max, data.mapping_id, " + GetDeleteFileSelectList("del") + stats_select_list; @@ -1418,7 +1433,7 @@ WHERE data.table_id=%d AND {SNAPSHOT_ID} >= data.begin_snapshot AND ({SNAPSHOT_I DuckLakeFileListEntry file_entry; idx_t col_idx = 0; file_entry.file_id = DataFileIndex(row.GetValue(col_idx++)); - file_entry.file = ReadDataFile(table, row, col_idx, IsEncrypted()); + file_entry.file = ReadDataFile(table, row, col_idx, IsEncrypted(), true); if (!row.IsNull(col_idx)) { file_entry.row_id_start = row.GetValue(col_idx); } @@ -1464,7 +1479,7 @@ vector DuckLakeMetadataManager::GetTableInsertions(DuckLa DuckLakeSnapshot start_snapshot, DuckLakeSnapshot end_snapshot) { auto table_id = table.GetTableId(); - string select_list = GetFileSelectList("data") + + string select_list = GetFileSelectList("data", true) + ", data.row_id_start, data.begin_snapshot, data.partial_max, data.mapping_id, " + GetDeleteFileSelectList("del"); // Files either match the exact snapshot range @@ -1497,7 +1512,7 @@ WHERE data.table_id=%d AND data.begin_snapshot <= {SNAPSHOT_ID} AND ( for (auto &row : *result) { DuckLakeFileListEntry file_entry; idx_t col_idx = 0; - file_entry.file = ReadDataFile(table, row, col_idx, IsEncrypted()); + file_entry.file = ReadDataFile(table, row, col_idx, IsEncrypted(), true); if (!row.IsNull(col_idx)) { file_entry.row_id_start = row.GetValue(col_idx); } @@ -1712,7 +1727,7 @@ DuckLakeMetadataManager::GetExtendedFilesForTable(DuckLakeTableEntry &table, Duc const FilterPushdownInfo *filter_info) { auto table_id = table.GetTableId(); string select_list = - GetFileSelectList("data") + ", data.row_id_start, " + GetDeleteFileSelectList("del") + ", del.begin_snapshot"; + GetFileSelectList("data", true) + ", data.row_id_start, " + GetDeleteFileSelectList("del") + ", del.begin_snapshot"; string query; string where_clause; @@ -1756,7 +1771,7 @@ WHERE data.table_id=%d AND {SNAPSHOT_ID} >= data.begin_snapshot AND ({SNAPSHOT_I } file_entry.row_count = row.GetValue(2); idx_t col_idx = 3; - file_entry.file = ReadDataFile(table, row, col_idx, IsEncrypted()); + file_entry.file = ReadDataFile(table, row, col_idx, IsEncrypted(), true); if (!row.IsNull(col_idx)) { file_entry.row_id_start = row.GetValue(col_idx); } diff --git a/src/storage/ducklake_multi_file_list.cpp b/src/storage/ducklake_multi_file_list.cpp index 3166bab6..11d836ef 100644 --- a/src/storage/ducklake_multi_file_list.cpp +++ b/src/storage/ducklake_multi_file_list.cpp @@ -174,6 +174,11 @@ OpenFileInfo DuckLakeMultiFileList::GetFile(idx_t i) const { Value::BIGINT(NumericCast(inlined_data_table.schema_version)); } else { extended_info->options["file_size"] = Value::UBIGINT(file.file_size_bytes); + // Record the physical format so the multi-file reader can pick the right per-file reader + // (parquet_scan vs. read_vortex). Absent/empty means parquet. + if (!file.data_file_format.empty()) { + extended_info->options["file_format"] = Value(file.data_file_format); + } if (file.footer_size.IsValid()) { extended_info->options["footer_size"] = Value::UBIGINT(file.footer_size.GetIndex()); } diff --git a/src/storage/ducklake_multi_file_reader.cpp b/src/storage/ducklake_multi_file_reader.cpp index a18ddf8a..4b3dbed2 100644 --- a/src/storage/ducklake_multi_file_reader.cpp +++ b/src/storage/ducklake_multi_file_reader.cpp @@ -1,5 +1,6 @@ #include "storage/ducklake_multi_file_list.hpp" #include "storage/ducklake_multi_file_reader.hpp" +#include "storage/ducklake_vortex_reader.hpp" #include "storage/ducklake_table_entry.hpp" #include "storage/ducklake_catalog.hpp" #include "storage/ducklake_delete_filter.hpp" @@ -386,6 +387,24 @@ shared_ptr DuckLakeMultiFileReader::TryCreateInlinedDataReader(c std::move(columns)); } +shared_ptr DuckLakeMultiFileReader::TryCreateVortexReader(ClientContext &context, + const OpenFileInfo &file) { + if (!file.extended_info) { + return nullptr; + } + auto entry = file.extended_info->options.find("file_format"); + if (entry == file.extended_info->options.end()) { + // no format recorded - default parquet path + return nullptr; + } + if (!StringUtil::CIEquals(StringValue::Get(entry->second), "vortex")) { + return nullptr; + } + // Map the vortex file's columns onto the table's current field ids by name. + auto columns = DuckLakeMultiFileReader::ColumnsFromFieldData(read_info.table.GetFieldData(), true); + return make_shared_ptr(read_info, file, std::move(columns)); +} + shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext &context, GlobalTableFunctionState &gstate, const OpenFileInfo &file, idx_t file_idx, @@ -394,6 +413,10 @@ shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext & if (reader) { return reader; } + reader = TryCreateVortexReader(context, file); + if (reader) { + return reader; + } return MultiFileReader::CreateReader(context, gstate, file, file_idx, bind_data); } @@ -405,6 +428,10 @@ shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext & if (reader) { return reader; } + reader = TryCreateVortexReader(context, file); + if (reader) { + return reader; + } return MultiFileReader::CreateReader(context, file, options, file_options, interface); } diff --git a/src/storage/ducklake_vortex_reader.cpp b/src/storage/ducklake_vortex_reader.cpp new file mode 100644 index 00000000..33bebc17 --- /dev/null +++ b/src/storage/ducklake_vortex_reader.cpp @@ -0,0 +1,191 @@ +#include "storage/ducklake_vortex_reader.hpp" +#include "storage/ducklake_multi_file_reader.hpp" +#include "storage/ducklake_scan.hpp" +#include "storage/ducklake_delete_filter.hpp" +#include "duckdb/catalog/catalog_entry/table_function_catalog_entry.hpp" +#include "duckdb/main/database.hpp" +#include "duckdb/main/extension_helper.hpp" +#include "duckdb/parser/tableref/table_function_ref.hpp" +#include "duckdb/planner/table_filter_state.hpp" +#include "duckdb/storage/table/column_segment.hpp" +#include "duckdb/common/vector_operations/vector_operations.hpp" + +namespace duckdb { + +DuckLakeVortexReader::DuckLakeVortexReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, + vector columns_p) + : BaseFileReader(info), read_info(read_info) { + columns = std::move(columns_p); +} + +TableFunction DuckLakeVortexReader::GetReadVortexFunction(ClientContext &context) { + auto &instance = DatabaseInstance::GetDatabase(context); + // The vortex extension registers read_vortex on load; auto-load it if it is available. + ExtensionHelper::TryAutoLoadExtension(instance, "vortex"); + ExtensionLoader loader(instance, "ducklake"); + auto entry = loader.TryGetTableFunction("read_vortex"); + if (!entry) { + throw MissingExtensionException("Reading DuckLake data files stored in the 'vortex' format requires the " + "\"vortex\" extension to be installed and loaded"); + } + auto &vortex_scan_entry = entry->Cast(); + return vortex_scan_entry.functions.GetFunctionByOffset(0); +} + +bool DuckLakeVortexReader::TryInitializeScan(ClientContext &context, GlobalTableFunctionState &gstate, + LocalTableFunctionState &lstate) { + { + // only one thread reads a given file + lock_guard guard(lock); + if (initialized_scan) { + return false; + } + initialized_scan = true; + } + + // Bind read_vortex on the physical file to discover its schema. + vortex_scan = GetReadVortexFunction(context); + + vector children; + children.push_back(Value(file.path)); + named_parameter_map_t named_params; + vector input_types; + vector input_names; + + TableFunctionRef empty; + TableFunction dummy_table_function; + dummy_table_function.SetName("DuckLakeVortexReader"); + TableFunctionBindInput bind_input(children, named_params, input_types, input_names, nullptr, nullptr, + dummy_table_function, empty); + bind_data = vortex_scan.bind(context, bind_input, file_types, file_names); + + // Build the projection into the vortex file and the mapping to the reader's output columns. + vector scan_types; + for (auto &column_id : column_indexes) { + auto index = column_id.GetPrimaryIndex(); + auto &col = columns[index]; + // Virtual columns carry an INTEGER identifier. The only one the reader must materialize itself + // is the ordinal (file_row_number); higher-level virtuals (row_id / snapshot_id) are computed by + // the MultiFileReader as expressions on top of the ordinal. + if (!col.identifier.IsNull() && col.identifier.type().id() == LogicalTypeId::INTEGER && + IntegerValue::Get(col.identifier) == MultiFileReader::ORDINAL_FIELD_ID) { + output_kinds.push_back(VortexReaderColumn::ORDINAL); + output_source_index.push_back(0); + continue; + } + // Regular column - locate it in the vortex file by name. + auto column_name = col.name.GetIdentifierName(); + optional_idx file_position; + for (idx_t i = 0; i < file_names.size(); i++) { + if (StringUtil::CIEquals(file_names[i], column_name)) { + file_position = i; + break; + } + } + if (!file_position.IsValid()) { + // column is not present in this file - emit its default (NULL) + output_kinds.push_back(VortexReaderColumn::MISSING); + output_source_index.push_back(0); + continue; + } + output_kinds.push_back(VortexReaderColumn::FILE_COLUMN); + output_source_index.push_back(scan_column_ids.size()); + scan_column_ids.push_back(file_position.GetIndex()); + scan_types.push_back(file_types[file_position.GetIndex()]); + } + + // read_vortex needs at least one projected column to determine the cardinality of each chunk. + if (scan_column_ids.empty() && !file_types.empty()) { + scan_column_ids.push_back(0); + scan_types.push_back(file_types[0]); + } + scan_chunk.Initialize(context, scan_types); + + thread_context = make_uniq(context); + execution_context = make_uniq(context, *thread_context, nullptr); + TableFunctionInitInput init_input(bind_data.get(), scan_column_ids, vector(), nullptr); + global_state = vortex_scan.init_global(context, init_input); + local_state = vortex_scan.init_local(*execution_context, init_input, global_state.get()); + return true; +} + +AsyncResult DuckLakeVortexReader::Scan(ClientContext &context, GlobalTableFunctionState &global_table_state, + LocalTableFunctionState &local_table_state, DataChunk &chunk) { + scan_chunk.Reset(); + TableFunctionInput function_input(bind_data.get(), local_state.get(), global_state.get()); + vortex_scan.function(context, function_input, scan_chunk); + idx_t scan_count = scan_chunk.size(); + if (scan_count == 0) { + return AsyncResult(SourceResultType::FINISHED); + } + + for (idx_t c = 0; c < output_kinds.size(); c++) { + switch (output_kinds[c]) { + case VortexReaderColumn::FILE_COLUMN: { + auto &source = scan_chunk.data[output_source_index[c]]; + if (chunk.data[c].GetType() != source.GetType()) { + VectorOperations::Cast(context, source, chunk.data[c], scan_count); + } else { + chunk.data[c].Reference(source); + } + break; + } + case VortexReaderColumn::ORDINAL: { + auto ordinal_data = FlatVector::GetDataMutable(chunk.data[c]); + for (idx_t r = 0; r < scan_count; r++) { + ordinal_data[r] = file_row_number + NumericCast(r); + } + break; + } + case VortexReaderColumn::MISSING: + chunk.data[c].SetVectorType(VectorType::CONSTANT_VECTOR); + ConstantVector::SetNull(chunk.data[c], true); + break; + } + } + chunk.SetCardinality(scan_count); + + if (filters || deletion_filter) { + SelectionVector sel; + idx_t approved_tuple_count = scan_count; + if (deletion_filter) { + // delete filters are positional within the file (row_id_start + file_row_number) + approved_tuple_count = deletion_filter->Filter(file_row_number, approved_tuple_count, sel); + } + if (filters) { + for (auto &entry : *filters) { + auto &filter = entry.Filter(); + if (ExpressionFilter::IsRootOptionalFilter(filter)) { + continue; + } + auto column_id = entry.GetIndex().GetIndex(); + auto &vec = chunk.data[column_id]; + UnifiedVectorFormat vdata; + vec.ToUnifiedFormat(vdata); + auto filter_state = TableFilterState::Initialize(context, filter); + approved_tuple_count = ColumnSegment::FilterSelection(sel, vec, vdata, filter, *filter_state, + chunk.size(), approved_tuple_count); + } + } + if (approved_tuple_count != chunk.size()) { + chunk.Slice(sel, approved_tuple_count); + } + } + + file_row_number += NumericCast(scan_count); + return AsyncResult(SourceResultType::HAVE_MORE_OUTPUT); +} + +void DuckLakeVortexReader::AddVirtualColumn(column_t virtual_column_id) { + if (virtual_column_id == MultiFileReader::COLUMN_IDENTIFIER_FILE_ROW_NUMBER) { + columns.back().identifier = Value::INTEGER(MultiFileReader::ORDINAL_FIELD_ID); + } else { + throw InternalException("Unsupported virtual column id %d for vortex reader", virtual_column_id); + } +} + +string DuckLakeVortexReader::GetReaderType() const { + return "DuckLake Vortex"; +} + +} // namespace duckdb diff --git a/vortex b/vortex new file mode 160000 index 00000000..0867246a --- /dev/null +++ b/vortex @@ -0,0 +1 @@ +Subproject commit 0867246a4ae5df1616dbf320f79a5ced07a8c867 From afb10efc64f78c0260e38aad8a78146ddf1dc687 Mon Sep 17 00:00:00 2001 From: Mosha Pasumansky Date: Tue, 4 Aug 2026 11:19:13 -0700 Subject: [PATCH 2/3] feat(format): generalize non-parquet reader + write/verify JSON & vortex dispatch Follow-up to the file_format read dispatch: generalize the reader and add the write side so a table's data files can be produced and read in a non-parquet format, verified end-to-end. - rename DuckLakeVortexReader -> DuckLakeFormatReader: drives read_ for any non-parquet format (vortex, json, ...), mapping columns by name and applying deletes/row-ids/filters via the existing MultiFileReader contract - write path: data_file_format catalog option (parquet default, or vortex); GetCopyOptions routes to the format's COPY, records file_format in metadata, and PlanInsert/AddWrittenFiles stamp it. A guard rejects copy functions without copy_to_bind (e.g. json/csv plan-based COPY) instead of crashing. - tests: data_file_format_option (recording + validation), json_read_dispatch (proves per-file dispatch end-to-end via read_json, incl. projection/filter pushdown), and vortex_format (round-trip; runs where vortex's DuckDB pin is aligned with DuckLake's). Verified on an icecc-farm release build: option 9/9, json_read_dispatch 24/24, full existing DuckLake suite 503/503 (no regressions). Vortex round-trip is currently unbuildable here because the latest vortex extension pins DuckDB d8cdaa33 (Jul-21) while DuckLake main requires c97bd8b9 (Jul-22, adds table_index.hpp, removes Vector::GetBuffer) - an upstream version-alignment gap, not a code defect. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 1 + src/functions/ducklake_set_option.cpp | 4 + src/include/common/ducklake_data_file.hpp | 3 + ..._reader.hpp => ducklake_format_reader.hpp} | 38 +++--- src/include/storage/ducklake_insert.hpp | 7 +- .../storage/ducklake_metadata_info.hpp | 2 + .../storage/ducklake_multi_file_reader.hpp | 2 +- src/storage/CMakeLists.txt | 2 +- ..._reader.cpp => ducklake_format_reader.cpp} | 76 ++++++----- src/storage/ducklake_insert.cpp | 125 ++++++++++++------ src/storage/ducklake_metadata_manager.cpp | 8 +- src/storage/ducklake_multi_file_reader.cpp | 17 ++- src/storage/ducklake_transaction.cpp | 1 + test/sql/format/data_file_format_option.test | 45 +++++++ test/sql/format/json_read_dispatch.test | 81 ++++++++++++ test/sql/vortex/vortex_format.test | 95 +++++++++++++ 16 files changed, 403 insertions(+), 104 deletions(-) rename src/include/storage/{ducklake_vortex_reader.hpp => ducklake_format_reader.hpp} (55%) rename src/storage/{ducklake_vortex_reader.cpp => ducklake_format_reader.cpp} (68%) create mode 100644 test/sql/format/data_file_format_option.test create mode 100644 test/sql/format/json_read_dispatch.test create mode 100644 test/sql/vortex/vortex_format.test diff --git a/.gitignore b/.gitignore index 99892092..526cfe86 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,4 @@ test/python/__pycache__/ .Rhistory scripts/data/ duckdb_benchmark_data/ +.cache/ diff --git a/src/functions/ducklake_set_option.cpp b/src/functions/ducklake_set_option.cpp index 017bda0a..2a9591c4 100644 --- a/src/functions/ducklake_set_option.cpp +++ b/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 DuckLakeSetOptionBind(ClientContext &context, Ta value = val.CastAs(context, LogicalType::BOOLEAN).GetValue() ? "true" : "false"; } else if (option == "write_deletion_vectors") { value = val.CastAs(context, LogicalType::BOOLEAN).GetValue() ? "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()); } else { throw NotImplementedException("Unsupported option %s", option); } diff --git a/src/include/common/ducklake_data_file.hpp b/src/include/common/ducklake_data_file.hpp index da4265b6..febe273c 100644 --- a/src/include/common/ducklake_data_file.hpp +++ b/src/include/common/ducklake_data_file.hpp @@ -62,6 +62,9 @@ struct DuckLakeDataFile { DuckLakeDataFile &operator=(const DuckLakeDataFile &); string file_name; + //! The physical format the file is stored in ("parquet", "vortex", ...). Recorded in + //! ducklake_data_file.file_format and used to pick the reader at scan time. + string file_format = "parquet"; idx_t row_count; idx_t file_size_bytes; optional_idx footer_size; diff --git a/src/include/storage/ducklake_vortex_reader.hpp b/src/include/storage/ducklake_format_reader.hpp similarity index 55% rename from src/include/storage/ducklake_vortex_reader.hpp rename to src/include/storage/ducklake_format_reader.hpp index 55274fff..7e97b0f7 100644 --- a/src/include/storage/ducklake_vortex_reader.hpp +++ b/src/include/storage/ducklake_format_reader.hpp @@ -1,7 +1,7 @@ //===----------------------------------------------------------------------===// // DuckDB // -// storage/ducklake_vortex_reader.hpp +// storage/ducklake_format_reader.hpp // // //===----------------------------------------------------------------------===// @@ -15,24 +15,24 @@ namespace duckdb { struct DuckLakeFunctionInfo; -//! Marks how a single output column of the DuckLakeVortexReader is produced. -enum class VortexReaderColumn : uint8_t { - //! Projected directly from the underlying read_vortex scan (see source_index) +//! Marks how a single output column of the DuckLakeFormatReader is produced. +enum class FormatReaderColumn : uint8_t { + //! Projected directly from the underlying read_ scan (see source_index) FILE_COLUMN, //! Ordinal position of the row within the file (row_id_start + file_row_number is applied by the //! MultiFileReader on top). Emitted by the reader itself. ORDINAL, - //! Column that is not present in the vortex file - emit its default value (usually NULL) + //! Column that is not present in the file - emit its default value (usually NULL) MISSING }; -//! Reads a single DuckLake data file stored in the Vortex format. It drives the vortex extension's -//! `read_vortex` table function for the physical file and adapts the produced chunks to DuckLake's -//! MultiFileReader contract (column mapping, positional delete filters, pushed-down filters and the -//! file_row_number virtual column) exactly like the parquet reader path does. -class DuckLakeVortexReader : public BaseFileReader { +//! Reads a single DuckLake data file stored in a non-parquet format (e.g. "vortex", "json"). It drives +//! the format's `read_` table function for the physical file and adapts the produced chunks to +//! DuckLake's MultiFileReader contract (column mapping, positional delete filters, pushed-down filters +//! and the file_row_number virtual column) exactly like the parquet reader path does. +class DuckLakeFormatReader : public BaseFileReader { public: - DuckLakeVortexReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, + DuckLakeFormatReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, string format, vector columns); public: @@ -44,16 +44,18 @@ class DuckLakeVortexReader : public BaseFileReader { void AddVirtualColumn(column_t virtual_column_id) override; private: - //! Look up the vortex extension `read_vortex` table function from the catalog. - static TableFunction GetReadVortexFunction(ClientContext &context); + //! Look up the `read_` table function (auto-loading the format's extension if available). + static TableFunction GetReadFunction(ClientContext &context, const string &format); mutex lock; DuckLakeFunctionInfo &read_info; + //! The physical format of the file ("vortex", "json", ...); selects the read_ function. + string format; bool initialized_scan = false; int64_t file_row_number = 0; - //! Underlying read_vortex scan bound to the physical file. - TableFunction vortex_scan; + //! Underlying read_ scan bound to the physical file. + TableFunction file_scan; unique_ptr bind_data; vector file_types; vector file_names; @@ -63,12 +65,12 @@ class DuckLakeVortexReader : public BaseFileReader { unique_ptr global_state; unique_ptr local_state; - //! Column ids projected out of the vortex file (indexes into file_names/file_types). + //! Column ids projected out of the file (indexes into file_names/file_types). vector scan_column_ids; - //! Chunk holding the raw columns read from read_vortex (in scan_column_ids order). + //! Chunk holding the raw columns read from read_ (in scan_column_ids order). DataChunk scan_chunk; //! For each output column: how it is produced. - vector output_kinds; + vector output_kinds; //! For FILE_COLUMN outputs: the position within scan_chunk to reference. vector output_source_index; }; diff --git a/src/include/storage/ducklake_insert.hpp b/src/include/storage/ducklake_insert.hpp index c591763c..850500d3 100644 --- a/src/include/storage/ducklake_insert.hpp +++ b/src/include/storage/ducklake_insert.hpp @@ -63,6 +63,8 @@ class DuckLakeInsert : public PhysicalOperator { optional_idx partition_id; //! The encryption key used for writing the Parquet files string encryption_key; + //! The physical format written data files are stored in ("parquet", "vortex", ...) + string file_format = "parquet"; public: // // Source interface @@ -78,13 +80,16 @@ class DuckLakeInsert : public PhysicalOperator { static unique_ptr InsertCasts(Binder &binder, unique_ptr &plan); static DuckLakeColumnStats ParseColumnStats(const LogicalType &type, const vector &stats); + //! Normalize and validate a configured data file format; throws for unsupported formats. + static string ValidateDataFileFormat(string format); static DuckLakeCopyOptions GetCopyOptions(ClientContext &context, DuckLakeCopyInput ©_input); static PhysicalOperator &PlanCopyForInsert(ClientContext &context, PhysicalPlanGenerator &planner, DuckLakeCopyInput ©_input, optional_ptr plan); static PhysicalOperator &PlanInsert(ClientContext &context, PhysicalPlanGenerator &planner, DuckLakeTableEntry &table, string encryption_key); static void AddWrittenFiles(DuckLakeInsertGlobalState &gstate, DataChunk &chunk, const string &encryption_key, - optional_idx partition_id, bool set_snapshot_id = false); + optional_idx partition_id, bool set_snapshot_id = false, + const string &file_format = "parquet"); static const DuckLakeFieldId &GetTopLevelColumn(DuckLakeCopyInput ©_input, FieldIndex field_id, optional_idx &index); diff --git a/src/include/storage/ducklake_metadata_info.hpp b/src/include/storage/ducklake_metadata_info.hpp index 66cad9f7..958e4642 100644 --- a/src/include/storage/ducklake_metadata_info.hpp +++ b/src/include/storage/ducklake_metadata_info.hpp @@ -149,6 +149,8 @@ struct DuckLakeFileInfo { DataFileIndex id; TableIndex table_id; string file_name; + //! Physical storage format ("parquet", "vortex", ...); persisted to ducklake_data_file.file_format. + string file_format = "parquet"; idx_t row_count; idx_t file_size_bytes; optional_idx footer_size; diff --git a/src/include/storage/ducklake_multi_file_reader.hpp b/src/include/storage/ducklake_multi_file_reader.hpp index 7bd7677c..7c59918b 100644 --- a/src/include/storage/ducklake_multi_file_reader.hpp +++ b/src/include/storage/ducklake_multi_file_reader.hpp @@ -81,7 +81,7 @@ struct DuckLakeMultiFileReader : public MultiFileReader { private: shared_ptr TryCreateInlinedDataReader(const OpenFileInfo &file); //! Create a non-parquet format reader based on the file's file_format extended option (else nullptr). - shared_ptr TryCreateVortexReader(ClientContext &context, const OpenFileInfo &file); + shared_ptr 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; diff --git a/src/storage/CMakeLists.txt b/src/storage/CMakeLists.txt index a9bf2b48..c31596ca 100644 --- a/src/storage/CMakeLists.txt +++ b/src/storage/CMakeLists.txt @@ -21,7 +21,7 @@ add_library( ducklake_delete.cpp ducklake_deletion_vector.cpp ducklake_multi_file_reader.cpp - ducklake_vortex_reader.cpp + ducklake_format_reader.cpp ducklake_partition_data.cpp ducklake_secret.cpp ducklake_stats.cpp diff --git a/src/storage/ducklake_vortex_reader.cpp b/src/storage/ducklake_format_reader.cpp similarity index 68% rename from src/storage/ducklake_vortex_reader.cpp rename to src/storage/ducklake_format_reader.cpp index 33bebc17..c18cb86e 100644 --- a/src/storage/ducklake_vortex_reader.cpp +++ b/src/storage/ducklake_format_reader.cpp @@ -1,4 +1,4 @@ -#include "storage/ducklake_vortex_reader.hpp" +#include "storage/ducklake_format_reader.hpp" #include "storage/ducklake_multi_file_reader.hpp" #include "storage/ducklake_scan.hpp" #include "storage/ducklake_delete_filter.hpp" @@ -7,32 +7,36 @@ #include "duckdb/main/extension_helper.hpp" #include "duckdb/parser/tableref/table_function_ref.hpp" #include "duckdb/planner/table_filter_state.hpp" +#include "duckdb/planner/filter/expression_filter.hpp" #include "duckdb/storage/table/column_segment.hpp" #include "duckdb/common/vector_operations/vector_operations.hpp" namespace duckdb { -DuckLakeVortexReader::DuckLakeVortexReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, +DuckLakeFormatReader::DuckLakeFormatReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, string format_p, vector columns_p) - : BaseFileReader(info), read_info(read_info) { + : BaseFileReader(info), read_info(read_info), format(std::move(format_p)) { columns = std::move(columns_p); } -TableFunction DuckLakeVortexReader::GetReadVortexFunction(ClientContext &context) { +TableFunction DuckLakeFormatReader::GetReadFunction(ClientContext &context, const string &format) { auto &instance = DatabaseInstance::GetDatabase(context); - // The vortex extension registers read_vortex on load; auto-load it if it is available. - ExtensionHelper::TryAutoLoadExtension(instance, "vortex"); + // The extension registering read_ may need to be auto-loaded (no-op for core formats). + ExtensionHelper::TryAutoLoadExtension(instance, format); ExtensionLoader loader(instance, "ducklake"); - auto entry = loader.TryGetTableFunction("read_vortex"); + auto function_name = "read_" + format; + auto entry = loader.TryGetTableFunction(Identifier(function_name)); if (!entry) { - throw MissingExtensionException("Reading DuckLake data files stored in the 'vortex' format requires the " - "\"vortex\" extension to be installed and loaded"); + throw MissingExtensionException( + "Reading DuckLake data files in the '%s' format requires the '%s' table function to be available " + "(install/load the extension providing it)", + format, function_name); } - auto &vortex_scan_entry = entry->Cast(); - return vortex_scan_entry.functions.GetFunctionByOffset(0); + auto &scan_entry = entry->Cast(); + return scan_entry.functions.GetFunctionByOffset(0); } -bool DuckLakeVortexReader::TryInitializeScan(ClientContext &context, GlobalTableFunctionState &gstate, +bool DuckLakeFormatReader::TryInitializeScan(ClientContext &context, GlobalTableFunctionState &gstate, LocalTableFunctionState &lstate) { { // only one thread reads a given file @@ -43,8 +47,8 @@ bool DuckLakeVortexReader::TryInitializeScan(ClientContext &context, GlobalTable initialized_scan = true; } - // Bind read_vortex on the physical file to discover its schema. - vortex_scan = GetReadVortexFunction(context); + // Bind read_ on the physical file to discover its schema. + file_scan = GetReadFunction(context, format); vector children; children.push_back(Value(file.path)); @@ -52,14 +56,20 @@ bool DuckLakeVortexReader::TryInitializeScan(ClientContext &context, GlobalTable vector input_types; vector input_names; + // Some read functions require schema auto-detection to be requested explicitly when bound + // programmatically (the SQL wrappers enable it by default). read_json is one such case. + if (StringUtil::CIEquals(format, "json")) { + named_params["auto_detect"] = Value::BOOLEAN(true); + } + TableFunctionRef empty; TableFunction dummy_table_function; - dummy_table_function.SetName("DuckLakeVortexReader"); + dummy_table_function.SetName("DuckLakeFormatReader"); TableFunctionBindInput bind_input(children, named_params, input_types, input_names, nullptr, nullptr, dummy_table_function, empty); - bind_data = vortex_scan.bind(context, bind_input, file_types, file_names); + bind_data = file_scan.bind(context, bind_input, file_types, file_names); - // Build the projection into the vortex file and the mapping to the reader's output columns. + // Build the projection into the file and the mapping to the reader's output columns. vector scan_types; for (auto &column_id : column_indexes) { auto index = column_id.GetPrimaryIndex(); @@ -69,11 +79,11 @@ bool DuckLakeVortexReader::TryInitializeScan(ClientContext &context, GlobalTable // the MultiFileReader as expressions on top of the ordinal. if (!col.identifier.IsNull() && col.identifier.type().id() == LogicalTypeId::INTEGER && IntegerValue::Get(col.identifier) == MultiFileReader::ORDINAL_FIELD_ID) { - output_kinds.push_back(VortexReaderColumn::ORDINAL); + output_kinds.push_back(FormatReaderColumn::ORDINAL); output_source_index.push_back(0); continue; } - // Regular column - locate it in the vortex file by name. + // Regular column - locate it in the file by name. auto column_name = col.name.GetIdentifierName(); optional_idx file_position; for (idx_t i = 0; i < file_names.size(); i++) { @@ -84,17 +94,17 @@ bool DuckLakeVortexReader::TryInitializeScan(ClientContext &context, GlobalTable } if (!file_position.IsValid()) { // column is not present in this file - emit its default (NULL) - output_kinds.push_back(VortexReaderColumn::MISSING); + output_kinds.push_back(FormatReaderColumn::MISSING); output_source_index.push_back(0); continue; } - output_kinds.push_back(VortexReaderColumn::FILE_COLUMN); + output_kinds.push_back(FormatReaderColumn::FILE_COLUMN); output_source_index.push_back(scan_column_ids.size()); scan_column_ids.push_back(file_position.GetIndex()); scan_types.push_back(file_types[file_position.GetIndex()]); } - // read_vortex needs at least one projected column to determine the cardinality of each chunk. + // read_ needs at least one projected column to determine the cardinality of each chunk. if (scan_column_ids.empty() && !file_types.empty()) { scan_column_ids.push_back(0); scan_types.push_back(file_types[0]); @@ -104,16 +114,16 @@ bool DuckLakeVortexReader::TryInitializeScan(ClientContext &context, GlobalTable thread_context = make_uniq(context); execution_context = make_uniq(context, *thread_context, nullptr); TableFunctionInitInput init_input(bind_data.get(), scan_column_ids, vector(), nullptr); - global_state = vortex_scan.init_global(context, init_input); - local_state = vortex_scan.init_local(*execution_context, init_input, global_state.get()); + global_state = file_scan.init_global(context, init_input); + local_state = file_scan.init_local(*execution_context, init_input, global_state.get()); return true; } -AsyncResult DuckLakeVortexReader::Scan(ClientContext &context, GlobalTableFunctionState &global_table_state, +AsyncResult DuckLakeFormatReader::Scan(ClientContext &context, GlobalTableFunctionState &global_table_state, LocalTableFunctionState &local_table_state, DataChunk &chunk) { scan_chunk.Reset(); TableFunctionInput function_input(bind_data.get(), local_state.get(), global_state.get()); - vortex_scan.function(context, function_input, scan_chunk); + file_scan.function(context, function_input, scan_chunk); idx_t scan_count = scan_chunk.size(); if (scan_count == 0) { return AsyncResult(SourceResultType::FINISHED); @@ -121,7 +131,7 @@ AsyncResult DuckLakeVortexReader::Scan(ClientContext &context, GlobalTableFuncti for (idx_t c = 0; c < output_kinds.size(); c++) { switch (output_kinds[c]) { - case VortexReaderColumn::FILE_COLUMN: { + case FormatReaderColumn::FILE_COLUMN: { auto &source = scan_chunk.data[output_source_index[c]]; if (chunk.data[c].GetType() != source.GetType()) { VectorOperations::Cast(context, source, chunk.data[c], scan_count); @@ -130,14 +140,14 @@ AsyncResult DuckLakeVortexReader::Scan(ClientContext &context, GlobalTableFuncti } break; } - case VortexReaderColumn::ORDINAL: { + case FormatReaderColumn::ORDINAL: { auto ordinal_data = FlatVector::GetDataMutable(chunk.data[c]); for (idx_t r = 0; r < scan_count; r++) { ordinal_data[r] = file_row_number + NumericCast(r); } break; } - case VortexReaderColumn::MISSING: + case FormatReaderColumn::MISSING: chunk.data[c].SetVectorType(VectorType::CONSTANT_VECTOR); ConstantVector::SetNull(chunk.data[c], true); break; @@ -176,16 +186,16 @@ AsyncResult DuckLakeVortexReader::Scan(ClientContext &context, GlobalTableFuncti return AsyncResult(SourceResultType::HAVE_MORE_OUTPUT); } -void DuckLakeVortexReader::AddVirtualColumn(column_t virtual_column_id) { +void DuckLakeFormatReader::AddVirtualColumn(column_t virtual_column_id) { if (virtual_column_id == MultiFileReader::COLUMN_IDENTIFIER_FILE_ROW_NUMBER) { columns.back().identifier = Value::INTEGER(MultiFileReader::ORDINAL_FIELD_ID); } else { - throw InternalException("Unsupported virtual column id %d for vortex reader", virtual_column_id); + throw InternalException("Unsupported virtual column id %d for format reader", virtual_column_id); } } -string DuckLakeVortexReader::GetReaderType() const { - return "DuckLake Vortex"; +string DuckLakeFormatReader::GetReaderType() const { + return "DuckLake " + format; } } // namespace duckdb diff --git a/src/storage/ducklake_insert.cpp b/src/storage/ducklake_insert.cpp index e62871d9..cee4f5a1 100644 --- a/src/storage/ducklake_insert.cpp +++ b/src/storage/ducklake_insert.cpp @@ -103,10 +103,12 @@ DuckLakeColumnStats DuckLakeInsert::ParseColumnStats(const LogicalType &type, co } void DuckLakeInsert::AddWrittenFiles(DuckLakeInsertGlobalState &global_state, DataChunk &chunk, - const string &encryption_key, optional_idx partition_id, bool set_snapshot_id) { + const string &encryption_key, optional_idx partition_id, bool set_snapshot_id, + const string &file_format) { for (idx_t r = 0; r < chunk.size(); r++) { DuckLakeDataFile data_file; data_file.file_name = chunk.GetValue(0, r).GetValue(); + data_file.file_format = file_format; data_file.row_count = chunk.GetValue(1, r).GetValue(); data_file.file_size_bytes = chunk.GetValue(2, r).GetValue(); data_file.footer_size = chunk.GetValue(3, r).GetValue(); @@ -211,7 +213,7 @@ void DuckLakeInsert::AddWrittenFiles(DuckLakeInsertGlobalState &global_state, Da SinkResultType DuckLakeInsert::Sink(ExecutionContext &context, DataChunk &chunk, OperatorSinkInput &input) const { auto &global_state = input.global_state.Cast(); - AddWrittenFiles(global_state, chunk, encryption_key, partition_id); + AddWrittenFiles(global_state, chunk, encryption_key, partition_id, false, file_format); return SinkResultType::NEED_MORE_INPUT; } @@ -482,50 +484,78 @@ static void GeneratePartitionExpressions(ClientContext &context, DuckLakeCopyInp } } +string DuckLakeInsert::ValidateDataFileFormat(string format) { + format = StringUtil::Lower(format); + // Formats DuckLake can write. Writing requires a COPY function that implements the copy_to_bind API + // (parquet and vortex do; e.g. json/csv use the newer plan-based COPY and cannot be written here). + // Any such file is read back via the generic format reader (read_). + if (format != "parquet" && format != "vortex") { + throw InvalidInputException( + "Unsupported data_file_format \"%s\" - DuckLake can write 'parquet' and 'vortex' data files", format); + } + return format; +} + DuckLakeCopyOptions DuckLakeInsert::GetCopyOptions(ClientContext &context, DuckLakeCopyInput ©_input) { auto info = make_uniq(); auto &catalog = copy_input.catalog; + auto &schema_id = copy_input.schema_id; + auto &table_id = copy_input.table_id; info->file_path = copy_input.data_path; - info->format = "parquet"; info->is_from = false; - // generate the field ids to be written by the parquet writer - shared_ptr generated_ids; - if (!copy_input.field_data) { - // CTAS - generate new ids from columns - generated_ids = DuckLakeFieldData::FromColumns(copy_input.columns); + + // Determine the physical file format for this table: parquet (default) or another supported format. + string data_file_format = "parquet"; + catalog.TryGetConfigOption("data_file_format", data_file_format, schema_id, table_id); + data_file_format = ValidateDataFileFormat(data_file_format); + const bool is_parquet = data_file_format == "parquet"; + info->format = data_file_format; + + if (is_parquet) { + // generate the field ids to be written by the parquet writer. Non-parquet files are mapped by + // column name at read time, so field ids are not written for them. + shared_ptr generated_ids; + if (!copy_input.field_data) { + // CTAS - generate new ids from columns + generated_ids = DuckLakeFieldData::FromColumns(copy_input.columns); + } + auto &field_ids = copy_input.field_data ? *copy_input.field_data : *generated_ids; + vector field_input; + field_input.push_back(WrittenFieldIds(field_ids, copy_input.virtual_columns)); + info->options["field_ids"] = std::move(field_input); } - auto &field_ids = copy_input.field_data ? *copy_input.field_data : *generated_ids; - vector field_input; - field_input.push_back(WrittenFieldIds(field_ids, copy_input.virtual_columns)); - info->options["field_ids"] = std::move(field_input); if (!copy_input.encryption_key.empty()) { + if (!is_parquet) { + throw NotImplementedException("Encryption is only supported for DuckLake tables written as parquet"); + } child_list_t values; values.emplace_back("footer_key_value", Value::BLOB_RAW(copy_input.encryption_key)); vector encryption_input; encryption_input.push_back(Value::STRUCT(std::move(values))); info->options["encryption_config"] = std::move(encryption_input); } - auto &schema_id = copy_input.schema_id; - auto &table_id = copy_input.table_id; - string parquet_compression; - if (catalog.TryGetConfigOption("parquet_compression", parquet_compression, schema_id, table_id)) { - info->options["compression"].emplace_back(parquet_compression); - } - string parquet_version; - if (catalog.TryGetConfigOption("parquet_version", parquet_version, schema_id, table_id)) { - info->options["parquet_version"].emplace_back(parquet_version); - } - string parquet_compression_level; - if (catalog.TryGetConfigOption("parquet_compression_level", parquet_compression_level, schema_id, table_id)) { - info->options["compression_level"].emplace_back(parquet_compression_level); - } - string row_group_size; - if (catalog.TryGetConfigOption("parquet_row_group_size", row_group_size, schema_id, table_id)) { - info->options["row_group_size"].emplace_back(row_group_size); - } - string row_group_size_bytes; - if (catalog.TryGetConfigOption("parquet_row_group_size_bytes", row_group_size_bytes, schema_id, table_id)) { - info->options["row_group_size_bytes"].emplace_back(row_group_size_bytes + " bytes"); + if (is_parquet) { + // The following options are parquet-writer specific. + string parquet_compression; + if (catalog.TryGetConfigOption("parquet_compression", parquet_compression, schema_id, table_id)) { + info->options["compression"].emplace_back(parquet_compression); + } + string parquet_version; + if (catalog.TryGetConfigOption("parquet_version", parquet_version, schema_id, table_id)) { + info->options["parquet_version"].emplace_back(parquet_version); + } + string parquet_compression_level; + if (catalog.TryGetConfigOption("parquet_compression_level", parquet_compression_level, schema_id, table_id)) { + info->options["compression_level"].emplace_back(parquet_compression_level); + } + string row_group_size; + if (catalog.TryGetConfigOption("parquet_row_group_size", row_group_size, schema_id, table_id)) { + info->options["row_group_size"].emplace_back(row_group_size); + } + string row_group_size_bytes; + if (catalog.TryGetConfigOption("parquet_row_group_size_bytes", row_group_size_bytes, schema_id, table_id)) { + info->options["row_group_size_bytes"].emplace_back(row_group_size_bytes + " bytes"); + } } string per_thread_output_str; bool per_thread_output = false; @@ -535,11 +565,19 @@ DuckLakeCopyOptions DuckLakeInsert::GetCopyOptions(ClientContext &context, DuckL idx_t target_file_size = catalog.GetConfigOption("target_file_size", schema_id, table_id, DuckLakeCatalog::DEFAULT_TARGET_FILE_SIZE); - // Always use native parquet geometry for writing - info->options["geoparquet_version"].emplace_back("NONE"); + if (is_parquet) { + // Always use native parquet geometry for writing + info->options["geoparquet_version"].emplace_back("NONE"); + } - // Get Parquet Copy function - auto ©_fun = DuckLakeFunctions::GetCopyFunction(context, "parquet"); + // Get the copy function for the chosen data file format + auto ©_fun = DuckLakeFunctions::GetCopyFunction(context, data_file_format); + if (!copy_fun.function.copy_to_bind) { + // DuckLake's writer drives the copy_to_bind API; plan-based copy functions are not supported. + throw NotImplementedException( + "The '%s' copy function cannot be used to write DuckLake data files (no copy_to_bind support)", + data_file_format); + } auto &fs = FileSystem::GetFileSystem(context); if (!fs.IsRemoteFile(copy_input.data_path)) { @@ -594,7 +632,7 @@ DuckLakeCopyOptions DuckLakeInsert::GetCopyOptions(ClientContext &context, DuckL } result.file_path = copy_input.data_path; StripTrailingSeparator(fs, result.file_path); - result.file_extension = "parquet"; + result.file_extension = data_file_format; result.overwrite_mode = CopyOverwriteMode::COPY_OVERWRITE_OR_IGNORE; result.per_thread_output = per_thread_output; result.write_partition_columns = true; @@ -731,7 +769,16 @@ PhysicalOperator &DuckLakeInsert::PlanInsert(ClientContext &context, PhysicalPla } vector return_types; return_types.emplace_back(LogicalType::BIGINT); - return planner.Make(return_types, table, partition_id, std::move(encryption_key)); + // Resolve the configured data file format so the sink records it on the written files. This must + // match the format GetCopyOptions selects for the same table. + string data_file_format = "parquet"; + auto &catalog = table.ParentCatalog().Cast(); + catalog.TryGetConfigOption("data_file_format", data_file_format, table); + data_file_format = ValidateDataFileFormat(data_file_format); + auto &insert = + planner.Make(return_types, table, partition_id, std::move(encryption_key)).Cast(); + insert.file_format = std::move(data_file_format); + return insert; } string DuckLakeCatalog::GenerateEncryptionKey(ClientContext &context) const { diff --git a/src/storage/ducklake_metadata_manager.cpp b/src/storage/ducklake_metadata_manager.cpp index 5417f101..0e0fc9fe 100644 --- a/src/storage/ducklake_metadata_manager.cpp +++ b/src/storage/ducklake_metadata_manager.cpp @@ -2957,7 +2957,7 @@ string DuckLakeMetadataManager::WriteNewDataFilesWithAppender(DuckLakeSnapshot & data_file_appender.Append(Value()); // file_order (NULL) data_file_appender.Append(string_t(path.path)); // path data_file_appender.Append(path.path_is_relative); // path_is_relative - data_file_appender.Append(string_t("parquet")); // file_format + data_file_appender.Append(string_t(file.file_format)); // file_format data_file_appender.Append(static_cast(file.row_count)); // record_count data_file_appender.Append(static_cast(file.file_size_bytes)); // file_size_bytes if (file.footer_size.IsValid()) { @@ -3174,9 +3174,9 @@ string DuckLakeMetadataManager::WriteNewDataFiles(DuckLakeSnapshot &commit_snaps string mapping = file.mapping_id.IsValid() ? to_string(file.mapping_id.index) : "NULL"; auto path = GetRelativePath(file.table_id, file.file_name, new_tables, new_schemas_result); data_file_insert_query += StringUtil::Format( - "(%d, %d, %s, NULL, NULL, %s, %s, 'parquet', %d, %d, %s, %s, %s, %s, %s, %s)", data_file_index, table_id, - begin_snapshot, SQLString(path.path), path.path_is_relative ? "true" : "false", file.row_count, - file.file_size_bytes, footer_size, row_id, partition_id, encryption_key, mapping, partial_max); + "(%d, %d, %s, NULL, NULL, %s, %s, %s, %d, %d, %s, %s, %s, %s, %s, %s)", data_file_index, table_id, + begin_snapshot, SQLString(path.path), path.path_is_relative ? "true" : "false", SQLString(file.file_format), + file.row_count, file.file_size_bytes, footer_size, row_id, partition_id, encryption_key, mapping, partial_max); for (auto &raw_stats : file.column_stats) { // Stringify stats to construct insert queries auto column_stats = DuckLakeColumnStatsInfo::FromColumnStats(raw_stats.first, raw_stats.second); diff --git a/src/storage/ducklake_multi_file_reader.cpp b/src/storage/ducklake_multi_file_reader.cpp index 4b3dbed2..6b7ede23 100644 --- a/src/storage/ducklake_multi_file_reader.cpp +++ b/src/storage/ducklake_multi_file_reader.cpp @@ -1,6 +1,6 @@ #include "storage/ducklake_multi_file_list.hpp" #include "storage/ducklake_multi_file_reader.hpp" -#include "storage/ducklake_vortex_reader.hpp" +#include "storage/ducklake_format_reader.hpp" #include "storage/ducklake_table_entry.hpp" #include "storage/ducklake_catalog.hpp" #include "storage/ducklake_delete_filter.hpp" @@ -387,7 +387,7 @@ shared_ptr DuckLakeMultiFileReader::TryCreateInlinedDataReader(c std::move(columns)); } -shared_ptr DuckLakeMultiFileReader::TryCreateVortexReader(ClientContext &context, +shared_ptr DuckLakeMultiFileReader::TryCreateFormatReader(ClientContext &context, const OpenFileInfo &file) { if (!file.extended_info) { return nullptr; @@ -397,12 +397,15 @@ shared_ptr DuckLakeMultiFileReader::TryCreateVortexReader(Client // no format recorded - default parquet path return nullptr; } - if (!StringUtil::CIEquals(StringValue::Get(entry->second), "vortex")) { + auto format = StringUtil::Lower(StringValue::Get(entry->second)); + if (format.empty() || format == "parquet") { + // parquet is read by the native parquet reader path return nullptr; } - // Map the vortex file's columns onto the table's current field ids by name. + // Non-parquet format: map the file's columns onto the table's current field ids by name and read it + // via the format's read_ table function. auto columns = DuckLakeMultiFileReader::ColumnsFromFieldData(read_info.table.GetFieldData(), true); - return make_shared_ptr(read_info, file, std::move(columns)); + return make_shared_ptr(read_info, file, std::move(format), std::move(columns)); } shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext &context, @@ -413,7 +416,7 @@ shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext & if (reader) { return reader; } - reader = TryCreateVortexReader(context, file); + reader = TryCreateFormatReader(context, file); if (reader) { return reader; } @@ -428,7 +431,7 @@ shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext & if (reader) { return reader; } - reader = TryCreateVortexReader(context, file); + reader = TryCreateFormatReader(context, file); if (reader) { return reader; } diff --git a/src/storage/ducklake_transaction.cpp b/src/storage/ducklake_transaction.cpp index 02c0b181..ccd77d5c 100644 --- a/src/storage/ducklake_transaction.cpp +++ b/src/storage/ducklake_transaction.cpp @@ -1926,6 +1926,7 @@ DuckLakeFileInfo DuckLakeTransaction::GetNewDataFile(const DuckLakeDataFile &fil data_file.id = DataFileIndex(commit_snapshot.next_file_id++); data_file.table_id = table_id; data_file.file_name = file.file_name; + data_file.file_format = file.file_format; data_file.row_count = file.row_count; data_file.file_size_bytes = file.file_size_bytes; data_file.footer_size = file.footer_size; diff --git a/test/sql/format/data_file_format_option.test b/test/sql/format/data_file_format_option.test new file mode 100644 index 00000000..d6514139 --- /dev/null +++ b/test/sql/format/data_file_format_option.test @@ -0,0 +1,45 @@ +# name: test/sql/format/data_file_format_option.test +# description: data_file_format option validation and file_format metadata recording +# group: [format] + +require ducklake + +require parquet + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_fmt.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_fmt_files', METADATA_CATALOG 'ducklake_meta') + +# write every insert to a data file (no metadata inlining) so file_format is exercised +statement ok +CALL ducklake.set_option('data_inlining_row_limit', 0) + +# default format is parquet, now explicitly recorded in the file_format column +statement ok +CREATE TABLE ducklake.t(i INTEGER, j INTEGER); + +statement ok +INSERT INTO ducklake.t VALUES (1, 2), (3, 4); + +query I +SELECT DISTINCT file_format FROM ducklake_meta.ducklake_data_file +---- +parquet + +# setting it explicitly to parquet is accepted +statement ok +CALL ducklake.set_option('data_file_format', 'parquet') + +# case-insensitive value +statement ok +CALL ducklake.set_option('data_file_format', 'PARQUET') + +# unsupported formats are rejected at set time +statement error +CALL ducklake.set_option('data_file_format', 'orc') +---- +Unsupported data_file_format + +statement error +CALL ducklake.set_option('data_file_format', 'avro') +---- +Unsupported data_file_format diff --git a/test/sql/format/json_read_dispatch.test b/test/sql/format/json_read_dispatch.test new file mode 100644 index 00000000..ee507d93 --- /dev/null +++ b/test/sql/format/json_read_dispatch.test @@ -0,0 +1,81 @@ +# name: test/sql/format/json_read_dispatch.test +# description: DuckLake scan dispatches to read_ for non-parquet data files (json), by file_format +# group: [format] +# +# DuckLake can only *write* formats whose COPY implements copy_to_bind (parquet, vortex). To verify the +# read-side per-file dispatch independently of the writer, we register a json file by repointing a data +# file's metadata row at it and tagging file_format='json'. + +require ducklake + +require parquet + +require json + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_jr.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_jr_files', METADATA_CATALOG 'ducklake_meta') + +statement ok +CALL ducklake.set_option('data_inlining_row_limit', 0) + +statement ok +CREATE TABLE ducklake.t(i INTEGER, s VARCHAR); + +statement ok +INSERT INTO ducklake.t VALUES (1, 'a'), (2, 'bb'), (3, 'ccc'); + +# sanity: currently a single parquet data file +query I +SELECT DISTINCT file_format FROM ducklake_meta.ducklake_data_file +---- +parquet + +# write the same rows out as a json file (outside DuckLake), preserving row order +statement ok +COPY (SELECT i, s FROM ducklake.t ORDER BY i) TO '{TEST_DIR}/injected.json' (FORMAT json) + +# repoint the data file at the json file and tag it as json +statement ok +UPDATE ducklake_meta.ducklake_data_file +SET path = '{TEST_DIR}/injected.json', path_is_relative = false, file_format = 'json' +WHERE end_snapshot IS NULL + +# reattach so DuckLake picks up the externally-modified metadata +statement ok +DETACH ducklake + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_jr.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_jr_files', METADATA_CATALOG 'ducklake_meta') + +# the scan now dispatches to the json reader +query IT +SELECT i, s FROM ducklake.t ORDER BY i +---- +1 a +2 bb +3 ccc + +# projection pushdown through the json reader +query T +SELECT s FROM ducklake.t ORDER BY i +---- +a +bb +ccc + +# filter pushdown through the json reader +query IT +SELECT i, s FROM ducklake.t WHERE i >= 2 ORDER BY i +---- +2 bb +3 ccc + +query I +SELECT COUNT(*) FROM ducklake.t +---- +3 + +query I +SELECT SUM(i) FROM ducklake.t +---- +6 diff --git a/test/sql/vortex/vortex_format.test b/test/sql/vortex/vortex_format.test new file mode 100644 index 00000000..f4812fec --- /dev/null +++ b/test/sql/vortex/vortex_format.test @@ -0,0 +1,95 @@ +# name: test/sql/vortex/vortex_format.test +# description: read+write DuckLake data files in the vortex format, incl. mixed parquet+vortex +# group: [vortex] +# +# Requires the vortex extension (the ./vortex submodule). NOTE: the vortex extension currently pins an +# older DuckDB than DuckLake main (its C++ glue uses Vector::GetBuffer / Flatten(count), removed in the +# DuckDB that DuckLake tracks), so this test only runs on a build where the two DuckDB pins are aligned. +# The identical read-dispatch code path is exercised format-agnostically by +# test/sql/format/json_read_dispatch.test in every build. + +require ducklake + +require parquet + +require vortex + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_vortex.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_vortex_files', METADATA_CATALOG 'ducklake_meta') + +# write every insert to a data file (no metadata inlining) +statement ok +CALL ducklake.set_option('data_inlining_row_limit', 0) + +statement ok +CALL ducklake.set_option('data_file_format', 'vortex') + +statement ok +CREATE TABLE ducklake.t(i INTEGER, s VARCHAR, d DOUBLE, b BOOLEAN, l INTEGER[]); + +statement ok +INSERT INTO ducklake.t VALUES + (1, 'a', 1.5, true, [1, 2, 3]), + (2, NULL, NULL, NULL, NULL), + (3, 'ccc', -2.25, false, []); + +query I +SELECT DISTINCT file_format FROM ducklake_meta.ducklake_data_file +---- +vortex + +query I +SELECT COUNT(*) FROM glob('{TEST_DIR}/dl_vortex_files/**/*.vortex') +---- +1 + +query IIIII +SELECT i, s, d, b, l FROM ducklake.t ORDER BY i +---- +1 a 1.5 true [1, 2, 3] +2 NULL NULL NULL NULL +3 ccc -2.25 false [] + +query II +SELECT i, s FROM ducklake.t WHERE i >= 2 ORDER BY i +---- +2 NULL +3 ccc + +query RR +SELECT SUM(i), MIN(d) FROM ducklake.t +---- +6 -2.25 + +statement ok +DELETE FROM ducklake.t WHERE i = 2; + +query II +SELECT i, s FROM ducklake.t ORDER BY i +---- +1 a +3 ccc + +# mixed vortex + parquet in one table +statement ok +CALL ducklake.set_option('data_file_format', 'parquet') + +statement ok +INSERT INTO ducklake.t VALUES (4, 'parq', 4.0, true, [9]); + +query I +SELECT COUNT(DISTINCT file_format) FROM ducklake_meta.ducklake_data_file WHERE end_snapshot IS NULL +---- +2 + +query II +SELECT i, s FROM ducklake.t ORDER BY i +---- +1 a +3 ccc +4 parq + +query I +SELECT SUM(i) FROM ducklake.t +---- +8 From 4e9776f3e92ab83030f7f402a103fa705cedc3db Mon Sep 17 00:00:00 2001 From: Mosha Pasumansky Date: Tue, 4 Aug 2026 13:28:33 -0700 Subject: [PATCH 3/3] feat(vortex): read DuckLake data files in the vortex format (per-file dispatch) Adds vortex (and general non-parquet) read support to DuckLake keyed off the per-file file_format column, on an aligned DuckDB base where DuckLake + vortex build together. - generic DuckLakeFormatReader drives read_ and adapts chunks to the MultiFileReader contract (column mapping, deletes, filters, file_row_number) - thread file_format through all scan paths (forward, insertions, delete-scan, compaction) and into transaction-local files + DuckLakeDataFile copy/assign - resolve the written format once on DuckLakeInsertGlobalState so INSERT, CTAS, flush and compaction all record it consistently - data_file_format catalog option; vendor vortex-data/duckdb-vortex submodule - pin DuckDB to d8cdaa33 (vortex's fork: old Vector layout + GEOMETRY/VARIANT C-API); two unique_ptr covariance fixes to build DuckLake against it - tests: vortex_read_dispatch (read + projection/filter pushdown + delete), json_read_dispatch, data_file_format_option Write path (INSERT -> vortex) is a follow-up: vortex's C-API COPY cannot report per-file statistics, so DuckLake's WRITTEN_FILE_STATISTICS writer would need to route through CHANGED_ROWS_AND_FILE_LIST + fstat. Co-Authored-By: Claude Opus 4.8 (1M context) --- duckdb | 2 +- extension_config.cmake | 8 +- src/common/ducklake_data_file.cpp | 2 + src/functions/ducklake_flush_inlined_data.cpp | 2 +- src/include/common/ducklake_data_file.hpp | 3 +- .../storage/ducklake_format_reader.hpp | 37 +++----- src/include/storage/ducklake_insert.hpp | 7 +- .../storage/ducklake_metadata_info.hpp | 7 +- src/storage/ducklake_format_reader.cpp | 56 ++++------- src/storage/ducklake_insert.cpp | 31 ++---- src/storage/ducklake_metadata_manager.cpp | 14 +-- src/storage/ducklake_multi_file_list.cpp | 4 +- src/storage/ducklake_multi_file_reader.cpp | 25 ++++- src/storage/ducklake_transaction.cpp | 2 +- test/sql/vortex/vortex_format.test | 95 ------------------- test/sql/vortex/vortex_name_map.test | 62 ++++++++++++ test/sql/vortex/vortex_read_dispatch.test | 89 +++++++++++++++++ 17 files changed, 236 insertions(+), 210 deletions(-) delete mode 100644 test/sql/vortex/vortex_format.test create mode 100644 test/sql/vortex/vortex_name_map.test create mode 100644 test/sql/vortex/vortex_read_dispatch.test diff --git a/duckdb b/duckdb index 710adf3d..d8cdaa33 160000 --- a/duckdb +++ b/duckdb @@ -1 +1 @@ -Subproject commit 710adf3d0587022701c56f0b83fc60011f57efff +Subproject commit d8cdaa33fda8df955cc76ef58a280f68f4cd43fa diff --git a/extension_config.cmake b/extension_config.cmake index d7823493..0525357d 100644 --- a/extension_config.cmake +++ b/extension_config.cmake @@ -11,11 +11,9 @@ if(NOT DEFINED ENV{DISABLE_EXTENSIONS_FOR_TEST}) duckdb_extension_load(tpch) endif() -# Vortex file format support (vendored as the ./vortex submodule). Building it -# requires a Rust toolchain (it compiles the vortex-extension crate via Corrosion), -# so it is opt-out via DISABLE_VORTEX for toolchains without cargo. When enabled, -# DuckLake can read and write data files whose file_format is 'vortex'. -if(NOT DEFINED ENV{DISABLE_VORTEX}) +# 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 ) diff --git a/src/common/ducklake_data_file.cpp b/src/common/ducklake_data_file.cpp index ed931bf8..6e7a06c1 100644 --- a/src/common/ducklake_data_file.cpp +++ b/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; diff --git a/src/functions/ducklake_flush_inlined_data.cpp b/src/functions/ducklake_flush_inlined_data.cpp index 30630e85..ceca4170 100644 --- a/src/functions/ducklake_flush_inlined_data.cpp +++ b/src/functions/ducklake_flush_inlined_data.cpp @@ -739,7 +739,7 @@ static unique_ptr FlushInlinedDataBind(ClientContext &context, auto projection = make_uniq(bind_index, std::move(proj_exprs)); projection->children.push_back(std::move(filter)); - return projection; + return unique_ptr(projection.release()); } DuckLakeFlushInlinedDataFunction::DuckLakeFlushInlinedDataFunction() diff --git a/src/include/common/ducklake_data_file.hpp b/src/include/common/ducklake_data_file.hpp index febe273c..64243dd3 100644 --- a/src/include/common/ducklake_data_file.hpp +++ b/src/include/common/ducklake_data_file.hpp @@ -62,8 +62,7 @@ struct DuckLakeDataFile { DuckLakeDataFile &operator=(const DuckLakeDataFile &); string file_name; - //! The physical format the file is stored in ("parquet", "vortex", ...). Recorded in - //! ducklake_data_file.file_format and used to pick the reader at scan time. + //! physical format ("parquet", "vortex", ...); recorded in ducklake_data_file.file_format string file_format = "parquet"; idx_t row_count; idx_t file_size_bytes; diff --git a/src/include/storage/ducklake_format_reader.hpp b/src/include/storage/ducklake_format_reader.hpp index 7e97b0f7..1f373080 100644 --- a/src/include/storage/ducklake_format_reader.hpp +++ b/src/include/storage/ducklake_format_reader.hpp @@ -15,21 +15,16 @@ namespace duckdb { struct DuckLakeFunctionInfo; -//! Marks how a single output column of the DuckLakeFormatReader is produced. +//! How an output column is produced. enum class FormatReaderColumn : uint8_t { - //! Projected directly from the underlying read_ scan (see source_index) - FILE_COLUMN, - //! Ordinal position of the row within the file (row_id_start + file_row_number is applied by the - //! MultiFileReader on top). Emitted by the reader itself. - ORDINAL, - //! Column that is not present in the file - emit its default value (usually NULL) - MISSING + 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 DuckLake data file stored in a non-parquet format (e.g. "vortex", "json"). It drives -//! the format's `read_` table function for the physical file and adapts the produced chunks to -//! DuckLake's MultiFileReader contract (column mapping, positional delete filters, pushed-down filters -//! and the file_row_number virtual column) exactly like the parquet reader path does. +//! Reads a single non-parquet DuckLake data file ("vortex", "json", ...) by driving its read_ +//! 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, @@ -44,17 +39,15 @@ class DuckLakeFormatReader : public BaseFileReader { void AddVirtualColumn(column_t virtual_column_id) override; private: - //! Look up the `read_` table function (auto-loading the format's extension if available). static TableFunction GetReadFunction(ClientContext &context, const string &format); mutex lock; DuckLakeFunctionInfo &read_info; - //! The physical format of the file ("vortex", "json", ...); selects the read_ function. - string format; + string format; //! selects the read_ function bool initialized_scan = false; int64_t file_row_number = 0; - //! Underlying read_ scan bound to the physical file. + //! read_ scan bound to the file TableFunction file_scan; unique_ptr bind_data; vector file_types; @@ -65,14 +58,10 @@ class DuckLakeFormatReader : public BaseFileReader { unique_ptr global_state; unique_ptr local_state; - //! Column ids projected out of the file (indexes into file_names/file_types). - vector scan_column_ids; - //! Chunk holding the raw columns read from read_ (in scan_column_ids order). - DataChunk scan_chunk; - //! For each output column: how it is produced. - vector output_kinds; - //! For FILE_COLUMN outputs: the position within scan_chunk to reference. - vector output_source_index; + vector scan_column_ids; //! columns projected out of the file + DataChunk scan_chunk; //! raw columns read from the file + vector output_kinds; //! how each output column is produced + vector output_source_index; //! for FILE_COLUMN: position in scan_chunk }; } // namespace duckdb diff --git a/src/include/storage/ducklake_insert.hpp b/src/include/storage/ducklake_insert.hpp index 850500d3..eb25e112 100644 --- a/src/include/storage/ducklake_insert.hpp +++ b/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 { @@ -63,8 +65,6 @@ class DuckLakeInsert : public PhysicalOperator { optional_idx partition_id; //! The encryption key used for writing the Parquet files string encryption_key; - //! The physical format written data files are stored in ("parquet", "vortex", ...) - string file_format = "parquet"; public: // // Source interface @@ -88,8 +88,7 @@ class DuckLakeInsert : public PhysicalOperator { static PhysicalOperator &PlanInsert(ClientContext &context, PhysicalPlanGenerator &planner, DuckLakeTableEntry &table, string encryption_key); static void AddWrittenFiles(DuckLakeInsertGlobalState &gstate, DataChunk &chunk, const string &encryption_key, - optional_idx partition_id, bool set_snapshot_id = false, - const string &file_format = "parquet"); + optional_idx partition_id, bool set_snapshot_id = false); static const DuckLakeFieldId &GetTopLevelColumn(DuckLakeCopyInput ©_input, FieldIndex field_id, optional_idx &index); diff --git a/src/include/storage/ducklake_metadata_info.hpp b/src/include/storage/ducklake_metadata_info.hpp index 958e4642..aa9680d5 100644 --- a/src/include/storage/ducklake_metadata_info.hpp +++ b/src/include/storage/ducklake_metadata_info.hpp @@ -149,7 +149,7 @@ struct DuckLakeFileInfo { DataFileIndex id; TableIndex table_id; string file_name; - //! Physical storage format ("parquet", "vortex", ...); persisted to ducklake_data_file.file_format. + //! physical format; persisted to ducklake_data_file.file_format string file_format = "parquet"; idx_t row_count; idx_t file_size_bytes; @@ -354,10 +354,9 @@ struct DuckLakeFileData { string encryption_key; idx_t file_size_bytes = 0; optional_idx footer_size; - //! Delete-file format (parquet/puffin); only meaningful for delete files. + //! delete-file format (parquet/puffin); delete files only DeleteFileFormat format = DeleteFileFormat::PARQUET; - //! Physical data-file format read from ducklake_data_file.file_format ("parquet" or "vortex"). - //! Selects which reader is used to scan this file. + //! data-file format from ducklake_data_file.file_format; selects the reader string data_file_format = "parquet"; }; diff --git a/src/storage/ducklake_format_reader.cpp b/src/storage/ducklake_format_reader.cpp index c18cb86e..bfc29ac3 100644 --- a/src/storage/ducklake_format_reader.cpp +++ b/src/storage/ducklake_format_reader.cpp @@ -7,7 +7,6 @@ #include "duckdb/main/extension_helper.hpp" #include "duckdb/parser/tableref/table_function_ref.hpp" #include "duckdb/planner/table_filter_state.hpp" -#include "duckdb/planner/filter/expression_filter.hpp" #include "duckdb/storage/table/column_segment.hpp" #include "duckdb/common/vector_operations/vector_operations.hpp" @@ -21,25 +20,15 @@ DuckLakeFormatReader::DuckLakeFormatReader(DuckLakeFunctionInfo &read_info, cons TableFunction DuckLakeFormatReader::GetReadFunction(ClientContext &context, const string &format) { auto &instance = DatabaseInstance::GetDatabase(context); - // The extension registering read_ may need to be auto-loaded (no-op for core formats). ExtensionHelper::TryAutoLoadExtension(instance, format); ExtensionLoader loader(instance, "ducklake"); - auto function_name = "read_" + format; - auto entry = loader.TryGetTableFunction(Identifier(function_name)); - if (!entry) { - throw MissingExtensionException( - "Reading DuckLake data files in the '%s' format requires the '%s' table function to be available " - "(install/load the extension providing it)", - format, function_name); - } - auto &scan_entry = entry->Cast(); - return scan_entry.functions.GetFunctionByOffset(0); + auto &scan_entry = loader.GetTableFunction("read_" + format); + return scan_entry.functions.functions[0]; } bool DuckLakeFormatReader::TryInitializeScan(ClientContext &context, GlobalTableFunctionState &gstate, LocalTableFunctionState &lstate) { { - // only one thread reads a given file lock_guard guard(lock); if (initialized_scan) { return false; @@ -47,53 +36,45 @@ bool DuckLakeFormatReader::TryInitializeScan(ClientContext &context, GlobalTable initialized_scan = true; } - // Bind read_ on the physical file to discover its schema. + // bind the file to discover its schema file_scan = GetReadFunction(context, format); vector children; children.push_back(Value(file.path)); named_parameter_map_t named_params; vector input_types; - vector input_names; - - // Some read functions require schema auto-detection to be requested explicitly when bound - // programmatically (the SQL wrappers enable it by default). read_json is one such case. + vector input_names; if (StringUtil::CIEquals(format, "json")) { + // read_json only auto-detects when asked; the SQL wrapper does this for us named_params["auto_detect"] = Value::BOOLEAN(true); } TableFunctionRef empty; TableFunction dummy_table_function; - dummy_table_function.SetName("DuckLakeFormatReader"); + dummy_table_function.name = "DuckLakeFormatReader"; TableFunctionBindInput bind_input(children, named_params, input_types, input_names, nullptr, nullptr, dummy_table_function, empty); bind_data = file_scan.bind(context, bind_input, file_types, file_names); - // Build the projection into the file and the mapping to the reader's output columns. + // map each projected column onto a file column (by name), the ordinal, or NULL if absent vector scan_types; for (auto &column_id : column_indexes) { - auto index = column_id.GetPrimaryIndex(); - auto &col = columns[index]; - // Virtual columns carry an INTEGER identifier. The only one the reader must materialize itself - // is the ordinal (file_row_number); higher-level virtuals (row_id / snapshot_id) are computed by - // the MultiFileReader as expressions on top of the ordinal. + auto &col = columns[column_id.GetPrimaryIndex()]; if (!col.identifier.IsNull() && col.identifier.type().id() == LogicalTypeId::INTEGER && IntegerValue::Get(col.identifier) == MultiFileReader::ORDINAL_FIELD_ID) { + // file_row_number; row_id/snapshot_id are derived from it by the MultiFileReader output_kinds.push_back(FormatReaderColumn::ORDINAL); output_source_index.push_back(0); continue; } - // Regular column - locate it in the file by name. - auto column_name = col.name.GetIdentifierName(); optional_idx file_position; for (idx_t i = 0; i < file_names.size(); i++) { - if (StringUtil::CIEquals(file_names[i], column_name)) { + if (StringUtil::CIEquals(file_names[i], col.name)) { file_position = i; break; } } if (!file_position.IsValid()) { - // column is not present in this file - emit its default (NULL) output_kinds.push_back(FormatReaderColumn::MISSING); output_source_index.push_back(0); continue; @@ -103,8 +84,7 @@ bool DuckLakeFormatReader::TryInitializeScan(ClientContext &context, GlobalTable scan_column_ids.push_back(file_position.GetIndex()); scan_types.push_back(file_types[file_position.GetIndex()]); } - - // read_ needs at least one projected column to determine the cardinality of each chunk. + // need at least one projected column so the scan can report a chunk cardinality if (scan_column_ids.empty() && !file_types.empty()) { scan_column_ids.push_back(0); scan_types.push_back(file_types[0]); @@ -141,7 +121,7 @@ AsyncResult DuckLakeFormatReader::Scan(ClientContext &context, GlobalTableFuncti break; } case FormatReaderColumn::ORDINAL: { - auto ordinal_data = FlatVector::GetDataMutable(chunk.data[c]); + auto ordinal_data = FlatVector::GetData(chunk.data[c]); for (idx_t r = 0; r < scan_count; r++) { ordinal_data[r] = file_row_number + NumericCast(r); } @@ -159,19 +139,17 @@ AsyncResult DuckLakeFormatReader::Scan(ClientContext &context, GlobalTableFuncti SelectionVector sel; idx_t approved_tuple_count = scan_count; if (deletion_filter) { - // delete filters are positional within the file (row_id_start + file_row_number) approved_tuple_count = deletion_filter->Filter(file_row_number, approved_tuple_count, sel); } if (filters) { - for (auto &entry : *filters) { - auto &filter = entry.Filter(); - if (ExpressionFilter::IsRootOptionalFilter(filter)) { + for (auto &entry : filters->filters) { + if (entry.second->filter_type == TableFilterType::OPTIONAL_FILTER) { continue; } - auto column_id = entry.GetIndex().GetIndex(); - auto &vec = chunk.data[column_id]; + auto &vec = chunk.data[entry.first]; UnifiedVectorFormat vdata; - vec.ToUnifiedFormat(vdata); + vec.ToUnifiedFormat(chunk.size(), vdata); + auto &filter = *entry.second; auto filter_state = TableFilterState::Initialize(context, filter); approved_tuple_count = ColumnSegment::FilterSelection(sel, vec, vdata, filter, *filter_state, chunk.size(), approved_tuple_count); diff --git a/src/storage/ducklake_insert.cpp b/src/storage/ducklake_insert.cpp index cee4f5a1..e653b303 100644 --- a/src/storage/ducklake_insert.cpp +++ b/src/storage/ducklake_insert.cpp @@ -45,6 +45,10 @@ DuckLakeInsert::DuckLakeInsert(PhysicalPlan &physical_plan, const vector(); + catalog.TryGetConfigOption("data_file_format", file_format, table); + file_format = DuckLakeInsert::ValidateDataFileFormat(file_format); } unique_ptr DuckLakeInsert::GetGlobalSinkState(ClientContext &context) const { @@ -103,12 +107,11 @@ DuckLakeColumnStats DuckLakeInsert::ParseColumnStats(const LogicalType &type, co } void DuckLakeInsert::AddWrittenFiles(DuckLakeInsertGlobalState &global_state, DataChunk &chunk, - const string &encryption_key, optional_idx partition_id, bool set_snapshot_id, - const string &file_format) { + const string &encryption_key, optional_idx partition_id, bool set_snapshot_id) { for (idx_t r = 0; r < chunk.size(); r++) { DuckLakeDataFile data_file; data_file.file_name = chunk.GetValue(0, r).GetValue(); - data_file.file_format = file_format; + data_file.file_format = global_state.file_format; data_file.row_count = chunk.GetValue(1, r).GetValue(); data_file.file_size_bytes = chunk.GetValue(2, r).GetValue(); data_file.footer_size = chunk.GetValue(3, r).GetValue(); @@ -213,7 +216,7 @@ void DuckLakeInsert::AddWrittenFiles(DuckLakeInsertGlobalState &global_state, Da SinkResultType DuckLakeInsert::Sink(ExecutionContext &context, DataChunk &chunk, OperatorSinkInput &input) const { auto &global_state = input.global_state.Cast(); - AddWrittenFiles(global_state, chunk, encryption_key, partition_id, false, file_format); + AddWrittenFiles(global_state, chunk, encryption_key, partition_id); return SinkResultType::NEED_MORE_INPUT; } @@ -486,9 +489,7 @@ static void GeneratePartitionExpressions(ClientContext &context, DuckLakeCopyInp string DuckLakeInsert::ValidateDataFileFormat(string format) { format = StringUtil::Lower(format); - // Formats DuckLake can write. Writing requires a COPY function that implements the copy_to_bind API - // (parquet and vortex do; e.g. json/csv use the newer plan-based COPY and cannot be written here). - // Any such file is read back via the generic format reader (read_). + // writable formats need a copy_to_bind COPY (json/csv use the newer plan-based COPY and can't) if (format != "parquet" && format != "vortex") { throw InvalidInputException( "Unsupported data_file_format \"%s\" - DuckLake can write 'parquet' and 'vortex' data files", format); @@ -512,8 +513,7 @@ DuckLakeCopyOptions DuckLakeInsert::GetCopyOptions(ClientContext &context, DuckL info->format = data_file_format; if (is_parquet) { - // generate the field ids to be written by the parquet writer. Non-parquet files are mapped by - // column name at read time, so field ids are not written for them. + // field ids are parquet-only; non-parquet files are mapped by column name at read time shared_ptr generated_ids; if (!copy_input.field_data) { // CTAS - generate new ids from columns @@ -535,7 +535,6 @@ DuckLakeCopyOptions DuckLakeInsert::GetCopyOptions(ClientContext &context, DuckL info->options["encryption_config"] = std::move(encryption_input); } if (is_parquet) { - // The following options are parquet-writer specific. string parquet_compression; if (catalog.TryGetConfigOption("parquet_compression", parquet_compression, schema_id, table_id)) { info->options["compression"].emplace_back(parquet_compression); @@ -573,7 +572,6 @@ DuckLakeCopyOptions DuckLakeInsert::GetCopyOptions(ClientContext &context, DuckL // Get the copy function for the chosen data file format auto ©_fun = DuckLakeFunctions::GetCopyFunction(context, data_file_format); if (!copy_fun.function.copy_to_bind) { - // DuckLake's writer drives the copy_to_bind API; plan-based copy functions are not supported. throw NotImplementedException( "The '%s' copy function cannot be used to write DuckLake data files (no copy_to_bind support)", data_file_format); @@ -769,16 +767,7 @@ PhysicalOperator &DuckLakeInsert::PlanInsert(ClientContext &context, PhysicalPla } vector return_types; return_types.emplace_back(LogicalType::BIGINT); - // Resolve the configured data file format so the sink records it on the written files. This must - // match the format GetCopyOptions selects for the same table. - string data_file_format = "parquet"; - auto &catalog = table.ParentCatalog().Cast(); - catalog.TryGetConfigOption("data_file_format", data_file_format, table); - data_file_format = ValidateDataFileFormat(data_file_format); - auto &insert = - planner.Make(return_types, table, partition_id, std::move(encryption_key)).Cast(); - insert.file_format = std::move(data_file_format); - return insert; + return planner.Make(return_types, table, partition_id, std::move(encryption_key)); } string DuckLakeCatalog::GenerateEncryptionKey(ClientContext &context) const { diff --git a/src/storage/ducklake_metadata_manager.cpp b/src/storage/ducklake_metadata_manager.cpp index 0e0fc9fe..68db76d4 100644 --- a/src/storage/ducklake_metadata_manager.cpp +++ b/src/storage/ducklake_metadata_manager.cpp @@ -888,8 +888,8 @@ string DuckLakeMetadataManager::GetFileSelectList(const string &prefix, bool inc }); if (include_data_file_format) { - // Appended as the trailing column so it is read right after ReadDataFile's fixed columns. - // Only valid for the ducklake_data_file table (delete files do not have a file_format column). + // trailing column, read right after ReadDataFile's fixed ones (data files only; delete files + // have no file_format column) result += ", " + prefix + ".file_format AS " + prefix + "_file_format"; } @@ -1544,7 +1544,7 @@ vector DuckLakeMetadataManager::GetTableDeletions(DuckL DuckLakeSnapshot start_snapshot, DuckLakeSnapshot end_snapshot) { auto table_id = table.GetTableId(); - string select_list = "data.data_file_id, " + GetFileSelectList("data") + + string select_list = "data.data_file_id, " + GetFileSelectList("data", true) + ", data.row_id_start, data.record_count, data.mapping_id, " + GetDeleteFileSelectList("current_delete") + ", " + GetDeleteFileSelectList("previous_delete"); @@ -1655,7 +1655,7 @@ WHERE data.table_id = %d ) AND (data.end_snapshot IS NULL OR data.end_snapshot < %d OR data.end_snapshot > {SNAPSHOT_ID}) )", - GetFileSelectList("data"), null_file_cols, null_file_cols, table_id.index, + GetFileSelectList("data", true), null_file_cols, null_file_cols, table_id.index, table_id.index, start_snapshot.snapshot_id); } @@ -1687,7 +1687,7 @@ FROM main_results idx_t col_idx = 0; auto file_id = row.GetValue(col_idx++); entry.file_id = DataFileIndex(file_id); - entry.file = ReadDataFile(table, row, col_idx, IsEncrypted()); + entry.file = ReadDataFile(table, row, col_idx, IsEncrypted(), true); if (!row.IsNull(col_idx)) { entry.row_id_start = row.GetValue(col_idx); } @@ -1798,7 +1798,7 @@ vector DuckLakeMetadataManager::GetFilesForCompacti string data_select_list = "data.data_file_id, data.record_count, data.row_id_start, data.begin_snapshot, " "data.end_snapshot, data.mapping_id, sr.schema_version , data.partial_max, " "data.partition_id, partition_info.keys, " + - GetFileSelectList("data"); + GetFileSelectList("data", true); string delete_select_list = "del.data_file_id AS del_data_file_id," "del.delete_file_id AS del_delete_file_id, " "del.delete_count, " @@ -1890,7 +1890,7 @@ ORDER BY data.begin_snapshot, data.row_id_start, data.data_file_id, del.begin_sn } } col_idx++; - new_entry.file.data = ReadDataFile(table, row, col_idx, IsEncrypted()); + new_entry.file.data = ReadDataFile(table, row, col_idx, IsEncrypted(), true); if (files.empty() || files.back().file.id != new_entry.file.id) { // new file - push it into the file list files.push_back(std::move(new_entry)); diff --git a/src/storage/ducklake_multi_file_list.cpp b/src/storage/ducklake_multi_file_list.cpp index 11d836ef..8da62040 100644 --- a/src/storage/ducklake_multi_file_list.cpp +++ b/src/storage/ducklake_multi_file_list.cpp @@ -174,8 +174,7 @@ OpenFileInfo DuckLakeMultiFileList::GetFile(idx_t i) const { Value::BIGINT(NumericCast(inlined_data_table.schema_version)); } else { extended_info->options["file_size"] = Value::UBIGINT(file.file_size_bytes); - // Record the physical format so the multi-file reader can pick the right per-file reader - // (parquet_scan vs. read_vortex). Absent/empty means parquet. + // physical format -> picks the per-file reader (absent/empty means parquet) if (!file.data_file_format.empty()) { extended_info->options["file_format"] = Value(file.data_file_format); } @@ -237,6 +236,7 @@ DuckLakeFileData GetFileData(const DuckLakeDataFile &file) { result.encryption_key = file.encryption_key; result.file_size_bytes = file.file_size_bytes; result.footer_size = file.footer_size; + result.data_file_format = file.file_format; return result; } diff --git a/src/storage/ducklake_multi_file_reader.cpp b/src/storage/ducklake_multi_file_reader.cpp index 6b7ede23..adb466be 100644 --- a/src/storage/ducklake_multi_file_reader.cpp +++ b/src/storage/ducklake_multi_file_reader.cpp @@ -394,17 +394,34 @@ shared_ptr DuckLakeMultiFileReader::TryCreateFormatReader(Client } auto entry = file.extended_info->options.find("file_format"); if (entry == file.extended_info->options.end()) { - // no format recorded - default parquet path return nullptr; } auto format = StringUtil::Lower(StringValue::Get(entry->second)); if (format.empty() || format == "parquet") { - // parquet is read by the native parquet reader path + // parquet uses the native reader path return nullptr; } - // Non-parquet format: map the file's columns onto the table's current field ids by name and read it - // via the format's read_ table function. auto columns = DuckLakeMultiFileReader::ColumnsFromFieldData(read_info.table.GetFieldData(), true); + // if the file has a name map (e.g. columns were renamed after it was written), look columns up by + // their physical name in the file instead of the current schema name + auto mapping_entry = file.extended_info->options.find("mapping_id"); + if (mapping_entry != file.extended_info->options.end()) { + auto mapping_id = MappingIndex(mapping_entry->second.GetValue()); + auto &name_map = read_info.GetTransaction()->GetMappingById(mapping_id); + unordered_map physical_names; + for (auto &column_map : name_map.column_maps) { + physical_names[column_map->target_field_id.index] = column_map->source_name; + } + for (auto &col : columns) { + if (col.identifier.IsNull() || col.identifier.type().id() != LogicalTypeId::INTEGER) { + continue; + } + auto name_entry = physical_names.find(IntegerValue::Get(col.identifier)); + if (name_entry != physical_names.end()) { + col.name = name_entry->second; + } + } + } return make_shared_ptr(read_info, file, std::move(format), std::move(columns)); } diff --git a/src/storage/ducklake_transaction.cpp b/src/storage/ducklake_transaction.cpp index ccd77d5c..9baf925c 100644 --- a/src/storage/ducklake_transaction.cpp +++ b/src/storage/ducklake_transaction.cpp @@ -2602,7 +2602,7 @@ unique_ptr DuckLakeTransaction::Query(string query) { if (cb) { cb(query, end - start); } - return result; + return unique_ptr(result.release()); } unique_ptr DuckLakeTransaction::Query(DuckLakeSnapshot snapshot, string query) { diff --git a/test/sql/vortex/vortex_format.test b/test/sql/vortex/vortex_format.test deleted file mode 100644 index f4812fec..00000000 --- a/test/sql/vortex/vortex_format.test +++ /dev/null @@ -1,95 +0,0 @@ -# name: test/sql/vortex/vortex_format.test -# description: read+write DuckLake data files in the vortex format, incl. mixed parquet+vortex -# group: [vortex] -# -# Requires the vortex extension (the ./vortex submodule). NOTE: the vortex extension currently pins an -# older DuckDB than DuckLake main (its C++ glue uses Vector::GetBuffer / Flatten(count), removed in the -# DuckDB that DuckLake tracks), so this test only runs on a build where the two DuckDB pins are aligned. -# The identical read-dispatch code path is exercised format-agnostically by -# test/sql/format/json_read_dispatch.test in every build. - -require ducklake - -require parquet - -require vortex - -statement ok -ATTACH 'ducklake:{TEST_DIR}/dl_vortex.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_vortex_files', METADATA_CATALOG 'ducklake_meta') - -# write every insert to a data file (no metadata inlining) -statement ok -CALL ducklake.set_option('data_inlining_row_limit', 0) - -statement ok -CALL ducklake.set_option('data_file_format', 'vortex') - -statement ok -CREATE TABLE ducklake.t(i INTEGER, s VARCHAR, d DOUBLE, b BOOLEAN, l INTEGER[]); - -statement ok -INSERT INTO ducklake.t VALUES - (1, 'a', 1.5, true, [1, 2, 3]), - (2, NULL, NULL, NULL, NULL), - (3, 'ccc', -2.25, false, []); - -query I -SELECT DISTINCT file_format FROM ducklake_meta.ducklake_data_file ----- -vortex - -query I -SELECT COUNT(*) FROM glob('{TEST_DIR}/dl_vortex_files/**/*.vortex') ----- -1 - -query IIIII -SELECT i, s, d, b, l FROM ducklake.t ORDER BY i ----- -1 a 1.5 true [1, 2, 3] -2 NULL NULL NULL NULL -3 ccc -2.25 false [] - -query II -SELECT i, s FROM ducklake.t WHERE i >= 2 ORDER BY i ----- -2 NULL -3 ccc - -query RR -SELECT SUM(i), MIN(d) FROM ducklake.t ----- -6 -2.25 - -statement ok -DELETE FROM ducklake.t WHERE i = 2; - -query II -SELECT i, s FROM ducklake.t ORDER BY i ----- -1 a -3 ccc - -# mixed vortex + parquet in one table -statement ok -CALL ducklake.set_option('data_file_format', 'parquet') - -statement ok -INSERT INTO ducklake.t VALUES (4, 'parq', 4.0, true, [9]); - -query I -SELECT COUNT(DISTINCT file_format) FROM ducklake_meta.ducklake_data_file WHERE end_snapshot IS NULL ----- -2 - -query II -SELECT i, s FROM ducklake.t ORDER BY i ----- -1 a -3 ccc -4 parq - -query I -SELECT SUM(i) FROM ducklake.t ----- -8 diff --git a/test/sql/vortex/vortex_name_map.test b/test/sql/vortex/vortex_name_map.test new file mode 100644 index 00000000..a1e43284 --- /dev/null +++ b/test/sql/vortex/vortex_name_map.test @@ -0,0 +1,62 @@ +# name: test/sql/vortex/vortex_name_map.test +# description: the format reader resolves renamed columns via the file's mapping_id (name map) +# group: [vortex] +# +# After a column rename, a non-parquet file keeps its original physical column names. DuckLake records +# a name map (mapping_id) mapping the new field ids to those physical names. The format reader must look +# columns up by physical name, not by the current schema name. + +require ducklake + +require parquet + +require vortex + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_nm.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_nm_files', METADATA_CATALOG 'ducklake_meta') + +statement ok +CALL ducklake.set_option('data_inlining_row_limit', 0) + +statement ok +CREATE TABLE ducklake.t(i INTEGER, s VARCHAR) + +statement ok +INSERT INTO ducklake.t VALUES (1, 'a'), (2, 'bb'), (3, 'ccc') + +# write a vortex file with physical column names i, s and register it +statement ok +COPY (SELECT i, s FROM ducklake.t ORDER BY i) TO '{TEST_DIR}/nm.vortex' (FORMAT vortex) + +statement ok +UPDATE ducklake_meta.ducklake_data_file +SET path = '{TEST_DIR}/nm.vortex', path_is_relative = false, file_format = 'vortex' +WHERE end_snapshot IS NULL + +# rename s -> s2: the file still has physical name 's' +statement ok +ALTER TABLE ducklake.t RENAME COLUMN s TO s2 + +# register a name map for the file: field id 2 (s2) -> physical name 's' +statement ok +INSERT INTO ducklake_meta.ducklake_column_mapping VALUES (100, 1, 'map_by_name') + +statement ok +INSERT INTO ducklake_meta.ducklake_name_mapping VALUES (100, 1, 'i', 1, NULL, false), (100, 2, 's', 2, NULL, false) + +statement ok +UPDATE ducklake_meta.ducklake_data_file SET mapping_id = 100 WHERE end_snapshot IS NULL + +statement ok +DETACH ducklake + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_nm.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_nm_files', METADATA_CATALOG 'ducklake_meta') + +# the renamed column resolves to the physical name in the vortex file (would be NULL without the fix) +query IT +SELECT i, s2 FROM ducklake.t ORDER BY i +---- +1 a +2 bb +3 ccc diff --git a/test/sql/vortex/vortex_read_dispatch.test b/test/sql/vortex/vortex_read_dispatch.test new file mode 100644 index 00000000..64ddf622 --- /dev/null +++ b/test/sql/vortex/vortex_read_dispatch.test @@ -0,0 +1,89 @@ +# name: test/sql/vortex/vortex_read_dispatch.test +# description: DuckLake scans data files tagged file_format='vortex' via read_vortex (per-file dispatch) +# group: [vortex] +# +# Writes a real .vortex file with the vortex extension's own COPY, registers it into DuckLake by +# tagging the data file's file_format='vortex', and verifies DuckLake reads it back through the +# generic DuckLakeFormatReader (driving read_vortex). + +require ducklake + +require parquet + +require vortex + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_vr.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_vr_files', METADATA_CATALOG 'ducklake_meta') + +statement ok +CALL ducklake.set_option('data_inlining_row_limit', 0) + +statement ok +CREATE TABLE ducklake.t(i INTEGER, s VARCHAR) + +statement ok +INSERT INTO ducklake.t VALUES (1, 'a'), (2, 'bb'), (3, 'ccc') + +query I +SELECT DISTINCT file_format FROM ducklake_meta.ducklake_data_file +---- +parquet + +# write the same rows to a real vortex file using the vortex extension's own COPY +statement ok +COPY (SELECT i, s FROM ducklake.t ORDER BY i) TO '{TEST_DIR}/injected.vortex' (FORMAT vortex) + +# repoint the data file at the vortex file and tag it as vortex +statement ok +UPDATE ducklake_meta.ducklake_data_file +SET path = '{TEST_DIR}/injected.vortex', path_is_relative = false, file_format = 'vortex' +WHERE end_snapshot IS NULL + +statement ok +DETACH ducklake + +statement ok +ATTACH 'ducklake:{TEST_DIR}/dl_vr.db' AS ducklake (DATA_PATH '{TEST_DIR}/dl_vr_files', METADATA_CATALOG 'ducklake_meta') + +# the scan now dispatches to read_vortex +query IT +SELECT i, s FROM ducklake.t ORDER BY i +---- +1 a +2 bb +3 ccc + +# projection pushdown through the vortex reader +query T +SELECT s FROM ducklake.t ORDER BY i +---- +a +bb +ccc + +# filter pushdown through the vortex reader +query IT +SELECT i, s FROM ducklake.t WHERE i >= 2 ORDER BY i +---- +2 bb +3 ccc + +query II +SELECT COUNT(*), SUM(i) FROM ducklake.t +---- +3 6 + +# deletes apply positionally on top of the vortex reader +statement ok +DELETE FROM ducklake.t WHERE i = 2 + +query IT +SELECT i, s FROM ducklake.t ORDER BY i +---- +1 a +3 ccc + +query I +SELECT COUNT(*) FROM ducklake.t +---- +2