Follow a leaf through a cast everywhere, not just where it is allowed

Reading a leaf as a number is allowed, so the check strips a cast before
deciding that. Two other places ask the same question -- which column a
read went into, and what to call it -- and neither stripped the cast. A
cast leaf therefore reached pushdown untracked, where a script over a
flat_object cannot be built, and the failure surfaced as

  HTTP 500, code UNKNOWN
  java.sql.SQLException: ... A flat_object field cannot be read by a
  pushed-down script: attributes

for a query such as

  source=otel | eval d = cast(attributes.duration_ms as double)
              | sort d | fields service

which now reports "Cannot sort by attributes.duration_ms" like the rest.
The three callers share one stripCasts so they cannot drift again.

Signed-off-by: Jialiang Liang <jiallian@amazon.com>
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHurdipg7MHo61tquK7BpD
Signed-off-by: Jialiang Liang <ryanleeang@gmail.com>

Author: RyanL1997

core/src/main/java/org/opensearch/sql/calcite/FlatObjectScopeValidator.java

@@ -158,7 +158,7 @@ private static Set<Integer> leafColumns(RelNode node, Set<String> fields) {
         List<RelDataTypeField> below = inputFields(project);
         List<RexNode> projects = project.getProjects();
         for (int i = 0; i < projects.size(); i++) {
-          RexNode expression = projects.get(i);
+          RexNode expression = stripCasts(projects.get(i));
           boolean fromField =
               isLeafRef(expression, below, fields)
                   || (expression instanceof RexInputRef ref
@@ -178,16 +178,26 @@ private static Set<Integer> leafColumns(RelNode node, Set<String> fields) {
    * text, so converting it is the parse a caller would do anyway, the value comes from {@code
    * _source} either way, and it is how a query asks for a leaf as a number.
    */
-  private static boolean isRead(
-      RexNode node, List<RelDataTypeField> inputFields, Set<String> fields, Set<Integer> leaves) {
-    if (node instanceof RexInputRef) {
-      return true;
-    }
+  /**
+   * Unwraps the casts a read may wear. {@code isRead} and {@link #leafColumns} have to agree on
+   * what counts as reading a leaf: a column one of them lets through while the other does not track
+   * is a column that reaches pushdown, where a script over a flat_object cannot be built.
+   */
+  private static RexNode stripCasts(RexNode node) {
     RexNode read = node;
     while (read instanceof RexCall cast
         && (cast.getKind() == SqlKind.CAST || cast.getKind() == SqlKind.SAFE_CAST)) {
       read = cast.getOperands().get(0);
     }
+    return read;
+  }
+
+  private static boolean isRead(
+      RexNode node, List<RelDataTypeField> inputFields, Set<String> fields, Set<Integer> leaves) {
+    if (node instanceof RexInputRef) {
+      return true;
+    }
+    RexNode read = stripCasts(node);
     return isLeafRef(read, inputFields, fields)
         || (read instanceof RexInputRef ref && leaves.contains(ref.getIndex()));
   }
@@ -262,7 +272,7 @@ private static String displayName(
       RelNode node, List<RelDataTypeField> inputFields, int index, Set<String> fields) {
     if (!node.getInputs().isEmpty() && node.getInput(0) instanceof Project project) {
       List<RelDataTypeField> below = inputFields(project);
-      RexNode expression = project.getProjects().get(index);
+      RexNode expression = stripCasts(project.getProjects().get(index));
       if (isLeafRef(expression, below, fields)) {
         return leafName((RexCall) expression, below);
       }

core/src/test/java/org/opensearch/sql/calcite/FlatObjectScopeValidatorTest.java

@@ -163,6 +163,20 @@ void sortingByALeafIsRejected() {
         "Cannot sort by attributes");
   }
 
+  // Reading a leaf as a number is allowed, so a cast is stripped when deciding that -- and it has to
+  // be stripped just the same when noting which column the read went into. Otherwise the column
+  // sorts by a leaf unchecked and the failure surfaces from pushdown, as a 500.
+  @Test
+  void sortingByACastLeafIsRejected() {
+    RexNode asDouble =
+        rexBuilder.makeCast(
+            TYPE_FACTORY.createSqlType(SqlTypeName.DOUBLE), leaf("duration_ms"), true, true);
+    rejected(
+        LogicalSort.create(
+            project(field(0), asDouble), RelCollations.of(new RelFieldCollation(1)), null, null),
+        "Cannot sort by attributes.duration_ms");
+  }
+
   @Test
   void groupingByALeafIsRejected() {
     RelNode projected = project(field(0), leaf("status_code"));