[VL] Preserve ordinals for nested Substrait field references

cpp/velox/substrait/SubstraitToVeloxExpr.cc

@@ -208,6 +208,55 @@ makeFieldAccessExpr(const std::string& name, const TypePtr& type, core::FieldAcc
   return std::make_shared<core::FieldAccessTypedExpr>(type, name);
 }
 
+core::TypedExprPtr
+makeOrdinalFieldReferenceExpr(uint32_t index, const RowTypePtr& inputType, core::TypedExprPtr input) {
+  const auto& type = inputType->childAt(index);
+  if (input) {
+    return std::make_shared<core::DereferenceTypedExpr>(type, std::move(input), index);
+  }
+
+  return std::make_shared<core::FieldAccessTypedExpr>(type, inputType->nameOf(index));
+}
+
+core::TypedExprPtr toVeloxOrdinalFieldReferenceExpr(
+    const ::substrait::Expression::FieldReference& substraitField,
+    const RowTypePtr& inputType) {
+  auto typeCase = substraitField.reference_type_case();
+  switch (typeCase) {
+    case ::substrait::Expression::FieldReference::ReferenceTypeCase::kDirectReference: {
+      const auto& directRef = substraitField.direct_reference();
+      core::TypedExprPtr fieldReference{nullptr};
+      const auto* tmp = &directRef.struct_field();
+
+      auto inputColumnType = inputType;
+      for (;;) {
+        auto idx = tmp->field();
+        VELOX_USER_CHECK(
+            idx >= 0 && static_cast<uint32_t>(idx) < inputColumnType->size(),
+            "Field reference index {} is out of range for the {}-field row type.",
+            idx,
+            inputColumnType->size());
+        const TypePtr childType = inputColumnType->childAt(idx);
+        fieldReference =
+            makeOrdinalFieldReferenceExpr(static_cast<uint32_t>(idx), inputColumnType, std::move(fieldReference));
+
+        if (!tmp->has_child()) {
+          break;
+        }
+
+        inputColumnType = asRowType(childType);
+        VELOX_USER_CHECK_NOT_NULL(
+            inputColumnType,
+            "Nested field reference into a non-struct type (e.g. an array or map element) is not supported.");
+        tmp = &tmp->child().struct_field();
+      }
+      return fieldReference;
+    }
+    default:
+      VELOX_NYI("Substrait conversion not supported for Reference '{}'", std::to_string(typeCase));
+  }
+}
+
 } // namespace
 
 using facebook::velox::variantToVector;
@@ -651,7 +700,7 @@ core::TypedExprPtr SubstraitVeloxExprConverter::toVeloxExpr(
     case ::substrait::Expression::RexTypeCase::kScalarFunction:
       return toVeloxExpr(substraitExpr.scalar_function(), inputType);
     case ::substrait::Expression::RexTypeCase::kSelection:
-      return toVeloxExpr(substraitExpr.selection(), inputType);
+      return toVeloxOrdinalFieldReferenceExpr(substraitExpr.selection(), inputType);
     case ::substrait::Expression::RexTypeCase::kCast:
       return toVeloxExpr(substraitExpr.cast(), inputType);
     case ::substrait::Expression::RexTypeCase::kIfThen:

cpp/velox/substrait/SubstraitToVeloxPlan.cc

@@ -592,6 +592,7 @@ core::PlanNodePtr SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
       if (substraitAggMask.ByteSizeLong() > 0) {
         mask = std::dynamic_pointer_cast<const core::FieldAccessTypedExpr>(
             exprConverter_->toVeloxExpr(substraitAggMask, inputType));
+        VELOX_USER_CHECK_NOT_NULL(mask, "Aggregation Operator only supports a top-level field mask.");
       }
     }
     const auto& aggFunction = measure.measure();

cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc

@@ -289,4 +289,78 @@ TEST_F(Substrait2VeloxPlanConversionTest, filterUpper) {
       planNode->toString(true, true));
 }
 
