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: support partition pruning for IndexJoin inner table #19990

Merged
merged 8 commits into from
Sep 29, 2020
Merged

executor: support partition pruning for IndexJoin inner table #19990

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
44e6f61
support runtime prune for index join inner table
Sep 15, 2020
c20e2db
fix make dev
Sep 15, 2020
01c7d8e
divide range into each partition according to datum
Sep 18, 2020
10ee034
fix ci
Sep 18, 2020
8723aba
clean code
Sep 24, 2020
583a96b
fix test
Sep 28, 2020
53b182a
Merge branch 'master' into innerprune
Sep 28, 2020
2c97de9
Merge branch 'master' into innerprune
ti-srebot Sep 29, 2020
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
223 changes: 206 additions & 17 deletions executor/builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2563,7 +2563,7 @@ func (b *executorBuilder) buildTableReader(v *plannercore.PhysicalTableReader) E
}
}

nextPartition := nextPartitionForTableReader{ret}
nextPartition := nextPartitionForTableReader{exec: ret}
exec, err := buildPartitionTable(b, ts.Table, &v.PartitionInfo, ret, nextPartition)
if err != nil {
b.err = err
Expand All @@ -2590,6 +2590,111 @@ func buildPartitionTable(b *executorBuilder, tblInfo *model.TableInfo, partition
}, nil
}

func buildIndexRangeForEachPartition(ctx sessionctx.Context, usedPartitions []table.PhysicalTable, contentPos []int64,
lookUpContent []*indexJoinLookUpContent, indexRanges []*ranger.Range, keyOff2IdxOff []int, cwc *plannercore.ColWithCmpFuncManager) (map[int64][]*ranger.Range, error) {
contentBucket := make(map[int64][]*indexJoinLookUpContent)
for _, p := range usedPartitions {
contentBucket[p.GetPhysicalID()] = make([]*indexJoinLookUpContent, 0, 8)
}
for i, pos := range contentPos {
if _, ok := contentBucket[pos]; ok {
contentBucket[pos] = append(contentBucket[pos], lookUpContent[i])
}
}
nextRange := make(map[int64][]*ranger.Range)
for _, p := range usedPartitions {
ranges, err := buildRangesForIndexJoin(ctx, contentBucket[p.GetPhysicalID()], indexRanges, keyOff2IdxOff, cwc)
if err != nil {
return nil, err
}
nextRange[p.GetPhysicalID()] = ranges
}
return nextRange, nil
}

func buildKVRangeForEachPartition(ctx sessionctx.Context, usedPartitions []table.PhysicalTable, contentPos []int64, isCommonHandle bool,
lookUpContents []*indexJoinLookUpContent, indexRanges []*ranger.Range, keyOff2IdxOff []int, cwc *plannercore.ColWithCmpFuncManager) (map[int64]kvRangeBuilder, error) {
rangeBuilders := make(map[int64]kvRangeBuilder)
contentBucket := make(map[int64][]*indexJoinLookUpContent)
for _, p := range usedPartitions {
contentBucket[p.GetPhysicalID()] = make([]*indexJoinLookUpContent, 0, 8)
}
for i, pos := range contentPos {
if _, ok := contentBucket[pos]; ok {
contentBucket[pos] = append(contentBucket[pos], lookUpContents[i])
}
}
for _, p := range usedPartitions {
if isCommonHandle {
rangeBuilders[p.GetPhysicalID()] = kvRangeBuilderFromFunc(func(pid int64) ([]kv.KeyRange, error) {
return buildKvRangesForIndexJoin(ctx, pid, -1, contentBucket[pid], indexRanges, keyOff2IdxOff, cwc)
})
} else {
handles := make([]kv.Handle, 0, len(contentBucket[p.GetPhysicalID()]))
for _, content := range contentBucket[p.GetPhysicalID()] {
handle := kv.IntHandle(content.keys[0].GetInt64())
handles = append(handles, handle)
}
rangeBuilders[p.GetPhysicalID()] = kvRangeBuilderFromHandles(handles)
}
}
return rangeBuilders, nil
}

