Close gaps from top/rare analytics-engine wiring (opensearch-project#5433)

ahkcs · web-flow · commit 817f7f3eddaf · 2026-05-19T12:19:37.000-07:00
* Forward PPL_SYNTAX_LEGACY_PREFERRED through UnifiedQueryContext

`RestUnifiedQueryAction.applyClusterOverrides` previously only forwarded
`PPL_REX_MAX_MATCH_LIMIT` into the per-request `UnifiedQueryContext`. As a
result, cluster-side updates to `plugins.ppl.syntax.legacy.preferred` were
ignored on the analytics-engine route: `PPLQueryParser` -&gt; `AstBuilder` -&gt;
`ArgumentFactory` read the legacy-preferred flag from the unified context's
settings map, which never received the override. This caused queries like
`top age` / `rare state` with `usenull` defaulting off to behave as if
`usenull=true` on the analytics route.

Refactor the override builder into a small helper and forward both
`PPL_REX_MAX_MATCH_LIMIT` and `PPL_SYNTAX_LEGACY_PREFERRED`. Future keys can
be added with a one-liner.

Fixes `CalciteTopCommandIT.testTopCommandLegacyFalse` and
`CalciteRareCommandIT.testRareCommandLegacyFalse` against the analytics
route (`-Dtests.analytics.force_routing=true`).

Signed-off-by: Kai Huang &lt;huangkaics@gmail.com&gt;

* Stabilize rare/top tie-break by appending field columns to ROW_NUMBER order

`CalciteRelNodeVisitor.visitRareTopN` lowers `rare`/`top` to a
`ROW_NUMBER() OVER (PARTITION BY ... ORDER BY count [DESC])` window. With
only the count column in the ORDER BY clause, ties at the same count
resolved via the upstream operator's insertion order, which differed
between backends (in-process Calcite vs. analytics-engine vs. Lucene
pushdown). On the analytics-engine route, `testRareWithGroup` failed
because ROW_NUMBER picked NV at count=8 while the test expected AR.

Append the rare/top field columns as secondary ASC keys so ties resolve
alphabetically and deterministically across backends. This matches the
behavior of the existing OpenSearch terms-aggregation pushdown, which
tie-breaks on `_key:asc`.

Update `RareTopPushdownRule` to accept the new shape: 1 or 2 order keys,
where the (optional) second key must be the rare/top target field in ASC
direction. The pushdown's wire payload is unchanged.

Update the matching unit-test expectations in `CalcitePPLRareTopNTest`
(11 RelNode/result/SparkSQL strings) and 5 explain YAML fixtures.

Fixes `CalciteRareCommandIT.testRareWithGroup` against the analytics
route (and removes the same class of tie-break flakiness across other
rare/top tests).

Signed-off-by: Kai Huang &lt;huangkaics@gmail.com&gt;

* 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 &gt; 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 &lt;huangkaics@gmail.com&gt;

---------

Signed-off-by: Kai Huang &lt;huangkaics@gmail.com&gt;
diff --git a/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java b/core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java
@@ -3071,14 +3071,31 @@ public RelNode visitRareTopN(RareTopN node, CalcitePlanContext context) {
       countField = context.relBuilder.field(countFieldName);
     }
 
+    // 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).
+    // Skip collection / map / struct 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. RelDataType#getComparability would be
+    // the natural Calcite primitive but every concrete RelDataTypeImpl subclass (including
+    // ArraySqlType / MapSqlType / RelRecordType) returns ALL by default in calcite 1.41, so
+    // it doesn't actually discriminate the cases we need to reject here.
+    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);
+
     RexNode rowNumberWindowOver =
         PlanUtils.makeOver(
             context,
             BuiltinFunctionName.ROW_NUMBER,
             null,
             List.of(),
             partitionKeys,
-            List.of(countField),
+            orderKeys,
             WindowFrame.toCurrentRow());
     context.relBuilder.projectPlus(
         context.relBuilder.alias(rowNumberWindowOver, ROW_NUMBER_COLUMN_FOR_RARE_TOP));
