[BUG] eventstats does not push down to OpenSearch (RexOver excluded from aggregation pushdown)

[RyanL1997]

Query Information

PPL Command/Query:

source=logs-* 
| where `@timestamp` > DATE_SUB(NOW(), INTERVAL 5 MINUTE)
| where severityText = 'ERROR'
| eventstats count() as total
| head 10
| fields severityText, total

Expected Result:
The global eventstats count() should push down to OpenSearch as track_total_hits (the same way stats count() does), so the shard returns one number and the coordinator scales it across documents — instead of streaming every matching document to the coordinator just to count them.

Actual Result:
Pushdown does not fire. The scan runs with requestedTotalSize=2147483647 and _source includes — i.e., every matching document streams to the coordinator, where a COUNT() OVER () window function counts them in memory. On a 47B-document index this exceeds query timeouts (the reported "FAIL on PI") and on smaller indices it returns 0 rows when the streaming exceeds limits ("0 rows on CAPE/WI"). The query returns correct results on tiny test data but the plan is unviable at production scale.

Dataset Information

Dataset/Schema Type

• OpenTelemetry (OTEL)

Index Mapping

{
 "index_patterns": ["logs-*"],
 "template": {
 "mappings": {
 "properties": {
 "@timestamp": { "type": "date" },
 "body": { "type": "text", "norms": false },
 "severityText": { "type": "keyword" }
 }
 }
 }
}

Sample Data

{ "@timestamp": "2026-05-28T17:44:20.000Z", "severityText": "ERROR", "body": "..." }
{ "@timestamp": "2026-05-28T17:44:21.000Z", "severityText": "INFO", "body": "..." }
{ "@timestamp": "2026-05-28T17:44:22.000Z", "severityText": "ERROR", "body": "..." }
{ "@timestamp": "2026-05-28T17:44:23.000Z", "severityText": "ERROR", "body": "..." }

Bug Description

Issue Summary:

eventstats (no BY, no partition) — equivalent to a global stats count() for purposes of the count value — produces a COUNT() OVER () window function instead of a flat aggregation, and the existing pushdown rules deliberately skip it.

Explain output for both shapes (Calcite engine, pushdown enabled):

stats count() as total (works, pushed down):

CalciteEnumerableIndexScan(
 PushDownContext=[[FILTER->..., AGGREGATION->...COUNT(), LIMIT->10000]],
 sourceBuilder={"size":0, ..., "track_total_hits":2147483647})

DSL has "size":0 and track_total_hits — the shard returns one number.

eventstats count() as total (fails to push down):