func prunePartitionForInnerExecutor(ctx sessionctx.Context, tbl table.Table, schema *expression.Schema, partitionInfo *plannercore.PartitionInfo,
lookUpContent []*indexJoinLookUpContent) (usedPartition []table.PhysicalTable, canPrune bool, contentPos []int64, err error) {
partitionTbl := tbl.(table.PartitionedTable)
locateKey := make([]types.Datum, schema.Len())
// TODO: condition based pruning can be do in advance.
condPruneResult, err := partitionPruning(ctx, partitionTbl, partitionInfo.PruningConds, partitionInfo.PartitionNames, partitionInfo.Columns, partitionInfo.ColumnNames)
if err != nil {
return nil, false, nil, err
}

// check whether can runtime prune.
type partitionExpr interface {
PartitionExpr() (*tables.PartitionExpr, error)
}
pe, err := tbl.(partitionExpr).PartitionExpr()
if err != nil {
return nil, false, nil, err
}
offsetMap := make(map[int]bool)
for _, offset := range lookUpContent[0].keyCols {
offsetMap[offset] = true
}
for _, offset := range pe.ColumnOffset {
if _, ok := offsetMap[offset]; !ok {
logutil.BgLogger().Warn("can not runtime prune in index join")
return condPruneResult, false, nil, nil
}
}

partitions := make(map[int64]table.PhysicalTable)
contentPos = make([]int64, len(lookUpContent))
for idx, content := range lookUpContent {
for i, date := range content.keys {
locateKey[content.keyCols[i]] = date
}
p, err := partitionTbl.GetPartitionByRow(ctx, locateKey)
if err != nil {
return nil, false, nil, err
}
if _, ok := partitions[p.GetPhysicalID()]; !ok {
partitions[p.GetPhysicalID()] = p
}
contentPos[idx] = p.GetPhysicalID()
}

usedPartition = make([]table.PhysicalTable, 0, len(partitions))
for _, p := range condPruneResult {
if _, ok := partitions[p.GetPhysicalID()]; ok {
usedPartition = append(usedPartition, p)
}
}
return usedPartition, true, contentPos, nil
}

