Force hash/merge join on no-BY rewrite to fix N-call OS round-trip blowup

A perf A/B on a local 20k-doc index uncovered a real problem in the
no-BY rewrite that was hidden by tests with bounded result sets.

Before this fix, the no-BY case emitted:
  EnumerableNestedLoopJoin(condition=[true], joinType=[inner])
    leftScan  (returns N rows)
    rightScan (returns 1 row — the COUNT() scalar)

Calcite's NestedLoopJoin contract calls Enumerable.enumerator() on the
right side once per left tuple. Each enumerator open on a
CalciteEnumerableIndexScan triggers a fresh OpenSearch _search request.
For a 10k-row left side that means 10k OpenSearch calls. On a remote
cluster (1-10ms RTT per call), the head-less query would take tens of
seconds.

Measured on the local node, 20k docs, single shard, no head:
  before: 10,004 OS calls per PPL query, ~1055ms wall
  after:    4 OS calls per PPL query,  ~174ms wall
That's ~6x faster wall and ~2500x fewer OS round-trips, with no
correctness change (results identical).

Fix: in the no-BY branch of rewriteWindowAsAggregateJoin, project a
literal-0 key column onto both sides (left: append after orig cols;
right: append after agg outputs) and join on equality. The equi-join
condition makes the planner pick EnumerableHashJoin, which drains the
single-row right side once into a hash table and probes per left row
in O(1).

Pushdown still fires on the right side — verified via EXPLAIN that the
right scan still carries `AGGREGATION->...COUNT()` and `size:0` in
PushDownContext; the literal-0 projection is a top-level wrapper that
doesn't disrupt the Aggregate→Scan operand chain
AggregateIndexScanRule.AGGREGATE_SCAN matches.

The BY case is unchanged — it already has an equi-join condition (or
IS NOT DISTINCT FROM for bucketNullable=true) which Calcite handles
correctly via EnumerableMergeJoin.

Test expectation updates:
  - CalcitePPLEventstatsTest.testEventstatsCount / testEventstatsAvg
  - CalcitePPLEventstatsEarliestLatestTest no-BY variants (4 tests)
  - explain_eventstats_dc.json (no-BY, pushdown)
  - explain_eventstats_earliest_latest_no_group.json (no-BY, both modes)
Outer LogicalProject now appears in the no-BY case because we must
strip the literal-0 key columns from the join output — it's no longer
a no-op passthrough that Calcite folds.

Signed-off-by: Jialiang Liang <jiallian@amazon.com>

Author: RyanL1997

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

