Support Streamstats command with calcite

core/src/main/java/org/opensearch/sql/analysis/Analyzer.java

@@ -88,6 +88,7 @@
 import org.opensearch.sql.ast.tree.Rex;
 import org.opensearch.sql.ast.tree.Sort;
 import org.opensearch.sql.ast.tree.Sort.SortOption;
+import org.opensearch.sql.ast.tree.StreamWindow;
 import org.opensearch.sql.ast.tree.SubqueryAlias;
 import org.opensearch.sql.ast.tree.TableFunction;
 import org.opensearch.sql.ast.tree.Timechart;
@@ -732,6 +733,11 @@ public LogicalPlan visitTrendline(Trendline node, AnalysisContext context) {
         computationsAndTypes.build());
   }
 
+  @Override
+  public LogicalPlan visitStreamWindow(StreamWindow node, AnalysisContext context) {
+    throw getOnlyForCalciteException("Streamstats");
+  }
+
   @Override
   public LogicalPlan visitFlatten(Flatten node, AnalysisContext context) {
     throw getOnlyForCalciteException("Flatten");

core/src/main/java/org/opensearch/sql/ast/AbstractNodeVisitor.java

@@ -76,6 +76,7 @@
 import org.opensearch.sql.ast.tree.Rex;
 import org.opensearch.sql.ast.tree.SPath;
 import org.opensearch.sql.ast.tree.Sort;
+import org.opensearch.sql.ast.tree.StreamWindow;
 import org.opensearch.sql.ast.tree.SubqueryAlias;
 import org.opensearch.sql.ast.tree.TableFunction;
 import org.opensearch.sql.ast.tree.Timechart;
@@ -399,6 +400,10 @@ public T visitWindow(Window window, C context) {
     return visitChildren(window, context);
   }
 
+  public T visitStreamWindow(StreamWindow node, C context) {
+    return visitChildren(node, context);
+  }
+
   public T visitJoin(Join node, C context) {
     return visitChildren(node, context);
   }

core/src/main/java/org/opensearch/sql/ast/tree/StreamWindow.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ast.tree;
+
+import com.google.common.collect.ImmutableList;
+import java.util.Collections;
+import java.util.List;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.opensearch.sql.ast.AbstractNodeVisitor;
+import org.opensearch.sql.ast.expression.Argument;
+import org.opensearch.sql.ast.expression.UnresolvedExpression;
+
+@Getter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+public class StreamWindow extends UnresolvedPlan {
+
+  private final List<UnresolvedExpression> windowFunctionList;
+  private final List<Argument> argExprList;
+  @ToString.Exclude private UnresolvedPlan child;
+
+  /** StreamWindow Constructor without optional argument. */
+  public StreamWindow(List<UnresolvedExpression> windowFunctionList) {
+    this(windowFunctionList, Collections.emptyList());
+  }
+
+  /** StreamWindow Constructor. */
+  public StreamWindow(List<UnresolvedExpression> windowFunctionList, List<Argument> argExprList) {
+    this.windowFunctionList = windowFunctionList;
+    this.argExprList = argExprList;
+  }
+
+  public boolean hasArgument() {
+    return !argExprList.isEmpty();
+  }
+
+  @Override
+  public StreamWindow attach(UnresolvedPlan child) {
+    this.child = child;
+    return this;
+  }
+
+  @Override
+  public List<UnresolvedPlan> getChild() {
+    return this.child == null ? ImmutableList.of() : ImmutableList.of(this.child);
+  }
+
+  @Override
+  public <T, C> T accept(AbstractNodeVisitor<T, C> nodeVisitor, C context) {
+    return nodeVisitor.visitStreamWindow(this, context);
+  }
+}

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

@@ -112,6 +112,7 @@
 import org.opensearch.sql.ast.tree.SPath;
 import org.opensearch.sql.ast.tree.Sort;
 import org.opensearch.sql.ast.tree.Sort.SortOption;
+import org.opensearch.sql.ast.tree.StreamWindow;
 import org.opensearch.sql.ast.tree.SubqueryAlias;
 import org.opensearch.sql.ast.tree.TableFunction;
 import org.opensearch.sql.ast.tree.Trendline;
@@ -1117,6 +1118,15 @@ public RelNode visitWindow(Window node, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
+  @Override
+  public RelNode visitStreamWindow(StreamWindow node, CalcitePlanContext context) {
+    visitChildren(node, context);
+    List<RexNode> overExpressions =
+        node.getWindowFunctionList().stream().map(w -> rexVisitor.analyze(w, context)).toList();
+    context.relBuilder.projectPlus(overExpressions);
+    return context.relBuilder.peek();
+  }
+
   @Override
   public RelNode visitFillNull(FillNull node, CalcitePlanContext context) {
     visitChildren(node, context);

docs/user/ppl/index.rst

@@ -112,6 +112,8 @@ The query start with search command and then flowing a set of command delimited
 
   - `stats command <cmd/stats.rst>`_
 
+  - `streamstats command <cmd/streamstats.rst>`_
+
   - `subquery (aka subsearch) command <cmd/subquery.rst>`_
 
   - `reverse command <cmd/reverse.rst>`_

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

@@ -27,7 +27,7 @@ public void init() throws Exception {
   }
 
   @Test
-  public void testEventstat() throws IOException {
+  public void testEventstats() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -57,7 +57,7 @@ public void testEventstat() throws IOException {
   }
 
   @Test
-  public void testEventstatWithNull() throws IOException {
+  public void testEventstatsWithNull() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -89,7 +89,7 @@ public void testEventstatWithNull() throws IOException {
   }
 
   @Test
-  public void testEventstatBy() throws IOException {
+  public void testEventstatsBy() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -119,7 +119,7 @@ public void testEventstatBy() throws IOException {
   }
 
   @Test
-  public void testEventstatByWithNull() throws IOException {
+  public void testEventstatsByWithNull() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -166,7 +166,7 @@ public void testEventstatByWithNull() throws IOException {
   }
 
   @Test
-  public void testEventstatBySpan() throws IOException {
+  public void testEventstatsBySpan() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -183,7 +183,7 @@ public void testEventstatBySpan() throws IOException {
   }
 
   @Test
-  public void testEventstatBySpanWithNull() throws IOException {
+  public void testEventstatsBySpanWithNull() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -202,7 +202,7 @@ public void testEventstatBySpanWithNull() throws IOException {
   }
 
   @Test
-  public void testEventstatByMultiplePartitions1() throws IOException {
+  public void testEventstatsByMultiplePartitions1() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -219,7 +219,7 @@ public void testEventstatByMultiplePartitions1() throws IOException {
   }
 
   @Test
-  public void testEventstatByMultiplePartitions2() throws IOException {
+  public void testEventstatsByMultiplePartitions2() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -236,7 +236,7 @@ public void testEventstatByMultiplePartitions2() throws IOException {
   }
 
   @Test
-  public void testEventstatByMultiplePartitionsWithNull1() throws IOException {
+  public void testEventstatsByMultiplePartitionsWithNull1() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -255,7 +255,7 @@ public void testEventstatByMultiplePartitionsWithNull1() throws IOException {
   }
 
   @Test
-  public void testEventstatByMultiplePartitionsWithNull2() throws IOException {
+  public void testEventstatsByMultiplePartitionsWithNull2() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -289,7 +289,7 @@ public void testUnsupportedWindowFunctions() {
   }
 
   @Test
-  public void testMultipleEventstat() throws IOException {
+  public void testMultipleEventstats() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -306,7 +306,7 @@ public void testMultipleEventstat() throws IOException {
   }
 
   @Test
-  public void testMultipleEventstatWithNull() throws IOException {
+  public void testMultipleEventstatsWithNull() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -325,7 +325,7 @@ public void testMultipleEventstatWithNull() throws IOException {
   }
 
   @Test
-  public void testMultipleEventstatWithEval() throws IOException {
+  public void testMultipleEventstatsWithEval() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -343,7 +343,7 @@ public void testMultipleEventstatWithEval() throws IOException {
   }
 
   @Test
-  public void testEventstatEmptyRows() throws IOException {
+  public void testEventstatsEmptyRows() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -363,7 +363,7 @@ public void testEventstatEmptyRows() throws IOException {
   }
 
   @Test
-  public void testEventstatVariance() throws IOException {
+  public void testEventstatsVariance() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -433,7 +433,7 @@ public void testEventstatVariance() throws IOException {
   }
 
   @Test
-  public void testEventstatVarianceWithNull() throws IOException {
+  public void testEventstatsVarianceWithNull() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -496,7 +496,7 @@ public void testEventstatVarianceWithNull() throws IOException {
   }
 
   @Test
-  public void testEventstatVarianceBy() throws IOException {
+  public void testEventstatsVarianceBy() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -513,7 +513,7 @@ public void testEventstatVarianceBy() throws IOException {
   }
 
   @Test
-  public void testEventstatVarianceBySpan() throws IOException {
+  public void testEventstatsVarianceBySpan() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -527,7 +527,7 @@ public void testEventstatVarianceBySpan() throws IOException {
   }
 
   @Test
-  public void testEventstatVarianceWithNullBy() throws IOException {
+  public void testEventstatsVarianceWithNullBy() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -576,7 +576,7 @@ public void testEventstatVarianceWithNullBy() throws IOException {
   }
 
   @Test
-  public void testEventstatDistinctCount() throws IOException {
+  public void testEventstatsDistinctCount() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -601,7 +601,7 @@ public void testEventstatDistinctCount() throws IOException {
   }
 
   @Test
-  public void testEventstatDistinctCountByCountry() throws IOException {
+  public void testEventstatsDistinctCountByCountry() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -627,7 +627,7 @@ public void testEventstatDistinctCountByCountry() throws IOException {
   }
 
   @Test
-  public void testEventstatDistinctCountFunction() throws IOException {
+  public void testEventstatsDistinctCountFunction() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -653,7 +653,7 @@ public void testEventstatDistinctCountFunction() throws IOException {
   }
 
   @Test
-  public void testEventstatDistinctCountWithNull() throws IOException {
+  public void testEventstatsDistinctCountWithNull() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -682,7 +682,7 @@ public void testEventstatDistinctCountWithNull() throws IOException {
 
   @Ignore
   @Test
-  public void testEventstatEarliestAndLatest() throws IOException {
+  public void testEventstatsEarliestAndLatest() throws IOException {
     JSONObject actual =
         executeQuery(
             String.format(