@@ -3102,6 +3119,15 @@ public RelNode visitRareTopN(RareTopN node, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
+  private static boolean isComparableOrderKey(RexNode key) {
+    RelDataType type = key.getType();
+    // SqlTypeUtil#isCollection covers ARRAY + MULTISET; RelDataType#isStruct covers ROW +
+    // STRUCTURED. MAP has no dedicated predicate, so check it explicitly.
+    return !SqlTypeUtil.isCollection(type)
+        && !type.isStruct()
+        && type.getSqlTypeName() != SqlTypeName.MAP;
+  }
+
   @Override
   public RelNode visitTableFunction(TableFunction node, CalcitePlanContext context) {
     throw new CalciteUnsupportedException("Table function is unsupported in Calcite");
diff --git a/integ-test/src/test/java/org/opensearch/sql/ppl/RareCommandIT.java b/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),
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_nested_agg_top_push.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_nested_agg_top_push.yaml
@@ -3,7 +3,7 @@ calcite:
     LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
       LogicalProject(address.city=[$0], count=[$1])
         LogicalFilter(condition=[<=($2, 10)])
-          LogicalProject(address.city=[$0], count=[$1], _row_number_rare_top_=[ROW_NUMBER() OVER (ORDER BY $1 DESC)])
+          LogicalProject(address.city=[$0], count=[$1], _row_number_rare_top_=[ROW_NUMBER() OVER (ORDER BY $1 DESC, $0)])
             LogicalAggregate(group=[{0}], count=[COUNT()])
               LogicalProject(address.city=[$3])
                 LogicalFilter(condition=[IS NOT NULL($3)])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_rare_usenull_false.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_rare_usenull_false.yaml
@@ -3,7 +3,7 @@ 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))])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_rare_usenull_true.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_rare_usenull_true.yaml
@@ -3,12 +3,12 @@ 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()])])
           CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 1},count=COUNT())], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"gender":{"terms":{"field":"gender.keyword","missing_bucket":true,"missing_order":"first","order":"asc"}}},{"state":{"terms":{"field":"state.keyword","missing_bucket":true,"missing_order":"first","order":"asc"}}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_top_usenull_false.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_top_usenull_false.yaml
@@ -3,7 +3,7 @@ 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))])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite/explain_top_usenull_true.yaml b/integ-test/src/test/resources/expectedOutput/calcite/explain_top_usenull_true.yaml
@@ -3,12 +3,12 @@ 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()])])
           CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#:LogicalAggregate.NONE.[](input=RelSubset#,group={0, 1},count=COUNT())], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":0,"timeout":"1m","aggregations":{"composite_buckets":{"composite":{"size":1000,"sources":[{"gender":{"terms":{"field":"gender.keyword","missing_bucket":true,"missing_order":"first","order":"asc"}}},{"state":{"terms":{"field":"state.keyword","missing_bucket":true,"missing_order":"first","order":"asc"}}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_rare_usenull_false.yaml b/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]])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_rare_usenull_true.yaml b/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]])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_top_usenull_false.yaml b/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]])
diff --git a/integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_top_usenull_true.yaml b/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]])
diff --git a/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/RareTopPushdownRule.java b/opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/RareTopPushdownRule.java
@@ -8,6 +8,7 @@
 import java.util.List;
 import java.util.function.Predicate;
 import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.logical.LogicalFilter;
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.calcite.rel.rules.SubstitutionRule;
@@ -49,22 +50,31 @@ protected void onMatchImpl(RelOptRuleCall call) {
       List<Integer> groupIndices = PlanUtils.getSelectColumns(windows.getFirst().partitionKeys);
       List<String> byList = groupIndices.stream().map(fieldNameList::get).toList();
 
-      if (windows.getFirst().orderKeys.size() != 1) {
+      // ORDER BY must be either the count column alone, or the count column followed by the
+      // rare/top target field ASC (the stable tie-break inserted by visitRareTopN). The pushdown
+      // to the OpenSearch terms aggregation naturally tie-breaks on `_key:asc`, so the latter
+      // shape lowers to the same wire request as the single-key shape.
+      List<RexFieldCollation> orderKeys = windows.getFirst().orderKeys;
+      if (orderKeys.isEmpty() || orderKeys.size() > 2) {
         return;
       }
-      RexFieldCollation orderKey = windows.getFirst().orderKeys.getFirst();
-      List<Integer> orderIndices = PlanUtils.getSelectColumns(List.of(orderKey.getKey()));
-      List<String> orderList = orderIndices.stream().map(fieldNameList::get).toList();
+      RexFieldCollation primaryOrderKey = orderKeys.getFirst();
+      List<Integer> primaryOrderIndices =
+          PlanUtils.getSelectColumns(List.of(primaryOrderKey.getKey()));
+      List<String> primaryOrderList = primaryOrderIndices.stream().map(fieldNameList::get).toList();
       List<String> targetList =
           fieldNameList.stream()
               .filter(Predicate.not(byList::contains))
-              .filter(Predicate.not(orderList::contains))
+              .filter(Predicate.not(primaryOrderList::contains))
               .toList();
       if (targetList.size() != 1) {
         return;
       }
       String targetName = targetList.getFirst();
-      digest = new RareTopDigest(targetName, byList, number, orderKey.getDirection());
+      if (hasIncompatibleTieBreak(orderKeys, fieldNameList, targetName)) {
+        return;
+      }
+      digest = new RareTopDigest(targetName, byList, number, primaryOrderKey.getDirection());
     } catch (Exception e) {
       return;
     }
@@ -75,6 +85,26 @@ protected void onMatchImpl(RelOptRuleCall call) {
     }
   }
 
