executor: support index_look_up_merge_join to speed up index_look_up_join

executor/builder.go

@@ -1529,33 +1529,90 @@ func (b *executorBuilder) buildIndexLookUpJoin(v *plannercore.PhysicalIndexJoin)
 	if defaultValues == nil {
 		defaultValues = make([]types.Datum, len(innerTypes))
 	}
+	outerKeyCols := make([]int, len(v.OuterJoinKeys))
+	for i := 0; i < len(v.OuterJoinKeys); i++ {
+		outerKeyCols[i] = v.OuterJoinKeys[i].Index
+	}
+	innerKeyCols := make([]int, len(v.InnerJoinKeys))
+	for i := 0; i < len(v.InnerJoinKeys); i++ {
+		innerKeyCols[i] = v.InnerJoinKeys[i].Index
+	}
+	metrics.ExecutorCounter.WithLabelValues("IndexLookUpJoin").Inc()
+
+	// If the inner join keys are the suffix set of the index, then return the IndexLookUpMergeJoin
+	var isIndexScan bool
+	var is *plannercore.PhysicalIndexScan
+	if ir, ok := innerPlan.(*plannercore.PhysicalIndexReader); ok {
+		is, isIndexScan = ir.IndexPlans[0].(*plannercore.PhysicalIndexScan)
+	}
+	if ilr, ok := innerPlan.(*plannercore.PhysicalIndexLookUpReader); ok {
+		is, isIndexScan = ilr.IndexPlans[0].(*plannercore.PhysicalIndexScan)
+	}
+	if isIndexScan {
+		idx := is.Index.Columns
+		isInnerKeysPrefix := true
+		for i, innerKey := range v.InnerJoinKeys {
+			if innerKey.ColName != idx[i].Name {
+				isInnerKeysPrefix = false
+				break
+			}
+		}
+		if isInnerKeysPrefix {
+			is.KeepOrder = true
+			// enforceOuterOrder means the outerJoinKeys must be the prefix of the prop items of PhysicalIndexJoin
+			enforceOuterOrder := v.KeepOuterOrder && len(v.OuterJoinKeys) < len(v.PropItems)
+			compareFuncs := make([]expression.CompareFunc, 0, len(v.OuterJoinKeys))
+			outerCompareFuncs := make([]expression.CompareFunc, 0, len(v.OuterJoinKeys))
+			for i := range v.OuterJoinKeys {
+				if enforceOuterOrder && v.PropItems[i].Col.ColName != v.OuterJoinKeys[i].ColName {
+					enforceOuterOrder = false
+				}
+				compareFuncs = append(compareFuncs, expression.GetCmpFunction(v.OuterJoinKeys[i], v.InnerJoinKeys[i]))
+				outerCompareFuncs = append(outerCompareFuncs, expression.GetCmpFunction(v.OuterJoinKeys[i], v.OuterJoinKeys[i]))
+			}
+			return &IndexLookUpMergeJoin{
+				baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ExplainID(), outerExec),
+				outerMergeCtx: outerMergeCtx{
+					rowTypes:          outerTypes,
+					filter:            outerFilter,
+					joinKeys:          v.OuterJoinKeys,
+					keyCols:           outerKeyCols,
+					enforceOuterOrder: enforceOuterOrder,
+					compareFuncs:      outerCompareFuncs,
+				},
+				innerMergeCtx: innerMergeCtx{
+					readerBuilder: &dataReaderBuilder{innerPlan, b},
+					rowTypes:      innerTypes,
+					joinKeys:      v.InnerJoinKeys,
+					keyCols:       innerKeyCols,
+					compareFuncs:  compareFuncs,
+				},
+				workerWg:      new(sync.WaitGroup),
+				joiner:        newJoiner(b.ctx, v.JoinType, v.OuterIndex == 1, defaultValues, v.OtherConditions, leftTypes, rightTypes),
+				indexRanges:   v.Ranges,
+				keyOff2IdxOff: v.KeyOff2IdxOff,
+			}
+		}
+	}
 	e := &IndexLookUpJoin{
 		baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ExplainID(), outerExec),
 		outerCtx: outerCtx{
 			rowTypes: outerTypes,
 			filter:   outerFilter,
+			keyCols:  outerKeyCols,
 		},
 		innerCtx: innerCtx{
 			readerBuilder: &dataReaderBuilder{innerPlan, b},
 			rowTypes:      innerTypes,
+			keyCols:       innerKeyCols,
 		},
 		workerWg:      new(sync.WaitGroup),
 		joiner:        newJoiner(b.ctx, v.JoinType, v.OuterIndex == 1, defaultValues, v.OtherConditions, leftTypes, rightTypes),
 		indexRanges:   v.Ranges,
 		keyOff2IdxOff: v.KeyOff2IdxOff,
 	}
-	outerKeyCols := make([]int, len(v.OuterJoinKeys))
-	for i := 0; i < len(v.OuterJoinKeys); i++ {
-		outerKeyCols[i] = v.OuterJoinKeys[i].Index
-	}
-	e.outerCtx.keyCols = outerKeyCols
-	innerKeyCols := make([]int, len(v.InnerJoinKeys))
-	for i := 0; i < len(v.InnerJoinKeys); i++ {
-		innerKeyCols[i] = v.InnerJoinKeys[i].Index
-	}
-	e.innerCtx.keyCols = innerKeyCols
 	e.joinResult = e.newFirstChunk()
-	metrics.ExecutorCounter.WithLabelValues("IndexLookUpJoin").Inc()
+
 	return e
 }