Restrict ROW_NUMBER tie-break to comparable columns + sync no-pushdown fixtures

Three follow-ups to the rare/top tie-break commit, surfaced by CI:

1. **Skip non-comparable tie-break columns.** `top address` on a nested-array
   field reached the in-process `EnumerableWindow` with `ORDER BY count DESC,
   address` and crashed at runtime — `class java.util.ArrayList cannot be cast
   to class java.lang.Comparable`. Filter array / multiset / map / struct (row)
   columns out of the tie-break list before appending. If all field columns are
   non-comparable, no tie-break is added (falling back to the original
   count-only ORDER BY).

2. **Sync the parallel `calcite_no_pushdown/` explain fixtures.**
   `PPLIntegTestCase.loadExpectedPlan` switches to
   `expectedOutput/calcite_no_pushdown/` when pushdown is disabled, which
   `CalciteNoPushdownIT` exercises. The previous commit only updated the
   `expectedOutput/calcite/` copies, so `CalciteNoPushdownIT > CalciteExplainIT`
   read stale (no-tie-break) YAML. Update the four `explain_{rare,top}_usenull_*`
   YAMLs under `calcite_no_pushdown/` to include the `, $N` tie-break key in
   both the logical plan and the `EnumerableWindow` collation.

3. **Collapse the pushdown-disabled branch in `RareCommandIT.testRareWithGroup`.**
   With the stable tie-break, both pushdown-enabled and pushdown-disabled paths
   now resolve the count=8 tie deterministically to `AR` (alphabetical). Drop
   the `isPushdownDisabled() ? rows("F", "NV", 8) : rows("F", "AR", 8)`
   conditional in favor of the single `AR` expectation.

Signed-off-by: Kai Huang <huangkaics@gmail.com>

Author: ahkcs

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