@@ -2165,17 +2165,45 @@ private RelNode rewriteWindowAsAggregateJoin(Window node, CalcitePlanContext con
     aggregateWithTrimming(groupList, aggExprList, context, !bucketNullable);
     RelNode rightAggregate = context.relBuilder.build();
 
-    // Join left and right. Cross-join for no-BY (right is a single scalar row); equi-join on each
-    // partition key for BY. The condition for bucketNullable=true is IS NOT DISTINCT FROM so the
-    // NULL bucket on each side matches; LEFT for bucketNullable=false so NULL-keyed left rows
-    // survive with NULL aggregate values (right has no NULL bucket to match).
-    context.relBuilder.push(leftInput);
-    context.relBuilder.push(rightAggregate);
+    // Join left and right. For BY: equi-join on each partition key (IS NOT DISTINCT FROM for
+    // bucketNullable=true so NULL buckets match; plain equality + LEFT join for false so NULL-keyed
+    // left rows survive with NULL agg values). For no-BY: a true cross-join with `condition=true`
+    // would force EnumerableNestedLoopJoin to re-open the right-side enumerator per left row —
+    // which is catastrophic when the right side is a CalciteEnumerableIndexScan (one OpenSearch
+    // request per left row, e.g. 10k OS calls for 10k matching docs). Instead, project a literal-0
+    // key on both sides and join on equality so Calcite picks EnumerableMergeJoin /
+    // EnumerableHashJoin, which drains the (single-row) right side once and probes per left row.
     int leftFieldCount = leftInput.getRowType().getFieldCount();
+    int rightGroupKeyCount = hasGroup ? groupList.size() : 0;
+    int aggCount = node.getWindowFunctionList().size();
+
+    // For no-BY, append the literal-0 key column to both sides AFTER the existing columns.
+    // Left side becomes [orig cols..., key]; right side becomes [agg outputs..., key].
+    RelNode leftForJoin = leftInput;
+    RelNode rightForJoin = rightAggregate;
+    int leftJoinKeyOffset = 0;
+    if (!hasGroup) {
+      context.relBuilder.push(leftInput);
+      context.relBuilder.projectPlus(
+          context.relBuilder.alias(context.relBuilder.literal(0), EVENTSTATS_NOGROUP_JOIN_KEY));
+      leftForJoin = context.relBuilder.build();
+      context.relBuilder.push(rightAggregate);
+      context.relBuilder.projectPlus(
+          context.relBuilder.alias(context.relBuilder.literal(0), EVENTSTATS_NOGROUP_JOIN_KEY));
+      rightForJoin = context.relBuilder.build();
+      leftJoinKeyOffset = 1;
+    }
+
+    context.relBuilder.push(leftForJoin);
+    context.relBuilder.push(rightForJoin);
 
     RexNode joinCondition;
     if (!hasGroup) {
-      joinCondition = context.relBuilder.literal(true);
+      // Equi-join on the literal-0 keys we just projected (positions: leftFieldCount on the left
+      // input post-projection, aggCount on the right after the agg outputs).
+      RexNode leftKey = context.relBuilder.field(2, 0, leftFieldCount);
+      RexNode rightKey = context.relBuilder.field(2, 1, aggCount);
+      joinCondition = context.relBuilder.equals(leftKey, rightKey);
     } else {
       List<RexNode> perKeyConditions = new ArrayList<>();
       for (UnresolvedExpression groupExpr : groupList) {
@@ -2199,18 +2227,20 @@ private RelNode rewriteWindowAsAggregateJoin(Window node, CalcitePlanContext con
     context.relBuilder.join(joinType, joinCondition);
 
     // Final projection: keep all original left columns, then append the aggregate output columns
-    // (skipping the right-side group key columns). The output row type matches what the legacy
-    // RexOver lowering produced: [left cols ..., agg outputs ...] with the user-supplied aliases.
-    int rightGroupKeyCount = hasGroup ? groupList.size() : 0;
-    int aggCount = node.getWindowFunctionList().size();
+    // (skipping the right-side group key columns AND the literal-0 key columns when present).
+    // Output row type matches what the legacy RexOver lowering produced: [left cols ..., agg
+    // outputs ...] with the user-supplied aliases.
     List<RexNode> finalProjects = new ArrayList<>();
     List<String> finalNames = new ArrayList<>();
     List<String> leftNames = leftInput.getRowType().getFieldNames();
     for (int i = 0; i < leftFieldCount; i++) {
       finalProjects.add(context.relBuilder.field(i));
       finalNames.add(leftNames.get(i));
     }
-    int rightAggStart = leftFieldCount + rightGroupKeyCount;
+    // Right-side agg outputs come BEFORE the literal-0 key (key was added to the end via
+    // projectPlus). Absolute position in the join row = leftWidth + (right-side group-key prefix).
+    int leftWidth = leftFieldCount + leftJoinKeyOffset;
+    int rightAggStart = leftWidth + rightGroupKeyCount;
     for (int i = 0; i < aggCount; i++) {
       finalProjects.add(context.relBuilder.field(rightAggStart + i));
       finalNames.add(extractAliasName(node.getWindowFunctionList().get(i)));
@@ -2219,6 +2249,8 @@ private RelNode rewriteWindowAsAggregateJoin(Window node, CalcitePlanContext con
     return context.relBuilder.peek();
   }
 
+  private static final String EVENTSTATS_NOGROUP_JOIN_KEY = "__eventstats_join_key__";
+
   /**
    * Returns true if {@code node} matches the shape PPL {@code eventstats} actually emits — all
    * window functions resolve to a windowable aggregation (present in both {@link

integ-test/src/test/resources/expectedOutput/calcite/explain_eventstats_dc.json

@@ -1,6 +1,6 @@
 {
   "calcite": {
-    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], distinct_states=[$17])\n    LogicalJoin(condition=[true], joinType=[inner])\n      CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n      LogicalAggregate(group=[{}], distinct_states=[DISTINCT_COUNT_APPROX($0)])\n        LogicalProject(state=[$7])\n          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
-    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableNestedLoopJoin(condition=[true], joinType=[inner])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"account_number\",\"firstname\",\"address\",\"balance\",\"gender\",\"city\",\"employer\",\"state\",\"age\",\"email\",\"lastname\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#48:LogicalAggregate.NONE.[](input=RelSubset#47,group={},distinct_states=DISTINCT_COUNT_APPROX($0))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"aggregations\":{\"distinct_states\":{\"cardinality\":{\"field\":\"state.keyword\"}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], distinct_states=[$18])\n    LogicalJoin(condition=[=($17, $19)], joinType=[inner])\n      LogicalProject(account_number=[$0], firstname=[$1], address=[$2], balance=[$3], gender=[$4], city=[$5], employer=[$6], state=[$7], age=[$8], email=[$9], lastname=[$10], _id=[$11], _index=[$12], _score=[$13], _maxscore=[$14], _sort=[$15], _routing=[$16], __eventstats_join_key__=[0])\n        CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n      LogicalProject(distinct_states=[$0], __eventstats_join_key__=[0])\n        LogicalAggregate(group=[{}], distinct_states=[DISTINCT_COUNT_APPROX($0)])\n          LogicalProject(state=[$7])\n            CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
+    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableCalc(expr#0..13=[{inputs}], account_number=[$t2], firstname=[$t3], address=[$t4], balance=[$t5], gender=[$t6], city=[$t7], employer=[$t8], state=[$t9], age=[$t10], email=[$t11], lastname=[$t12], distinct_states=[$t0])\n    EnumerableHashJoin(condition=[=($1, $13)], joinType=[inner])\n      EnumerableCalc(expr#0=[{inputs}], expr#1=[0], proj#0..1=[{exprs}])\n        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[AGGREGATION->rel#2342:LogicalAggregate.NONE.[](input=RelSubset#2341,group={},distinct_states=DISTINCT_COUNT_APPROX($0))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"aggregations\":{\"distinct_states\":{\"cardinality\":{\"field\":\"state.keyword\"}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n      EnumerableCalc(expr#0..10=[{inputs}], expr#11=[0], proj#0..11=[{exprs}])\n        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[account_number, firstname, address, balance, gender, city, employer, state, age, email, lastname]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"account_number\",\"firstname\",\"address\",\"balance\",\"gender\",\"city\",\"employer\",\"state\",\"age\",\"email\",\"lastname\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
   }
 }

integ-test/src/test/resources/expectedOutput/calcite/explain_eventstats_earliest_latest_no_group.json

@@ -1,6 +1,6 @@
 {
   "calcite": {
-    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(created_at=[$0], server=[$1], @timestamp=[$2], message=[$3], level=[$4], earliest_message=[$11], latest_message=[$12])\n    LogicalJoin(condition=[true], joinType=[inner])\n      CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n      LogicalAggregate(group=[{}], earliest_message=[ARG_MIN($0, $1)], latest_message=[ARG_MAX($0, $1)])\n        LogicalProject(message=[$3], @timestamp=[$2])\n          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n",
-    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableNestedLoopJoin(condition=[true], joinType=[inner])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[PROJECT->[created_at, server, @timestamp, message, level]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"created_at\",\"server\",\"@timestamp\",\"message\",\"level\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n    CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[AGGREGATION->rel#333:LogicalAggregate.NONE.[](input=RelSubset#332,group={},earliest_message=ARG_MIN($0, $1),latest_message=ARG_MAX($0, $1))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"aggregations\":{\"earliest_message\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false,\"_source\":false,\"fields\":[{\"field\":\"message\"}],\"sort\":[{\"@timestamp\":{\"order\":\"asc\"}}]}},\"latest_message\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false,\"_source\":false,\"fields\":[{\"field\":\"message\"}],\"sort\":[{\"@timestamp\":{\"order\":\"desc\"}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(created_at=[$0], server=[$1], @timestamp=[$2], message=[$3], level=[$4], earliest_message=[$12], latest_message=[$13])\n    LogicalJoin(condition=[=($11, $14)], joinType=[inner])\n      LogicalProject(created_at=[$0], server=[$1], @timestamp=[$2], message=[$3], level=[$4], _id=[$5], _index=[$6], _score=[$7], _maxscore=[$8], _sort=[$9], _routing=[$10], __eventstats_join_key__=[0])\n        CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n      LogicalProject(earliest_message=[$0], latest_message=[$1], __eventstats_join_key__=[0])\n        LogicalAggregate(group=[{}], earliest_message=[ARG_MIN($0, $1)], latest_message=[ARG_MAX($0, $1)])\n          LogicalProject(message=[$3], @timestamp=[$2])\n            CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n",
+    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableCalc(expr#0..8=[{inputs}], created_at=[$t3], server=[$t4], @timestamp=[$t5], message=[$t6], level=[$t7], earliest_message=[$t0], latest_message=[$t1])\n    EnumerableHashJoin(condition=[=($2, $8)], joinType=[inner])\n      EnumerableCalc(expr#0..1=[{inputs}], expr#2=[0], proj#0..2=[{exprs}])\n        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[AGGREGATION->rel#2527:LogicalAggregate.NONE.[](input=RelSubset#2526,group={},earliest_message=ARG_MIN($0, $1),latest_message=ARG_MAX($0, $1))], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":0,\"timeout\":\"1m\",\"aggregations\":{\"earliest_message\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false,\"_source\":false,\"fields\":[{\"field\":\"message\"}],\"sort\":[{\"@timestamp\":{\"order\":\"asc\"}}]}},\"latest_message\":{\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false,\"explain\":false,\"_source\":false,\"fields\":[{\"field\":\"message\"}],\"sort\":[{\"@timestamp\":{\"order\":\"desc\"}}]}}}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n      EnumerableCalc(expr#0..4=[{inputs}], expr#5=[0], proj#0..5=[{exprs}])\n        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]], PushDownContext=[[PROJECT->[created_at, server, @timestamp, message, level]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"created_at\",\"server\",\"@timestamp\",\"message\",\"level\"]}}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
   }
 }

integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/explain_eventstats_earliest_latest_no_group.json

@@ -1,6 +1,6 @@
 {
   "calcite": {
-    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(created_at=[$0], server=[$1], @timestamp=[$2], message=[$3], level=[$4], earliest_message=[$11], latest_message=[$12])\n    LogicalJoin(condition=[true], joinType=[inner])\n      CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n      LogicalAggregate(group=[{}], earliest_message=[ARG_MIN($0, $1)], latest_message=[ARG_MAX($0, $1)])\n        LogicalProject(message=[$3], @timestamp=[$2])\n          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n",
-    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableNestedLoopJoin(condition=[true], joinType=[inner])\n    EnumerableCalc(expr#0..10=[{inputs}], proj#0..4=[{exprs}])\n      CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n    EnumerableAggregate(group=[{}], earliest_message=[ARG_MIN($3, $2)], latest_message=[ARG_MAX($3, $2)])\n      CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n"
+    "logical": "LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])\n  LogicalProject(created_at=[$0], server=[$1], @timestamp=[$2], message=[$3], level=[$4], earliest_message=[$12], latest_message=[$13])\n    LogicalJoin(condition=[=($11, $14)], joinType=[inner])\n      LogicalProject(created_at=[$0], server=[$1], @timestamp=[$2], message=[$3], level=[$4], _id=[$5], _index=[$6], _score=[$7], _maxscore=[$8], _sort=[$9], _routing=[$10], __eventstats_join_key__=[0])\n        CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n      LogicalProject(earliest_message=[$0], latest_message=[$1], __eventstats_join_key__=[0])\n        LogicalAggregate(group=[{}], earliest_message=[ARG_MIN($0, $1)], latest_message=[ARG_MAX($0, $1)])\n          LogicalProject(message=[$3], @timestamp=[$2])\n            CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n",
+    "physical": "EnumerableLimit(fetch=[10000])\n  EnumerableCalc(expr#0..8=[{inputs}], created_at=[$t3], server=[$t4], @timestamp=[$t5], message=[$t6], level=[$t7], earliest_message=[$t0], latest_message=[$t1])\n    EnumerableHashJoin(condition=[=($2, $8)], joinType=[inner])\n      EnumerableCalc(expr#0..1=[{inputs}], expr#2=[0], proj#0..2=[{exprs}])\n        EnumerableAggregate(group=[{}], earliest_message=[ARG_MIN($3, $2)], latest_message=[ARG_MAX($3, $2)])\n          CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n      EnumerableCalc(expr#0..10=[{inputs}], expr#11=[0], proj#0..4=[{exprs}], __eventstats_join_key__=[$t11])\n        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_logs]])\n"
   }
 }