backport opensearch-project#4831

Signed-off-by: Xinyu Hao <haoxinyu@amazon.com>

Author: ishaoxy

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

@@ -1747,31 +1747,31 @@ public RelNode visitStreamWindow(StreamWindow node, CalcitePlanContext context)
                     .as(ROW_NUMBER_COLUMN_FOR_STREAMSTATS);
             context.relBuilder.projectPlus(streamSeq);
 
-      if (!node.isBucketNullable()) {
-        // construct groupNotNull predicate
-        List<RexNode> groupByList =
-            groupList.stream().map(expr -> rexVisitor.analyze(expr, context)).collect(Collectors.toList());
-        List<RexNode> notNullList =
-            PlanUtils.getSelectColumns(groupByList).stream()
-                .map(context.relBuilder::field)
-                .map(context.relBuilder::isNotNull)
-                .collect(Collectors.toList());
-        RexNode groupNotNull = context.relBuilder.and(notNullList);
+        if (!node.isBucketNullable()) {
+            // construct groupNotNull predicate
+            List<RexNode> groupByList =
+                groupList.stream().map(expr -> rexVisitor.analyze(expr, context)).collect(Collectors.toList());
+            List<RexNode> notNullList =
+                PlanUtils.getSelectColumns(groupByList).stream()
+                    .map(context.relBuilder::field)
+                    .map(context.relBuilder::isNotNull)
+                    .collect(Collectors.toList());
+            RexNode groupNotNull = context.relBuilder.and(notNullList);
 
-        // wrap each expr: CASE WHEN groupNotNull THEN rawExpr ELSE CAST(NULL AS rawType) END
-        List<RexNode> wrappedOverExprs =
-            wrapWindowFunctionsWithGroupNotNull(overExpressions, groupNotNull, context);
-        context.relBuilder.projectPlus(wrappedOverExprs);
-      } else {
-        context.relBuilder.projectPlus(overExpressions);
-      }
+            // wrap each expr: CASE WHEN groupNotNull THEN rawExpr ELSE CAST(NULL AS rawType) END
+            List<RexNode> wrappedOverExprs =
+                wrapWindowFunctionsWithGroupNotNull(overExpressions, groupNotNull, context);
+            context.relBuilder.projectPlus(wrappedOverExprs);
+            } else {
+              context.relBuilder.projectPlus(overExpressions);
+            }
 
-      // resort when there is by condition
-      context.relBuilder.sort(context.relBuilder.field(ROW_NUMBER_COLUMN_FOR_STREAMSTATS));
-      context.relBuilder.projectExcept(context.relBuilder.field(ROW_NUMBER_COLUMN_FOR_STREAMSTATS));
-    } else {
-      context.relBuilder.projectPlus(overExpressions);
-    }
+            // resort when there is by condition
+            context.relBuilder.sort(context.relBuilder.field(ROW_NUMBER_COLUMN_FOR_STREAMSTATS));
+            context.relBuilder.projectExcept(context.relBuilder.field(ROW_NUMBER_COLUMN_FOR_STREAMSTATS));
+        } else {
+          context.relBuilder.projectPlus(overExpressions);
+        }
 
         return context.relBuilder.peek();
     }
@@ -1985,7 +1985,7 @@ private RexNode buildGroupFilter(
         RexCorrelVariable correl) {
         // build conjunctive equality filters: right.g_i = outer.g_i
         if (groupList.isEmpty()) {
-          return null;
+            return null;
         }
         List<RexNode> equalsList =
             groupList.stream()