Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

executor: fill correlated column value in late materialization filter conditions #49244

Merged
merged 2 commits into from
Dec 13, 2023
Merged

executor: fill correlated column value in late materialization filter conditions #49244

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a8767d5
executor: fill correlated column value in late materialization filter…
Lloyd-Pottiger Dec 7, 2023
38a9869
add some comments
Lloyd-Pottiger Dec 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 12 additions & 7 deletions pkg/executor/builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2972,13 +2972,18 @@ func markChildrenUsedCols(outputCols []*expression.Column, childSchemas ...*expr

func (*executorBuilder) corColInDistPlan(plans []plannercore.PhysicalPlan) bool {
for _, p := range plans {
x, ok := p.(*plannercore.PhysicalSelection)
if !ok {
continue
}
for _, cond := range x.Conditions {
if len(expression.ExtractCorColumns(cond)) > 0 {
return true
switch x := p.(type) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why mpp plan works fine?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

mpp returns a MPPGather when buildTableReader

tidb/pkg/executor/builder.go

Lines 3414 to 3435 in cf23666

func (b *executorBuilder) buildTableReader(v *plannercore.PhysicalTableReader) exec.Executor {
failpoint.Inject("checkUseMPP", func(val failpoint.Value) {
if !b.ctx.GetSessionVars().InRestrictedSQL && val.(bool) != useMPPExecution(b.ctx, v) {
if val.(bool) {
b.err = errors.New("expect mpp but not used")
} else {
b.err = errors.New("don't expect mpp but we used it")
}
failpoint.Return(nil)
}
})
useMPP := useMPPExecution(b.ctx, v)
useTiFlashBatchCop := v.ReadReqType == plannercore.BatchCop
useTiFlash := useMPP || useTiFlashBatchCop
if useTiFlash {
if _, isTiDBZoneLabelSet := config.GetGlobalConfig().Labels[placement.DCLabelKey]; b.ctx.GetSessionVars().TiFlashReplicaRead != tiflash.AllReplicas && !isTiDBZoneLabelSet {
b.ctx.GetSessionVars().StmtCtx.AppendWarning(errors.Errorf("the variable tiflash_replica_read is ignored, because the entry TiDB[%s] does not set the zone attribute and tiflash_replica_read is '%s'", config.GetGlobalConfig().AdvertiseAddress, tiflash.GetTiFlashReplicaRead(b.ctx.GetSessionVars().TiFlashReplicaRead)))
}
}
if useMPP {
return b.buildMPPGather(v)
}

MPPGather will always call ConstructTreeBasedDistExec when ConstructDAGReq

tidb/pkg/executor/internal/builder/builder_utils.go

Lines 46 to 59 in cf23666

func ConstructDAGReq(ctx sessionctx.Context, plans []plannercore.PhysicalPlan, storeType kv.StoreType) (dagReq *tipb.DAGRequest, err error) {
dagReq = &tipb.DAGRequest{}
dagReq.TimeZoneName, dagReq.TimeZoneOffset = timeutil.Zone(ctx.GetSessionVars().Location())
sc := ctx.GetSessionVars().StmtCtx
if sc.RuntimeStatsColl != nil {
collExec := true
dagReq.CollectExecutionSummaries = &collExec
}
dagReq.Flags = sc.PushDownFlags()
if storeType == kv.TiFlash {
var executors []*tipb.Executor
executors, err = ConstructTreeBasedDistExec(ctx, plans[0])
dagReq.RootExecutor = executors[0]
} else {

For TableReaderExecutor, only when e.corColInFilter = true will call ConstructTreeBasedDistExec

tidb/pkg/executor/table_reader.go

Lines 142 to 161 in cf23666

func (e *TableReaderExecutor) Open(ctx context.Context) error {
r, ctx := tracing.StartRegionEx(ctx, "TableReaderExecutor.Open")
defer r.End()
failpoint.Inject("mockSleepInTableReaderNext", func(v failpoint.Value) {
ms := v.(int)
time.Sleep(time.Millisecond * time.Duration(ms))
})
if e.memTracker != nil {
e.memTracker.Reset()
} else {
e.memTracker = memory.NewTracker(e.ID(), -1)
}
e.memTracker.AttachTo(e.Ctx().GetSessionVars().StmtCtx.MemTracker)
var err error
if e.corColInFilter {
if e.storeType == kv.TiFlash {
execs, err := builder.ConstructTreeBasedDistExec(e.Ctx(), e.tablePlan)
if err != nil {

case *plannercore.PhysicalSelection:
for _, cond := range x.Conditions {
if len(expression.ExtractCorColumns(cond)) > 0 {
return true
}
}
case *plannercore.PhysicalTableScan:
for _, cond := range x.LateMaterializationFilterCondition {
if len(expression.ExtractCorColumns(cond)) > 0 {
return true
}
}
}
}
Expand Down
4 changes: 3 additions & 1 deletion pkg/executor/table_reader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,8 @@ type TableReaderExecutor struct {
byItems []*util.ByItems
paging bool
storeType kv.StoreType
// corColInFilter tells whether there's correlated column in filter.
// corColInFilter tells whether there's correlated column in filter (both conditions in PhysicalSelection and LateMaterializationFilterCondition in PhysicalTableScan)
// If true, we will need to revise the dagPB (fill correlated column value in filter) each time call Open().
corColInFilter bool
// corColInAccess tells whether there's correlated column in access conditions.
corColInAccess bool
Expand Down Expand Up @@ -156,6 +157,7 @@ func (e *TableReaderExecutor) Open(ctx context.Context) error {

var err error
if e.corColInFilter {
// If there's correlated column in filter, need to rewrite dagPB
if e.storeType == kv.TiFlash {
execs, err := builder.ConstructTreeBasedDistExec(e.Ctx(), e.tablePlan)
if err != nil {
Expand Down
6 changes: 3 additions & 3 deletions pkg/planner/core/explain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,11 +218,11 @@ func (p *PhysicalTableScan) OperatorInfo(normalized bool) string {
}
if p.SCtx().GetSessionVars().EnableLateMaterialization && len(p.filterCondition) > 0 && p.StoreType == kv.TiFlash {
buffer.WriteString("pushed down filter:")
if len(p.lateMaterializationFilterCondition) > 0 {
if len(p.LateMaterializationFilterCondition) > 0 {
if normalized {
buffer.Write(expression.SortedExplainNormalizedExpressionList(p.lateMaterializationFilterCondition))
buffer.Write(expression.SortedExplainNormalizedExpressionList(p.LateMaterializationFilterCondition))
} else {
buffer.Write(expression.SortedExplainExpressionList(p.SCtx(), p.lateMaterializationFilterCondition))
buffer.Write(expression.SortedExplainExpressionList(p.SCtx(), p.LateMaterializationFilterCondition))
}
} else {
buffer.WriteString("empty")
Expand Down
4 changes: 2 additions & 2 deletions pkg/planner/core/physical_plans.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -829,10 +829,10 @@ type PhysicalTableScan struct {
// AccessCondition is used to calculate range.
AccessCondition []expression.Expression
filterCondition []expression.Expression
// lateMaterializationFilterCondition is used to record the filter conditions
// LateMaterializationFilterCondition is used to record the filter conditions
// that are pushed down to table scan from selection by late materialization.
// TODO: remove this field after we support pushing down selection to coprocessor.
lateMaterializationFilterCondition []expression.Expression
LateMaterializationFilterCondition []expression.Expression

Table *model.TableInfo
Columns []*model.ColumnInfo
Expand Down
8 changes: 4 additions & 4 deletions pkg/planner/core/plan_to_pb.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,9 +248,9 @@ func (p *PhysicalTableScan) ToPB(ctx sessionctx.Context, storeType kv.StoreType)
tsExec.KeepOrder = &keepOrder
tsExec.IsFastScan = &(ctx.GetSessionVars().TiFlashFastScan)

if len(p.lateMaterializationFilterCondition) > 0 {
if len(p.LateMaterializationFilterCondition) > 0 {
client := ctx.GetClient()
conditions, err := expression.ExpressionsToPBList(ctx, p.lateMaterializationFilterCondition, client)
conditions, err := expression.ExpressionsToPBList(ctx, p.LateMaterializationFilterCondition, client)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -287,9 +287,9 @@ func (p *PhysicalTableScan) partitionTableScanToPBForFlash(ctx sessionctx.Contex
telemetry.CurrentTiflashTableScanWithFastScanCount.Inc()
}

if len(p.lateMaterializationFilterCondition) > 0 {
if len(p.LateMaterializationFilterCondition) > 0 {
client := ctx.GetClient()
conditions, err := expression.ExpressionsToPBList(ctx, p.lateMaterializationFilterCondition, client)
conditions, err := expression.ExpressionsToPBList(ctx, p.LateMaterializationFilterCondition, client)
if err != nil {
return nil, err
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/planner/core/tiflash_selection_late_materialization.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,7 +255,7 @@ func predicatePushDownToTableScanImpl(sctx sessionctx.Context, physicalSelection
// remove the pushed down conditions from selection
removeSpecificExprsFromSelection(physicalSelection, selectedConds)
// add the pushed down conditions to table scan
physicalTableScan.lateMaterializationFilterCondition = selectedConds
physicalTableScan.LateMaterializationFilterCondition = selectedConds
// Update the row count of table scan after pushing down the conditions.
physicalTableScan.StatsInfo().RowCount *= selectedSelectivity
}