executor: fill correlated column value in late materialization filter…

… conditions (pingcap#49244) (pingcap#49431)

close pingcap#49241

executor/builder.go

@@ -3103,13 +3103,18 @@ func constructDAGReq(ctx sessionctx.Context, plans []plannercore.PhysicalPlan, s
 
 func (b *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) {
+ 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
+ }
  }
  }
  }

executor/table_reader.go

@@ -110,7 +110,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
@@ -154,6 +155,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 := constructDistExecForTiFlash(e.ctx, e.tablePlan)
  if err != nil {

planner/core/explain.go

@@ -216,11 +216,11 @@ func (p *PhysicalTableScan) OperatorInfo(normalized bool) string {
  }
  if p.ctx.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.lateMaterializationFilterCondition))
+ buffer.Write(expression.SortedExplainExpressionList(p.LateMaterializationFilterCondition))
  }
  } else {
  buffer.WriteString("empty")

planner/core/physical_plans.go

@@ -808,10 +808,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

planner/core/plan_to_pb.go

@@ -228,10 +228,10 @@ 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 {
  sc := ctx.GetSessionVars().StmtCtx
  client := ctx.GetClient()
- conditions, err := expression.ExpressionsToPBList(sc, p.lateMaterializationFilterCondition, client)
+ conditions, err := expression.ExpressionsToPBList(sc, p.LateMaterializationFilterCondition, client)
  if err != nil {
  return nil, err
  }
@@ -261,10 +261,10 @@ func (p *PhysicalTableScan) partitionTableScanToPBForFlash(ctx sessionctx.Contex
  telemetry.CurrentTiflashTableScanWithFastScanCount.Inc()
  }
 
- if len(p.lateMaterializationFilterCondition) > 0 {
+ if len(p.LateMaterializationFilterCondition) > 0 {
  sc := ctx.GetSessionVars().StmtCtx
  client := ctx.GetClient()
- conditions, err := expression.ExpressionsToPBList(sc, p.lateMaterializationFilterCondition, client)
+ conditions, err := expression.ExpressionsToPBList(sc, p.LateMaterializationFilterCondition, client)
  if err != nil {
  return nil, err
  }

planner/core/tiflash_selection_late_materialization.go

@@ -249,7 +249,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.stats.RowCount *= selectedSelectivity
 }