diff --git a/.github/workflows/MainDistributionPipeline.yml b/.github/workflows/MainDistributionPipeline.yml index ffebc661..be607d5a 100644 --- a/.github/workflows/MainDistributionPipeline.yml +++ b/.github/workflows/MainDistributionPipeline.yml @@ -14,15 +14,15 @@ concurrency: jobs: duckdb-stable-build: name: Build extension binaries - uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@main # NOTE: manually update ref :( + uses: duckdb/extension-ci-tools/.github/workflows/_extension_distribution.yml@main secrets: inherit # required for writing to vcpkg binary cache with: # Main config extension_name: &ext_name delta - duckdb_version: &duckdb_ref v1.5.2 + duckdb_version: &duckdb_ref main ci_tools_version: &ci_tools_ref main enable_rust: true - exclude_archs: 'wasm_mvp;wasm_eh;wasm_threads;windows_amd64_rtools;windows_amd64_mingw;linux_amd64_musl' + exclude_archs: "wasm_mvp;wasm_eh;wasm_threads;windows_amd64_rtools;windows_amd64_mingw;linux_amd64_musl" # Config to write vcpkg binary cache extra_toolchains: ${{ github.event_name != 'pull_request' && ';downgraded_aws_cli;python3;' || 'python3' }} @@ -30,27 +30,26 @@ jobs: vcpkg_binary_sources: ${{ github.event_name != 'pull_request' && vars.VCPKG_BINARY_SOURCES || '' }} reduced_ci_mode: ${{ github.event_name == 'pull_request' && 'enabled' || 'disabled'}} - duckdb-stable-deploy: name: Deploy extension binaries needs: duckdb-stable-build - uses: duckdb/extension-ci-tools/.github/workflows/_extension_deploy.yml@main # NOTE: manually update ref :( + uses: duckdb/extension-ci-tools/.github/workflows/_extension_deploy.yml@main secrets: inherit with: extension_name: *ext_name duckdb_version: *duckdb_ref ci_tools_version: *ci_tools_ref - exclude_archs: 'wasm_mvp;wasm_eh;wasm_threads;windows_amd64_rtools;windows_amd64_mingw;linux_amd64_musl' + exclude_archs: "wasm_mvp;wasm_eh;wasm_threads;windows_amd64_rtools;windows_amd64_mingw;linux_amd64_musl" # deploy_latest: ${{ startsWith(github.ref, 'refs/tags/v') || github.ref == 'refs/heads/main' }} deploy_versioned: ${{ startsWith(github.ref, 'refs/tags/v') || github.ref == 'refs/heads/main' }} reduced_ci_mode: ${{ github.event_name == 'pull_request' && 'enabled' || 'disabled'}} code-quality-check: name: Code Quality Check - uses: duckdb/extension-ci-tools/.github/workflows/_extension_code_quality.yml@main # NOTE: manually update ref :( + uses: duckdb/extension-ci-tools/.github/workflows/_extension_code_quality.yml@main with: extension_name: *ext_name duckdb_version: *duckdb_ref ci_tools_version: *ci_tools_ref - format_checks: format - extra_toolchains: 'python3' + format_checks: format + extra_toolchains: "python3" diff --git a/duckdb b/duckdb index 8a585197..2daa4fc9 160000 --- a/duckdb +++ b/duckdb @@ -1 +1 @@ -Subproject commit 8a5851971fae891f292c2714d86046ee018e9737 +Subproject commit 2daa4fc9a48c638d10dcc613b1ed15d56c66f5b0 diff --git a/src/delta_extension.cpp b/src/delta_extension.cpp index acefa36c..15b35d0f 100644 --- a/src/delta_extension.cpp +++ b/src/delta_extension.cpp @@ -65,15 +65,17 @@ static unique_ptr DeltaCatalogAttach(optional_ptr string commit_fun_name = "__internal_delta_ccv2_commit_staged"; CatalogEntryRetriever retriever(context); - EntryLookupInfo lookup_info(CatalogType::TABLE_FUNCTION_ENTRY, commit_fun_name); - auto fun = retriever.GetEntry(res->parent_catalog_name, schema, lookup_info, OnEntryNotFound::RETURN_NULL); + EntryLookupInfo lookup_info( + CatalogType::TABLE_FUNCTION_ENTRY, + QualifiedName(Identifier(res->parent_catalog_name), Identifier(schema), Identifier(commit_fun_name))); + auto fun = retriever.GetEntry(lookup_info, OnEntryNotFound::RETURN_NULL); if (!fun) { throw InternalException("Parent catalog does not have a __internal_delta_ccv2_commit_staged function"); } res->commit_function = fun->Cast(); } - res->SetDefaultTable(DEFAULT_SCHEMA, res->GetInternalTableName()); + res->SetDefaultTable(Identifier::DefaultSchema(), Identifier(res->GetInternalTableName())); return std::move(res); } diff --git a/src/delta_macros.cpp b/src/delta_macros.cpp index c18f81d3..729bb116 100644 --- a/src/delta_macros.cpp +++ b/src/delta_macros.cpp @@ -58,16 +58,17 @@ void DeltaMacros::RegisterTableMacro(ExtensionLoader &loader, const string &name auto func = make_uniq(std::move(node)); for (auto ¶m : params) { - func->parameters.push_back(make_uniq(param)); + func->parameters.push_back(make_uniq(Identifier(param))); } for (auto ¶m : named_params) { - func->default_parameters[param.first] = make_uniq(param.second); + func->default_parameters[Identifier(param.first)] = make_uniq(Value(param.second)); } CreateMacroInfo info(CatalogType::TABLE_MACRO_ENTRY); - info.schema = DEFAULT_SCHEMA; - info.name = name; + info.SetQualifiedName( + QualifiedName(info.GetQualifiedName().Catalog(), Identifier::DefaultSchema(), info.GetQualifiedName().Name())); + info.SetFunctionName(Identifier(name)); info.temporary = true; info.internal = true; info.macros.push_back(std::move(func)); @@ -76,11 +77,9 @@ void DeltaMacros::RegisterTableMacro(ExtensionLoader &loader, const string &name } static DefaultMacro delta_macros[] = { - {DEFAULT_SCHEMA, - "parse_delta_filter_logline", - {"x", nullptr}, - {{nullptr, nullptr}}, - "x::STRUCT(path VARCHAR, type VARCHAR, filters_before VARCHAR[], filters_after VARCHAR[], files_before BIGINT, " + {DEFAULT_SCHEMA, "parse_delta_filter_logline", + "(x) AS x::STRUCT(path VARCHAR, type VARCHAR, filters_before VARCHAR[], filters_after VARCHAR[], files_before " + "BIGINT, " "files_after BIGINT)"}, }; diff --git a/src/delta_utils.cpp b/src/delta_utils.cpp index 48fdb79a..a0ef85cb 100644 --- a/src/delta_utils.cpp +++ b/src/delta_utils.cpp @@ -4,9 +4,17 @@ #include #include "delta_log_types.hpp" +#include "functions/delta_scan/delta_multi_file_list.hpp" #include "duckdb/common/operator/decimal_cast_operators.hpp" #include "duckdb.hpp" +#include "duckdb/planner/expression/bound_columnref_expression.hpp" +#include "duckdb/planner/expression/bound_comparison_expression.hpp" +#include "duckdb/planner/expression/bound_conjunction_expression.hpp" +#include "duckdb/planner/expression/bound_constant_expression.hpp" +#include "duckdb/planner/expression/bound_function_expression.hpp" +#include "duckdb/planner/expression/bound_operator_expression.hpp" +#include "duckdb/function/scalar/struct_utils.hpp" #include "duckdb/common/extension_type_info.hpp" #include "duckdb/common/types/decimal.hpp" #include "duckdb/main/database.hpp" @@ -279,7 +287,7 @@ void ExpressionVisitor::VisitStructLiteral(void *state, uintptr_t sibling_list_i } for (idx_t i = 0; i < children_keys->size(); i++) { - (*children_values)[i]->alias = (*children_keys)[i]->ToString(); + (*children_values)[i]->SetAlias(Identifier((*children_keys)[i]->ToString())); } unique_ptr expression = make_uniq("struct_pack", std::move(*children_values)); @@ -331,8 +339,8 @@ void ExpressionVisitor::VisitLiteralMap(void *state, uintptr_t sibling_list_id, ErrorData("DuckDB only supports parsing Map literals from delta kernel that consist for constants!"); return; } - key_values.push_back(key_field->Cast().value); - key_type = key_field->Cast().value.type(); + key_values.push_back(key_field->Cast().GetValue()); + key_type = key_field->Cast().GetValue().type(); } vector value_values; @@ -343,8 +351,8 @@ void ExpressionVisitor::VisitLiteralMap(void *state, uintptr_t sibling_list_id, ErrorData("DuckDB only supports parsing Map literals from delta kernel that consist for constants!"); return; } - value_values.push_back(value_field->Cast().value); - value_type = value_field->Cast().value.type(); + value_values.push_back(value_field->Cast().GetValue()); + value_type = value_field->Cast().GetValue().type(); } unique_ptr expression = @@ -423,7 +431,7 @@ void ExpressionVisitor::VisitColumnExpression(void *state, uintptr_t sibling_lis col_ref_string = col_ref_string.substr(1, col_ref_string.size() - 2); } - auto expression = make_uniq(col_ref_string); + auto expression = make_uniq(Identifier(col_ref_string)); static_cast(state)->AppendToList(sibling_list_id, std::move(expression)); } @@ -648,7 +656,7 @@ void SchemaVisitor::VisitStruct(SchemaVisitor *state, uintptr_t sibling_list_id, child_list_t children_types; for (const auto &child_col_def : children) { - children_types.push_back({child_col_def.name, child_col_def.type}); + children_types.emplace_back(child_col_def.name, child_col_def.type); } auto struct_type = LogicalType::STRUCT(children_types); @@ -734,7 +742,7 @@ void SchemaVisitor::AppendToList(uintptr_t id, ffi::KernelStringSlice name, Delt } // Inject the name - child.name = string(name.ptr, name.len); + child.name = Identifier(string(name.ptr, name.len)); it->second.emplace_back(std::move(child)); } @@ -915,7 +923,7 @@ string KernelUtils::FetchFromStringMap(ffi::Handle engi return val; } -vector> & +vector> KernelUtils::UnpackTransformExpression(const vector> &parsed_expression) { if (parsed_expression.size() != 1) { throw IOException("Unexpected size of transformation expression returned by delta kernel: %d", @@ -923,26 +931,33 @@ KernelUtils::UnpackTransformExpression(const vector } const auto &root_expression = parsed_expression.get(0); - if (root_expression->type != ExpressionType::FUNCTION) { - throw IOException("Unexpected type of root expression returned by delta kernel: %d", root_expression->type); + if (root_expression->GetExpressionType() != ExpressionType::FUNCTION) { + throw IOException("Unexpected type of root expression returned by delta kernel: %d", + root_expression->GetExpressionType()); } - if (root_expression->Cast().function_name != "delta_kernel_transform_expression") { + if (root_expression->Cast().FunctionName() != "delta_kernel_transform_expression") { throw IOException("Unexpected function of root expression returned by delta kernel: %s", - root_expression->Cast().function_name); + root_expression->Cast().FunctionName()); } - return root_expression->Cast().children; + vector> children; + for (const auto &child : root_expression->Cast().GetArguments()) { + children.push_back(child.GetExpression().Copy()); + } + return children; } PredicateVisitor::PredicateVisitor(const vector &columns, - optional_ptr filters) { + optional_ptr filters) { predicate = this; visitor = (uintptr_t(*)(void *, ffi::KernelExpressionVisitorState *)) & VisitPredicate; if (filters) { - for (auto &filter : filters->filters) { - column_filters[columns[filter.first].name] = filter.second.get(); + for (auto &entry : *filters) { + auto &column = columns[entry.first]; + column_filters[column.name.GetIdentifierName()] = entry.second.get(); + column_types[column.name.GetIdentifierName()] = column.type; } } } @@ -978,8 +993,8 @@ uintptr_t PredicateVisitor::VisitPredicate(PredicateVisitor *predicate, ffi::Ker return ffi::visit_predicate_and(state, &eit); } -uintptr_t PredicateVisitor::VisitConstantFilter(const string &col_name, const ConstantFilter &filter, - ffi::KernelExpressionVisitorState *state) { +uintptr_t PredicateVisitor::VisitConstantFilter(const string &col_name, ExpressionType comparison_type, + const Value &value, ffi::KernelExpressionVisitorState *state) { auto maybe_left = ffi::visit_expression_column(state, KernelUtils::ToDeltaString(col_name), DuckDBEngineError::AllocateError); @@ -991,7 +1006,6 @@ uintptr_t PredicateVisitor::VisitConstantFilter(const string &col_name, const Co } uintptr_t right = ~0; - auto &value = filter.constant; switch (value.type().id()) { case LogicalType::BIGINT: right = visit_expression_literal_long(state, BigIntValue::Get(value)); @@ -1071,7 +1085,7 @@ uintptr_t PredicateVisitor::VisitConstantFilter(const string &col_name, const Co } // TODO support other comparison types? - switch (filter.comparison_type) { + switch (comparison_type) { case ExpressionType::COMPARE_LESSTHAN: return visit_predicate_lt(state, left, right); case ExpressionType::COMPARE_LESSTHANOREQUALTO: @@ -1097,20 +1111,40 @@ uintptr_t PredicateVisitor::VisitConstantFilter(const string &col_name, const Co } } -uintptr_t PredicateVisitor::VisitAndFilter(const string &col_name, const ConjunctionAndFilter &filter, - ffi::KernelExpressionVisitorState *state) { - auto it = filter.child_filters.begin(); - auto end = filter.child_filters.end(); - auto get_next = [this, col_name, state, &it, &end]() -> uintptr_t { - if (it == end) { - return 0; +// Resolves the (possibly struct-nested) column being filtered to the dot-separated path the kernel +// expects. `base` is the top-level column name this filter is keyed on: a bare column subject yields +// `base`, while struct field access (struct_extract / struct_extract_at) appends the nested field +// names, e.g. base "i" with subject i.a.b -> "i.a.b". Returns false for unsupported subjects. +static bool ResolveFilterColumnPath(const Expression &expr, const string &base, string &result) { + switch (expr.GetExpressionClass()) { + case ExpressionClass::BOUND_REF: + case ExpressionClass::BOUND_COLUMN_REF: + result = base; + return true; + case ExpressionClass::BOUND_FUNCTION: { + auto &func = expr.Cast(); + const auto &name = func.Function().GetName(); + if (name != "struct_extract" && name != "struct_extract_at") { + return false; } - auto &child_filter = *it++; - - return VisitFilter(col_name, *child_filter, state); - }; - auto eit = EngineIteratorFromCallable(get_next); - return visit_predicate_and(state, &eit); + if (func.GetChildren().empty()) { + return false; + } + auto &struct_type = func.GetChildren()[0]->GetReturnType(); + idx_t child_idx; + if (struct_type.id() != LogicalTypeId::STRUCT || !TryGetStructExtractChildIndex(func, child_idx)) { + return false; + } + string parent; + if (!ResolveFilterColumnPath(*func.GetChildren()[0], base, parent)) { + return false; + } + result = parent + "." + StructType::GetChildName(struct_type, child_idx).GetIdentifierName(); + return true; + } + default: + return false; + } } uintptr_t PredicateVisitor::VisitIsNull(const string &col_name, ffi::KernelExpressionVisitorState *state) { @@ -1130,48 +1164,90 @@ uintptr_t PredicateVisitor::VisitIsNotNull(const string &col_name, ffi::KernelEx return ffi::visit_predicate_not(state, VisitIsNull(col_name, state)); } -uintptr_t PredicateVisitor::VisitStructExtractFilter(const string &col_name, const StructFilter &filter, - ffi::KernelExpressionVisitorState *state) { - // Build the full dot-separated path by recursing through nested StructFilters. - // E.g. col "i" with StructFilter{child_name="a", child_filter=StructFilter{child_name="b", leaf}} - // becomes "i.a.b". visit_expression_column splits on "." to construct the kernel ColumnName. - string full_path = col_name + "." + filter.child_name; - const TableFilter *child = filter.child_filter.get(); - while (child->filter_type == TableFilterType::STRUCT_EXTRACT) { - const auto &nested = static_cast(*child); - full_path += "." + nested.child_name; - child = nested.child_filter.get(); +uintptr_t PredicateVisitor::VisitFilterExpression(const string &col_name, const Expression &expr, + ffi::KernelExpressionVisitorState *state) { + // A filter subject that is a bare reference to a nested (struct/list/map) column can't be expressed to the kernel: + // the same predicate also arrives in struct_extract form (which resolves to the full leaf path, e.g. i.a.b), so + // the bare form (resolving only to the top-level column) is skipped. + auto type_entry = column_types.find(col_name); + bool base_is_nested = type_entry != column_types.end() && type_entry->second.IsNested(); + + if (BoundComparisonExpression::IsComparison(expr)) { + auto &comparison = expr.Cast(); + auto comparison_type = comparison.GetExpressionType(); + auto &left = BoundComparisonExpression::Left(comparison); + auto &right = BoundComparisonExpression::Right(comparison); + string path; + if (left.GetExpressionClass() == ExpressionClass::BOUND_CONSTANT && + ResolveFilterColumnPath(right, col_name, path) && !(path == col_name && base_is_nested)) { + return VisitConstantFilter(path, FlipComparisonExpression(comparison_type), + left.Cast().GetValue(), state); + } + if (right.GetExpressionClass() == ExpressionClass::BOUND_CONSTANT && + ResolveFilterColumnPath(left, col_name, path) && !(path == col_name && base_is_nested)) { + return VisitConstantFilter(path, comparison_type, right.Cast().GetValue(), state); + } + return ~0; } - return VisitFilter(full_path, *child, state); -} -uintptr_t PredicateVisitor::VisitFilter(const string &col_name, const TableFilter &filter, - ffi::KernelExpressionVisitorState *state) { - switch (filter.filter_type) { - case TableFilterType::CONSTANT_COMPARISON: - return VisitConstantFilter(col_name, static_cast(filter), state); - case TableFilterType::CONJUNCTION_AND: - return VisitAndFilter(col_name, static_cast(filter), state); - case TableFilterType::IS_NULL: - return VisitIsNull(col_name, state); - case TableFilterType::IS_NOT_NULL: - return VisitIsNotNull(col_name, state); - case TableFilterType::STRUCT_EXTRACT: - return VisitStructExtractFilter(col_name, static_cast(filter), state); - // TODO: implement once kernel can do arbitrary expressions - case TableFilterType::EXPRESSION_FILTER: - // TODO: implement once kernel adds support for IN filters / arbitrary expressions - case TableFilterType::IN_FILTER: - // TODO: figure out if this is ever useful - case TableFilterType::DYNAMIC_FILTER: - // TODO: can we even push these down? - case TableFilterType::CONJUNCTION_OR: - case TableFilterType::OPTIONAL_FILTER: + switch (expr.GetExpressionClass()) { + case ExpressionClass::BOUND_CONJUNCTION: { + auto &conjunction = expr.Cast(); + if (conjunction.GetExpressionType() != ExpressionType::CONJUNCTION_AND) { + return ~0; + } + auto it = conjunction.GetChildren().begin(); + auto end = conjunction.GetChildren().end(); + auto get_next = [this, col_name, state, &it, &end]() -> uintptr_t { + if (it == end) { + return 0; + } + return VisitFilterExpression(col_name, *(*it++), state); + }; + auto eit = EngineIteratorFromCallable(get_next); + return visit_predicate_and(state, &eit); + } + case ExpressionClass::BOUND_OPERATOR: { + auto &op = expr.Cast(); + string path; + if (op.GetChildren().size() != 1 || !ResolveFilterColumnPath(*op.GetChildren()[0], col_name, path) || + (path == col_name && base_is_nested)) { + return ~0; + } + if (op.GetExpressionType() == ExpressionType::OPERATOR_IS_NULL) { + return VisitIsNull(path, state); + } + if (op.GetExpressionType() == ExpressionType::OPERATOR_IS_NOT_NULL) { + return VisitIsNotNull(path, state); + } + return ~0; + } + case ExpressionClass::BOUND_FUNCTION: { + auto &func = expr.Cast(); + if (func.Function().GetName() == OptionalFilterScalarFun::NAME && func.BindInfo()) { + auto &data = func.BindInfo()->Cast(); + if (data.child_filter_expr) { + return VisitFilterExpression(col_name, *data.child_filter_expr, state); + } + } + if (func.Function().GetName() == SelectivityOptionalFilterScalarFun::NAME && func.BindInfo()) { + auto &data = func.BindInfo()->Cast(); + if (data.child_filter_expr) { + return VisitFilterExpression(col_name, *data.child_filter_expr, state); + } + } + return ~0; + } default: return ~0; } } +uintptr_t PredicateVisitor::VisitFilter(const string &col_name, const ExpressionFilter &filter, + ffi::KernelExpressionVisitorState *state) { + return VisitFilterExpression(col_name, *filter.expr, state); +} + void LoggerCallback::Initialize(DatabaseInstance &db_p) { auto &instance = GetInstance(); unique_lock lck(instance.lock); diff --git a/src/functions/delta_domain_metadata.cpp b/src/functions/delta_domain_metadata.cpp index 0e0eeeed..54dc9211 100644 --- a/src/functions/delta_domain_metadata.cpp +++ b/src/functions/delta_domain_metadata.cpp @@ -23,7 +23,7 @@ static unique_ptr DeltaDomainMetadataBind(ClientContext &context, idx_t version = DConstants::INVALID_INDEX; for (auto &kv : input.named_parameters) { - auto loption = StringUtil::Lower(kv.first); + auto loption = StringUtil::Lower(kv.first.GetIdentifierName()); if (loption == "version") { version = kv.second.GetValue(); } @@ -32,7 +32,7 @@ static unique_ptr DeltaDomainMetadataBind(ClientContext &context, auto file_list = make_uniq(context, input_string, version); // Trigger snapshot initialization - vector _n; + vector _n; vector _t; file_list->Bind(_t, _n); diff --git a/src/functions/delta_metadata_scan.cpp b/src/functions/delta_metadata_scan.cpp index 3ee43c4e..b31593dd 100644 --- a/src/functions/delta_metadata_scan.cpp +++ b/src/functions/delta_metadata_scan.cpp @@ -26,7 +26,7 @@ static void AddFileInfo(OpenFileInfo &file_info, DeltaFileMetaData &metadata, ve auto partitions = metadata.partition_map; InsertionOrderPreservingMap map; for (auto &kv : partitions) { - map[kv.first] = kv.second.ToString(); + map[kv.first.GetIdentifierName()] = kv.second.ToString(); } row_values.emplace_back(Value::MAP(map)); @@ -71,7 +71,7 @@ static unique_ptr DeltaFileListBind(ClientContext &context, TableF DeltaFileListOptions options; for (auto &kv : input.named_parameters) { - auto loption = StringUtil::Lower(kv.first); + auto loption = StringUtil::Lower(kv.first.GetIdentifierName()); auto &val = kv.second; if (loption == "version") { version = val.GetValue(); @@ -87,7 +87,7 @@ static unique_ptr DeltaFileListBind(ClientContext &context, TableF auto file_list = make_uniq(context, input_string, version); // TODO: this is weird - vector _n; + vector _n; vector _t; file_list->Bind(_t, _n); diff --git a/src/functions/delta_scan/delta_multi_file_list.cpp b/src/functions/delta_scan/delta_multi_file_list.cpp index 4ce9e205..97ea47fa 100644 --- a/src/functions/delta_scan/delta_multi_file_list.cpp +++ b/src/functions/delta_scan/delta_multi_file_list.cpp @@ -18,6 +18,7 @@ #include "duckdb/parser/constraints/not_null_constraint.hpp" #include +#include #include "duckdb/planner/constraints/bound_not_null_constraint.hpp" @@ -78,8 +79,8 @@ static ffi::EngineBuilder *CreateBuilder(ClientContext &context, const string &p if (res.HasError()) { res.Throw(); } - // Use multi-threaded tokio executor (required for checkpoint support) - ffi::set_builder_with_multithreaded_executor(return_value, 0, 0); + // DIAGNOSTIC (revert before merge): single-threaded (default) executor to test the Windows tokio hang + // ffi::set_builder_with_multithreaded_executor(return_value, 0, 0); return return_value; } @@ -186,8 +187,8 @@ static ffi::EngineBuilder *CreateBuilder(ClientContext &context, const string &p "create an R2 or GCS secret containing the credentials for this endpoint and try again."); } - // Use multi-threaded tokio executor (required for checkpoint support) - ffi::set_builder_with_multithreaded_executor(builder, 0, 0); + // DIAGNOSTIC (revert before merge): single-threaded (default) executor to test the Windows tokio hang + // ffi::set_builder_with_multithreaded_executor(builder, 0, 0); return builder; } const auto &kv_secret = dynamic_cast(*secret_match.secret_entry->secret); @@ -329,8 +330,8 @@ static ffi::EngineBuilder *CreateBuilder(ClientContext &context, const string &p } set_option(builder, "container_name", bucket); } - // Use multi-threaded tokio executor (required for checkpoint support) - ffi::set_builder_with_multithreaded_executor(builder, 0, 0); + // DIAGNOSTIC (revert before merge): single-threaded (default) executor to test the Windows tokio hang + // ffi::set_builder_with_multithreaded_executor(builder, 0, 0); return builder; } @@ -346,30 +347,37 @@ static void KernelPartitionStringVisitor(ffi::NullableCvoid engine_context, ffi: static unordered_map FindPartitionValues(ParsedExpression &transformation, const vector &cols) { - if (transformation.Cast().function_name != "delta_kernel_transform_expression") { + if (transformation.Cast().FunctionName() != "delta_kernel_transform_expression") { throw IOException("Unexpected function of root expression returned by delta kernel: %s", - transformation.Cast().function_name); + transformation.Cast().FunctionName()); } unordered_map res; // Iterate the children of the transform - for (auto &child : transformation.Cast().children) { - auto &transform_op = child->Cast(); - if (transform_op.function_name != "delta_transform_op") { + for (auto &child : transformation.Cast().GetArguments()) { + auto &transform_op = child.GetExpression().Cast(); + if (transform_op.FunctionName() != "delta_transform_op") { throw IOException("Unexpected function for delta_transform_op returned by delta kernel: %s", - child->Cast().function_name); + child.GetExpression().Cast().FunctionName()); } bool is_replace = false; string field_name; vector values; - for (auto &transform_op_child : transform_op.children) { - if (transform_op_child->GetExpressionType() == ExpressionType::COMPARE_EQUAL) { - auto name = - transform_op_child->Cast().left->Cast().GetName(); - auto value = transform_op_child->Cast().right->Cast().value; + for (auto &transform_op_child : transform_op.GetArguments()) { + if (transform_op_child.GetExpression().GetExpressionType() == ExpressionType::COMPARE_EQUAL) { + auto name = transform_op_child.GetExpression() + .Cast() + .Left() + .Cast() + .GetName(); + auto value = transform_op_child.GetExpression() + .Cast() + .Right() + .Cast() + .GetValue(); if (name == "is_replace") { is_replace = value.GetValue(); @@ -381,12 +389,12 @@ static unordered_map FindPartitionValues(ParsedExpression &transfo throw InternalException("Unexpected name for delta_transform_op returned by delta kernel: %s", name); } - } else if (transform_op_child->GetExpressionType() == ExpressionType::VALUE_CONSTANT) { - values.push_back(transform_op_child->Cast().value); + } else if (transform_op_child.GetExpression().GetExpressionType() == ExpressionType::VALUE_CONSTANT) { + values.push_back(transform_op_child.GetExpression().Cast().GetValue()); } else { throw NotImplementedException( "Unexpected expression for delta_transform_op returned by delta kernel: %s", - transform_op_child->ToString()); + transform_op_child.GetExpression().ToString()); } } @@ -482,12 +490,12 @@ void ScanDataCallBack::VisitCallbackInternal(ffi::NullableCvoid engine_context, auto transform_partitions = FindPartitionValues(*(*parsed_transformation_expression)[0], snapshot.global_columns); - case_insensitive_map_t constant_map; + identifier_map_t constant_map; for (idx_t i = 0; i < snapshot.partitions.size(); ++i) { const auto &partition_id = context->snapshot.partition_ids[i]; const auto &partition_name = context->snapshot.partitions[i]; - constant_map[partition_name] = transform_partitions[partition_id]; + constant_map[Identifier(partition_name)] = transform_partitions[partition_id]; } snapshot.metadata.back()->partition_map = std::move(constant_map); snapshot.metadata.back()->transform_expression = @@ -604,12 +612,12 @@ static bool ExtractHasNullConstraintsInArrays(const vector &return_types, vector &names) { +void DeltaMultiFileList::Bind(vector &return_types, vector &names) { unique_lock lck(lock); if (have_bound) { for (const auto &field : global_columns) { - names.push_back(field.name); + names.push_back(Identifier(field.name)); return_types.push_back(field.type); } return; @@ -624,7 +632,7 @@ void DeltaMultiFileList::Bind(vector &return_types, vector } for (const auto &field : visited_schema) { - names.push_back(field.name); + names.push_back(Identifier(field.name)); return_types.push_back(field.type); } @@ -811,21 +819,24 @@ void DeltaMultiFileList::EnsureScanInitialized() const { } } -unique_ptr DeltaMultiFileList::PushdownInternal(ClientContext &context, - TableFilterSet &new_filters) const { +unique_ptr DeltaMultiFileList::PushdownInternal(ClientContext &context, TableFilterSet &new_filters, + vector column_indexes) const { auto filtered_list = make_uniq(context, paths[0].path, version); - TableFilterSet result_filter_set; + DeltaTableFilters result_filter_set; // Add pre-existing filters - for (auto &entry : table_filters.filters) { - result_filter_set.PushFilter(ColumnIndex(entry.first), entry.second->Copy()); + for (auto &entry : table_filters) { + result_filter_set.PushFilter(entry.first, entry.second->Copy()); } // Add new filters - for (auto &entry : new_filters.filters) { - if (entry.first < global_columns.size()) { - result_filter_set.PushFilter(ColumnIndex(entry.first), entry.second->Copy()); + for (auto &entry : new_filters) { + auto &column_id = column_indexes[entry.GetIndex()]; + if (column_id < global_columns.size()) { + auto &filter = + ExpressionFilter::GetExpressionFilter(entry.Filter(), "DeltaMultiFileList::PushdownInternal"); + result_filter_set.PushFilter(column_id, filter.Copy()); } } @@ -874,11 +885,11 @@ unique_ptr DeltaMultiFileList::ComplexFilterPushdown(ClientContex vector pushdown_results; auto filter_set = combiner.GenerateTableScanFilters(info.column_indexes, pushdown_results); - if (filter_set.filters.empty()) { + if (!filter_set.HasFilters()) { return nullptr; } - auto filtered_list = PushdownInternal(context, filter_set); + auto filtered_list = PushdownInternal(context, filter_set, info.column_ids); ReportFilterPushdown(context, *filtered_list, info.column_ids, "constant", info); @@ -922,64 +933,64 @@ void DeltaMultiFileList::ReportFilterPushdown(ClientContext &context, DeltaMulti new_total = new_list.GetTotalFileCount(); if (should_report_explain_output) { - if (!mfr_info->extra_info.total_files.IsValid()) { + // Filter pushdown can run multiple times for one query (e.g. RemoveUnusedColumns re-runs it on the + // already-filtered list), so old_total shrinks across passes. Keep the largest (the true pre-filter total). + if (!mfr_info->extra_info.total_files.IsValid() || + old_total > mfr_info->extra_info.total_files.GetIndex()) { mfr_info->extra_info.total_files = old_total; - } else if (mfr_info->extra_info.total_files.GetIndex() != old_total) { - throw InternalException( - "Error encountered when analyzing filtered out files for delta scan: total_files inconsistent!"); } + // Likewise keep the smallest post-filter count across passes (the most files skipped). if (!mfr_info->extra_info.filtered_files.IsValid() || - mfr_info->extra_info.filtered_files.GetIndex() >= new_total) { + new_total < mfr_info->extra_info.filtered_files.GetIndex()) { mfr_info->extra_info.filtered_files = new_total; - } else { - throw InternalException( - "Error encountered when analyzing filtered out files for delta scan: filtered_files inconsistent!"); } } } - // Report the new filters - vector old_filters_value_list; - for (auto &f : table_filters.filters) { - auto &column_index = f.first; - auto &filter = f.second; - if (column_index < global_columns.size()) { - auto &col_name = global_columns[column_index].name; - old_filters_value_list.push_back(filter->ToString(col_name)); + // Collect a filter set's rendered strings, sorted for deterministic output (the underlying map is unordered). + auto collect_filter_strings = [&](const DeltaTableFilters &tf) { + vector result; + for (auto &entry : tf) { + auto column_id = entry.first; + if (column_id < global_columns.size()) { + result.push_back(entry.second->ToString(global_columns[column_id].name.GetIdentifierName())); + } } - } - auto old_filters_value = Value::LIST(LogicalType::VARCHAR, old_filters_value_list); - - // Report the new filters - vector filters_value_list; - for (auto &f : new_list.table_filters.filters) { - auto &column_index = f.first; - auto &filter = f.second; - if (column_index < global_columns.size()) { - auto &col_name = global_columns[column_index].name; - filters_value_list.push_back(filter->ToString(col_name)); + std::sort(result.begin(), result.end()); + return result; + }; + + auto to_value_list = [](const vector &strings) { + vector values; + for (auto &s : strings) { + values.push_back(Value(s)); } - } - auto filters_value = Value::LIST(LogicalType::VARCHAR, filters_value_list); + return Value::LIST(LogicalType::VARCHAR, values); + }; + + // Report the pre- and post-pushdown filters + auto old_filters_value = to_value_list(collect_filter_strings(table_filters)); + auto new_filter_strings = collect_filter_strings(new_list.table_filters); + auto filters_value = to_value_list(new_filter_strings); if (should_report_explain_output) { string files_string; - for (auto &filter : filters_value_list) { - files_string += filter.ToString() + "\n"; + for (auto &filter : new_filter_strings) { + files_string += filter + "\n"; } mfr_info->extra_info.file_filters = files_string.substr(0, files_string.size() - 1); } if (should_log) { child_list_t struct_fields; - struct_fields.push_back({"path", Value(GetPath())}); - struct_fields.push_back({"type", Value(pushdown_type)}); - struct_fields.push_back({"filters_before", old_filters_value}); - struct_fields.push_back({"filters_after", filters_value}); + struct_fields.emplace_back("path", Value(GetPath())); + struct_fields.emplace_back("type", Value(pushdown_type)); + struct_fields.emplace_back("filters_before", old_filters_value); + struct_fields.emplace_back("filters_after", filters_value); if (new_total != DConstants::INVALID_INDEX) { - struct_fields.push_back({"files_before", Value::BIGINT(old_total)}); - struct_fields.push_back({"files_after", Value::BIGINT(new_total)}); + struct_fields.emplace_back("files_before", Value::BIGINT(old_total)); + struct_fields.emplace_back("files_after", Value::BIGINT(new_total)); } auto struct_value = Value::STRUCT(struct_fields); logger.WriteLog(delta_log_type, log_level, struct_value.ToString()); @@ -988,31 +999,33 @@ void DeltaMultiFileList::ReportFilterPushdown(ClientContext &context, DeltaMulti unique_ptr DeltaMultiFileList::DynamicFilterPushdown(ClientContext &context, const MultiFileOptions &options, - const vector &names, const vector &types, + const vector &names, const vector &types, const vector &column_ids, TableFilterSet &filters) const { auto pushdown_mode = GetDeltaFilterPushdownMode(context, options); if (pushdown_mode == DeltaFilterPushdownMode::NONE || pushdown_mode == DeltaFilterPushdownMode::CONSTANT_ONLY) { return nullptr; } - if (filters.filters.empty()) { + if (!filters.HasFilters()) { return nullptr; } TableFilterSet filters_copy; - for (auto &filter : filters.filters) { - auto column_id = column_ids[filter.first]; - auto previously_pushed_down_filter = this->table_filters.filters.find(column_id); - if (previously_pushed_down_filter != this->table_filters.filters.end() && - filter.second->Equals(*previously_pushed_down_filter->second)) { + for (auto &entry : filters) { + auto proj_id = entry.GetIndex(); + auto &filter = + ExpressionFilter::GetExpressionFilter(entry.Filter(), "DeltaMultiFileList::DynamicFilterPushdown"); + auto column_id = column_ids[proj_id]; + auto previously_pushed_down_filter = table_filters.TryGetFilterByColumnIndex(column_id); + if (previously_pushed_down_filter && filter.Equals(*previously_pushed_down_filter)) { // Skip filters that we already have pushed down continue; } - filters_copy.PushFilter(ColumnIndex(column_id), filter.second->Copy()); + filters_copy.PushFilter(proj_id, filter.Copy()); } - if (!filters_copy.filters.empty()) { - auto new_snap = PushdownInternal(context, filters_copy); + if (filters_copy.HasFilters()) { + auto new_snap = PushdownInternal(context, filters_copy, column_ids); ReportFilterPushdown(context, *new_snap, column_ids, "dynamic", nullptr); return std::move(new_snap); } diff --git a/src/functions/delta_scan/delta_multi_file_reader.cpp b/src/functions/delta_scan/delta_multi_file_reader.cpp index 37a01199..d38ad28e 100644 --- a/src/functions/delta_scan/delta_multi_file_reader.cpp +++ b/src/functions/delta_scan/delta_multi_file_reader.cpp @@ -55,7 +55,7 @@ void FinalizeBindBaseOverride(MultiFileReaderData &reader_data, const MultiFileO // create a map of name -> column index auto &local_columns = reader_data.reader->GetColumns(); auto &filename = reader_data.reader->GetFileName(); - case_insensitive_map_t name_map; + identifier_map_t name_map; if (file_options.union_by_name) { for (idx_t col_idx = 0; col_idx < local_columns.size(); col_idx++) { auto &column = local_columns[col_idx]; @@ -104,7 +104,7 @@ void FinalizeBindBaseOverride(MultiFileReaderData &reader_data, const MultiFileO } bool DeltaMultiFileReader::Bind(MultiFileOptions &options, MultiFileList &files, vector &return_types, - vector &names, MultiFileReaderBindData &bind_data) { + vector &names, MultiFileReaderBindData &bind_data) { auto &delta_snapshot = dynamic_cast(files); auto log_tail_setting = options.custom_options.find("log_tail"); @@ -118,7 +118,7 @@ bool DeltaMultiFileReader::Bind(MultiFileOptions &options, MultiFileList &files, } void DeltaMultiFileReader::BindOptions(MultiFileOptions &options, MultiFileList &files, - vector &return_types, vector &names, + vector &return_types, vector &names, MultiFileReaderBindData &bind_data) { // Disable all other multifilereader options options.auto_detect_hive_partitioning = false; @@ -137,8 +137,8 @@ void DeltaMultiFileReader::BindOptions(MultiFileOptions &options, MultiFileList auto partitions = snapshot.GetPartitionColumns(); for (auto &part : partitions) { idx_t hive_partitioning_index; - auto lookup = std::find_if(names.begin(), names.end(), - [&](const string &col_name) { return StringUtil::CIEquals(col_name, part); }); + auto lookup = + std::find_if(names.begin(), names.end(), [&](const Identifier &col_name) { return col_name == part; }); if (lookup != names.end()) { // hive partitioning column also exists in file - override auto idx = NumericCast(lookup - names.begin()); @@ -147,6 +147,9 @@ void DeltaMultiFileReader::BindOptions(MultiFileOptions &options, MultiFileList throw IOException("Delta Snapshot returned partition column that is not present in the schema"); } bind_data.hive_partitioning_indexes.emplace_back(part, hive_partitioning_index); + // Register the partition column's type so pre-open filter skipping resolves the partition value as the + // column type instead of defaulting to VARCHAR (which crashes typed ExpressionFilter evaluation). + options.hive_types_schema[part] = return_types[hive_partitioning_index]; } } @@ -264,7 +267,7 @@ DeltaMultiFileReader::InitializeGlobalState(ClientContext &context, const MultiF } auto global_name = global_columns[global_id].name; - selected_columns.insert({global_name, i}); + selected_columns.insert({global_name.GetIdentifierName(), i}); } auto res = make_uniq(extra_columns, &file_list); diff --git a/src/functions/delta_scan/delta_scan.cpp b/src/functions/delta_scan/delta_scan.cpp index f4b45ce8..64949b30 100644 --- a/src/functions/delta_scan/delta_scan.cpp +++ b/src/functions/delta_scan/delta_scan.cpp @@ -113,10 +113,10 @@ TableFunctionSet DeltaFunctions::GetDeltaScanFunction(ExtensionLoader &loader) { function.named_parameters["pushdown_filters"] = LogicalType::VARCHAR; function.named_parameters["log_tail"] = KernelUtils::GetLogPathType(); - function.name = "delta_scan"; + function.SetName("delta_scan"); } - parquet_scan_copy.name = "delta_scan"; + parquet_scan_copy.SetName("delta_scan"); return parquet_scan_copy; } diff --git a/src/functions/delta_transaction_utils/idempotency_helpers.cpp b/src/functions/delta_transaction_utils/idempotency_helpers.cpp index 4a5558f9..0f5896dd 100644 --- a/src/functions/delta_transaction_utils/idempotency_helpers.cpp +++ b/src/functions/delta_transaction_utils/idempotency_helpers.cpp @@ -91,7 +91,7 @@ static unique_ptr DeltaGetTransactionVersionBind(ClientContext &co names.emplace_back("version"); // TODO: support catalog.schema.table format - EntryLookupInfo lookup(CatalogType::TABLE_ENTRY, path); + EntryLookupInfo lookup(CatalogType::TABLE_ENTRY, Identifier(path)); auto lookup_result = Catalog::GetEntry(context, "", "", lookup, OnEntryNotFound::RETURN_NULL); if (lookup_result.get()) { res->delta_table_entry = lookup_result->Cast(); @@ -113,7 +113,7 @@ static unique_ptr DeltaSetTransactionVersionBind(ClientContext &co res->expected_version = input.inputs[3]; // TODO: support catalog.schema.table format - EntryLookupInfo lookup(CatalogType::TABLE_ENTRY, path); + EntryLookupInfo lookup(CatalogType::TABLE_ENTRY, Identifier(path)); auto lookup_result = Catalog::GetEntry(context, "", "", lookup, OnEntryNotFound::RETURN_NULL); if (lookup_result.get()) { res->delta_table_entry = lookup_result->Cast(); diff --git a/src/functions/expression_functions.cpp b/src/functions/expression_functions.cpp index bb2f9768..1227c33e 100644 --- a/src/functions/expression_functions.cpp +++ b/src/functions/expression_functions.cpp @@ -15,11 +15,11 @@ static void AddTestExpressions(vector &result, vectorGetExpressionType() == ExpressionType::FUNCTION) { - for (auto &expr : expression->Cast().children) { - result.push_back(expr->ToString()); + for (auto &arg : expression->Cast().GetArguments()) { + result.push_back(arg.GetExpression().ToString()); } } else if (expression->GetExpressionType() == ExpressionType::CONJUNCTION_AND) { - for (auto &expr : expression->Cast().children) { + for (auto &expr : expression->Cast().GetChildren()) { result.push_back(expr->ToString()); } } else { @@ -46,7 +46,7 @@ static void GetDeltaTestExpression(DataChunk &input, ExpressionState &state, Vec ScalarFunctionSet DeltaFunctions::GetExpressionFunction(ExtensionLoader &loader) { ScalarFunctionSet result; - result.name = "get_delta_test_expression"; + result.SetName("get_delta_test_expression"); ScalarFunction getvar({}, LogicalType::LIST(LogicalType::VARCHAR), GetDeltaTestExpression, nullptr, nullptr); result.AddFunction(getvar); diff --git a/src/functions/metadata_functions.cpp b/src/functions/metadata_functions.cpp index a2add2cf..70947bef 100644 --- a/src/functions/metadata_functions.cpp +++ b/src/functions/metadata_functions.cpp @@ -44,7 +44,7 @@ static void MetadataFunctionExecute(ClientContext &context, TableFunctionInput & output.SetCardinality(count); } -DeltaBaseMetadataFunction::DeltaBaseMetadataFunction(string name_p, table_function_bind_t bind) +DeltaBaseMetadataFunction::DeltaBaseMetadataFunction(Identifier name_p, table_function_bind_t bind) : TableFunction(std::move(name_p), {LogicalType::VARCHAR}, MetadataFunctionExecute, bind, MetadataFunctionInit) { } diff --git a/src/functions/util_functions.cpp b/src/functions/util_functions.cpp index a4da2403..987ae3f3 100644 --- a/src/functions/util_functions.cpp +++ b/src/functions/util_functions.cpp @@ -33,7 +33,7 @@ static void GetWriteDataFunction(DataChunk &input, ExpressionState &state, Vecto ScalarFunctionSet DeltaFunctions::GetWriteFileFunction(ExtensionLoader &loader) { ScalarFunctionSet result; - result.name = "write_blob"; + result.SetName("write_blob"); ScalarFunction write_file({LogicalType::VARCHAR, LogicalType::BLOB}, {LogicalType::BOOLEAN}, GetWriteDataFunction, nullptr, nullptr); diff --git a/src/include/delta_functions.hpp b/src/include/delta_functions.hpp index 2f40e5fa..4d9780c1 100644 --- a/src/include/delta_functions.hpp +++ b/src/include/delta_functions.hpp @@ -15,7 +15,7 @@ class ExtensionLoader; class DeltaBaseMetadataFunction : public TableFunction { public: - DeltaBaseMetadataFunction(string name, table_function_bind_t bind); + DeltaBaseMetadataFunction(Identifier name, table_function_bind_t bind); }; class DeltaFileListFunction : public DeltaBaseMetadataFunction { diff --git a/src/include/delta_utils.hpp b/src/include/delta_utils.hpp index 90598039..f65988a8 100644 --- a/src/include/delta_utils.hpp +++ b/src/include/delta_utils.hpp @@ -4,8 +4,8 @@ #define DEFINE_DEFAULT_ENGINE_BASE 1 #include "delta_kernel_ffi.hpp" #include "duckdb/common/enum_util.hpp" -#include "duckdb/planner/filter/conjunction_filter.hpp" -#include "duckdb/planner/filter/constant_filter.hpp" +#include "duckdb/planner/filter/expression_filter.hpp" +#include "duckdb/planner/filter/table_filter_functions.hpp" #include "duckdb/planner/expression.hpp" #include "duckdb/parser/expression/constant_expression.hpp" #include "duckdb/common/multi_file/multi_file_data.hpp" @@ -92,7 +92,7 @@ struct KernelUtils { return none; } - static vector> & + static vector> UnpackTransformExpression(const vector> &parsed_expression); }; @@ -461,29 +461,33 @@ struct SharedKernelPointer { }; typedef SharedKernelPointer SharedKernelSnapshot; +struct DeltaTableFilters; class PredicateVisitor : public ffi::EnginePredicate { public: - PredicateVisitor(const vector &columns, optional_ptr filters); + PredicateVisitor(const vector &columns, + optional_ptr filters); ErrorData error_data; private: - unordered_map column_filters; + unordered_map> column_filters; + // Top-level column types, keyed by the same name as column_filters, used to reject filters whose subject is a bare + // reference to a nested (struct/list/map) column - the path to the scalar leaf is not recoverable in that form. + unordered_map column_types; static uintptr_t VisitPredicate(PredicateVisitor *predicate, ffi::KernelExpressionVisitorState *state); - uintptr_t VisitConstantFilter(const string &col_name, const ConstantFilter &filter, + uintptr_t VisitConstantFilter(const string &col_name, ExpressionType comparison_type, const Value &value, ffi::KernelExpressionVisitorState *state); - uintptr_t VisitAndFilter(const string &col_name, const ConjunctionAndFilter &filter, - ffi::KernelExpressionVisitorState *state); + uintptr_t VisitFilterExpression(const string &col_name, const Expression &expr, + ffi::KernelExpressionVisitorState *state); uintptr_t VisitIsNull(const string &col_name, ffi::KernelExpressionVisitorState *state); uintptr_t VisitIsNotNull(const string &col_name, ffi::KernelExpressionVisitorState *state); - uintptr_t VisitStructExtractFilter(const string &col_name, const StructFilter &filter, - ffi::KernelExpressionVisitorState *state); - uintptr_t VisitFilter(const string &col_name, const TableFilter &filter, ffi::KernelExpressionVisitorState *state); + uintptr_t VisitFilter(const string &col_name, const ExpressionFilter &filter, + ffi::KernelExpressionVisitorState *state); }; // Singleton class to forward logs to DuckDB diff --git a/src/include/functions/delta_scan/delta_multi_file_list.hpp b/src/include/functions/delta_scan/delta_multi_file_list.hpp index 003887cf..2b8ab34c 100644 --- a/src/include/functions/delta_scan/delta_multi_file_list.hpp +++ b/src/include/functions/delta_scan/delta_multi_file_list.hpp @@ -16,6 +16,7 @@ #include "duckdb/common/multi_file/multi_file_data.hpp" #include "duckdb/common/multi_file/multi_file_list.hpp" #include "duckdb/parser/constraints/not_null_constraint.hpp" +#include "duckdb/planner/filter/expression_filter.hpp" namespace duckdb { @@ -37,11 +38,48 @@ struct DeltaFileMetaData { idx_t cardinality = DConstants::INVALID_INDEX; ffi::KernelBoolSlice selection_vector = {nullptr, 0}; - case_insensitive_map_t partition_map; + identifier_map_t partition_map; unique_ptr>> transform_expression; }; +struct DeltaTableFilters { + using filter_set_t = unordered_map>; + using iterator = filter_set_t::iterator; + using const_iterator = filter_set_t::const_iterator; + +public: + bool HasFilters() const { + return !table_filters.empty(); + } + void PushFilter(column_t column_idx, unique_ptr table_filter) { + table_filters[column_idx] = std::move(table_filter); + } + optional_ptr TryGetFilterByColumnIndex(column_t column_idx) const { + auto entry = table_filters.find(column_idx); + if (entry == table_filters.end()) { + return nullptr; + } + return entry->second.get(); + } + + iterator begin() { // NOLINT: match stl API + return table_filters.begin(); + } + iterator end() { // NOLINT: match stl API + return table_filters.end(); + } + const_iterator begin() const { // NOLINT: match stl API + return table_filters.begin(); + } + const_iterator end() const { // NOLINT: match stl API + return table_filters.end(); + } + +private: + filter_set_t table_filters; +}; + // Constraint only for internal delta extension use // Todo: refactor to use duckdb constraint classes, updating the DuckDB side NotNullConstraint class NestedNotNullConstraint { @@ -65,17 +103,18 @@ class DeltaMultiFileList : public SimpleMultiFileList { //! MultiFileList API public: - void Bind(vector &return_types, vector &names); + void Bind(vector &return_types, vector &names); unique_ptr ComplexFilterPushdown(ClientContext &context, const MultiFileOptions &options, MultiFilePushdownInfo &info, vector> &filters) const override; unique_ptr DynamicFilterPushdown(ClientContext &context, const MultiFileOptions &options, - const vector &names, const vector &types, + const vector &names, const vector &types, const vector &column_ids, TableFilterSet &filters) const override; - unique_ptr PushdownInternal(ClientContext &context, TableFilterSet &new_filters) const; + unique_ptr PushdownInternal(ClientContext &context, TableFilterSet &new_filters, + vector column_indexes) const; vector GetAllFiles() const override; FileExpandResult GetExpandResult() const override; @@ -148,7 +187,7 @@ class DeltaMultiFileList : public SimpleMultiFileList { mutable vector> metadata; mutable vector resolved_files; - mutable TableFilterSet table_filters; + mutable DeltaTableFilters table_filters; mutable vector not_null_constraints; mutable bool has_null_constraints_in_arrays = false; diff --git a/src/include/functions/delta_scan/delta_multi_file_reader.hpp b/src/include/functions/delta_scan/delta_multi_file_reader.hpp index d56b7a0b..5ed4acbe 100644 --- a/src/include/functions/delta_scan/delta_multi_file_reader.hpp +++ b/src/include/functions/delta_scan/delta_multi_file_reader.hpp @@ -33,12 +33,12 @@ struct DeltaMultiFileReader : public MultiFileReader { //! Override the regular parquet bind using the MultiFileReader Bind. The bind from these are what DuckDB's file //! readers will try read - bool Bind(MultiFileOptions &options, MultiFileList &files, vector &return_types, vector &names, - MultiFileReaderBindData &bind_data) override; + bool Bind(MultiFileOptions &options, MultiFileList &files, vector &return_types, + vector &names, MultiFileReaderBindData &bind_data) override; //! Override the Options bind void BindOptions(MultiFileOptions &options, MultiFileList &files, vector &return_types, - vector &names, MultiFileReaderBindData &bind_data) override; + vector &names, MultiFileReaderBindData &bind_data) override; unique_ptr InitializeGlobalState(ClientContext &context, const MultiFileOptions &file_options, diff --git a/src/storage/delta_insert.cpp b/src/storage/delta_insert.cpp index f48d9b3a..eadde223 100644 --- a/src/storage/delta_insert.cpp +++ b/src/storage/delta_insert.cpp @@ -286,7 +286,7 @@ string DeltaInsert::GetName() const { InsertionOrderPreservingMap DeltaInsert::ParamsToString() const { InsertionOrderPreservingMap result; - result["Table Name"] = table ? table->name : info->Base().table; + result["Table Name"] = table ? table->name.GetIdentifierName() : info->Base().GetTableName().GetIdentifierName(); return result; } @@ -297,8 +297,8 @@ static optional_ptr TryGetCopyFunction(DatabaseInstanc D_ASSERT(!name.empty()); auto &system_catalog = Catalog::GetSystemCatalog(db); auto data = CatalogTransaction::GetSystemTransaction(db); - auto &schema = system_catalog.GetSchema(data, DEFAULT_SCHEMA); - return schema.GetEntry(data, CatalogType::COPY_FUNCTION_ENTRY, name)->Cast(); + auto &schema = system_catalog.GetSchema(data, Identifier::DefaultSchema()); + return schema.GetEntry(data, CatalogType::COPY_FUNCTION_ENTRY, Identifier(name))->Cast(); } PhysicalOperator &DeltaCatalog::PlanInsert(ClientContext &context, PhysicalPlanGenerator &planner, LogicalInsert &op, @@ -316,7 +316,7 @@ PhysicalOperator &DeltaCatalog::PlanInsert(ClientContext &context, PhysicalPlanG // In child catalog mode, the LogicalInsert will not actually contain the table entry, so we need to look it up // here CatalogEntryRetriever retriever(context); - EntryLookupInfo lookup_info(CatalogType::TABLE_ENTRY, default_table); + EntryLookupInfo lookup_info(CatalogType::TABLE_ENTRY, Identifier(default_table)); auto default_table_entry = LookupEntry(retriever, default_schema, lookup_info, OnEntryNotFound::THROW_EXCEPTION); table_entry = default_table_entry.entry->Cast(); @@ -359,7 +359,8 @@ PhysicalOperator &DeltaCatalog::PlanInsert(ClientContext &context, PhysicalPlanG auto names_to_write = columns.GetColumnNames(); auto types_to_write = columns.GetColumnTypes(); - auto function_data = copy_fun->function.copy_to_bind(context, bind_input, names_to_write, types_to_write); + auto function_data = + copy_fun->function.copy_to_bind(context, bind_input, StringsToIdentifiers(names_to_write), types_to_write); auto &insert = planner.Make(op, *table_entry, op.column_index_map); @@ -394,11 +395,10 @@ PhysicalOperator &DeltaCatalog::PlanInsert(ClientContext &context, PhysicalPlanG physical_copy_ref.file_extension = "parquet"; physical_copy_ref.overwrite_mode = CopyOverwriteMode::COPY_OVERWRITE_OR_IGNORE; physical_copy_ref.per_thread_output = false; - physical_copy_ref.rotate = false; physical_copy_ref.return_type = CopyFunctionReturnType::WRITTEN_FILE_STATISTICS; physical_copy_ref.write_partition_columns = true; physical_copy_ref.children.push_back(*plan); - physical_copy_ref.names = names_to_write; + physical_copy_ref.names = StringsToIdentifiers(names_to_write); physical_copy_ref.expected_types = types_to_write; physical_copy_ref.hive_file_pattern = true; diff --git a/src/storage/delta_schema_entry.cpp b/src/storage/delta_schema_entry.cpp index b2a18a59..fbcda943 100644 --- a/src/storage/delta_schema_entry.cpp +++ b/src/storage/delta_schema_entry.cpp @@ -42,10 +42,10 @@ optional_ptr DeltaSchemaEntry::CreateFunction(CatalogTransaction t } void DeltaUnqualifyColumnRef(ParsedExpression &expr) { - if (expr.type == ExpressionType::COLUMN_REF) { + if (expr.GetExpressionType() == ExpressionType::COLUMN_REF) { auto &colref = expr.Cast(); - auto name = std::move(colref.column_names.back()); - colref.column_names = {std::move(name)}; + auto name = std::move(colref.ColumnNamesMutable().back()); + colref.ColumnNamesMutable() = {std::move(name)}; return; } ParsedExpressionIterator::EnumerateChildren(expr, DeltaUnqualifyColumnRef); @@ -117,17 +117,17 @@ unique_ptr DeltaSchemaEntry::CreateTableEntry(ClientContext &co // Get the names and types from the delta snapshot vector return_types; - vector names; + vector names; snapshot->Bind(return_types, names); // TODO: forward nullability constraints CreateTableInfo table_info; for (idx_t i = 0; i < return_types.size(); i++) { - table_info.columns.AddColumn(ColumnDefinition(names[i], return_types[i])); + table_info.columns.AddColumn(ColumnDefinition(Identifier(names[i]), return_types[i])); } - table_info.table = - !delta_catalog.internal_table_name.empty() ? delta_catalog.internal_table_name : catalog.GetName(); + table_info.SetTableName(!delta_catalog.internal_table_name.empty() ? Identifier(delta_catalog.internal_table_name) + : catalog.GetName()); // Copy over constraints to table info TODO: these are incompatible currently // table_info.constraints = snapshot->not_null_constraints;} diff --git a/src/storage/delta_table_entry.cpp b/src/storage/delta_table_entry.cpp index ac50afbd..1831d369 100644 --- a/src/storage/delta_table_entry.cpp +++ b/src/storage/delta_table_entry.cpp @@ -36,7 +36,7 @@ TableFunction DeltaTableEntry::GetScanFunctionInternal(ClientContext &context, u auto &system_catalog = Catalog::GetSystemCatalog(db); auto data = CatalogTransaction::GetSystemTransaction(db); - auto &schema = system_catalog.GetSchema(data, DEFAULT_SCHEMA); + auto &schema = system_catalog.GetSchema(data, Identifier::DefaultSchema()); auto catalog_entry = schema.GetEntry(data, CatalogType::TABLE_FUNCTION_ENTRY, "delta_scan"); if (!catalog_entry) { throw InvalidInputException("Function with name \"%s\" not found in ExtensionLoader::GetTableFunction", name); @@ -64,13 +64,13 @@ TableFunction DeltaTableEntry::GetScanFunctionInternal(ClientContext &context, u } function_info->snapshot = this->snapshot; - function_info->table_name = delta_catalog.GetName(); + function_info->table_name = delta_catalog.GetName().GetIdentifierName(); delta_scan_function.function_info = std::move(function_info); vector inputs = {delta_catalog.GetDBPath()}; named_parameter_map_t param_map; vector return_types; - vector names; + vector names; TableFunctionRef empty_ref; // Propagate settings @@ -80,7 +80,9 @@ TableFunction DeltaTableEntry::GetScanFunctionInternal(ClientContext &context, u TableFunctionBindInput bind_input(inputs, param_map, return_types, names, nullptr, nullptr, delta_scan_function, empty_ref); - auto result = delta_scan_function.bind(context, bind_input, return_types, names); + vector bind_names; + auto result = delta_scan_function.bind(context, bind_input, return_types, bind_names); + names = StringsToIdentifiers(bind_names); bind_data = std::move(result); return delta_scan_function; @@ -90,7 +92,7 @@ case_insensitive_map_t> DeltaTableEntry::GetNotN case_insensitive_map_t> result; for (auto &constraint : snapshot->GetNestedNotNullConstraints()) { auto &col = GetColumn(constraint.index); - auto &item = result[col.Name()]; + auto &item = result[col.Name().GetIdentifierName()]; item.push_back(constraint); } return result; diff --git a/src/storage/delta_transaction.cpp b/src/storage/delta_transaction.cpp index 7af0fe6a..58db4319 100644 --- a/src/storage/delta_transaction.cpp +++ b/src/storage/delta_transaction.cpp @@ -43,12 +43,6 @@ static void *allocate_string(const struct ffi::KernelStringSlice slice) { } struct DeltaCommitInfo { -public: - DeltaCommitInfo() { - buffer.Initialize(Allocator::DefaultAllocator(), GetTypes()); - buffer.SetCardinality(0); - } - public: static vector GetTypes() { return {LogicalType::MAP(LogicalType::VARCHAR, LogicalType::VARCHAR)}; @@ -59,15 +53,7 @@ struct DeltaCommitInfo { public: void Append(Value commit_info_map) { - idx_t current_size = buffer.size(); - idx_t current_capacity = buffer.GetCapacity(); - - if (current_size == current_capacity) { - buffer.SetCapacity(2 * current_capacity); - } - - buffer.SetValue(0, current_size, commit_info_map); - buffer.SetCardinality(current_size + 1); + values.push_back(std::move(commit_info_map)); } void (*release)(); @@ -79,6 +65,14 @@ struct DeltaCommitInfo { ffi::ArrowFFIData ToArrow(optional_ptr context) { LoggerCallback::TryLog("delta", LogLevel::LOG_TRACE, "Delta ToArrow debug: created CommitInfo"); + DataChunk buffer; + idx_t capacity = MaxValue(values.size(), STANDARD_VECTOR_SIZE); + buffer.Initialize(Allocator::DefaultAllocator(), GetTypes(), capacity); + for (idx_t i = 0; i < values.size(); i++) { + buffer.SetValue(0, i, values[i]); + } + buffer.SetCardinality(values.size()); + ffi::ArrowFFIData ffi_data; unordered_map> extension_types; ClientProperties props("UTC", ArrowOffsetSize::REGULAR, false, false, false, ArrowFormatVersion::V1_0, context); @@ -90,7 +84,7 @@ struct DeltaCommitInfo { } private: - DataChunk buffer; + vector values; }; struct StatNode { @@ -152,22 +146,25 @@ static Value CreateValueLogicalTypeFromStatNode(const unordered_map &outstanding_appends) { const DeltaDataFile *first_file = outstanding_appends.empty() ? nullptr : &outstanding_appends[0]; buffer_types = GetTypes(first_file); - buffer = make_uniq(); - buffer->Initialize(Allocator::DefaultAllocator(), buffer_types); for (const auto &file : outstanding_appends) { auto table_path = snapshot.GetPath(); @@ -250,21 +245,11 @@ struct WriteMetaData { } void Append(const string &path, Value partition_values, const DeltaDataFile &file) { - idx_t current_size = buffer->size(); - idx_t current_capacity = buffer->GetCapacity(); - - if (current_size == current_capacity) { - buffer->SetCapacity(2 * current_capacity); - } - - auto stats = CreateStatsValue(file, true); - - buffer->SetValue(0, current_size, path); - buffer->SetValue(1, current_size, partition_values); - buffer->SetValue(2, current_size, Value::BIGINT(file.file_size_bytes)); - buffer->SetValue(3, current_size, Value::BIGINT(Timestamp::GetEpochMs(file.last_modified_time))); - buffer->SetValue(4, current_size, stats); - buffer->SetCardinality(current_size + 1); + paths.push_back(Value(path)); + partition_values_list.push_back(std::move(partition_values)); + sizes.push_back(Value::BIGINT(file.file_size_bytes)); + modification_times.push_back(Value::BIGINT(Timestamp::GetEpochMs(file.last_modified_time))); + stats.push_back(CreateStatsValue(file, true)); } void (*release)(); @@ -276,11 +261,24 @@ struct WriteMetaData { ffi::ArrowFFIData ToArrow(ClientContext &context) { LoggerCallback::TryLog("delta", LogLevel::LOG_TRACE, "Delta ToArrow debug: created WriteMetaData"); + DataChunk buffer; + idx_t n = paths.size(); + idx_t capacity = MaxValue(n, STANDARD_VECTOR_SIZE); + buffer.Initialize(Allocator::DefaultAllocator(), buffer_types, capacity); + for (idx_t i = 0; i < n; i++) { + buffer.SetValue(0, i, paths[i]); + buffer.SetValue(1, i, partition_values_list[i]); + buffer.SetValue(2, i, sizes[i]); + buffer.SetValue(3, i, modification_times[i]); + buffer.SetValue(4, i, stats[i]); + } + buffer.SetCardinality(n); + ffi::ArrowFFIData ffi_data; unordered_map> extension_types; ClientProperties props("UTC", ArrowOffsetSize::REGULAR, false, false, false, ArrowFormatVersion::V1_0, optional_ptr(&context)); - ArrowConverter::ToArrowArray(*buffer, (ArrowArray *)(&ffi_data.array), props, extension_types); + ArrowConverter::ToArrowArray(buffer, (ArrowArray *)(&ffi_data.array), props, extension_types); ArrowConverter::ToArrowSchema((ArrowSchema *)(&ffi_data.schema), buffer_types, GetNames(), props); ffi_data.array.release = reinterpret_cast(InstrumentedRelease); @@ -288,8 +286,13 @@ struct WriteMetaData { return ffi_data; } +private: vector buffer_types; - unique_ptr buffer; + vector paths; + vector partition_values_list; + vector sizes; + vector modification_times; + vector stats; }; vector DeltaTransaction::GetWriteSchema(ClientContext &context) { diff --git a/test/sql/dat/custom_parameters.test b/test/sql/dat/custom_parameters.test index 2ac854c3..c036d1d1 100644 --- a/test/sql/dat/custom_parameters.test +++ b/test/sql/dat/custom_parameters.test @@ -42,5 +42,5 @@ WHERE filename != 'henk' and letter = 'd' query IIIII SELECT filter_type, files_before, files_after, filters_before, filters_after FROM delta_filter_pushdown_log() order by filter_type; ---- -constant 2 1 [] ['letter=\'d\''] -dynamic 1 1 ['letter=\'d\''] ['letter=\'d\''] +constant 2 1 [] ['(letter = \'d\')'] +dynamic 1 1 ['(letter = \'d\')'] ['(letter = \'d\')'] diff --git a/test/sql/generated/file_skipping_all_types.test b/test/sql/generated/file_skipping_all_types.test index 197231ba..7cc2fe13 100644 --- a/test/sql/generated/file_skipping_all_types.test +++ b/test/sql/generated/file_skipping_all_types.test @@ -19,7 +19,7 @@ WHERE value2 > 2.5 and value3 < 3.5 ---- -analyzed_plan :.*File Filters:.*value1>0.5.*value2>2.5.*value3<3.5.*Scanning Files: 1/5.* +analyzed_plan :.*File Filters:.*value1 > 0.5.*value2 > 2.5.*value3 < 3.5.*Scanning Files: 1/5.* query III SELECT value1, value2, value3 @@ -46,7 +46,7 @@ EXPLAIN ANALYZE SELECT * FROM delta_scan('./data/generated/test_file_skipping/bool/delta_lake') WHERE value1=false ---- -analyzed_plan :.*File Filters:.*value1=false.*Scanning Files: 1/2.* +analyzed_plan :.*File Filters:.*value1 = false.*Scanning Files: 1/2.* query II EXPLAIN ANALYZE SELECT part @@ -66,7 +66,7 @@ WHERE value2 > 2 and value3 < 4 ---- -analyzed_plan :.*File Filters:.*value1>1.*value2>2.*value3<4.*Scanning Files: 1/5.* +analyzed_plan :.*File Filters:.*value1 > 1.*value2 > 2.*value3 < 4.*Scanning Files: 1/5.* query III SELECT value1, value2, value3 @@ -96,7 +96,7 @@ WHERE value2 = '2' and value3 = '2' ---- -analyzed_plan :.*File Filters:.*value1='2'.*value2='2'.*value3='2'.*Scanning Files: 1/5.* +analyzed_plan :.*File Filters:.*value1 = '2'.*value2 = '2'.*value3 = '2'.*Scanning Files: 1/5.* query III SELECT value1, value2, value3 @@ -124,7 +124,7 @@ FROM delta_scan('./data/generated/test_file_skipping/date/delta_lake') WHERE value1 = '1994-01-01'::DATE ---- -analyzed_plan :.*File Filters:.*value1='1994-01-01'.*Scanning Files: 1/5.* +analyzed_plan :.*File Filters:.*value1 = '1994-01-01'.*Scanning Files: 1/5.* query II SELECT value1, part @@ -148,7 +148,7 @@ EXPLAIN ANALYZE SELECT value1, part FROM delta_scan('./data/generated/test_file_skipping/decimal/delta_lake') WHERE value1 = 3.00::DECIMAL(10,2) ---- -analyzed_plan :.*File Filters:.*value1=3.*Scanning Files: 1/5.* +analyzed_plan :.*File Filters:.*value1 = 3.*Scanning Files: 1/5.* query II SELECT value1, part diff --git a/test/sql/generated/file_skipping_dynamic.test b/test/sql/generated/file_skipping_dynamic.test index fead510a..a661a13b 100644 --- a/test/sql/generated/file_skipping_dynamic.test +++ b/test/sql/generated/file_skipping_dynamic.test @@ -23,7 +23,7 @@ query IIIII SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ---- -constant [] ['value2=102', 'value3=1002'] 5 1 +constant [] ['(value2 = 102)', '(value3 = 1002)'] 5 1 # Clear logging storage statement ok @@ -41,7 +41,7 @@ query IIIII SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ---- -constant [] ['value3=1002'] 5 1 +constant [] ['(value3 = 1002)'] 5 1 # Clear logging storage statement ok @@ -59,7 +59,7 @@ query IIIII SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ---- -dynamic [] ['value2=102'] 5 1 +dynamic [] ['(value2 = 102)'] 5 1 # Clear logging storage statement ok @@ -77,7 +77,7 @@ query IIIII SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ---- -dynamic [] ['value1=13'] 5 1 +dynamic [] ['(value1 = 13)'] 5 1 # Clear logging storage statement ok @@ -95,7 +95,7 @@ query IIIII SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ---- -dynamic [] ['part=2'] 5 1 +dynamic [] ['(part = 2)'] 5 1 # Clear logging storage statement ok @@ -115,8 +115,8 @@ SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ORDER BY filter_type ---- -constant [] ['value1=13'] 5 1 -dynamic ['value1=13'] ['value1=13', 'value2=103'] 1 1 +constant [] ['(value1 = 13)'] 5 1 +dynamic ['(value1 = 13)'] ['(value1 = 13)', '(value2 = 103)'] 1 1 # Clear logging storage statement ok @@ -136,8 +136,8 @@ SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ORDER BY filter_type ---- -constant [] ['value1=13'] 5 1 -dynamic ['value1=13'] ['value1=13 AND value1=13 AND value1=13'] 1 1 +constant [] ['(value1 = 13)'] 5 1 +dynamic ['(value1 = 13)'] ['((value1 = 13) AND (value1 = 13))'] 1 1 # Clear logging storage statement ok @@ -167,5 +167,5 @@ SELECT filter_type, filters_before, filters_after, files_before, files_after FROM delta_filter_pushdown_log() ORDER BY filter_type ---- -constant [] ['value1=12'] NULL NULL -dynamic [] ['value1=12'] NULL NULL +constant [] ['(value1 = 12)'] NULL NULL +dynamic [] ['(value1 = 12)'] NULL NULL diff --git a/test/sql/generated/partitioned_large.test b/test/sql/generated/partitioned_large.test index b8506bfc..b1aa24a1 100644 --- a/test/sql/generated/partitioned_large.test +++ b/test/sql/generated/partitioned_large.test @@ -17,7 +17,7 @@ CREATE VIEW t AS SELECT part::INT as part, sum(i) as value query II EXPLAIN FROM t ---- -physical_plan :.*PARTITIONED_AGGREGATE.* +physical_plan :.*Partitioned Aggregate.* # With a projection and delta constant column query II @@ -54,7 +54,7 @@ CREATE VIEW t2 AS SELECT part::INT as part, sum(i) as value query II EXPLAIN FROM t2 ---- -physical_plan :.*PARTITIONED_AGGREGATE.* +physical_plan :.*Partitioned Aggregate.* # With a projection and delta constant column query II @@ -94,7 +94,7 @@ CREATE VIEW t3 AS SELECT part::INT as part, sum(i) as value query II EXPLAIN FROM t3 ---- -physical_plan :.*PARTITIONED_AGGREGATE.* +physical_plan :.*Partitioned Aggregate.* # Then for ATTACH, with default settings partition info pushdown statement ok @@ -109,5 +109,5 @@ CREATE VIEW t4 AS SELECT part::INT as part, sum(i) as value query II EXPLAIN FROM t4 ---- -physical_plan :.*PARTITIONED_AGGREGATE.* +physical_plan :.*Partitioned Aggregate.* diff --git a/test/sql/inlined/variant/basic.test b/test/sql/inlined/variant/basic.test index a142da9a..a8b6d3e3 100644 --- a/test/sql/inlined/variant/basic.test +++ b/test/sql/inlined/variant/basic.test @@ -28,7 +28,7 @@ select * from delta_table; query I select stats(data) from delta_table; ---- -shredding_state: INCONSISTENT[Has Null: true, Has No Null: true] +{'has_no_null': true, 'has_null': true, 'shredding_state': INCONSISTENT} query I select variant_typeof(data) from delta_table; diff --git a/test/sql/main/writing/incremental_snapshot.test b/test/sql/main/writing/incremental_snapshot.test index 5522a48a..dbd27a1f 100644 --- a/test/sql/main/writing/incremental_snapshot.test +++ b/test/sql/main/writing/incremental_snapshot.test @@ -1,5 +1,5 @@ # name: test/sql/main/writing/incremental_snapshot.test -# description: Test that incremental snapshot loading does not re-read previously loaded log files +# description: Incremental snapshot loading reads only checkpoint-bounded commit files # group: [writing] require parquet @@ -9,7 +9,14 @@ require delta require notwindows # ----------------------------------------------------------------------------- -# setup +# setup: table with checkpoints at v2 and v4, plus a trailing commit v5 +# +# Log layout produced below: +# 0000.json 0001.json (v0, v1) +# 0002.checkpoint.parquet 0002.json (v2 + checkpoint@v2) +# 0003.json (v3) +# 0004.checkpoint.parquet 0004.json (v4 + checkpoint@v4) +# 0005.json (v5, trailing) # # simple_table copy gives us a v0 with 10 rows @@ -19,103 +26,118 @@ FROM copy_dir('data/inlined/simple_table', '__TEST_DIR__/main/writing/incrementa statement ok ATTACH '__TEST_DIR__/main/writing/incremental_snapshot/delta_lake' AS t (TYPE delta); -# 2 inserts makes v1 and v2 +# v1, v2, then a checkpoint at v2 statement ok INSERT INTO t VALUES (10); statement ok INSERT INTO t VALUES (11); +statement ok +CHECKPOINT t; + +# v3, v4, then a checkpoint at v4 +statement ok +INSERT INTO t VALUES (12); + +statement ok +INSERT INTO t VALUES (13); + +statement ok +CHECKPOINT t; + +# v5, trailing commit after the last checkpoint +statement ok +INSERT INTO t VALUES (14); + statement ok DETACH t; -# Enable DeltaKernel logging to track metadata/log file reads +# Enable DeltaKernel trace logging so we can observe which commit files each read touches. statement ok -CALL enable_logging(level = 'trace'); +CALL enable_logging(level='trace', storage='memory'); statement ok SET delta_kernel_logging=true; - # ----------------------------------------------------------------------------- -# test: Variant 1: separate ATTACH calls per version +# test: step forward v2 -> v4 -> v5, asserting checkpoint-bounded commit reads +# +# The kernel emits a "Provisionally selecting ... NNNN.json" trace only for JSON commits +# in the resolved log segment; a checkpoint bounds that segment. So a read AT a checkpoint +# version references NO commit json, and a read past a checkpoint references only the +# trailing commit(s) after it. Pre-checkpoint commits must never be re-read. # -# Cold load at v1: reads log entries 0–1 statement ok -ATTACH '__TEST_DIR__/main/writing/incremental_snapshot/delta_lake' AS t1 (TYPE delta, VERSION 1); - -query I -SELECT count() FROM t1; ----- -11 +ATTACH '__TEST_DIR__/main/writing/incremental_snapshot/delta_lake' AS t3 (TYPE delta); -# Clear logs to track the next op; we'll track kernel logs for version file read refs, -# (e.g. 00000000000000000000.json) +# --- v2: served entirely from checkpoint@v2; no commit json read --- statement ok CALL truncate_duckdb_logs(); -# Incremental load at v2 via separate ATTACH: should reuse v1 snapshot, only reading log entry 2. -statement ok -ATTACH '__TEST_DIR__/main/writing/incremental_snapshot/delta_lake' AS t2 (TYPE delta, VERSION 2); - query I -SELECT count() FROM t2; +SELECT count() FROM t3 AT (VERSION => 2); ---- 12 -# v0/v1 log files must NOT be re-read by the incremental path. +# commits 0..2 are covered by 0002.checkpoint.parquet and must NOT be read query I SELECT count() FROM duckdb_logs WHERE type = 'DeltaKernel' -AND message LIKE '%00000000000000000000.json%'; ----- -0 - -query I -SELECT count() FROM duckdb_logs -WHERE type = 'DeltaKernel' -AND message LIKE '%00000000000000000001.json%'; +AND (message LIKE '%00000000000000000000.json%' + OR message LIKE '%00000000000000000001.json%' + OR message LIKE '%00000000000000000002.json%'); ---- 0 +# --- v4: served entirely from checkpoint@v4; no commit json read --- statement ok CALL truncate_duckdb_logs(); -# ----------------------------------------------------------------------------- -# test: Variant 2: AT (VERSION => n) on a plain ATTACH -# - -statement ok -ATTACH '__TEST_DIR__/main/writing/incremental_snapshot/delta_lake' AS t3 (TYPE delta); +query I +SELECT count() FROM t3 AT (VERSION => 4); +---- +14 -# Cold read at v1 via AT clause query I -SELECT count() FROM t3 AT (VERSION => 1); +SELECT count() FROM duckdb_logs +WHERE type = 'DeltaKernel' +AND (message LIKE '%00000000000000000000.json%' + OR message LIKE '%00000000000000000001.json%' + OR message LIKE '%00000000000000000002.json%' + OR message LIKE '%00000000000000000003.json%' + OR message LIKE '%00000000000000000004.json%'); ---- -11 +0 +# --- v5: checkpoint@v4 plus the single trailing commit 0005.json --- statement ok CALL truncate_duckdb_logs(); -# Incremental read at v2 via AT clause: should reuse v1 snapshot query I -SELECT count() FROM t3 AT (VERSION => 2); +SELECT count() FROM t3 AT (VERSION => 5); ---- -12 +15 -# Same filename-based assertions as Variant 1. +# the trailing commit after checkpoint@v4 IS read. This also guards the negative assertions +# above from passing vacuously on a build whose kernel emits no trace at all. query I -SELECT count() FROM duckdb_logs +SELECT count() > 0 FROM duckdb_logs WHERE type = 'DeltaKernel' -AND message LIKE '%00000000000000000000.json%'; +AND message LIKE '%00000000000000000005.json%'; ---- -0 +true +# nothing at or before checkpoint@v4 is re-read to reach v5 query I SELECT count() FROM duckdb_logs WHERE type = 'DeltaKernel' -AND message LIKE '%00000000000000000001.json%'; +AND (message LIKE '%00000000000000000000.json%' + OR message LIKE '%00000000000000000001.json%' + OR message LIKE '%00000000000000000002.json%' + OR message LIKE '%00000000000000000003.json%' + OR message LIKE '%00000000000000000004.json%'); ---- 0 @@ -123,30 +145,29 @@ statement ok CALL truncate_duckdb_logs(); # ----------------------------------------------------------------------------- -# test: Variant 3: backward time travel (fresh builder fallback) +# test: backward time travel falls back to a fresh (non-incremental) builder # statement ok ATTACH '__TEST_DIR__/main/writing/incremental_snapshot/delta_lake' AS t4 (TYPE delta); -# Load HEAD to populate the schema-level cached_table at v2 (12 rows) +# Load HEAD (v5, 15 rows) to populate the schema-level cached_table query I SELECT count() FROM t4; ---- -12 +15 statement ok CALL truncate_duckdb_logs(); -# Request v1 (< v2): kernel rejects builder_from for backward travel, so -# InitializeSnapshot falls back to a fresh builder. Verify data is correct. +# Request v1 (< cached v5): the kernel rejects builder_from for backward travel, so +# InitializeSnapshot falls back to a fresh builder. Data must still be correct. query I SELECT count() FROM t4 AT (VERSION => 1); ---- 11 -# Check duckdb delta's own logs for 'incremental=false', to confirm -# non-incremental load +# Confirm the non-incremental fallback via delta's own boundary log query I SELECT count() > 0 FROM duckdb_logs WHERE type = 'delta.DeltaMultiFileList'