func buildNoRangeIndexReader(b *executorBuilder, v *plannercore.PhysicalIndexReader) (*IndexReaderExecutor, error) {
dagReq, streaming, err := b.constructDAGReq(v.IndexPlans, kv.TiKV)
if err != nil {
Expand Down Expand Up @@ -3038,10 +3143,33 @@ func (builder *dataReaderBuilder) buildTableReaderForIndexJoin(ctx context.Conte
e.kvRangeBuilder = kvRangeBuilderFromFunc(func(pid int64) ([]kv.KeyRange, error) {
return buildKvRangesForIndexJoin(e.ctx, pid, -1, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
})
nextPartition := nextPartitionForTableReader{e}
return buildPartitionTable(builder.executorBuilder, tbInfo, &v.PartitionInfo, e, nextPartition)
nextPartition := nextPartitionForTableReader{exec: e, innerPartitionInfo: &innerPartitionInfo{isFullPartition: true}}
tbl, _ := builder.executorBuilder.is.TableByID(tbInfo.ID)
usedPartition, canPrune, contentPos, err := prunePartitionForInnerExecutor(builder.executorBuilder.ctx, tbl, e.Schema(), &v.PartitionInfo, lookUpContents)
if err != nil {
return nil, err
}
if len(usedPartition) != 0 {
if canPrune {
rangeBuilders, err := buildKVRangeForEachPartition(e.ctx, usedPartition, contentPos, v.IsCommonHandle, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
if err != nil {
return nil, err
}
nextPartition.rangeBuilders = rangeBuilders
nextPartition.isFullPartition = false
}
partitionExec := &PartitionTableExecutor{
baseExecutor: *e.base(),
partitions: usedPartition,
nextPartition: nextPartition,
}
return partitionExec, nil
}
ret := &TableDualExec{baseExecutor: *e.base()}
return ret, err
}
handles := make([]kv.Handle, 0, len(lookUpContents))
validLookUpContents := make([]*indexJoinLookUpContent, 0, len(lookUpContents))
for _, content := range lookUpContents {
isValidHandle := true
handle := kv.IntHandle(content.keys[0].GetInt64())
Expand All @@ -3053,6 +3181,7 @@ func (builder *dataReaderBuilder) buildTableReaderForIndexJoin(ctx context.Conte
}
if isValidHandle {
handles = append(handles, handle)
validLookUpContents = append(validLookUpContents, content)
}
}

Expand All @@ -3062,10 +3191,31 @@ func (builder *dataReaderBuilder) buildTableReaderForIndexJoin(ctx context.Conte
if !builder.ctx.GetSessionVars().UseDynamicPartitionPrune() {
return builder.buildTableReaderFromHandles(ctx, e, handles, canReorderHandles)
}

e.kvRangeBuilder = kvRangeBuilderFromHandles(handles)
nextPartition := nextPartitionForTableReader{e}
return buildPartitionTable(builder.executorBuilder, tbInfo, &v.PartitionInfo, e, nextPartition)
nextPartition := nextPartitionForTableReader{exec: e, innerPartitionInfo: &innerPartitionInfo{isFullPartition: true}}
tbl, _ := builder.executorBuilder.is.TableByID(tbInfo.ID)
usedPartition, canPrune, contentPos, err := prunePartitionForInnerExecutor(builder.executorBuilder.ctx, tbl, e.Schema(), &v.PartitionInfo, validLookUpContents)
if err != nil {
return nil, err
}
if len(usedPartition) != 0 {
if canPrune {
rangeBuilders, err := buildKVRangeForEachPartition(e.ctx, usedPartition, contentPos, v.IsCommonHandle, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
if err != nil {
return nil, err
}
nextPartition.rangeBuilders = rangeBuilders
nextPartition.isFullPartition = false
}
partitionExec := &PartitionTableExecutor{
baseExecutor: *e.base(),
partitions: usedPartition,
nextPartition: nextPartition,
}
return partitionExec, nil
}
ret := &TableDualExec{baseExecutor: *e.base()}
return ret, err
}

type kvRangeBuilderFromFunc func(pid int64) ([]kv.KeyRange, error)
Expand Down Expand Up @@ -3148,15 +3298,35 @@ func (builder *dataReaderBuilder) buildIndexReaderForIndexJoin(ctx context.Conte
return e, err
}

e.ranges, err = buildRangesForIndexJoin(e.ctx, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
nextPartition := nextPartitionForIndexReader{exec: e, innerPartitionInfo: &innerPartitionInfo{isFullPartition: true}}
tbl, _ := builder.executorBuilder.is.TableByID(tbInfo.ID)
usedPartition, canPrune, contentPos, err := prunePartitionForInnerExecutor(builder.executorBuilder.ctx, tbl, e.Schema(), &v.PartitionInfo, lookUpContents)
if err != nil {
return nil, err
}
nextPartition := nextPartitionForIndexReader{exec: e}
ret, err := buildPartitionTable(builder.executorBuilder, tbInfo, &v.PartitionInfo, e, nextPartition)
if err != nil {
return nil, err
if len(usedPartition) != 0 {
if canPrune {
rangeMap, err := buildIndexRangeForEachPartition(e.ctx, usedPartition, contentPos, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
if err != nil {
return nil, err
}
nextPartition.isFullPartition = false
nextPartition.nextRange = rangeMap
} else {
e.ranges, err = buildRangesForIndexJoin(e.ctx, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
if err != nil {
return nil, err
}
}
partitionExec := &PartitionTableExecutor{
baseExecutor: *e.base(),
partitions: usedPartition,
nextPartition: nextPartition,
}
err = partitionExec.Open(ctx)
return partitionExec, err
}
ret := &TableDualExec{baseExecutor: *e.base()}
err = ret.Open(ctx)
return ret, err
}
Expand All @@ -3177,16 +3347,35 @@ func (builder *dataReaderBuilder) buildIndexLookUpReaderForIndexJoin(ctx context
err = e.open(ctx)
return e, err
}

e.ranges, err = buildRangesForIndexJoin(e.ctx, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
nextPartition := nextPartitionForIndexLookUp{exec: e, innerPartitionInfo: &innerPartitionInfo{isFullPartition: true}}
tbl, _ := builder.executorBuilder.is.TableByID(tbInfo.ID)
usedPartition, canPrune, contentPos, err := prunePartitionForInnerExecutor(builder.executorBuilder.ctx, tbl, e.Schema(), &v.PartitionInfo, lookUpContents)
if err != nil {
return nil, err
}
nextPartition := nextPartitionForIndexLookUp{exec: e}
ret, err := buildPartitionTable(builder.executorBuilder, tbInfo, &v.PartitionInfo, e, nextPartition)
if err != nil {
return nil, err
if len(usedPartition) != 0 {
if canPrune {
rangeMap, err := buildIndexRangeForEachPartition(e.ctx, usedPartition, contentPos, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
if err != nil {
return nil, err
}
nextPartition.isFullPartition = false
nextPartition.nextRange = rangeMap
} else {
e.ranges, err = buildRangesForIndexJoin(e.ctx, lookUpContents, indexRanges, keyOff2IdxOff, cwc)
if err != nil {
return nil, err
}
}
partitionExec := &PartitionTableExecutor{
baseExecutor: *e.base(),
partitions: usedPartition,
nextPartition: nextPartition,
}
err = partitionExec.Open(ctx)
return partitionExec, err
}
ret := &TableDualExec{baseExecutor: *e.base()}
err = ret.Open(ctx)
return ret, err
}
Expand Down
7 changes: 4 additions & 3 deletions executor/index_lookup_join.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -492,8 +492,9 @@ func (iw *innerWorker) run(ctx context.Context, wg *sync.WaitGroup) {
}

type indexJoinLookUpContent struct {
keys []types.Datum
row chunk.Row
keys []types.Datum
row chunk.Row
keyCols []int
}

func (iw *innerWorker) handleTask(ctx context.Context, task *lookUpJoinTask) error {
Expand Down Expand Up @@ -558,7 +559,7 @@ func (iw *innerWorker) constructLookupContent(task *lookUpJoinTask) ([]*indexJoi
// dLookUpKey is sorted and deduplicated at sortAndDedupLookUpContents.
// So we don't need to do it here.
}
lookUpContents = append(lookUpContents, &indexJoinLookUpContent{keys: dLookUpKey, row: chk.GetRow(rowIdx)})
lookUpContents = append(lookUpContents, &indexJoinLookUpContent{keys: dLookUpKey, row: chk.GetRow(rowIdx), keyCols: iw.keyCols})
}
}

Expand Down
38 changes: 37 additions & 1 deletion executor/partition_table.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
plannercore "github.com/pingcap/tidb/planner/core"
"github.com/pingcap/tidb/table"
"github.com/pingcap/tidb/util/chunk"
"github.com/pingcap/tidb/util/ranger"
"github.com/pingcap/tipb/go-tipb"
)

Expand All @@ -40,36 +41,71 @@ type nextPartition interface {
nextPartition(context.Context, table.PhysicalTable) (Executor, error)
}

type innerPartitionInfo struct {
isFullPartition bool
nextRange map[int64][]*ranger.Range
}

type innerNextPartition interface {
nextPartition
GetInnerPartitionInfo() *innerPartitionInfo
}

type nextPartitionForTableReader struct {
exec *TableReaderExecutor
*innerPartitionInfo
rangeBuilders map[int64]kvRangeBuilder
exec *TableReaderExecutor
}

func (n nextPartitionForTableReader) GetInnerPartitionInfo() *innerPartitionInfo {
return n.innerPartitionInfo
}

func (n nextPartitionForTableReader) nextPartition(ctx context.Context, tbl table.PhysicalTable) (Executor, error) {
n.exec.table = tbl
n.exec.kvRanges = n.exec.kvRanges[:0]
if n.innerPartitionInfo != nil && !n.isFullPartition {
n.exec.kvRangeBuilder = n.rangeBuilders[tbl.GetPhysicalID()]
}
if err := updateDAGRequestTableID(ctx, n.exec.dagPB, tbl.Meta().ID, tbl.GetPhysicalID()); err != nil {
return nil, err
}
return n.exec, nil
}

type nextPartitionForIndexLookUp struct {
*innerPartitionInfo
exec *IndexLookUpExecutor
}

func (n nextPartitionForIndexLookUp) GetInnerPartitionInfo() *innerPartitionInfo {
return n.innerPartitionInfo
}

func (n nextPartitionForIndexLookUp) nextPartition(ctx context.Context, tbl table.PhysicalTable) (Executor, error) {
n.exec.table = tbl
if n.innerPartitionInfo != nil && !n.isFullPartition {
n.exec.ranges = n.nextRange[tbl.GetPhysicalID()]
}
return n.exec, nil
}

type nextPartitionForIndexReader struct {
*innerPartitionInfo
exec *IndexReaderExecutor
}

func (n nextPartitionForIndexReader) GetInnerPartitionInfo() *innerPartitionInfo {
return n.innerPartitionInfo
}

func (n nextPartitionForIndexReader) nextPartition(ctx context.Context, tbl table.PhysicalTable) (Executor, error) {
exec := n.exec
exec.table = tbl
exec.physicalTableID = tbl.GetPhysicalID()
if n.innerPartitionInfo != nil && !n.isFullPartition {
exec.ranges = n.nextRange[tbl.GetPhysicalID()]
}
return exec, nil
}

Expand Down
Loading