+TEST_F(Substrait2VeloxPlanConversionTest, aggregateMaskMustBeTopLevelField) {
+  const auto makeAggregateRel = [](bool nestedMask) {
+    ::substrait::Rel rel;
+    auto* aggregate = rel.mutable_aggregate();
+    aggregate->mutable_common()->mutable_direct();
+
+    auto* read = aggregate->mutable_input()->mutable_read();
+    read->mutable_common()->mutable_direct();
+    auto* schema = read->mutable_base_schema();
+    for (const auto* name : {"nested", "mask", "value"}) {
+      schema->add_names(name);
+    }
+
+    auto* nestedType = schema->mutable_struct_()->add_types()->mutable_struct_();
+    nestedType->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+    nestedType->add_names("");
+    nestedType->add_names("");
+    nestedType->add_types()->mutable_i64()->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+    nestedType->add_types()->mutable_bool_()->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+    schema->mutable_struct_()->add_types()->mutable_bool_()->set_nullability(
+        ::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+    schema->mutable_struct_()->add_types()->mutable_i64()->set_nullability(
+        ::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+
+    auto* measure = aggregate->add_measures();
+    auto* maskField =
+        measure->mutable_filter()->mutable_selection()->mutable_direct_reference()->mutable_struct_field();
+    maskField->set_field(nestedMask ? 0 : 1);
+    if (nestedMask) {
+      maskField->mutable_child()->mutable_struct_field()->set_field(1);
+    }
+
+    auto* function = measure->mutable_measure();
+    function->set_function_reference(1);
+    function->set_phase(::substrait::AGGREGATION_PHASE_INITIAL_TO_RESULT);
+    function->set_invocation(::substrait::AggregateFunction::AGGREGATION_INVOCATION_ALL);
+    function->add_arguments()
+        ->mutable_value()
+        ->mutable_selection()
+        ->mutable_direct_reference()
+        ->mutable_struct_field()
+        ->set_field(2);
+    function->mutable_output_type()->mutable_i64()->set_nullability(::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+    return rel;
+  };
+
+  const auto makeConverter = [&] {
+    auto converter = std::make_shared<SubstraitToVeloxPlanConverter>(
+        pool(),
+        veloxCfg_.get(),
+        std::vector<std::shared_ptr<ResultIterator>>{},
+        VeloxConnectorIds{.hive = facebook::velox::exec::test::kHiveConnectorId},
+        std::nullopt,
+        std::nullopt,
+        /*validationMode=*/true);
+    converter->constructFunctionMap(std::unordered_map<uint64_t, std::string>{{1, "sum:opt_i64"}});
+    return converter;
+  };
+
+  auto plan = makeConverter()->toVeloxPlan(makeAggregateRel(/*nestedMask=*/false));
+  auto aggregation = std::dynamic_pointer_cast<const core::AggregationNode>(plan);
+  ASSERT_NE(aggregation, nullptr);
+  ASSERT_EQ(aggregation->aggregates().size(), 1);
+  ASSERT_NE(aggregation->aggregates().front().mask, nullptr);
+  EXPECT_TRUE(aggregation->aggregates().front().mask->isInputColumn());
+  EXPECT_EQ(aggregation->aggregates().front().mask->name(), "n0_1");
+
+  // A nested selection converts to a DereferenceTypedExpr, which cannot be an
+  // AggregationNode mask. Reject it instead of silently dropping the filter.
+  VELOX_ASSERT_USER_THROW(
+      makeConverter()->toVeloxPlan(makeAggregateRel(/*nestedMask=*/true)),
+      "Aggregation Operator only supports a top-level field mask.");
+}
+
 } // namespace gluten

cpp/velox/tests/SubstraitVeloxExprConverterTest.cc

@@ -18,12 +18,17 @@
 #include "substrait/SubstraitToVeloxExpr.h"
 
 #include "velox/common/base/tests/GTestUtils.h"
+#include "velox/exec/tests/utils/AssertQueryBuilder.h"
+#include "velox/exec/tests/utils/OperatorTestBase.h"
+#include "velox/exec/tests/utils/PlanBuilder.h"
 #include "velox/type/Type.h"
 
 using namespace facebook::velox;
 
 namespace gluten {
 
+class SubstraitVeloxExprConverterExecutionTest : public exec::test::OperatorTestBase {};
+
 // Regression test for a SIGSEGV in
 // SubstraitVeloxExprConverter::toVeloxExpr(Expression::FieldReference, ...).
 // The direct-reference loop descends one nested struct_field at a time with
@@ -66,4 +71,60 @@ TEST(SubstraitVeloxExprConverterTest, fieldReferenceIndexOutOfRangeThrows) {
   VELOX_ASSERT_USER_THROW(SubstraitVeloxExprConverter::toVeloxExpr(fieldReference, inputType), "out of range");
 }
 
+TEST_F(SubstraitVeloxExprConverterExecutionTest, ordinalFieldReferenceIntoNonStructThrows) {
+  auto inputType = ROW({"arr"}, {ARRAY(INTEGER())});
+
+  ::substrait::Expression substraitExpr;
+  auto* structField = substraitExpr.mutable_selection()->mutable_direct_reference()->mutable_struct_field();
+  structField->set_field(0);
+  structField->mutable_child()->mutable_struct_field()->set_field(0);
+
+  const std::unordered_map<uint64_t, std::string> functionMap;
+  SubstraitVeloxExprConverter converter(pool(), functionMap);
+  VELOX_ASSERT_THROW(converter.toVeloxExpr(substraitExpr, inputType), "Nested field reference into a non-struct type");
+}
+
+TEST_F(SubstraitVeloxExprConverterExecutionTest, ordinalFieldReferenceIndexOutOfRangeThrows) {
+  auto inputType = ROW({"a", "b"}, {INTEGER(), INTEGER()});
+  const std::unordered_map<uint64_t, std::string> functionMap;
+  SubstraitVeloxExprConverter converter(pool(), functionMap);
+
+  for (const auto index : {-1, 5}) {
+    SCOPED_TRACE(index);
+    ::substrait::Expression substraitExpr;
+    substraitExpr.mutable_selection()->mutable_direct_reference()->mutable_struct_field()->set_field(index);
+    VELOX_ASSERT_USER_THROW(converter.toVeloxExpr(substraitExpr, inputType), "out of range");
+  }
+}
+
+TEST_F(SubstraitVeloxExprConverterExecutionTest, nestedFieldReferenceUsesOrdinalForUnnamedFields) {
+  auto accumulator = makeRowVector(
+      {"", ""}, {makeFlatVector<int128_t>({12345, 67890}, DECIMAL(22, 2)), makeFlatVector<bool>({false, true})});
+  auto input = makeRowVector({"acc"}, {accumulator});
+
+  // Nested ROW fields can have duplicate or empty names. A name-based lookup
+  // would bind both references to field 0 and return HUGEINT for field 1.
+  ::substrait::Expression substraitExpr;
+  auto* structField = substraitExpr.mutable_selection()->mutable_direct_reference()->mutable_struct_field();
+  structField->set_field(0);
+  structField->mutable_child()->mutable_struct_field()->set_field(1);
+
+  const std::unordered_map<uint64_t, std::string> functionMap;
+  SubstraitVeloxExprConverter converter(pool(), functionMap);
+  auto expression = converter.toVeloxExpr(substraitExpr, asRowType(input->type()));
+  auto dereference = std::dynamic_pointer_cast<const core::DereferenceTypedExpr>(expression);
+  ASSERT_NE(dereference, nullptr);
+  EXPECT_EQ(dereference->index(), 1);
+  EXPECT_EQ(dereference->type()->kind(), TypeKind::BOOLEAN);
+
+  auto inputField = std::dynamic_pointer_cast<const core::FieldAccessTypedExpr>(dereference->inputs().front());
+  ASSERT_NE(inputField, nullptr);
+  EXPECT_TRUE(inputField->isInputColumn());
+  EXPECT_EQ(inputField->name(), "acc");
+
+  auto plan = exec::test::PlanBuilder().values({input}).projectExpressions({expression}).planNode();
+  auto result = exec::test::AssertQueryBuilder(plan).copyResults(pool());
+  test::assertEqualVectors(makeFlatVector<bool>({false, true}), result->childAt(0));
+}
+
 } // namespace gluten