diff --git a/CMakeLists.txt b/CMakeLists.txt index 6b2a950..96088ac 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -145,7 +145,7 @@ set(CARGO_PROFILE "$,dev,release>") ExternalProject_Add( ${KERNEL_NAME} GIT_REPOSITORY "https://github.com/delta-io/delta-kernel-rs" - GIT_TAG v0.21.0 + GIT_TAG v0.25.0 # Prints the env variables passed to the cargo build to the terminal, useful # in debugging because passing them through CMake is an error-prone mess CONFIGURE_COMMAND ${CMAKE_COMMAND} -E env ${RUST_UNSET_ENV_VARS} diff --git a/src/delta_utils.cpp b/src/delta_utils.cpp index cc20956..17c781c 100644 --- a/src/delta_utils.cpp +++ b/src/delta_utils.cpp @@ -72,6 +72,9 @@ ffi::EngineExpressionVisitor ExpressionVisitor::CreateVisitor(ExpressionVisitor visitor.visit_literal_binary = &VisitBinaryLiteral; visitor.visit_literal_null = &VisitNullLiteral; visitor.visit_literal_array = &VisitArrayLiteral; + // visit_array constructs a non-literal ARRAY[...] expression; the child-list-to-list_value() + // logic is identical to the literal-array case. + visitor.visit_array = &VisitArrayLiteral; visitor.visit_and = VisitVariadicExpression(); visitor.visit_or = VisitVariadicExpression(); @@ -93,8 +96,8 @@ ffi::EngineExpressionVisitor ExpressionVisitor::CreateVisitor(ExpressionVisitor visitor.visit_column = VisitColumnExpression; visitor.visit_struct_expr = VisitStructExpression; - visitor.visit_transform_expr = VisitTransformExpression; - visitor.visit_field_transform = VisitFieldTransform; + visitor.visit_struct_patch_expr = VisitStructPatchExpression; + visitor.visit_field_patch = VisitFieldPatch; visitor.visit_literal_struct = VisitStructLiteral; @@ -257,7 +260,10 @@ void ExpressionVisitor::VisitBinaryLiteral(void *state, uintptr_t sibling_list_i auto expression = make_uniq(Value::BLOB(buffer, len)); static_cast(state)->AppendToList(sibling_list_id, std::move(expression)); } -void ExpressionVisitor::VisitNullLiteral(void *state, uintptr_t sibling_list_id) { +void ExpressionVisitor::VisitNullLiteral(void *state, uintptr_t sibling_list_id, uint8_t type_tag, uint8_t precision, + uint8_t scale) { + // type_tag/precision/scale identify the kernel-side data type of the null; we don't need it + // since DuckDB's untyped NULL constant is cast to the correct type by the surrounding context. auto expression = make_uniq(Value()); static_cast(state)->AppendToList(sibling_list_id, std::move(expression)); } @@ -448,66 +454,105 @@ void ExpressionVisitor::VisitStructExpression(void *state, uintptr_t sibling_lis state_cast->AppendToList(sibling_list_id, std::move(expression)); } -void ExpressionVisitor::VisitTransformExpression(void *state, uintptr_t sibling_list_id, uintptr_t input_path_list_id, - uintptr_t child_list_id) { +unique_ptr ExpressionVisitor::MakeStructPatchOp(const string &kind, const string *field_name, + FieldList &&insertions, bool keep_input, + bool optional) { + // Encode as "delta_transform_op(, keep_input, optional, field_name, kind)". + // `kind` is one of "prepend"/"field"/"append" and disambiguates the unnamed prepend/append + // patches (position-only, no field_name/keep_input/optional semantics) from named field + // patches. See FindPartitionValues in delta_multi_file_list.cpp for the consumer of this + // encoding. + FieldList children_values = std::move(insertions); + + children_values.push_back( + make_uniq(ExpressionType::COMPARE_EQUAL, make_uniq("keep_input"), + make_uniq(Value::BOOLEAN(keep_input)))); + children_values.push_back(make_uniq(ExpressionType::COMPARE_EQUAL, + make_uniq("optional"), + make_uniq(Value::BOOLEAN(optional)))); + + unique_ptr field_name_val; + if (field_name) { + field_name_val = make_uniq(Value(*field_name)); + } else { + field_name_val = make_uniq(Value()); + } + children_values.push_back(make_uniq( + ExpressionType::COMPARE_EQUAL, make_uniq("field_name"), std::move(field_name_val))); + + children_values.push_back(make_uniq(ExpressionType::COMPARE_EQUAL, + make_uniq("kind"), + make_uniq(Value(kind)))); + + return make_uniq("delta_transform_op", std::move(children_values)); +} + +void ExpressionVisitor::VisitStructPatchExpression(void *state, uintptr_t sibling_list_id, uintptr_t input_path_list_id, + uintptr_t prepended_field_list_id, uintptr_t field_patch_list_id, + uintptr_t appended_field_list_id) { auto state_cast = static_cast(state); - auto children_values = state_cast->TakeFieldList(child_list_id); - if (!children_values) { + auto field_patches = state_cast->TakeFieldList(field_patch_list_id); + if (!field_patches) { + return; + } + auto prepended = state_cast->TakeFieldList(prepended_field_list_id); + if (!prepended) { + return; + } + auto appended = state_cast->TakeFieldList(appended_field_list_id); + if (!appended) { return; } - if (input_path_list_id) { - auto input_path = state_cast->TakeFieldList(input_path_list_id); + FieldList children_values; + if (!prepended->empty()) { + children_values.push_back( + state_cast->MakeStructPatchOp("prepend", nullptr, std::move(*prepended), false, false)); + } + for (auto &field_patch : *field_patches) { + children_values.push_back(std::move(field_patch)); + } + if (!appended->empty()) { + children_values.push_back(state_cast->MakeStructPatchOp("append", nullptr, std::move(*appended), false, false)); + } - if (input_path->size() != 1) { - state_cast->error = ErrorData("Expected exactly one input path for transform expression"); - return; - } - children_values->push_back(make_uniq(ExpressionType::COMPARE_EQUAL, - make_uniq("input_path"), - std::move(input_path->front()))); + // Unlike prepended/field_patch/appended lists, the kernel always allocates a (possibly empty) + // input-path list, even when there is no input path: 0 items means "no path", not "no list". + auto input_path = state_cast->TakeFieldList(input_path_list_id); + if (!input_path) { + return; + } + if (input_path->size() == 1) { + children_values.push_back(make_uniq(ExpressionType::COMPARE_EQUAL, + make_uniq("input_path"), + std::move(input_path->front()))); + } else if (!input_path->empty()) { + state_cast->error = ErrorData("Expected zero or one input path for struct patch expression"); + return; } unique_ptr expression = - make_uniq("delta_kernel_transform_expression", std::move(*children_values)); + make_uniq("delta_kernel_transform_expression", std::move(children_values)); state_cast->AppendToList(sibling_list_id, std::move(expression)); } -void ExpressionVisitor::VisitFieldTransform(void *state, uintptr_t sibling_list_id, - const ffi::KernelStringSlice *field_name, uintptr_t expr_list_id, - bool is_replace) { +void ExpressionVisitor::VisitFieldPatch(void *state, uintptr_t sibling_list_id, ffi::KernelStringSlice field_name, + uintptr_t insertion_expr_list_id, bool keep_input, bool optional) { auto state_cast = static_cast(state); - unique_ptr children_values; - - if (expr_list_id) { - children_values = state_cast->TakeFieldList(expr_list_id); - if (!children_values) { + FieldList insertions; + if (insertion_expr_list_id) { + auto taken = state_cast->TakeFieldList(insertion_expr_list_id); + if (!taken) { return; } - } else { - children_values = make_uniq(); + insertions = std::move(*taken); } - // Create is_insert_value - children_values->push_back( - make_uniq(ExpressionType::COMPARE_EQUAL, make_uniq("is_replace"), - make_uniq(Value::BOOLEAN(is_replace)))); - - // Create field name expr - unique_ptr field_name_val; - if (field_name) { - string field_name_str = KernelUtils::FromDeltaString(*field_name); - field_name_val = make_uniq(Value(field_name_str)); - } else { - field_name_val = make_uniq(Value()); - } - children_values->push_back(make_uniq( - ExpressionType::COMPARE_EQUAL, make_uniq("field_name"), std::move(field_name_val))); - - unique_ptr expression = - make_uniq("delta_transform_op", std::move(*children_values)); + string field_name_str = KernelUtils::FromDeltaString(field_name); + auto expression = + state_cast->MakeStructPatchOp("field", &field_name_str, std::move(insertions), keep_input, optional); state_cast->AppendToList(sibling_list_id, std::move(expression)); } @@ -574,6 +619,7 @@ ffi::EngineSchemaVisitor SchemaVisitor::CreateSchemaVisitor(SchemaVisitor &state visitor.visit_date = VisitSimpleType(); visitor.visit_timestamp = VisitSimpleType(); visitor.visit_timestamp_ntz = VisitSimpleType(); + visitor.visit_void = VisitSimpleType(); visitor.visit_variant = (void (*)(void *data, uintptr_t sibling_list_id, ffi::KernelStringSlice name, bool is_nullable, const ffi::CStringMap *metadata)) & VisitVariant; diff --git a/src/functions/delta_scan/delta_multi_file_list.cpp b/src/functions/delta_scan/delta_multi_file_list.cpp index 3f01498..15aaf4d 100644 --- a/src/functions/delta_scan/delta_multi_file_list.cpp +++ b/src/functions/delta_scan/delta_multi_file_list.cpp @@ -363,8 +363,13 @@ static unordered_map FindPartitionValues(ParsedExpression &transfo child.GetExpression().Cast().FunctionName()); } - bool is_replace = false; + // kind: "prepend" | "field" | "append" (see MakeStructPatchOp in delta_utils.cpp). + // keep_input/optional only apply to kind == "field". + string kind; + bool keep_input = false; + bool optional = false; string field_name; + bool has_field_name = false; vector values; for (auto &transform_op_child : transform_op.GetArguments()) { @@ -380,11 +385,16 @@ static unordered_map FindPartitionValues(ParsedExpression &transfo .Cast() .GetValue(); - if (name == "is_replace") { - is_replace = value.GetValue(); + if (name == "kind") { + kind = value.ToString(); + } else if (name == "keep_input") { + keep_input = value.GetValue(); + } else if (name == "optional") { + optional = value.GetValue(); } else if (name == "field_name") { if (!value.IsNull()) { field_name = value.ToString(); + has_field_name = true; } } else { throw InternalException("Unexpected name for delta_transform_op returned by delta kernel: %s", @@ -399,22 +409,28 @@ static unordered_map FindPartitionValues(ParsedExpression &transfo } } - /// NOTE: Treating list id 0 as an empty list yields a simplified truth table: - /// - /// |field_name? |is_replace? |meaning| + /// |kind |keep_input? |meaning| /// |-|-|-| - /// | NO | * | Prepend a (possibly empty) list of expressions to the output - /// | YES | NO | Insert a (possibly empty) list of expressions after the named input field - /// | YES | YES | Replace the named input field with a (possibly empty) list of expressions + /// |prepend | * | Prepend a (possibly empty) list of expressions to the output + /// |append | * | Append a (possibly empty) list of expressions to the output + /// |field | YES | Insert a (possibly empty) list of expressions after the named input field + /// |field | NO | Replace the named input field with a (possibly empty) list of expressions // TODO: broken for multiple transform expressions? idx_t index_to_insert = 0; - if (!field_name.empty()) { + if (kind == "append") { + index_to_insert = cols.size(); + } else if (kind == "field" && has_field_name) { + bool found = false; for (idx_t i = 0; i < cols.size(); ++i) { if (field_name == cols[i].name) { - index_to_insert = is_replace ? i : i + 1; + index_to_insert = keep_input ? i + 1 : i; + found = true; break; } } + if (!found && optional) { + continue; + } } for (idx_t i = 0; i < values.size(); ++i) { diff --git a/src/include/delta_utils.hpp b/src/include/delta_utils.hpp index f65988a..e618a6f 100644 --- a/src/include/delta_utils.hpp +++ b/src/include/delta_utils.hpp @@ -139,7 +139,8 @@ class ExpressionVisitor : public ffi::EngineExpressionVisitor { static void VisitDateLiteral(void *state, uintptr_t sibling_list_id, int32_t value); static void VisitStringLiteral(void *state, uintptr_t sibling_list_id, ffi::KernelStringSlice value); static void VisitBinaryLiteral(void *state, uintptr_t sibling_list_id, const uint8_t *buffer, uintptr_t len); - static void VisitNullLiteral(void *state, uintptr_t sibling_list_id); + static void VisitNullLiteral(void *state, uintptr_t sibling_list_id, uint8_t type_tag, uint8_t precision, + uint8_t scale); static void VisitArrayLiteral(void *state, uintptr_t sibling_list_id, uintptr_t child_id); static void VisitStructLiteral(void *data, uintptr_t sibling_list_id, uintptr_t child_field_list_value, uintptr_t child_value_list_id); @@ -147,10 +148,13 @@ class ExpressionVisitor : public ffi::EngineExpressionVisitor { uint8_t precision, uint8_t scale); static void VisitColumnExpression(void *state, uintptr_t sibling_list_id, ffi::KernelStringSlice name); static void VisitStructExpression(void *state, uintptr_t sibling_list_id, uintptr_t child_list_id); - static void VisitTransformExpression(void *data, uintptr_t sibling_list_id, uintptr_t input_path_list_id, - uintptr_t child_list_id); - static void VisitFieldTransform(void *data, uintptr_t sibling_list_id, const ffi::KernelStringSlice *field_name, - uintptr_t expr_list_id, bool is_replace); + static void VisitStructPatchExpression(void *data, uintptr_t sibling_list_id, uintptr_t input_path_list_id, + uintptr_t prepended_field_list_id, uintptr_t field_patch_list_id, + uintptr_t appended_field_list_id); + static void VisitFieldPatch(void *data, uintptr_t sibling_list_id, ffi::KernelStringSlice field_name, + uintptr_t insertion_expr_list_id, bool keep_input, bool optional); + static unique_ptr MakeStructPatchOp(const string &kind, const string *field_name, + FieldList &&insertions, bool keep_input, bool optional); static void VisitNotExpression(void *state, uintptr_t sibling_list_id, uintptr_t child_list_id); static void VisitIsNullExpression(void *state, uintptr_t sibling_list_id, uintptr_t child_list_id); static void VisitLiteralMap(void *data, uintptr_t sibling_list_id, uintptr_t key_list_id, uintptr_t value_list_id); diff --git a/src/storage/delta_transaction.cpp b/src/storage/delta_transaction.cpp index 07e7e88..b916aab 100644 --- a/src/storage/delta_transaction.cpp +++ b/src/storage/delta_transaction.cpp @@ -303,7 +303,8 @@ vector DeltaTransaction::GetWriteSchema(ClientCo InitializeTransaction(context); } - auto write_context = ffi::get_write_context(kernel_transaction.get()); + auto write_context = write_entry.get()->snapshot->TryUnpackKernelResult(ffi::get_unpartitioned_write_context( + kernel_transaction.get(), write_entry.get()->snapshot->extern_engine.get())); auto result = SchemaVisitor::VisitWriteContextSchema(write_entry.get()->snapshot->extern_engine.get(), write_context); return result; @@ -469,7 +470,7 @@ void DeltaTransaction::Commit(ClientContext &context) { // We have some special error handling here to ensure the error created by DuckDB is properly thrown here, // because we can't throw it across the FFI boundary, we need to store it in the transaction - uint64_t commit_result; + ffi::Handle commit_result; DUCKDB_LOG_INTERNAL(context, "delta.Commit", LogLevel::LOG_DEBUG, "Committing %s", table_entry->snapshot->GetPath()); @@ -481,6 +482,8 @@ void DeltaTransaction::Commit(ClientContext &context) { } else { res.Throw(); } + } else { + ffi::free_committed_transaction(commit_result); } } } diff --git a/test/sql/generated/delta_list_files.test b/test/sql/generated/delta_list_files.test index f166d72..405cc8a 100644 --- a/test/sql/generated/delta_list_files.test +++ b/test/sql/generated/delta_list_files.test @@ -25,8 +25,8 @@ SELECT data_file[-8:], cardinality, partitions, have_deletes FROM delta_list_fil query IIIII SELECT data_file[-8:], cardinality, partitions, have_deletes, transform_expression FROM delta_list_files('./data/generated/simple_partitioned/delta_lake', transform_expression=1) order by partitions; ---- -.parquet 5 {part=0} false ['delta_kernel_transform_expression(delta_transform_op(0, (is_replace = false), (field_name = \'i\')))'] -.parquet 5 {part=1} false ['delta_kernel_transform_expression(delta_transform_op(1, (is_replace = false), (field_name = \'i\')))'] +.parquet 5 {part=0} false ['delta_kernel_transform_expression(delta_transform_op(0, (keep_input = true), (optional = false), (field_name = \'i\'), (kind = \'field\')))'] +.parquet 5 {part=1} false ['delta_kernel_transform_expression(delta_transform_op(1, (keep_input = true), (optional = false), (field_name = \'i\'), (kind = \'field\')))'] query IIIII SELECT data_file[-8:], cardinality, partitions, have_deletes, delete_count FROM delta_list_files('./data/generated/simple_partitioned/delta_lake', delete_count=1) order by partitions;