[WIP] Delta v0.25.0 update

CMakeLists.txt

@@ -145,7 +145,7 @@ set(CARGO_PROFILE "$<IF:$<CONFIG:Debug>,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}

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<ExpressionType::CONJUNCTION_AND, ConjunctionExpression>();
 	visitor.visit_or = VisitVariadicExpression<ExpressionType::CONJUNCTION_OR, ConjunctionExpression>();
@@ -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<ConstantExpression>(Value::BLOB(buffer, len));
 	static_cast<ExpressionVisitor *>(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<ConstantExpression>(Value());
 	static_cast<ExpressionVisitor *>(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<ParsedExpression> ExpressionVisitor::MakeStructPatchOp(const string &kind, const string *field_name,
+                                                                  FieldList &&insertions, bool keep_input,
+                                                                  bool optional) {
+	// Encode as "delta_transform_op(<insertion values...>, 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<ComparisonExpression>(ExpressionType::COMPARE_EQUAL, make_uniq<ColumnRefExpression>("keep_input"),
+	                                    make_uniq<ConstantExpression>(Value::BOOLEAN(keep_input))));
+	children_values.push_back(make_uniq<ComparisonExpression>(ExpressionType::COMPARE_EQUAL,
+	                                                          make_uniq<ColumnRefExpression>("optional"),
+	                                                          make_uniq<ConstantExpression>(Value::BOOLEAN(optional))));
+
+	unique_ptr<ParsedExpression> field_name_val;
+	if (field_name) {
+		field_name_val = make_uniq<ConstantExpression>(Value(*field_name));
+	} else {
+		field_name_val = make_uniq<ConstantExpression>(Value());
+	}
+	children_values.push_back(make_uniq<ComparisonExpression>(
+	    ExpressionType::COMPARE_EQUAL, make_uniq<ColumnRefExpression>("field_name"), std::move(field_name_val)));
+
+	children_values.push_back(make_uniq<ComparisonExpression>(ExpressionType::COMPARE_EQUAL,
+	                                                          make_uniq<ColumnRefExpression>("kind"),
+	                                                          make_uniq<ConstantExpression>(Value(kind))));
+
+	return make_uniq<FunctionExpression>("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<ExpressionVisitor *>(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<ComparisonExpression>(ExpressionType::COMPARE_EQUAL,
-		                                                           make_uniq<ColumnRefExpression>("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<ComparisonExpression>(ExpressionType::COMPARE_EQUAL,
+		                                                          make_uniq<ColumnRefExpression>("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<ParsedExpression> expression =
-	    make_uniq<FunctionExpression>("delta_kernel_transform_expression", std::move(*children_values));
+	    make_uniq<FunctionExpression>("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<ExpressionVisitor *>(state);
 
-	unique_ptr<FieldList> 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<FieldList>();
+		insertions = std::move(*taken);
 	}
 
-	// Create is_insert_value
-	children_values->push_back(
-	    make_uniq<ComparisonExpression>(ExpressionType::COMPARE_EQUAL, make_uniq<ColumnRefExpression>("is_replace"),
-	                                    make_uniq<ConstantExpression>(Value::BOOLEAN(is_replace))));
-
-	// Create field name expr
-	unique_ptr<ParsedExpression> field_name_val;
-	if (field_name) {
-		string field_name_str = KernelUtils::FromDeltaString(*field_name);
-		field_name_val = make_uniq<ConstantExpression>(Value(field_name_str));
-	} else {
-		field_name_val = make_uniq<ConstantExpression>(Value());
-	}
-	children_values->push_back(make_uniq<ComparisonExpression>(
-	    ExpressionType::COMPARE_EQUAL, make_uniq<ColumnRefExpression>("field_name"), std::move(field_name_val)));
-
-	unique_ptr<ParsedExpression> expression =
-	    make_uniq<FunctionExpression>("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<LogicalType::DATE>();
 	visitor.visit_timestamp = VisitSimpleType<LogicalType::TIMESTAMP_TZ>();
 	visitor.visit_timestamp_ntz = VisitSimpleType<LogicalType::TIMESTAMP>();
+	visitor.visit_void = VisitSimpleType<LogicalType::SQLNULL>();
 	visitor.visit_variant = (void (*)(void *data, uintptr_t sibling_list_id, ffi::KernelStringSlice name,
 	                                  bool is_nullable, const ffi::CStringMap *metadata)) &
 	                        VisitVariant;

src/functions/delta_scan/delta_multi_file_list.cpp

@@ -363,8 +363,13 @@ static unordered_map<idx_t, Value> FindPartitionValues(ParsedExpression &transfo
 			                  child.GetExpression().Cast<FunctionExpression>().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<Value> values;
 
 		for (auto &transform_op_child : transform_op.GetArguments()) {
@@ -380,11 +385,16 @@ static unordered_map<idx_t, Value> FindPartitionValues(ParsedExpression &transfo
 				                 .Cast<ConstantExpression>()
 				                 .GetValue();
 
-				if (name == "is_replace") {
-					is_replace = value.GetValue<bool>();
+				if (name == "kind") {
+					kind = value.ToString();
+				} else if (name == "keep_input") {
+					keep_input = value.GetValue<bool>();
+				} else if (name == "optional") {
+					optional = value.GetValue<bool>();
 				} 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<idx_t, Value> 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) {

src/include/delta_utils.hpp

@@ -139,18 +139,22 @@ 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);
 	static void VisitDecimalLiteral(void *state, uintptr_t sibling_list_id, int64_t value_ms, uint64_t value_ls,
 	                                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<ParsedExpression> 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);

src/storage/delta_transaction.cpp

@@ -303,7 +303,8 @@ vector<DeltaMultiFileColumnDefinition> 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<ffi::ExclusiveCommittedTransaction> 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);
 			}
 		}
 	}

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;