+  /**
+   * Returns {@code true} when a second {@link RexFieldCollation} is present but is not the stable
+   * tie-break shape this rule recognises (rare/top target field, ASC). A single-key shape (no
+   * tie-break) is always accepted because the OpenSearch terms aggregation tie-breaks on {@code
+   * _key:asc} natively. {@code orderKeys} arity is pre-checked by the caller.
+   */
+  private static boolean hasIncompatibleTieBreak(
+      List<RexFieldCollation> orderKeys, List<String> fieldNameList, String targetName) {
+    if (orderKeys.size() != 2) {
+      return false;
+    }
+    RexFieldCollation tieBreakKey = orderKeys.get(1);
+    if (tieBreakKey.getDirection() != RelFieldCollation.Direction.ASCENDING) {
+      return true;
+    }
+    List<Integer> tieBreakIndices = PlanUtils.getSelectColumns(List.of(tieBreakKey.getKey()));
+    List<String> tieBreakList = tieBreakIndices.stream().map(fieldNameList::get).toList();
+    return tieBreakList.size() != 1 || !tieBreakList.getFirst().equals(targetName);
+  }
+
   @Value.Immutable
   public interface Config extends OpenSearchRuleConfig {
     RareTopPushdownRule.Config DEFAULT =
diff --git a/plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java b/plugin/src/main/java/org/opensearch/sql/plugin/rest/RestUnifiedQueryAction.java
@@ -190,23 +190,25 @@ private UnifiedQueryContext buildContext(QueryType queryType, boolean profiling)
    * setting(String, Object)} API, keeping {@link UnifiedQueryContext} decoupled from any specific
    * {@link org.opensearch.sql.common.setting.Settings} implementation.
    *
-   * <p>Currently scoped to {@code plugins.ppl.rex.max_match.limit} — required so the unified path
-   * honors {@code _cluster/settings} updates for {@code rex max_match} (CalciteRexCommandIT's
-   * testRexMaxMatchConfigurableLimit). Add keys here if a future PR / IT depends on cluster-side
-   * fidelity for one of the other planning settings.
+   * <p>Add keys here if a future PR / IT depends on cluster-side fidelity for one of the other
+   * planning settings.
    */
   private UnifiedQueryContext.Builder applyClusterOverrides(UnifiedQueryContext.Builder builder) {
-    Object rexLimit =
-        pluginSettings.getSettingValue(
-            org.opensearch.sql.common.setting.Settings.Key.PPL_REX_MAX_MATCH_LIMIT);
-    if (rexLimit != null) {
-      builder.setting(
-          org.opensearch.sql.common.setting.Settings.Key.PPL_REX_MAX_MATCH_LIMIT.getKeyValue(),
-          rexLimit);
-    }
+    forwardClusterSetting(
+        builder, org.opensearch.sql.common.setting.Settings.Key.PPL_REX_MAX_MATCH_LIMIT);
+    forwardClusterSetting(
+        builder, org.opensearch.sql.common.setting.Settings.Key.PPL_SYNTAX_LEGACY_PREFERRED);
     return builder;
   }
 
+  private void forwardClusterSetting(
+      UnifiedQueryContext.Builder builder, org.opensearch.sql.common.setting.Settings.Key key) {
+    Object value = pluginSettings.getSettingValue(key);
+    if (value != null) {
+      builder.setting(key.getKeyValue(), value);
+    }
+  }
+
   /**
    * Extract the source index name by parsing the query and visiting the AST to find the Relation
    * node. Uses the context's parser which supports both PPL and SQL.
diff --git a/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLRareTopNTest.java b/ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLRareTopNTest.java
