*: Support stream aggregation in new plan #4481
Conversation
| @@ -380,7 +380,7 @@ func (s *testPlanSuite) TestDAGPlanBuilderSubquery(c *C) { | |||
| // Test Nested sub query. | |||
| { | |||
There was a problem hiding this comment.
add test for stream agg + index join and stream agg + merge join and stream agg + limit and stream agg + sort
plan/logical_plans.go
Outdated
| // groupByCols stores the columns that are group-by items. | ||
| groupByCols []*expression.Column | ||
|
|
||
| possibleProperties [][]*expression.Column | ||
| childCount float64 // childCount is the child plan's count. | ||
| cardinality float64 |
There was a problem hiding this comment.
cardinality is always equal to LogicalAggregation's profile's count.
plan/logical_plans.go
Outdated
| // groupByCols stores the columns that are group-by items. | ||
| groupByCols []*expression.Column | ||
|
|
||
| possibleProperties [][]*expression.Column | ||
| childCount float64 // childCount is the child plan's count. |
There was a problem hiding this comment.
maybe inputCount is a better name. childCount seems to indicate the number of children.
plan/new_physical_plan_builder.go
Outdated
| @@ -1090,26 +1090,80 @@ func (p *TopN) getChildrenPossibleProps(prop *requiredProp) [][]*requiredProp { | |||
| return props | |||
| } | |||
|
|
|||
| func (p *LogicalAggregation) getStreamAggs() []PhysicalPlan { | |||
| for _, aggFunc := range p.AggFuncs { | |||
| if aggFunc.GetMode() == expression.FinalMode { | |||
There was a problem hiding this comment.
This code seems never be touched ?
There was a problem hiding this comment.
We do decompose will have FinalMode aggregation.
| return nil | ||
| } | ||
| } | ||
| // group by a + b is not interested in any order. |
plan/task.go
Outdated
| } | ||
| } | ||
| task = finishCopTask(cop, p.ctx, p.allocator) | ||
| task.addCost(task.count()*cpuFactor + p.cardinality*hashAggMemFactor) |
There was a problem hiding this comment.
use p.statsProfile().count to replace p.cardinality
plan/stats.go
Outdated
| @@ -81,7 +81,6 @@ func (p *DataSource) getStatsProfileByFilter(conds expression.CNFExprs) *statsPr | |||
| } | |||
| selectivity, err := p.statisticTable.Selectivity(p.ctx, conds) | |||
| if err != nil { | |||
| log.Warnf("An error happened: %v, we have to use the default selectivity", err.Error()) | |||
plan/new_physical_plan_builder.go
Outdated
|
|
||
| reqProp := &requiredProp{taskTp: rootTaskType, cols: p.propKeys, expectedCnt: prop.expectedCnt * p.childCount / p.profile.count} | ||
| if !prop.isEmpty() { | ||
| if prop.desc { |
plan/new_physical_plan_builder.go
Outdated
| agg.SetSchema(p.schema.Clone()) | ||
| agg.profile = p.profile | ||
| aggs = append(aggs, agg) | ||
| if len(p.possibleProperties) == 0 { |
plan/new_physical_plan_builder.go
Outdated
| func (p *LogicalAggregation) generatePhysicalPlans() []PhysicalPlan { | ||
| ha := PhysicalAggregation{ | ||
| aggs := make([]PhysicalPlan, 0, 2) |
There was a problem hiding this comment.
2 is not reasonable, len(p.possibleProperties) + 1 is the upper bound.
|
@hanfei1991 PTAL |
| }, | ||
| { | ||
| sql: "select count(*) from t where e > 1 group by b", | ||
| best: "TableReader(Table(t)->Sel([gt(test.t.e, 1)])->HashAgg)->HashAgg", |
There was a problem hiding this comment.
Why not use index on e for filtering or use index on b for streaming aggregation?
There was a problem hiding this comment.
using index will cause a double read.
|
LGTM |
| best: "IndexLookUp(Index(t.b_c)[[-inf <nil>,20 +inf]], Table(t)->HashAgg)->HashAgg", | ||
| }, | ||
| { | ||
| sql: "select count(e) from t where t.b <= 30", |
There was a problem hiding this comment.
the case that t.b <= 30 and t.b <= 40 is not necessary?
There was a problem hiding this comment.
There are old tests. @hanfei1991
|
/run-all-test |
|
PTAL @winoros |
winoros
added
status/LGT2
support stream aggregation in the new plan and clean the tests.