EnumerableLimit(fetch=[10000])
 EnumerableLimit(fetch=[10])
 EnumerableWindow(window#0=[window(aggs [COUNT()])])
 CalciteEnumerableIndexScan(
 PushDownContext=[[PROJECT->[@timestamp, severityText],
 FILTER->AND(>(@ts,...), =(severityText,'ERROR')),
 PROJECT->[severityText]]],
 sourceBuilder={"_source":{"includes":["severityText"]},
 "query":{"bool":{"must":[{"range":...},{"term":{"severityText":"ERROR"}}]}}},
 requestedTotalSize=2147483647)

DSL has _source includes and requestedTotalSize=MAX_INT — every matching document streams up to the coordinator. EnumerableWindow then counts them.

Root cause:
PPL eventstats is parsed in AstBuilder.visitEventstatsCommand (ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java:498) as a Window node, which Calcite lowers to a RexOver. The aggregation pushdown path in AggregateIndexScanRule (opensearch/.../planner/rules/AggregateIndexScanRule.java:122,171) explicitly excludes RexOver from pushdown — windows are treated as opaque coordinator-side computations.

This treatment is correct for windows that need partitioning or ordering. But eventstats <agg> without a BY clause produces an unpartitioned, unordered window — OVER () — which is semantically equivalent to: "compute a single global aggregate, broadcast it to every input row." The first half of that (the global aggregate) is exactly what OpenSearch can do natively in the shard.

The result is pathological at scale: instead of one round-trip returning a single number, the coordinator pulls every matching document.

Steps to Reproduce:

1. Apply the index template above; index a few sample documents.
2. Set plugins.calcite.enabled: true, plugins.calcite.pushdown.enabled: true.
3. Run EXPLAIN on the failing query — observe EnumerableWindow and the absence of an aggregations block in the DSL.
4. Replace eventstats with stats and re-run EXPLAIN — observe the correct aggregations + track_total_hits shape.

Environment Information

OpenSearch Version: 3.7.0-SNAPSHOT (reproduced on main, commit ec433f4ff).

Additional Details:

• Calcite engine: plugins.calcite.enabled = true
• Pushdown: plugins.calcite.pushdown.enabled = true
• Code locations:
  • ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java:498 — eventstats becomes a Window node
  • opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java:122,171 — pushdown rule excludes RexOver

Possible fix:
Add a rule that recognizes RexOver with empty partition/order/frame as a global aggregate and rewrites it to:

Project(input.*, broadcast(globalAgg))
 Aggregate(group=[], aggs=[<agg>])
 <input>

The aggregate is then pushable through the existing AggregateIndexScanRule. The broadcast step on the coordinator is one extra column join that costs nothing — far cheaper than streaming the entire matched set.

Same fix applies to eventstats count() by <field>: OVER (PARTITION BY <field>) → Aggregate(group=[<field>], aggs=[<agg>]) joined back on <field>. Both shapes are common in observability queries.

[RyanL1997]

@ryan-gh-bot can you do a rca and propose a fix? I would like to review it before you actually open the PR. You can just post it here.

[ryan-gh-bot]

Working on /@triage (invoked by @RyanL1997) — posting result when done.

[ryan-gh-bot]

[RCA] eventstats lowers to a LogicalProject containing RexOver, which the aggregation pushdown rule explicitly excludes

The reporter's diagnosis is correct. PPL eventstats always emits a Window AST node, which lowers to a LogicalProject whose expressions are RexOvers. Every AggregateIndexScanRule.Config registered today rejects projects that match LogicalProject::containsOver, so the global-aggregate case (no BY, no order, full frame) — which is semantically a flat aggregate — never reaches the pushdown path.

Environment

• Tenant: sql (RyanL1997/sql fork; upstream opensearch-project/sql)
• Refs checked:
  • target/main @ d099d3eff
• Verification: not run live; the issue body already contains an EXPLAIN trace from main @ ec433f4ff showing the exact symptom (EnumerableWindow over requestedTotalSize=2147483647), and the same shape is locked in by an existing unit test (see Evidence).

Suspected location

• ppl/src/main/java/org/opensearch/sql/ppl/parser/AstBuilder.java L498-L526 — visitEventstatsCommand unconditionally returns a Window AST node, including when groupList is empty.
• core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java L2105-L2134 — visitWindow calls relBuilder.projectPlus(overExpressions), producing the LogicalProject(..., RexOver) shape.
• core/src/main/java/org/opensearch/sql/calcite/CalciteRexNodeVisitor.java L581-L654 — visitWindowFunction builds the RexOver via PlanUtils.makeOver, with partitions/orderKeys taken straight from the AST. For eventstats count() (no BY) both lists are empty.
• opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java L121-L131 — Config.DEFAULT predicate Predicate.not(LogicalProject::containsOver) excludes any project containing a RexOver.
• opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/AggregateIndexScanRule.java L171-L182 — same exclusion in the BUCKET_NON_NULL_AGG config; same exclusion repeats at L202 and L214 for the …WITH_UDF variant.
• ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLEventstatsTest.java L19-L31 — existing test asserts the broken shape: count()=[COUNT() OVER ()] over a LogicalTableScan.

Analysis

The query path for eventstats <agg> (no BY) is:

1. AstBuilder.visitEventstatsCommand returns new Window(windowFunctionList, groupList, bucketNullable). When the BY clause is absent, groupList is empty.
2. CalciteRelNodeVisitor.visitWindow calls rexVisitor.analyze(...) on each entry to produce RexOvers, then relBuilder.projectPlus(overExpressions). The result is a single LogicalProject whose new columns are RexOvers with empty partition keys, empty order keys, and a default unbounded frame.
3. During pushdown optimization, every AggregateIndexScanRule.Config requires Predicate.not(LogicalProject::containsOver) on the project under the aggregate. The eventstats project always carries a RexOver, so no rule fires. The window is preserved as EnumerableWindow, executed coordinator-side, and the underlying CalciteEnumerableIndexScan retains requestedTotalSize=Integer.MAX_VALUE because no aggregate replaced it.

The exclusion is correct for window functions that genuinely need partition/order semantics — those cannot be expressed as a single aggregation. But for an unpartitioned, unordered, fully unbounded OVER () the row-level value is, by definition, the global aggregate broadcast to every input row. That global aggregate is exactly the shape OpenSearch handles natively in a single shard request (size:0, track_total_hits).

The same reasoning extends to eventstats <agg> by <field> (a OVER (PARTITION BY <field>) window with no order and full frame): the row-level value is the per-group aggregate broadcast to each group's rows, equivalent to a terms-bucketed aggregation joined back on <field>. Both shapes can be expressed in pushable Calcite without losing semantics.

Evidence

• Grep/read: traced the path end-to-end. visitEventstatsCommand at AstBuilder.java L498-L526 constructs the Window AST. visitWindow at CalciteRelNodeVisitor.java L2105 lowers to projectPlus(overExpressions). containsOver exclusions in AggregateIndexScanRule.java appear at L125, L175, L202, L214.
• Cross-check vs. working sibling: visitStatsCommand at AstBuilder.java L463-L495 returns new Aggregation(...), lowering to LogicalAggregate and matching Config.AGGREGATE_SCAN. This is the path that pushes down — exactly as the reporter shows in the working stats count() EXPLAIN.
• Regression check: not performed — the reporter does not claim a prior working version. The exclusion in AggregateIndexScanRule predates the eventstats push-down request, so this is a missing optimization, not a regression.
• Live verification: not performed — the reporter included EXPLAIN output for both shapes against current main @ ec433f4ff, and the unit test testEventstatsCount explicitly asserts the broken plan shape, so the symptom is reproducible by inspection.

Confidence: High
Two independent signals converge: (a) static reading shows a single, unambiguous code path where eventstats becomes a project-with-RexOver and four pushdown configs uniformly reject it, and (d) the existing unit test asserts the exact LogicalProject(..., COUNT() OVER ()) shape that the reporter quotes from the runtime EXPLAIN.

Next steps

Proposed fix (please review before I open a PR). Add a Calcite rule that runs before AggregateIndexScanRule and rewrites a LogicalProject containing only "globally unbounded" RexOvers into an Aggregate + Join shape that the existing pushdown rule already handles. This matches the rule-level approach the issue suggests, and it generalizes to the SQL SELECT col, COUNT() OVER () FROM t shape too.

Match criteria (intentionally strict, to avoid touching legitimate window semantics):

• The project's new (non-passthrough) expressions are all RexOvers.
• For each RexOver: partitionKeys.isEmpty(), orderKeys.isEmpty(), and the frame is unbounded on both sides (lowerBound.isUnbounded() and upperBound.isUnbounded()). This is the shape visitEventstatsCommand produces when groupList is empty; it is also WindowFrame.rowsUnbounded() from PlanUtils.makeOver L178-L191.
• The remaining project expressions are passthroughs (RexInputRefs) so the input row is preserved unchanged. (Conservative; relax later if needed.)

Rewrite (logical):

LogicalProject(input.*, RexOver(<agg>) OVER ())
 <input>
 ↓
LogicalProject(input.*, agg_col)
 LogicalJoin(condition=true, INNER)
 <input>
 LogicalAggregate(group=[], aggs=[<agg>])
 <input>

The new LogicalAggregate is then matched by the existing AggregateIndexScanRule.Config.AGGREGATE_SCAN (no project under it), pushes into CalciteEnumerableIndexScan as a flat size:0 + track_total_hits request, and the cross-join with the resulting single-row relation runs coordinator-side as a constant broadcast.

Files to change:

1. New: opensearch/src/main/java/org/opensearch/sql/opensearch/planner/rules/GlobalWindowToAggregateRule.java — the new rule. Precedent for inspecting RexOver shapes inside a LogicalProject lives in RareTopPushdownRule.java L34-L88 and the helper PlanUtils.getRexWindowFromProject.
2. Modify: OpenSearchIndexRules.OPEN_SEARCH_PUSHDOWN_RULES — register the new rule before AGGREGATE_INDEX_SCAN.
3. Modify: CalcitePPLEventstatsTest.testEventstatsCount — the asserted "expectedLogical" string must change once the rewrite runs in this test path; if the test runs at a stage before pushdown rules fire, leave it and add a new expectation at a later stage. (Will confirm during implementation.)
4. New: a pushdown unit test verifying the rewritten plan uses LogicalAggregate and that the resulting OpenSearch DSL contains "size":0 and track_total_hits (mirror the stats count() baseline from the issue's EXPLAIN).

Open questions worth deciding before I write the PR:

• Scope. Land just the OVER () (no partition) case in the first PR, or include OVER (PARTITION BY <field>) as well? The BY case needs an Aggregate(group=[<field>]) JOIN ON input.field = agg.field shape with explicit handling of bucket_nullable=false (the existing wrapWindowFunctionsWithGroupNotNull path in visitWindow), so it is meaningfully more work and review surface. My preference is to land the no-BY case first since that's what the bug title and the reproducer call out, then a follow-up PR for BY.
• Rule placement. I'm proposing the rule in OPEN_SEARCH_PUSHDOWN_RULES (gated by plugins.calcite.pushdown.enabled). It's behavior-preserving so it could also live in OPEN_SEARCH_NON_PUSHDOWN_RULES, but there's no point rewriting if pushdown is off — keeping it gated avoids extra optimizer work.
• Upstream-vs-fork. Issue and target are on RyanL1997/sql. If the intent is to upstream this, file paths and tests are identical; only the PR target changes.

Let me know which scope to take and I'll open the PR against target/main.

---

Automated response from @ryan-gh-bot (maintainer-triggered agent). About this agent.