@@ -3044,7 +3044,13 @@ public RelNode visitRareTopN(RareTopN node, CalcitePlanContext context) {
     // Append the rare/top field columns as secondary order keys so ties in the count column
     // resolve deterministically. Without this, ROW_NUMBER's tie-break is insertion-order
     // dependent and varies between backends (e.g. analytics-engine vs in-process Calcite).
-    List<RexNode> tieBreakKeys = rexVisitor.analyze(fieldList, context);
+    // Skip array / struct / map columns — Calcite's executor can't compare those at runtime
+    // (`ArrayList cannot be cast to Comparable`), and pushdown to the OpenSearch terms
+    // aggregation only supports scalar keys.
+    List<RexNode> tieBreakKeys =
+        rexVisitor.analyze(fieldList, context).stream()
+            .filter(CalciteRelNodeVisitor::isComparableOrderKey)
+            .toList();
     List<RexNode> orderKeys = new ArrayList<>(tieBreakKeys.size() + 1);
     orderKeys.add(countField);
     orderKeys.addAll(tieBreakKeys);
@@ -3080,6 +3086,15 @@ public RelNode visitRareTopN(RareTopN node, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
+  private static boolean isComparableOrderKey(RexNode key) {
+    SqlTypeName typeName = key.getType().getSqlTypeName();
+    return typeName != SqlTypeName.ARRAY
+        && typeName != SqlTypeName.MULTISET
+        && typeName != SqlTypeName.MAP
+        && typeName != SqlTypeName.ROW
+        && typeName != SqlTypeName.STRUCTURED;
+  }
+
   @Override
   public RelNode visitTableFunction(TableFunction node, CalcitePlanContext context) {
     throw new CalciteUnsupportedException("Table function is unsupported in Calcite");

integ-test/src/test/java/org/opensearch/sql/ppl/RareCommandIT.java

@@ -59,7 +59,7 @@ public void testRareWithGroup() throws IOException {
           rows("F", "OK", 7),
           rows("F", "KS", 7),
           rows("F", "CO", 7),
-          isPushdownDisabled() ? rows("F", "NV", 8) : rows("F", "AR", 8),
+          rows("F", "AR", 8),
           rows("M", "NE", 5),
           rows("M", "RI", 5),
           rows("M", "NV", 5),

integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_rare_usenull_false.yaml

@@ -3,15 +3,15 @@ calcite:
     LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
       LogicalProject(gender=[$0], state=[$1], count=[$2])
         LogicalFilter(condition=[<=($3, 2)])
-          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2)])
+          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2, $1)])
             LogicalAggregate(group=[{0, 1}], count=[COUNT()])
               LogicalProject(gender=[$4], state=[$7])
                 LogicalFilter(condition=[AND(IS NOT NULL($4), IS NOT NULL($7))])
                   CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
   physical: |
     EnumerableLimit(fetch=[10000])
       EnumerableCalc(expr#0..3=[{inputs}], expr#4=[2], expr#5=[<=($t3, $t4)], proj#0..2=[{exprs}], $condition=[$t5])
-        EnumerableWindow(window#0=[window(partition {0} order by [2] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
+        EnumerableWindow(window#0=[window(partition {0} order by [2, 1] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
           EnumerableAggregate(group=[{4, 7}], count=[COUNT()])
             EnumerableCalc(expr#0..16=[{inputs}], expr#17=[IS NOT NULL($t4)], expr#18=[IS NOT NULL($t7)], expr#19=[AND($t17, $t18)], proj#0..16=[{exprs}], $condition=[$t19])
               CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])

integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_rare_usenull_true.yaml

@@ -3,13 +3,13 @@ calcite:
     LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
       LogicalProject(gender=[$0], state=[$1], count=[$2])
         LogicalFilter(condition=[<=($3, 2)])
-          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2)])
+          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2, $1)])
             LogicalAggregate(group=[{0, 1}], count=[COUNT()])
               LogicalProject(gender=[$4], state=[$7])
                 CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
   physical: |
     EnumerableLimit(fetch=[10000])
       EnumerableCalc(expr#0..3=[{inputs}], expr#4=[2], expr#5=[<=($t3, $t4)], proj#0..2=[{exprs}], $condition=[$t5])
-        EnumerableWindow(window#0=[window(partition {0} order by [2] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
+        EnumerableWindow(window#0=[window(partition {0} order by [2, 1] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
           EnumerableAggregate(group=[{4, 7}], count=[COUNT()])
             CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])

integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_top_usenull_false.yaml

@@ -3,15 +3,15 @@ calcite:
     LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
       LogicalProject(gender=[$0], state=[$1], count=[$2])
         LogicalFilter(condition=[<=($3, 2)])
-          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2 DESC)])
+          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2 DESC, $1)])
             LogicalAggregate(group=[{0, 1}], count=[COUNT()])
               LogicalProject(gender=[$4], state=[$7])
                 LogicalFilter(condition=[AND(IS NOT NULL($4), IS NOT NULL($7))])
                   CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
   physical: |
     EnumerableLimit(fetch=[10000])
       EnumerableCalc(expr#0..3=[{inputs}], expr#4=[2], expr#5=[<=($t3, $t4)], proj#0..2=[{exprs}], $condition=[$t5])
-        EnumerableWindow(window#0=[window(partition {0} order by [2 DESC] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
+        EnumerableWindow(window#0=[window(partition {0} order by [2 DESC, 1] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
           EnumerableAggregate(group=[{4, 7}], count=[COUNT()])
             EnumerableCalc(expr#0..16=[{inputs}], expr#17=[IS NOT NULL($t4)], expr#18=[IS NOT NULL($t7)], expr#19=[AND($t17, $t18)], proj#0..16=[{exprs}], $condition=[$t19])
               CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])

integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_top_usenull_true.yaml

@@ -3,13 +3,13 @@ calcite:
     LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
       LogicalProject(gender=[$0], state=[$1], count=[$2])
         LogicalFilter(condition=[<=($3, 2)])
-          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2 DESC)])
+          LogicalProject(gender=[$0], state=[$1], count=[$2], _row_number_rare_top_=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $2 DESC, $1)])
             LogicalAggregate(group=[{0, 1}], count=[COUNT()])
               LogicalProject(gender=[$4], state=[$7])
                 CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])
   physical: |
     EnumerableLimit(fetch=[10000])
       EnumerableCalc(expr#0..3=[{inputs}], expr#4=[2], expr#5=[<=($t3, $t4)], proj#0..2=[{exprs}], $condition=[$t5])
-        EnumerableWindow(window#0=[window(partition {0} order by [2 DESC] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
+        EnumerableWindow(window#0=[window(partition {0} order by [2 DESC, 1] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])])
           EnumerableAggregate(group=[{4, 7}], count=[COUNT()])
             CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])