feat: support inc window filter

internal/topo/planner/incAggPlan.go

@@ -22,14 +22,14 @@ type IncWindowPlan struct {
 	Length      int
 	Dimensions  ast.Dimensions
 	IncAggFuncs []*ast.Field
+	Condition   ast.Expr
 }
 
 func (p *IncWindowPlan) BuildExplainInfo() {
 	info := "wType:"
 	info += p.WType.String()
-	info += ", "
 	if len(p.Dimensions) > 0 {
-		info += "Dimension:["
+		info += ", Dimension:["
 		for i, dimension := range p.Dimensions {
 			if dimension.Expr != nil {
 				info += dimension.Expr.String()
@@ -40,6 +40,11 @@ func (p *IncWindowPlan) BuildExplainInfo() {
 		}
 		info += "]"
 	}
+	if p.Condition != nil {
+		info += ", filter:["
+		info += p.Condition.String()
+		info += "]"
+	}
 	info += ", funcs:["
 	for i, aggFunc := range p.IncAggFuncs {
 		if i > 0 {

internal/topo/planner/plan_explain_test.go

@@ -16,6 +16,7 @@ package planner
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
 	"testing"
 
@@ -58,7 +59,7 @@ func TestExplainPlan(t *testing.T) {
 		{
 			sql: `select count(a) from stream group by countwindow(2)`,
 			explain: `{"op":"ProjectPlan_0","info":"Fields:[ Call:{ name:bypass, args:[$$default.inc_agg_col_1] } ]"}
-	{"op":"IncAggWindowPlan_1","info":"wType:COUNT_WINDOW, , funcs:[Call:{ name:inc_count, args:[stream.a] }->inc_agg_col_1]"}
+	{"op":"IncAggWindowPlan_1","info":"wType:COUNT_WINDOW, funcs:[Call:{ name:inc_count, args:[stream.a] }->inc_agg_col_1]"}
 			{"op":"DataSourcePlan_2","info":"StreamName: stream, StreamFields:[ a ]"}`,
 		},
 		{
@@ -98,6 +99,12 @@ func TestExplainPlan(t *testing.T) {
 							{"op":"DataSourcePlan_4","info":"StreamName: stream, StreamFields:[ a, b ]"}
 							{"op":"DataSourcePlan_5","info":"StreamName: sharedStream, StreamFields:[ a, b ]"}`,
 		},
+		{
+			sql: `SELECT count(*) from stream group by countWindow(4) filter (where a > 1) `,
+			explain: `{"op":"ProjectPlan_0","info":"Fields:[ Call:{ name:bypass, args:[$$default.inc_agg_col_1] } ]"}
+	{"op":"IncAggWindowPlan_1","info":"wType:COUNT_WINDOW, filter:[binaryExpr:{ stream.a > 1 }], funcs:[Call:{ name:inc_count, args:[*] }->inc_agg_col_1]"}
+			{"op":"DataSourcePlan_2","info":"StreamName: stream, StreamFields:[ a, b ]"}`,
+		},
 	}
 	for _, tc := range testcases {
 		stmt, err := xsql.NewParser(strings.NewReader(tc.sql)).Parse()
@@ -108,6 +115,10 @@ func TestExplainPlan(t *testing.T) {
 		require.NoError(t, err)
 		explain, err := ExplainFromLogicalPlan(p, "")
 		require.NoError(t, err)
+		if tc.explain == "" {
+			fmt.Println(explain)
+			continue
+		}
 		require.Equal(t, tc.explain, explain, tc.sql)
 	}
 }

internal/topo/planner/planner.go

@@ -231,6 +231,11 @@
 	case *AnalyticFuncsPlan:
 		op = Transform(&operator.AnalyticFuncsOp{Funcs: t.funcs, FieldFuncs: t.fieldFuncs}, fmt.Sprintf("%d_analytic", newIndex), options)
 	case *IncWindowPlan:
+		if t.Condition != nil {
+			wfilterOp := Transform(&operator.FilterOp{Condition: t.Condition}, fmt.Sprintf("%d_windowFilter", newIndex), options)
+			tp.AddOperator(inputs, wfilterOp)
+			inputs = []node.Emitter{wfilterOp}
+		}
 		op, err = node.NewWindowIncAggOp(fmt.Sprintf("%d_inc_agg_window", newIndex), &node.WindowConfig{
 			Type:        t.WType,
 			CountLength: t.Length,
@@ -416,6 +421,7 @@
 					Length:      int(w.Length.Val),
 					Dimensions:  dimensions.GetGroups(),
 					IncAggFuncs: incAggFields,
+					Condition:   w.Filter,
 				}.Init()
 				incWp.SetChildren(children)
 				children = []LogicalPlan{incWp}