Stabilize CalciteStreamstatsCommandIT on the analytics-engine route (#5582)

Gate the streamstats ITs that diverge on the analytics-engine route
(parquet-backed composite store, DataFusion backend) so the route runs
green, while keeping every test active on the v2/Calcite path. Mechanism
follows the established capability-gating pattern (#5560): an in-test
@RequiresCapability(...) annotation plus a matching integTestRemote
excludeTestsMatching entry; both are no-ops on the v2 route.

Triaged single-shard and multi-shard (num_shards=3) analytics runs against
the v2 baseline. Three groups:

- DOC_MUTATION (4 tests): testStreamstatsGlobalWithNull,
  testStreamstatsGlobalWithNullBucket, testStreamstatsResetWithNull,
  testStreamstatsResetWithNullBucket seed state via PUT+DELETE. Doc-level
  DELETE is unsupported on the parquet store, and same-_id PUT is
  append-only, so the leaked doc inflated the row counts of every sibling
  test that reads the shared index. Gated with the same DOC_MUTATION
  capability the three existing mutation tests already carry.

- CHAINED_STREAMSTATS_BY (4 tests): chaining two streamstats where an
  upstream stage partitions `by` a group emits a ROW_NUMBER() sequence
  column from each stage; the Substrait converter names both physical
  columns identically, so the stacked schema has a duplicate/ambiguous
  field name (500) or, for chained window streamstats, non-deterministic
  values. The Calcite logical plan is correct; the alias is lost in
  Substrait conversion. Fails single- and multi-shard.

- STREAMSTATS_SORT_NOT_HONORED (1 test): testStreamstatsAndSort. The window
  is computed over the backend scan order, ignoring a preceding `| sort`
  (the OVER clause carries no explicit ORDER BY), so the per-row aggregates
  diverge from the v2/Calcite path.

Pass rate on the single-shard analytics route, CalciteStreamstatsCommandIT:

| metric    | before | after |
|-----------|--------|-------|
| tests run | 47     | 42    |
| failures  | 12     | 0     |
| skipped   | 3      | 7     |

(The before-failures count is inflated by the DOC_MUTATION leak described
above; the four leaking tests plus their downstream row-count victims all
clear once gated.)

v2 route (:integTest): 47 run, 0 failed, 0 skipped — gates are no-ops
off the analytics route.

Twelve further tests fail only on the multi-shard route (they pass
single-shard) due to cross-shard fragment-order non-determinism in the
streamstats window gather; that is an engine-side gap and is left
unchanged here rather than gated, to avoid skipping passing tests on the
single-shard route.

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

Author: ahkcs

integ-test/build.gradle

@@ -1328,6 +1328,20 @@ task integTestRemote(type: RestIntegTestTask) {
 
             // === Excludes: asserts a Lucene pushdown fragment absent on the AE route ===
             excludeTestsMatching '*CalciteSortCommandIT.testPushdownSortCastToDoubleExpression'
+
+            // === Excludes: CalciteStreamstatsCommandIT route divergences ===
+            // Each test also carries an in-test @RequiresCapability(...) recording the reason.
+            //  - CHAINED_STREAMSTATS_BY: chaining two streamstats where an upstream stage has `by`
+            //    emits two ROW_NUMBER() sequence columns the Substrait converter names identically,
+            //    so the stacked schema has a duplicate/ambiguous field name (500) or, for chained
+            //    window streamstats, non-deterministic values. Fails single- and multi-shard.
+            excludeTestsMatching '*CalciteStreamstatsCommandIT.testMultipleStreamstats'
+            excludeTestsMatching '*CalciteStreamstatsCommandIT.testMultipleStreamstatsWithWindow'
+            excludeTestsMatching '*CalciteStreamstatsCommandIT.testMultipleStreamstatsWithNull1'
+            excludeTestsMatching '*CalciteStreamstatsCommandIT.testMultipleStreamstatsWithEval'
+            //  - STREAMSTATS_SORT_NOT_HONORED: streamstats computes its window over the backend scan
+            //    order, ignoring a preceding `| sort` (the OVER clause has no explicit ORDER BY).
+            excludeTestsMatching '*CalciteStreamstatsCommandIT.testStreamstatsAndSort'
         }
     }
 

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteStreamstatsCommandIT.java

@@ -6,7 +6,9 @@
 package org.opensearch.sql.calcite.remote;
 
 import static org.opensearch.sql.legacy.TestsConstants.*;
+import static org.opensearch.sql.util.Capability.CHAINED_STREAMSTATS_BY;
 import static org.opensearch.sql.util.Capability.DOC_MUTATION;
+import static org.opensearch.sql.util.Capability.STREAMSTATS_SORT_NOT_HONORED;
 import static org.opensearch.sql.util.MatcherUtils.*;
 
 import java.io.IOException;
@@ -558,6 +560,7 @@ public void testStreamstatsGlobal() throws IOException {
   }
 
   @Test
+  @RequiresCapability(DOC_MUTATION)
   public void testStreamstatsGlobalWithNull() throws IOException {
     final int docId = 7;
     Request insertRequest =
@@ -613,6 +616,7 @@ public void testStreamstatsGlobalWithNull() throws IOException {
   }
 
   @Test
+  @RequiresCapability(DOC_MUTATION)
   public void testStreamstatsGlobalWithNullBucket() throws IOException {
     final int docId = 7;
     Request insertRequest =
@@ -718,6 +722,7 @@ public void testStreamstatsReset() throws IOException {
   }
 
   @Test
+  @RequiresCapability(DOC_MUTATION)
   public void testStreamstatsResetWithNull() throws IOException {
     final int docId = 7;
     Request insertRequest =
@@ -773,6 +778,7 @@ public void testStreamstatsResetWithNull() throws IOException {
   }
 
   @Test
+  @RequiresCapability(DOC_MUTATION)
   public void testStreamstatsResetWithNullBucket() throws IOException {
     final int docId = 7;
     Request insertRequest =
@@ -845,6 +851,7 @@ public void testUnsupportedWindowFunctions() {
   }
 
   @Test
+  @RequiresCapability(CHAINED_STREAMSTATS_BY)
   public void testMultipleStreamstats() throws IOException {
     JSONObject actual =
         executeQuery(
@@ -863,6 +870,7 @@ public void testMultipleStreamstats() throws IOException {
   }
 
   @Test
+  @RequiresCapability(CHAINED_STREAMSTATS_BY)
   public void testMultipleStreamstatsWithWindow() throws IOException {
     // Test case from GitHub issue #4800: chained streamstats with window=2
     JSONObject actual =
@@ -899,6 +907,7 @@ public void testMultipleStreamstatsWithWindow() throws IOException {
   // causing Calcite's RelDecorrelator to fail on duplicate correlate references.
 
   @Test
+  @RequiresCapability(CHAINED_STREAMSTATS_BY)
   public void testMultipleStreamstatsWithNull1() throws IOException {
     JSONObject actual =
         executeQuery(
@@ -1008,6 +1017,7 @@ public void testStreamstatsAndEventstats() throws IOException {
   }
 
   @Test
+  @RequiresCapability(STREAMSTATS_SORT_NOT_HONORED)
   public void testStreamstatsAndSort() throws IOException {
     JSONObject actual =
         executeQuery(
@@ -1074,6 +1084,7 @@ public void testWhereInWithStreamstatsSubquery() throws IOException {
   }
 
   @Test
+  @RequiresCapability(CHAINED_STREAMSTATS_BY)
   public void testMultipleStreamstatsWithEval() throws IOException {
     JSONObject actual =
         executeQuery(

integ-test/src/test/java/org/opensearch/sql/util/Capability.java

@@ -469,7 +469,37 @@ public enum Capability {
   LUCENE_PUSHDOWN_EXPLAIN(
       "A test asserting a Lucene-specific pushdown fragment in the explain plan (e.g. SORT->[...])"
           + " can't pass on the analytics-engine route: the DataFusion scan produces a different"
-          + " plan with no Lucene pushdown fragment.");
+          + " plan with no Lucene pushdown fragment."),
+
+  /**
+   * Chaining two {@code streamstats} commands where an upstream stage partitions {@code by} a group
+   * fails on the analytics-engine route. Each {@code streamstats ... by} stage projects a {@code
+   * ROW_NUMBER() OVER ()} sequence column to order its window; the Calcite plan aliases these
+   * distinctly ({@code __stream_seq__}), but the Substrait converter names both physical columns
+   * after the operator ({@code "row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"}),
+   * so the stacked schema has a duplicate/ambiguous field name. Verified: it surfaces as a 500
+   * ({@code Schema contains duplicate unqualified field name ...} / streaming-fragment failure) or,
+   * for chained {@code window} streamstats, non-deterministic window values. A single {@code
+   * streamstats by} (or a chain where only the final stage has {@code by}) works.
+   */
+  CHAINED_STREAMSTATS_BY(
+      "Chaining two streamstats where an upstream stage partitions by a group fails on the"
+          + " analytics-engine route: both stages emit a ROW_NUMBER() sequence column the Substrait"
+          + " converter names identically, producing a duplicate/ambiguous field name (500) or"
+          + " non-deterministic window values."),
+
+  /**
+   * {@code streamstats} computes its running/window aggregate over the backend scan order on the
+   * analytics-engine route, ignoring a preceding {@code | sort}. The {@code OVER} clause carries no
+   * explicit {@code ORDER BY} (streamstats orders by encounter order by design), so DataFusion
+   * evaluates the window in scan order rather than the sorted order the v2/Calcite path honors.
+   * Verified: {@code sort age | streamstats window=2 avg(age)} yields window values computed in
+   * insertion order, not age order, so the per-row aggregates diverge.
+   */
+  STREAMSTATS_SORT_NOT_HONORED(
+      "streamstats computes its window over the backend scan order on the analytics-engine route,"
+          + " ignoring a preceding | sort (the OVER clause has no explicit ORDER BY), so the window"
+          + " values diverge from the v2/Calcite path which honors the sort.");
 
   private final String reason;