Support Streamstats command with calcite (opensearch-project#4297)

* support streamstats simply

Signed-off-by: Xinyu Hao <[email protected]>

* add some tests

Signed-off-by: Xinyu Hao <[email protected]>

* add UT

Signed-off-by: Xinyu Hao <[email protected]>

* fix some error

Signed-off-by: Xinyu Hao <[email protected]>

* add global

Signed-off-by: Xinyu Hao <[email protected]>

* implement global

Signed-off-by: Xinyu Hao <[email protected]>

* implement reset

Signed-off-by: Xinyu Hao <[email protected]>

* implement all the arguments

Signed-off-by: Xinyu Hao <[email protected]>

* fix test

Signed-off-by: Xinyu Hao <[email protected]>

* add all IT, UT and rst doc

Signed-off-by: Xinyu Hao <[email protected]>

* fix anonymizer test

Signed-off-by: Xinyu Hao <[email protected]>

* fix doctest

Signed-off-by: Xinyu Hao <[email protected]>

* modify doc and IT

Signed-off-by: Xinyu Hao <[email protected]>

* add explainIT

Signed-off-by: Xinyu Hao <[email protected]>

* fix import

Signed-off-by: Xinyu Hao <[email protected]>

* fix typo

Signed-off-by: Xinyu Hao <[email protected]>

* fix doctest

Signed-off-by: Xinyu Hao <[email protected]>

* fix explainIT yaml format

Signed-off-by: Xinyu Hao <[email protected]>

* fix dc nopushdown explainIT

Signed-off-by: Xinyu Hao <[email protected]>

* add explainIT for path2 and path3

Signed-off-by: Xinyu Hao <[email protected]>

* typo error

Signed-off-by: Xinyu Hao <[email protected]>

* handle resort case

Signed-off-by: Xinyu Hao <[email protected]>

* fix IT

Signed-off-by: Xinyu Hao <[email protected]>

* change row_num

Signed-off-by: Xinyu Hao <[email protected]>

* Rule out aggregator from PPLAggregateMergeRule

Signed-off-by: Xinyu Hao <[email protected]>

* Rule out aggregator from PPLAggregateMergeRule

Signed-off-by: Xinyu Hao <[email protected]>

* fix explainIT

Signed-off-by: Xinyu Hao <[email protected]>

---------

Signed-off-by: Xinyu Hao <[email protected]>
Signed-off-by: Yuanchun Shen <[email protected]>
Co-authored-by: Yuanchun Shen <[email protected]>

Author: 2 people

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

@@ -92,6 +92,7 @@
 import org.opensearch.sql.ast.tree.Search;
 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;
@@ -748,6 +749,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

@@ -79,6 +79,7 @@
 import org.opensearch.sql.ast.tree.SPath;
 import org.opensearch.sql.ast.tree.Search;
 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;
@@ -410,6 +411,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,71 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.ast.tree;
+
+import com.google.common.collect.ImmutableList;
+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.UnresolvedExpression;
+
+@Getter
+@ToString
+@EqualsAndHashCode(callSuper = false)
+public class StreamWindow extends UnresolvedPlan {
+
+  private final List<UnresolvedExpression> windowFunctionList;
+  private final List<UnresolvedExpression> groupList;
+  private final boolean current;
+  private final int window;
+  private final boolean global;
+  private final UnresolvedExpression resetBefore;
+  private final UnresolvedExpression resetAfter;
+  @ToString.Exclude private UnresolvedPlan child;
+
+  /** StreamWindow Constructor. */
+  public StreamWindow(
+      List<UnresolvedExpression> windowFunctionList,
+      List<UnresolvedExpression> groupList,
+      boolean current,
+      int window,
+      boolean global,
+      UnresolvedExpression resetBefore,
+      UnresolvedExpression resetAfter) {
+    this.windowFunctionList = windowFunctionList;
+    this.groupList = groupList;
+    this.current = current;
+    this.window = window;
+    this.global = global;
+    this.resetBefore = resetBefore;
+    this.resetAfter = resetAfter;
+  }
+
+  public boolean isCurrent() {
+    return current;
+  }
+
+  public boolean isGlobal() {
+    return global;
+  }
+
+  @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);
+  }
+}