fix(embedding): preserve retained casts and protect destructive MO inputs

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<uint64_t>& 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<substrait::ReferenceRel const&>(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<substrait::ReadRel const&>(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<const duckdb::google::protobuf::FieldDescriptor*> 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<uint64_t> 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

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<cudf::column>(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 =

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]")
 {

test/cpp/expression/test_value.cpp

@@ -53,8 +53,13 @@ using sirius::value;
 
 static_assert(std::is_default_constructible_v<value>,
               "sirius::value must default-construct (NULL is the natural default).");
-static_assert(std::variant_size_v<value> == 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<value> == 22,
+              "sirius::value has exactly 22 alternatives including the MO coefficient.");
+static_assert(
+  std::is_same_v<std::variant_alternative_t<21, value>, 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)

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<int, int>{1, 5}});
+  auto ro             = input->to_read_only();
+  auto& in_repr       = ro.get_data()->cast<gpu_table_representation>();
+  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::node>(sirius::ast::reference{0, varchar});
+    auto identity =
+      std::make_unique<sirius::ast::node>(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<char>() != tv.column(0).child(0).head<char>());
+
+    for (std::string value : {std::string{}, std::string{"hello"}, std::string{"界é"}}) {
+      auto literal =
+        std::make_unique<sirius::ast::node>(sirius::ast::constant{sirius::value{value}, varchar});
+      auto cast =
+        std::make_unique<sirius::ast::node>(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<std::string>(tv.num_rows(), value));
+    }
+    auto null =
+      std::make_unique<sirius::ast::node>(sirius::ast::constant{sirius::value{}, varchar});
+    auto cast =
+      std::make_unique<sirius::ast::node>(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]")
 {