planner/cascades: Add ImplementationRule for Window

planner/cascades/implementation_rules.go

@@ -80,6 +80,9 @@ var defaultImplementationMap = map[memo.Operand][]ImplementationRule{
 	memo.OperandMaxOneRow: {
 		&ImplMaxOneRow{},
 	},
+	memo.OperandWindow: {
+		&ImplWindow{},
+	},
 }
 
 // ImplTableDual implements LogicalTableDual as PhysicalTableDual.
@@ -519,3 +522,38 @@ func (r *ImplMaxOneRow) OnImplement(expr *memo.GroupExpr, reqProp *property.Phys
 		&property.PhysicalProperty{ExpectedCnt: 2})
 	return impl.NewMaxOneRowImpl(physicalMaxOneRow), nil
 }
+
+// ImplWindow implements LogicalWindow to PhysicalWindow
+type ImplWindow struct {
+}
+
+// Match implements ImplementationRule Match interface.
+func (w *ImplWindow) Match(expr *memo.GroupExpr, prop *property.PhysicalProperty) (matched bool) {
+	lw := expr.ExprNode.(*plannercore.LogicalWindow)
+	var byItems []property.Item
+	byItems = append(byItems, lw.PartitionBy...)
+	byItems = append(byItems, lw.OrderBy...)
+	childProperty := &property.PhysicalProperty{ExpectedCnt: math.MaxFloat64, Items: byItems, Enforced: true}
+	return prop.IsPrefix(childProperty)
+}
+
+// OnImplement implements ImplementationRule OnImplement interface
+func (w *ImplWindow) OnImplement(expr *memo.GroupExpr, reqProp *property.PhysicalProperty) (memo.Implementation, error) {
+	lw := expr.ExprNode.(*plannercore.LogicalWindow)
+	var byItems []property.Item
+	byItems = append(byItems, lw.PartitionBy...)
+	byItems = append(byItems, lw.OrderBy...)
+	physicalWindow := plannercore.PhysicalWindow{
+		WindowFuncDescs: lw.WindowFuncDescs,
+		PartitionBy:     lw.PartitionBy,
+		OrderBy:         lw.OrderBy,
+		Frame:           lw.Frame,
+	}.Init(
+		lw.SCtx(),
+		expr.Group.Prop.Stats,
+		lw.SelectBlockOffset(),
+		&property.PhysicalProperty{ExpectedCnt: math.MaxFloat64, Items: byItems, Enforced: true},
+	)
+	physicalWindow.SetSchema(lw.Schema())
+	return impl.NewWindowImpl(physicalWindow), nil
+}

planner/implementation/simple_plans.go

@@ -203,3 +203,19 @@ func (impl *MaxOneRowImpl) CalcCost(outCount float64, children ...memo.Implement
 func NewMaxOneRowImpl(maxOneRow *plannercore.PhysicalMaxOneRow) *MaxOneRowImpl {
 	return &MaxOneRowImpl{baseImpl{plan: maxOneRow}}
 }
+
+// WindowImpl is the implementation of PhysicalWindow.
+type WindowImpl struct {
+	baseImpl
+}
+
+// NewWindowImpl creates a new WindowImpl.
+func NewWindowImpl(window *plannercore.PhysicalWindow) *WindowImpl {
+	return &WindowImpl{baseImpl{plan: window}}
+}
+
+// CalcCost implements Implementation CalcCost interface.
+func (impl *WindowImpl) CalcCost(outCount float64, children ...memo.Implementation) float64 {
+	impl.cost = children[0].GetCost()
+	return impl.cost
+}