[Analytics Engine] Wire window-function support for PPL top / rare (#21593)

* [Analytics Engine] Wire window-function support for PPL top / rare

PPL `top` and `rare` lower (via `CalciteRelNodeVisitor#visitRareTopN`) to a
`LogicalProject` containing
`ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...)`. On the analytics-engine
route, `OpenSearchProjectRule.annotateExpr` treated the `RexOver` as an
ordinary `RexCall` and fell into the scalar-function viability path. That
path keys on `ScalarFunction.fromSqlOperatorWithFallback`, which returns
null for window-only aggregates like `ROW_NUMBER`, so every `top` / `rare`
query died with "No backend supports scalar function [ROW_NUMBER] among
[datafusion]" before substrait emission.

Add `EngineCapability.WINDOW` and detect `RexOver` ahead of the standard
RexCall branch in `OpenSearchProjectRule#annotateExpr`. When the child's
viable backends include a WINDOW-capable backend, wrap the call in an
`AnnotatedProjectExpression` exactly like other annotated calls — the
existing strip path's `OperatorAnnotation::unwrap` returns the original
`RexOver` unchanged, so isthmus's `RexExpressionConverter#visitOver` emits
an inline substrait `WindowFunctionInvocation` (the substrait standard
catalog already binds `ROW_NUMBER`, so DataFusion's substrait consumer
decodes it natively — no separate substrait Window rel needed).

WINDOW is intentionally coarse (one boolean per backend rather than a
per-window-function `WindowCapability`). The substrait standard catalog
already constrains which window aggregates the backend's substrait
consumer can decode; a runtime decode failure is a clearer error than a
duplicated registry split between SPI and backend. The existing
`AggregateFunction` / `AggregateCapability` model remains right for
ordinary aggregates; once a second backend with different window support
lands, or a window-only function carries a non-standard call shape,
WINDOW can be promoted to a per-function class without disturbing the
planner-side annotation flow.

Result on `CalciteTopCommandIT` / `CalciteRareCommandIT` against the
force-routed analytics-engine path:
- `CalciteTopCommandIT`:  0/6 → 5/6 (only legacy-preferred test out-of-scope)
- `CalciteRareCommandIT`: 0/5 → 3/5 (legacy-preferred + sort-tie out-of-scope)
- combined: 0/11 → 8/11

Tests added in `ProjectRuleTests`:
- `testRowNumberOverPartitionByOrderByMarksAsWindow` — happy path
- `testRowNumberWithoutWindowCapabilityErrors` — capability-gap message
  names the window function
- `testStripAnnotationsPreservesRexOver` — strip path returns plain
  `LogicalProject(RexOver)` so isthmus can decode it

Signed-off-by: Kai Huang <ahkcs@amazon.com>

* [QA] Add self-contained Top + Rare CommandITs

Self-contained integration tests under sandbox/qa/analytics-engine-rest
mirroring CalciteTopCommandIT / CalciteRareCommandIT (and their PPL
bases TopCommandIT / RareCommandIT) from the opensearch-project/sql
repository. Each test sends a PPL query through POST /_analytics/ppl
(exposed by test-ppl-frontend), exercising the same UnifiedQueryPlanner
→ CalciteRelNodeVisitor#visitRareTopN → ROW_NUMBER() OVER (...) →
Substrait WindowFunctionInvocation → DataFusion pipeline that this
PR enables via EngineCapability.WINDOW.

TopCommandIT (5 tests on the calcs dataset):
- testTopWithoutGroup: top str0 → 3 rows with distinct counts (deterministic)
- testTopNWithoutGroup: top 1 str0 → 1 row
- testTopNWithGroup: top 1 str3 by str0 → 3 rows (one per str0 group)
- testTopCommandUseNull: top int0 → 8 buckets (7 non-null + 1 null)
- testTopCommandUseNullFalse: top usenull=false int0 → 7 non-null buckets

RareCommandIT (4 tests):
- testRareWithoutGroup: rare str0 → 3 rows ASC by count (deterministic)
- testRareWithGroup: rare str3 by str0 → 5 distinct (str0, str3) pairs
- testRareCommandUseNull: rare int0 → 8 buckets
- testRareCommandUseNullFalse: rare usenull=false int0 → 7 non-null buckets

The legacy-preferred=false variants from CalciteTopCommandIT /
CalciteRareCommandIT are not mirrored — they require a cluster-level
PPL syntax setting toggle that the QA harness doesn't expose. The
default usenull behaviour (PPL_SYNTAX_LEGACY_PREFERRED=true) is the
production-default code path and is covered by the four usenull
variants above.

All 9 pass against the QA test cluster (`./gradlew
:sandbox:qa:analytics-engine-rest:integTest --tests "*TopCommandIT"
--tests "*RareCommandIT"`).

Signed-off-by: Kai Huang <ahkcs@amazon.com>

---------

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/MockDataFusionBackend.java

@@ -140,7 +140,16 @@ protected Set<JoinCapability> joinCapabilities() {
     protected Set<WindowCapability> windowCapabilities() {
         return Set.of(
             new WindowCapability(
-                Set.of(WindowFunction.SUM, WindowFunction.AVG, WindowFunction.COUNT, WindowFunction.MIN, WindowFunction.MAX),
+                Set.of(
+                    WindowFunction.SUM,
+                    WindowFunction.AVG,
+                    WindowFunction.COUNT,
+                    WindowFunction.MIN,
+                    WindowFunction.MAX,
+                    // ROW_NUMBER backs PPL `top` / `rare` / `dedup` lowering
+                    // (ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...)).
+                    WindowFunction.ROW_NUMBER
+                ),
                 Set.of(PARQUET_DATA_FORMAT)
             )
         );

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/ProjectRuleTests.java

@@ -8,12 +8,18 @@
 
 package org.opensearch.analytics.planner;
 
+import com.google.common.collect.ImmutableList;
 import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.plan.RelOptUtil;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexFieldCollation;
+import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexOver;
+import org.apache.calcite.rex.RexWindowBounds;
+import org.apache.calcite.rex.RexWindowExclusion;
 import org.apache.calcite.sql.SqlFunction;
 import org.apache.calcite.sql.SqlFunctionCategory;
 import org.apache.calcite.sql.SqlKind;
@@ -323,6 +329,135 @@ private static void assertNoAnnotationInTree(RexNode node) {
         }
     }
 
+    // ---- Window functions ----
+
+    /**
+     * PPL's {@code top}/{@code rare}/{@code streamstats} commands lower to
+     * {@code ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...)} inside a {@code LogicalProject}.
+     * The project rule narrows viable backends via {@link
+     * org.opensearch.analytics.spi.BackendCapabilityProvider#windowCapabilities()} on the rule's
+     * window-narrowing pass; the {@link RexOver} itself is left unannotated so that
+     * strip-annotations / isthmus's {@code RexExpressionConverter#visitOver} can decode it
+     * directly into a substrait {@code WindowFunctionInvocation}.
+     */
+    public void testRowNumberOverPartitionByOrderByMarksAsWindow() {
+        RexNode rowNumber = makeRowNumberOver(/*partitionField*/ 0, /*orderField*/ 1);
+        OpenSearchProject result = runProject(
+            rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 0),
+            rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.INTEGER), 1),
+            rowNumber
+        );
+        assertTrue(result.getViableBackends().contains(MockDataFusionBackend.NAME));
+        // RexOver and pass-through field refs must NOT be annotated — RexOver dispatches through
+        // RexExpressionConverter#visitOver downstream, which doesn't recognize the wrapper.
+        assertFalse("Field ref must not be annotated", result.getProjects().get(0) instanceof AnnotatedProjectExpression);
+        assertFalse("Field ref must not be annotated", result.getProjects().get(1) instanceof AnnotatedProjectExpression);
+        assertFalse("RexOver must not be annotated", result.getProjects().get(2) instanceof AnnotatedProjectExpression);
+        assertTrue(
+            "Third project expression must remain a RexOver, was " + result.getProjects().get(2).getClass().getSimpleName(),
+            result.getProjects().get(2) instanceof RexOver
+        );
+    }
+
+    /**
+     * When no viable backend declares a {@link org.opensearch.analytics.spi.WindowCapability}
+     * covering {@code ROW_NUMBER}, the planner must surface a capability-gap error at plan
+     * time rather than failing later in substrait emission.
+     */
+    public void testRowNumberWithoutWindowCapabilityErrors() {
+        MockDataFusionBackend dfNoWindow = new MockDataFusionBackend() {
+            @Override
+            protected Set<org.opensearch.analytics.spi.WindowCapability> windowCapabilities() {
+                return Set.of();
+            }
+        };
+        RelOptTable table = mockTable(
+            "test_index",
+            new String[] { "name", "value" },
+            new SqlTypeName[] { SqlTypeName.VARCHAR, SqlTypeName.INTEGER }
+        );
+        RexNode rowNumber = makeRowNumberOver(/*partitionField*/ 0, /*orderField*/ 1);
+        LogicalProject project = LogicalProject.create(
+            stubScan(table),
+            List.of(),
+            List.of(rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 0), rowNumber),
+            List.of("name", "rn")
+        );
+        PlannerContext context = buildContext("parquet", nameValueFields(), List.of(dfNoWindow, LUCENE));
+        IllegalStateException exception = expectThrows(IllegalStateException.class, () -> runPlanner(project, context));
+        assertTrue(
+            "Expected planner to surface window-function capability gap, got: " + exception.getMessage(),
+            exception.getMessage().contains("No backend supports window functions") && exception.getMessage().contains("ROW_NUMBER")
+        );
+    }
+
+    /**
+     * Strip-annotations on a project containing a {@link RexOver} hoists each unique window
+     * call into a child {@link LogicalProject} (see {@code OpenSearchProject#liftNestedRexOver})
+     * and rewrites the outer expression to a {@link RexInputRef} into the hoisted slot. The
+     * outer project must carry no annotation wrappers, and the inner project must preserve the
+     * {@link RexOver} verbatim so isthmus's {@code RexExpressionConverter#visitOver} can decode
+     * it into a substrait {@code WindowFunctionInvocation}.
+     */
+    public void testStripAnnotationsPreservesRexOver() {
+        RexNode rowNumber = makeRowNumberOver(/*partitionField*/ 0, /*orderField*/ 1);
+        OpenSearchProject annotated = runProject(rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.VARCHAR), 0), rowNumber);
+        RelNode stripped = annotated.stripAnnotations(annotated.getInputs());
+        assertTrue("Stripped plan must be a plain LogicalProject", stripped instanceof LogicalProject);
+        List<RexNode> outerExprs = ((LogicalProject) stripped).getProjects();
+        for (RexNode expr : outerExprs) {
+            assertNoAnnotationInTree(expr);
+        }
+        assertTrue(
+            "Hoisted outer expr must be a RexInputRef into the inner Project, was " + outerExprs.get(1).getClass().getSimpleName(),
+            outerExprs.get(1) instanceof RexInputRef
+        );
+
+        // The inner Project (hoisted from liftNestedRexOver) must hold the original RexOver
+        // at the appended slot so substrait emission sees the WindowFunction at top level.
+        RelNode innerInput = stripped.getInputs().get(0);
+        assertTrue("Inner plan must be a LogicalProject carrying the hoisted RexOver", innerInput instanceof LogicalProject);
+        List<RexNode> innerExprs = ((LogicalProject) innerInput).getProjects();
+        boolean foundRexOver = false;
+        for (RexNode innerExpr : innerExprs) {
+            if (innerExpr instanceof RexOver) {
+                foundRexOver = true;
+                break;
+            }
+        }
+        assertTrue("Inner Project must contain the hoisted RexOver, exprs=" + innerExprs, foundRexOver);
+    }
+
+    /**
+     * Builds a {@code ROW_NUMBER() OVER (PARTITION BY $partitionField ORDER BY $orderField)}
+     * RexOver against the (VARCHAR, INTEGER) stub-scan schema.
+     */
+    private RexNode makeRowNumberOver(int partitionField, int orderField) {
+        RexNode partition = rexBuilder.makeInputRef(
+            typeFactory.createSqlType(partitionField == 0 ? SqlTypeName.VARCHAR : SqlTypeName.INTEGER),
+            partitionField
+        );
+        RexFieldCollation order = new RexFieldCollation(
+            rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.INTEGER), orderField),
+            Set.of()
+        );
+        return rexBuilder.makeOver(
+            typeFactory.createSqlType(SqlTypeName.BIGINT),
+            SqlStdOperatorTable.ROW_NUMBER,
+            List.of(),
+            List.of(partition),
+            ImmutableList.of(order),
+            RexWindowBounds.UNBOUNDED_PRECEDING,
+            RexWindowBounds.CURRENT_ROW,
+            RexWindowExclusion.EXCLUDE_NO_OTHER,
+            true,
+            true,
+            false,
+            false,
+            false
+        );
+    }
+
     // ---- Mixed backends in one projection ----
 
     public void testMixedBackendsInProjection() {

sandbox/qa/analytics-engine-rest/src/test/java/org/opensearch/analytics/qa/RareCommandIT.java

@@ -0,0 +1,141 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.qa;
+
+import org.opensearch.client.Request;
+import org.opensearch.client.Response;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Self-contained integration test for PPL {@code rare} on the analytics-engine route.
+ *
+ * <p>Mirrors {@code CalciteRareCommandIT} / {@code RareCommandIT} from the
+ * {@code opensearch-project/sql} repository. {@code rare} lowers the same way as
+ * {@code top} via {@code CalciteRelNodeVisitor#visitRareTopN} — a
+ * {@code LogicalProject} containing {@code ROW_NUMBER() OVER (PARTITION BY ... ORDER BY count ASC)} —
+ * so it exercises the same {@code EngineCapability.WINDOW} flag in
+ * {@code OpenSearchProjectRule} that this PR introduces.
+ */
+public class RareCommandIT extends AnalyticsRestTestCase {
+
+    private static final Dataset DATASET = new Dataset("calcs", "calcs");
+
+    private static boolean dataProvisioned = false;
+
+    private void ensureDataProvisioned() throws IOException {
+        if (dataProvisioned == false) {
+            DatasetProvisioner.provision(client(), DATASET);
+            dataProvisioned = true;
+        }
+    }
+
+    // ── rare without group ─────────────────────────────────────────────────────
+
+    public void testRareWithoutGroup() throws IOException {
+        // calcs.str0 has 3 distinct values with distinct counts → rare order is
+        // deterministic ASC by count: {FURNITURE: 2, OFFICE SUPPLIES: 6, TECHNOLOGY: 9}.
+        assertRowsEqual(
+            "source=" + DATASET.indexName + " | rare str0",
+            row("FURNITURE", 2),
+            row("OFFICE SUPPLIES", 6),
+            row("TECHNOLOGY", 9)
+        );
+    }
+
+    public void testRareWithGroup() throws IOException {
+        // rare str3 per str0 group. calcs.str3 distribution by str0:
+        //   FURNITURE       → {e: 2}                  → 1 row
+        //   OFFICE SUPPLIES → {e: 4, null: 2}         → 2 rows
+        //   TECHNOLOGY      → {null: 5, e: 4}         → 2 rows
+        // Total: 5 distinct (str0, str3) pairs under the default rare limit.
+        assertNumOfRows("source=" + DATASET.indexName + " | rare str3 by str0", 5);
+    }
+
+    // ── usenull behaviour ──────────────────────────────────────────────────────
+
+    public void testRareCommandUseNull() throws IOException {
+        // calcs.int0 distribution: 6 nulls; 7 distinct non-null values {1, 3, 4 (×3),
+        // 7, 8 (×3), 10, 11}. Default usenull=true → 8 bucket categories (7 + null).
+        assertNumOfRows("source=" + DATASET.indexName + " | rare int0", 8);
+    }
+
+    public void testRareCommandUseNullFalse() throws IOException {
+        // usenull=false drops the null bucket → 7 result rows.
+        assertNumOfRows("source=" + DATASET.indexName + " | rare usenull=false int0", 7);
+    }
+
+    // ── helpers ─────────────────────────────────────────────────────────────────
+
+    private static List<Object> row(Object... values) {
+        return Arrays.asList(values);
+    }
+
+    @SafeVarargs
+    @SuppressWarnings("varargs")
+    private final void assertRowsEqual(String ppl, List<Object>... expected) throws IOException {
+        Map<String, Object> response = executePpl(ppl);
+        @SuppressWarnings("unchecked")
+        List<List<Object>> actualRows = (List<List<Object>>) response.get("rows");
+        assertNotNull("Response missing 'rows' for query: " + ppl, actualRows);
+        assertEquals("Row count mismatch for query: " + ppl, expected.length, actualRows.size());
+        for (int i = 0; i < expected.length; i++) {
+            List<Object> want = expected[i];
+            List<Object> got = actualRows.get(i);
+            assertEquals(
+                "Column count mismatch at row " + i + " for query: " + ppl,
+                want.size(),
+                got.size()
+            );
+            for (int j = 0; j < want.size(); j++) {
+                assertCellEquals(
+                    "Cell mismatch at row " + i + ", col " + j + " for query: " + ppl,
+                    want.get(j),
+                    got.get(j)
+                );
+            }
+        }
+    }
+
+    private void assertNumOfRows(String ppl, int expectedRows) throws IOException {
+        Map<String, Object> response = executePpl(ppl);
+        @SuppressWarnings("unchecked")
+        List<List<Object>> actualRows = (List<List<Object>>) response.get("rows");
+        assertNotNull("Response missing 'rows' for query: " + ppl, actualRows);
+        assertEquals("Row count mismatch for query: " + ppl, expectedRows, actualRows.size());
+    }
+
+    /** Numeric-tolerant cell comparator (Jackson returns Integer/Long/Double interchangeably). */
+    private static void assertCellEquals(String message, Object expected, Object actual) {
+        if (expected == null || actual == null) {
+            assertEquals(message, expected, actual);
+            return;
+        }
+        if (expected instanceof Number && actual instanceof Number) {
+            double e = ((Number) expected).doubleValue();
+            double a = ((Number) actual).doubleValue();
+            if (Double.compare(e, a) != 0) {
+                fail(message + ": expected <" + expected + "> but was <" + actual + ">");
+            }
+            return;
+        }
+        assertEquals(message, expected, actual);
+    }
+
+    private Map<String, Object> executePpl(String ppl) throws IOException {
+        ensureDataProvisioned();
+        Request request = new Request("POST", "/_analytics/ppl");
+        request.setJsonEntity("{\"query\": \"" + escapeJson(ppl) + "\"}");
+        Response response = client().performRequest(request);
+        return assertOkAndParse(response, "PPL: " + ppl);
+    }
+}