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/.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/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 2bb8bb13..0525357d 100644 --- a/extension_config.cmake +++ b/extension_config.cmake @@ -11,6 +11,14 @@ if(NOT DEFINED ENV{DISABLE_EXTENSIONS_FOR_TEST}) duckdb_extension_load(tpch) endif() +# Vortex format support (./vortex submodule). Opt-in: needs a Rust toolchain and DuckDB pinned to +# vortex's fork (see .gitmodules), so it is off in the standard distribution CI. Enable with ENABLE_VORTEX=1. +if(DEFINED ENV{ENABLE_VORTEX}) + duckdb_extension_load(vortex + SOURCE_DIR ${CMAKE_CURRENT_LIST_DIR}/vortex + ) +endif() + set(EXTENSION_CONFIG_BASE_DIR "${CMAKE_CURRENT_LIST_DIR}/.github/config/extensions/") if($ENV{ENABLE_SQLITE_SCANNER}) include("${EXTENSION_CONFIG_BASE_DIR}/sqlite_scanner.cmake") 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/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..64243dd3 100644 --- a/src/include/common/ducklake_data_file.hpp +++ b/src/include/common/ducklake_data_file.hpp @@ -62,6 +62,8 @@ struct DuckLakeDataFile { DuckLakeDataFile &operator=(const DuckLakeDataFile &); string file_name; + //! physical format ("parquet", "vortex", ...); recorded in ducklake_data_file.file_format + string file_format = "parquet"; idx_t row_count; idx_t file_size_bytes; optional_idx footer_size; diff --git a/src/include/storage/ducklake_format_reader.hpp b/src/include/storage/ducklake_format_reader.hpp new file mode 100644 index 00000000..1f373080 --- /dev/null +++ b/src/include/storage/ducklake_format_reader.hpp @@ -0,0 +1,67 @@ +//===----------------------------------------------------------------------===// +// DuckDB +// +// storage/ducklake_format_reader.hpp +// +// +//===----------------------------------------------------------------------===// + +#pragma once + +#include "duckdb/common/multi_file/base_file_reader.hpp" +#include "duckdb/function/table_function.hpp" +#include "duckdb/parallel/thread_context.hpp" + +namespace duckdb { +struct DuckLakeFunctionInfo; + +//! How an output column is produced. +enum class FormatReaderColumn : uint8_t { + FILE_COLUMN, //! read from the file (see output_source_index) + ORDINAL, //! file_row_number, emitted by the reader + MISSING //! not in the file -> NULL +}; + +//! Reads a single non-parquet DuckLake data file ("vortex", "json", ...) by driving its read_ +//! table function and adapting the chunks to DuckLake's MultiFileReader contract (column mapping, +//! delete filters, pushed-down filters, file_row_number), like the parquet reader path does. +class DuckLakeFormatReader : public BaseFileReader { +public: + DuckLakeFormatReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, string format, + vector columns); + +public: + bool TryInitializeScan(ClientContext &context, GlobalTableFunctionState &gstate, + LocalTableFunctionState &lstate) override; + AsyncResult Scan(ClientContext &context, GlobalTableFunctionState &global_state, + LocalTableFunctionState &local_state, DataChunk &chunk) override; + string GetReaderType() const override; + void AddVirtualColumn(column_t virtual_column_id) override; + +private: + static TableFunction GetReadFunction(ClientContext &context, const string &format); + + mutex lock; + DuckLakeFunctionInfo &read_info; + string format; //! selects the read_ function + bool initialized_scan = false; + int64_t file_row_number = 0; + + //! read_ scan bound to the file + TableFunction file_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; + + 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 c591763c..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 { @@ -78,6 +80,8 @@ 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); diff --git a/src/include/storage/ducklake_metadata_info.hpp b/src/include/storage/ducklake_metadata_info.hpp index 23ededb5..aa9680d5 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 format; persisted to ducklake_data_file.file_format + string file_format = "parquet"; idx_t row_count; idx_t file_size_bytes; optional_idx footer_size; @@ -352,7 +354,10 @@ struct DuckLakeFileData { string encryption_key; idx_t file_size_bytes = 0; optional_idx footer_size; + //! delete-file format (parquet/puffin); delete files only DeleteFileFormat format = DeleteFileFormat::PARQUET; + //! data-file format from ducklake_data_file.file_format; selects the reader + string data_file_format = "parquet"; }; enum class DuckLakeDataType { 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..7c59918b 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 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 41e6ac9e..c31596ca 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_format_reader.cpp ducklake_partition_data.cpp ducklake_secret.cpp ducklake_stats.cpp diff --git a/src/storage/ducklake_format_reader.cpp b/src/storage/ducklake_format_reader.cpp new file mode 100644 index 00000000..bfc29ac3 --- /dev/null +++ b/src/storage/ducklake_format_reader.cpp @@ -0,0 +1,179 @@ +#include "storage/ducklake_format_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 { + +DuckLakeFormatReader::DuckLakeFormatReader(DuckLakeFunctionInfo &read_info, const OpenFileInfo &info, string format_p, + vector columns_p) + : BaseFileReader(info), read_info(read_info), format(std::move(format_p)) { + columns = std::move(columns_p); +} + +TableFunction DuckLakeFormatReader::GetReadFunction(ClientContext &context, const string &format) { + auto &instance = DatabaseInstance::GetDatabase(context); + ExtensionHelper::TryAutoLoadExtension(instance, format); + ExtensionLoader loader(instance, "ducklake"); + auto &scan_entry = loader.GetTableFunction("read_" + format); + return scan_entry.functions.functions[0]; +} + +bool DuckLakeFormatReader::TryInitializeScan(ClientContext &context, GlobalTableFunctionState &gstate, + LocalTableFunctionState &lstate) { + { + lock_guard guard(lock); + if (initialized_scan) { + return false; + } + initialized_scan = true; + } + + // 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; + 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.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); + + // 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 &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; + } + optional_idx file_position; + for (idx_t i = 0; i < file_names.size(); i++) { + if (StringUtil::CIEquals(file_names[i], col.name)) { + file_position = i; + break; + } + } + if (!file_position.IsValid()) { + output_kinds.push_back(FormatReaderColumn::MISSING); + output_source_index.push_back(0); + continue; + } + 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()]); + } + // 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]); + } + 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 = file_scan.init_global(context, init_input); + local_state = file_scan.init_local(*execution_context, init_input, global_state.get()); + return true; +} + +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()); + file_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 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); + } else { + chunk.data[c].Reference(source); + } + break; + } + case FormatReaderColumn::ORDINAL: { + 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); + } + break; + } + case FormatReaderColumn::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) { + approved_tuple_count = deletion_filter->Filter(file_row_number, approved_tuple_count, sel); + } + if (filters) { + for (auto &entry : filters->filters) { + if (entry.second->filter_type == TableFilterType::OPTIONAL_FILTER) { + continue; + } + auto &vec = chunk.data[entry.first]; + UnifiedVectorFormat 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); + } + } + 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 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 format reader", virtual_column_id); + } +} + +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..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 { @@ -107,6 +111,7 @@ void DuckLakeInsert::AddWrittenFiles(DuckLakeInsertGlobalState &global_state, Da 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 = 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(); @@ -482,50 +487,74 @@ static void GeneratePartitionExpressions(ClientContext &context, DuckLakeCopyInp } } +string DuckLakeInsert::ValidateDataFileFormat(string format) { + format = StringUtil::Lower(format); + // 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); + } + 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) { + // 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 + 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) { + 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 +564,18 @@ 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) { + 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 +630,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; diff --git a/src/storage/ducklake_metadata_manager.cpp b/src/storage/ducklake_metadata_manager.cpp index 2b7c0e8e..68db76d4 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) { + // 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"; + } + 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); } @@ -1529,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"); @@ -1640,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); } @@ -1672,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); } @@ -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); } @@ -1783,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, " @@ -1875,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)); @@ -2942,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()) { @@ -3159,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_list.cpp b/src/storage/ducklake_multi_file_list.cpp index 3166bab6..8da62040 100644 --- a/src/storage/ducklake_multi_file_list.cpp +++ b/src/storage/ducklake_multi_file_list.cpp @@ -174,6 +174,10 @@ 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); + // 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); + } if (file.footer_size.IsValid()) { extended_info->options["footer_size"] = Value::UBIGINT(file.footer_size.GetIndex()); } @@ -232,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 a18ddf8a..adb466be 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_format_reader.hpp" #include "storage/ducklake_table_entry.hpp" #include "storage/ducklake_catalog.hpp" #include "storage/ducklake_delete_filter.hpp" @@ -386,6 +387,44 @@ shared_ptr DuckLakeMultiFileReader::TryCreateInlinedDataReader(c std::move(columns)); } +shared_ptr DuckLakeMultiFileReader::TryCreateFormatReader(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()) { + return nullptr; + } + auto format = StringUtil::Lower(StringValue::Get(entry->second)); + if (format.empty() || format == "parquet") { + // parquet uses the native reader path + return nullptr; + } + 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)); +} + shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext &context, GlobalTableFunctionState &gstate, const OpenFileInfo &file, idx_t file_idx, @@ -394,6 +433,10 @@ shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext & if (reader) { return reader; } + reader = TryCreateFormatReader(context, file); + if (reader) { + return reader; + } return MultiFileReader::CreateReader(context, gstate, file, file_idx, bind_data); } @@ -405,6 +448,10 @@ shared_ptr DuckLakeMultiFileReader::CreateReader(ClientContext & if (reader) { return reader; } + reader = TryCreateFormatReader(context, file); + if (reader) { + return reader; + } return MultiFileReader::CreateReader(context, file, options, file_options, interface); } diff --git a/src/storage/ducklake_transaction.cpp b/src/storage/ducklake_transaction.cpp index 02c0b181..9baf925c 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; @@ -2601,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/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_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 diff --git a/vortex b/vortex new file mode 160000 index 00000000..0867246a --- /dev/null +++ b/vortex @@ -0,0 +1 @@ +Subproject commit 0867246a4ae5df1616dbf320f79a5ced07a8c867