diff --git a/src/embedding/plan.cpp b/src/embedding/plan.cpp index e0ef615410..c77eddf58c 100644 --- a/src/embedding/plan.cpp +++ b/src/embedding/plan.cpp @@ -376,6 +376,59 @@ void inspect_message(duckdb::google::protobuf::Message const& message, } } } +// ReferenceRel is inlined by the importer. A single wire ReadRel can therefore +// become several consumers of the same destructive MO input. Count reachable +// consumption rather than only declarations, before any native reader starts. +void inspect_consumption(duckdb::google::protobuf::Message const& message, + substrait::Plan const& plan, + int ordinal, + query_state const& query, + std::set& consumed, + std::size_t& visited, + int depth) +{ + if (++visited > 65536 || depth > 256) + throw failure(SIRIUS_UNSUPPORTED, "embedded reference expansion exceeds its bound"); + auto const name = message.GetDescriptor()->full_name(); + if (name == "substrait.ReferenceRel") { + auto const source = static_cast(message).subtree_ordinal(); + if (source < 0 || source >= ordinal || !plan.relations(source).has_rel()) + throw failure(SIRIUS_INVALID_ARGUMENT, "invalid embedded subtree reference"); + inspect_consumption( + plan.relations(source).rel(), plan, source, query, consumed, visited, depth + 1); + return; + } + if (name == "substrait.ReadRel") { + auto const& read = static_cast(message); + auto const id = binding_id(read.named_table().names(1)); + auto binding = std::find_if(query.bindings.begin(), query.bindings.end(), [id](auto const& b) { + return b.binding_id == id; + }); + if (binding != query.bindings.end() && binding->source_kind == SIRIUS_READ_MO && + !consumed.insert(id).second) + throw failure(SIRIUS_INVALID_ARGUMENT, + "a destructive MO read binding cannot be consumed through shared references"); + } + auto const* reflection = message.GetReflection(); + std::vector fields; + reflection->ListFields(message, &fields); + for (auto const* field : fields) { + if (field->cpp_type() != duckdb::google::protobuf::FieldDescriptor::CPPTYPE_MESSAGE) continue; + if (field->is_repeated()) { + for (int i = 0; i < reflection->FieldSize(message, field); ++i) + inspect_consumption(reflection->GetRepeatedMessage(message, field, i), + plan, + ordinal, + query, + consumed, + visited, + depth + 1); + } else { + inspect_consumption( + reflection->GetMessage(message, field), plan, ordinal, query, consumed, visited, depth + 1); + } + } +} } // namespace void validate_embedded_plan(std::string_view bytes, query_state const& query) @@ -437,5 +490,12 @@ void validate_embedded_plan(std::string_view bytes, query_state const& query) inspect_message(plan.relations(i), i, query, reads, functions, profile); if (reads.size() != query.bindings.size()) throw failure(SIRIUS_INVALID_ARGUMENT, "registered and planned read binding sets differ"); + std::set consumed; + std::size_t visited = 0; + inspect_consumption(root.input(), plan, plan.relations_size() - 1, query, consumed, visited, 0); + for (auto const& binding : query.bindings) + if (binding.source_kind == SIRIUS_READ_MO && !consumed.contains(binding.binding_id)) + throw failure(SIRIUS_INVALID_ARGUMENT, + "registered MO read binding has no reachable consumer"); } } // namespace sirius::embedding diff --git a/src/expression_evaluator/specializations/cast.cpp b/src/expression_evaluator/specializations/cast.cpp index 9aace651bf..5a8a216ca5 100644 --- a/src/expression_evaluator/specializations/cast.cpp +++ b/src/expression_evaluator/specializations/cast.cpp @@ -87,6 +87,23 @@ evaluate_result expression_evaluator::evaluate(sirius::ast::cast const& alt, eva child = evaluate_result( cudf::make_column_from_scalar(child.get_scalar(), _input_table.num_rows(), _stream, _mr)); } + // Exact Substrait preparation deliberately retains semantic casts instead + // of running DuckDB's ordinary optimizer. VARCHAR -> VARCHAR is an identity + // operation, but cuDF's unary cast accepts only fixed-width columns. Keep the + // value/validity and give the result an independent owner; no representation + // tunnel or conversion of other variable-width types is authorized here. + if (alt.kind == sirius::ast::cast_kind::semantic && + alt.target_type.id() == sirius::type_id::VARCHAR && + alt.child->return_type().id() == sirius::type_id::VARCHAR && + child.get_column_view().type().id() == cudf::type_id::STRING) { + auto result_column = child.is_owned_column() + ? child.release_column() + : std::make_unique(child.get_column_view(), _stream, _mr); + if (mode == evaluation_mode::AST) { + return materialize_as_ast_column(std::move(result_column)); + } + return evaluate_result(std::move(result_column)); + } // Only planner-certified carrier restoration may tunnel through the narrowed representation. // A semantic cast delegates to cuDF and is never reinterpreted as a physical DATE restore. auto result_column = diff --git a/test/cpp/embedding/test_native_binding.cpp b/test/cpp/embedding/test_native_binding.cpp index 408084c4f9..94afa2c253 100644 --- a/test/cpp/embedding/test_native_binding.cpp +++ b/test/cpp/embedding/test_native_binding.cpp @@ -101,6 +101,44 @@ TEST_CASE("embedded plan admission rejects schema and arbitrary reads", "[native sirius::embedding::failure); } +TEST_CASE("embedded shared references cannot replay destructive MO inputs", "[native_binding]") +{ + auto query = bound_query(); + auto original = one_read(); + auto plan = original; + plan.clear_relations(); + *plan.add_relations()->mutable_rel() = original.relations(0).root().input(); + auto* root = plan.add_relations()->mutable_root(); + root->add_names("c"); + root->mutable_input()->mutable_reference()->set_subtree_ordinal(0); + REQUIRE_NOTHROW(sirius::embedding::validate_embedded_plan(plan.SerializeAsString(), query)); + + auto single = plan; + root->add_names("second"); + query.contract->outputs.push_back({23, 0, 0, false, "second"}); + root->mutable_input()->Clear(); + auto* cross = root->mutable_input()->mutable_cross(); + cross->mutable_left()->mutable_reference()->set_subtree_ordinal(0); + cross->mutable_right()->mutable_reference()->set_subtree_ordinal(0); + REQUIRE_THROWS_AS(sirius::embedding::validate_embedded_plan(plan.SerializeAsString(), query), + sirius::embedding::failure); + + // An immutable TAE binding remains replayable through the importer. + query.bindings[0].source_kind = SIRIUS_READ_TAE; + REQUIRE_NOTHROW(sirius::embedding::validate_embedded_plan(plan.SerializeAsString(), query)); + + // A declared but unreachable destructive producer must not start and block. + query = bound_query(); + auto second = query.bindings[0]; + second.binding_id = 2; + query.bindings.push_back(std::move(second)); + plan = single; + *plan.mutable_relations(1)->mutable_root()->mutable_input() = + one_read("2").relations(0).root().input(); + REQUIRE_THROWS_AS(sirius::embedding::validate_embedded_plan(plan.SerializeAsString(), query), + sirius::embedding::failure); +} + TEST_CASE("embedded admission preserves allowed sort selections and scalar filters", "[native_binding]") { diff --git a/test/cpp/expression/test_value.cpp b/test/cpp/expression/test_value.cpp index 5252b5efab..9ebc5ddcd1 100644 --- a/test/cpp/expression/test_value.cpp +++ b/test/cpp/expression/test_value.cpp @@ -53,8 +53,13 @@ using sirius::value; static_assert(std::is_default_constructible_v, "sirius::value must default-construct (NULL is the natural default)."); -static_assert(std::variant_size_v == 21, - "sirius::value has exactly 21 alternatives (D-01 — locked ABI)."); +// The original 21 alternatives retain their indexes; merged MO exact support +// appends its coefficient carrier rather than changing those existing slots. +static_assert(std::variant_size_v == 22, + "sirius::value has exactly 22 alternatives including the MO coefficient."); +static_assert( + std::is_same_v, sirius::mo_decimal::coefficient>, + "The MO exact coefficient must occupy the appended variant slot."); // ============================================================================ // Round-trip every supported type_id (CONTEXT.md test plan item 1) diff --git a/test/cpp/expression_evaluator/test_expression_evaluator.cpp b/test/cpp/expression_evaluator/test_expression_evaluator.cpp index 6bec2e1bf0..72e4c66bcb 100644 --- a/test/cpp/expression_evaluator/test_expression_evaluator.cpp +++ b/test/cpp/expression_evaluator/test_expression_evaluator.cpp @@ -2633,6 +2633,46 @@ TEST_CASE("native_ast - comparison EQUAL (MATERIALIZE)", } } +TEST_CASE("native_ast - semantic VARCHAR identity casts retain values and ownership", + "[expression_evaluator_ast_native][varchar_identity_cast]") +{ + auto* space = get_default_gpu_space(); + REQUIRE(space != nullptr); + auto input = + make_input_batch(*space, {cudf::data_type{cudf::type_id::STRING}}, {std::pair{1, 5}}); + auto ro = input->to_read_only(); + auto& in_repr = ro.get_data()->cast(); + auto tv = in_repr.get_table_view(); + auto const varchar = sirius::logical_type::make(sirius::type_id::VARCHAR); + auto const expected = copy_string_column_to_host(tv.column(0)); + + for (auto strategy : {MAT, exp_strategy_enum::AST_INTERPRET, exp_strategy_enum::AST_JIT}) { + auto reference = std::make_unique(sirius::ast::reference{0, varchar}); + auto identity = + std::make_unique(sirius::ast::cast{std::move(reference), varchar, false}); + auto result = run_native_ast(*space, identity.get(), tv, strategy); + REQUIRE(result->view().column(0).type().id() == cudf::type_id::STRING); + REQUIRE(copy_string_column_to_host(result->view().column(0)) == expected); + REQUIRE(result->view().column(0).child(0).head() != tv.column(0).child(0).head()); + + for (std::string value : {std::string{}, std::string{"hello"}, std::string{"界é"}}) { + auto literal = + std::make_unique(sirius::ast::constant{sirius::value{value}, varchar}); + auto cast = + std::make_unique(sirius::ast::cast{std::move(literal), varchar, false}); + auto strings = run_native_ast(*space, cast.get(), tv, strategy); + REQUIRE(copy_string_column_to_host(strings->view().column(0)) == + std::vector(tv.num_rows(), value)); + } + auto null = + std::make_unique(sirius::ast::constant{sirius::value{}, varchar}); + auto cast = + std::make_unique(sirius::ast::cast{std::move(null), varchar, false}); + auto nulls = run_native_ast(*space, cast.get(), tv, strategy); + REQUIRE(nulls->view().column(0).null_count() == tv.num_rows()); + } +} + TEST_CASE("native_ast - comparison LESS_THAN (AST_INTERPRET)", "[expression_evaluator_ast_native][comparison]") {