[Backport 2.19-dev] Support Streamstats command with calcite

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;
@@ -747,6 +748,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

@@ -80,6 +80,7 @@
 import org.opensearch.sql.ast.tree.Search;
 import org.opensearch.sql.ast.tree.Sort;
 import org.opensearch.sql.ast.tree.SPath;
+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;
@@ -411,6 +412,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);
+  }
+}