planner, sessionctx: support alternative logical plans | tidb-test=pr/2752

pkg/executor/test/simpletest/BUILD.bazel

@@ -9,7 +9,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 12,
+    shard_count = 13,
     deps = [
         "//pkg/config",
         "//pkg/errno",

pkg/parser/ast/BUILD.bazel

@@ -45,6 +45,7 @@ go_test(
         "functions_test.go",
         "misc_test.go",
         "procedure_test.go",
+        "stats_test.go",
         "util_test.go",
     ],
     embed = [":ast"],

pkg/planner/BUILD.bazel

@@ -17,13 +17,15 @@ go_library(
         "//pkg/planner/core",
         "//pkg/planner/core/base",
         "//pkg/planner/core/resolve",
+        "//pkg/planner/core/rule",
         "//pkg/planner/indexadvisor",
         "//pkg/planner/planctx",
         "//pkg/planner/property",
         "//pkg/planner/util/debugtrace",
         "//pkg/planner/util/optimizetrace",
         "//pkg/privilege",
         "//pkg/sessionctx",
+        "//pkg/sessionctx/stmtctx",
         "//pkg/sessionctx/variable",
         "//pkg/types",
         "//pkg/util",

pkg/planner/core/casetest/correlated/BUILD.bazel

@@ -8,9 +8,11 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
+    shard_count = 3,
     deps = [
         "//pkg/testkit",
         "//pkg/testkit/testsetup",
+        "@com_github_stretchr_testify//require",
         "@org_uber_go_goleak//:goleak",
     ],
 )

pkg/planner/core/casetest/correlated/correlated_test.go

@@ -15,9 +15,12 @@
 package correlated
 
 import (
+	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/stretchr/testify/require"
 )
 
 func TestCorrelatedSubquery(t *testing.T) {
@@ -71,6 +74,78 @@ WHERE NOT (tlc07c2a51.col_1>=
               HAVING tlc07c2a51.col_6>0)) ;`).Check(testkit.Rows("1", "1", "1", "1", "1", "1", "1", "1", "1", "1"))
 }
 
+func TestNaturalJoinWithCorrelatedSubquery(tt *testing.T) {
+	testkit.RunTestUnderCascades(tt, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
+		tk.MustExec("use test")
+		tk.MustExec("drop table if exists t")
+		tk.MustExec("create table t (a int)")
+		// Keep duplicate and NULL rows so the regression also pins multiplicity
+		// and NULL-handling for the correlated EXISTS predicate.
+		tk.MustExec("insert into t values (1), (1), (2), (null)")
+
+		sql := `select t1.a
+			from t t1 natural join t t2
+			where exists (select 1 from t t3 where t3.a = t1.a)
+			order by t1.a`
+		tk.MustQuery(sql).Check(testkit.Rows("1", "1", "1", "1", "2"))
+
+		if cascades == "on" {
+			return
+		}
+
+		t.Run("AlternativeLogicalPlansChooseApply", func(t *testing.T) {
+			tk.MustExec("use test")
+			tk.MustExec("drop table if exists alt_pick_t1, alt_pick_t2, alt_pick_t3")
+			tk.MustExec("create table alt_pick_t1(a int primary key)")
+			tk.MustExec("create table alt_pick_t2(a int, b int, key idx_a(a))")
+			tk.MustExec("create table alt_pick_t3(a int, c int, key idx_a(a))")
+			tk.MustExec("insert into alt_pick_t1 values (1), (2)")
+
+			vals2 := make([]string, 0, 200)
+			vals3 := make([]string, 0, 200)
+			for i := 0; i < 200; i++ {
+				vals2 = append(vals2, fmt.Sprintf("(%d, %d)", i%100, i))
+				vals3 = append(vals3, fmt.Sprintf("(%d, %d)", i%100, i))
+			}
+			tk.MustExec("insert into alt_pick_t2 values " + strings.Join(vals2, ","))
+			tk.MustExec("insert into alt_pick_t3 values " + strings.Join(vals3, ","))
+			tk.MustExec("analyze table alt_pick_t1, alt_pick_t2, alt_pick_t3")
+
+			tk.MustExec("set @@tidb_opt_enable_alternative_logical_plans=off")
+			sql := "select alt_pick_t1.a, (select count(*) from alt_pick_t2 join alt_pick_t3 on alt_pick_t2.a = alt_pick_t3.a where alt_pick_t2.a = alt_pick_t1.a) as cnt from alt_pick_t1 order by alt_pick_t1.a"
+			explainSQL := "explain format = 'brief' " + sql
+
+			offPlan := tk.MustQuery(explainSQL).Rows()
+			tk.MustQuery(sql).Check(testkit.Rows("1 4", "2 4"))
+			require.False(t, rowsContainText(offPlan, "Apply"), rowsText(offPlan))
+
+			tk.MustExec("set @@tidb_opt_enable_alternative_logical_plans=on")
+			onPlan := tk.MustQuery(explainSQL).Rows()
+			tk.MustQuery(sql).Check(testkit.Rows("1 4", "2 4"))
+			require.True(t, rowsContainText(onPlan, "Apply"), rowsText(onPlan))
+		})
+
+		t.Run("AlternativeLogicalPlansSkipSecondRoundWhenIndexJoinExists", func(t *testing.T) {
+			tk.MustExec("use test")
+			tk.MustExec("set @@tidb_opt_enable_alternative_logical_plans=on")
+			tk.MustExec("drop table if exists alt_skip_t1, alt_skip_t2")
+			tk.MustExec("create table alt_skip_t1(a int primary key)")
+			tk.MustExec("create table alt_skip_t2(a int, b int, key idx_a(a))")
+			tk.MustExec("insert into alt_skip_t1 values (1), (2), (3)")
+			tk.MustExec("insert into alt_skip_t2 values (1, 1), (1, 2), (2, 3), (3, 4)")
+			tk.MustExec("analyze table alt_skip_t1, alt_skip_t2")
+
+			sql := "select alt_skip_t1.a from alt_skip_t1 where exists (select 1 from alt_skip_t2 where alt_skip_t2.a = alt_skip_t1.a and alt_skip_t2.b > 0) order by alt_skip_t1.a"
+			plan := tk.MustQuery("explain format = 'brief' " + sql).Rows()
+			require.False(t, rowsContainText(plan, "Apply"), rowsText(plan))
+			tk.MustQuery(sql).Check(testkit.Rows("1", "2", "3"))
+			stmtCtx := tk.Session().GetSessionVars().StmtCtx
+			require.True(t, stmtCtx.AlternativeLogicalPlanDecorrelatedApply)
+			require.True(t, stmtCtx.AlternativeLogicalPlanSameOrderIndexJoin)
+		})
+	})
+}
+
 func TestWrongDecorrelate(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)
@@ -84,3 +159,26 @@ func TestWrongDecorrelate(t *testing.T) {
 		" 30025.20000000000000000000 60121022342",
 		"X 6.23000000000000000000 60021022342"))
 }
+
+func rowsContainText(rows [][]any, needle string) bool {
+	for _, row := range rows {
+		for _, col := range row {
+			if strings.Contains(fmt.Sprint(col), needle) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func rowsText(rows [][]any) string {
+	lines := make([]string, 0, len(rows))
+	for _, row := range rows {
+		vals := make([]string, 0, len(row))
+		for _, col := range row {
+			vals = append(vals, fmt.Sprint(col))
+		}
+		lines = append(lines, strings.Join(vals, " "))
+	}
+	return strings.Join(lines, "\n")
+}

pkg/planner/core/exhaust_physical_plans.go

@@ -567,6 +567,8 @@ func constructIndexJoin(
 		CompareFilters:   compareFilters,
 		OuterHashKeys:    outerHashKeys,
 		InnerHashKeys:    innerHashKeys,
+		// Only count candidates that keep the original Apply outer/inner order.
+		FromDecorrelatedApply: p.FromDecorrelatedApply && outerIdx == 0,
 	}.Init(p.SCtx(), p.StatsInfo().ScaleByExpectCnt(prop.ExpectedCnt), p.QueryBlockOffset(), chReqProps...)
 	if path != nil {
 		join.IdxColLens = path.IdxColLens

pkg/planner/core/logical_plan_builder.go

@@ -3749,7 +3749,7 @@ func (b *PlanBuilder) pushHintWithoutTableWarning(hint *ast.TableOptimizerHint)
 }
 
 func (b *PlanBuilder) pushTableHints(hints []*ast.TableOptimizerHint, currentLevel int) {
-	hints = b.hintProcessor.GetCurrentStmtHints(hints, currentLevel)
+	hints = b.hintProcessor.GetCurrentStmtHints(hints, currentLevel, b.hintState)
 	sessionVars := b.ctx.GetSessionVars()
 	currentDB := sessionVars.CurrentDB
 	warnHandler := sessionVars.StmtCtx
@@ -4628,7 +4628,7 @@ func (b *PlanBuilder) buildDataSource(ctx context.Context, tn *ast.TableName, as
 				// Because of the nested views, so we should check the left table list in hint when build the data source from the view inside the current view.
 				currentQBNameMap4View[qbName] = viewQBNameHintTable[1:]
 				currentViewHints[qbName] = b.hintProcessor.ViewQBNameToHints[qbName]
-				b.hintProcessor.ViewQBNameUsed[qbName] = struct{}{}
+				b.hintProcessor.MarkViewQBNameUsed(qbName, b.hintState)
 			}
 		}
 		return b.BuildDataSourceFromView(ctx, dbName, tableInfo, currentQBNameMap4View, currentViewHints)
@@ -5109,18 +5109,21 @@ func (b *PlanBuilder) BuildDataSourceFromView(ctx context.Context, dbName pmodel
 
 	hintProcessor.ViewQBNameToTable = qbNameMap4View
 	hintProcessor.ViewQBNameToHints = viewHints
-	hintProcessor.ViewQBNameUsed = make(map[string]struct{})
-	hintProcessor.QBOffsetToHints = currentQbHints
 	hintProcessor.QBNameToSelOffset = currentQbNameMap
+	hintState := hintProcessor.NewBuildState()
+	hintState.QBOffsetToHints = currentQbHints
 
 	originHintProcessor := b.hintProcessor
+	originHintState := b.hintState
 	originPlannerSelectBlockAsName := b.ctx.GetSessionVars().PlannerSelectBlockAsName.Load()
 	b.hintProcessor = hintProcessor
+	b.hintState = hintState
 	newPlannerSelectBlockAsName := make([]ast.HintTable, hintProcessor.MaxSelectStmtOffset()+1)
 	b.ctx.GetSessionVars().PlannerSelectBlockAsName.Store(&newPlannerSelectBlockAsName)
 	defer func() {
-		b.hintProcessor.HandleUnusedViewHints()
+		b.hintProcessor.SetWarns(b.hintProcessor.HandleUnusedViewHints(b.hintState, nil))
 		b.hintProcessor = originHintProcessor
+		b.hintState = originHintState
 		b.ctx.GetSessionVars().PlannerSelectBlockAsName.Store(originPlannerSelectBlockAsName)
 	}()
 	nodeW := resolve.NewNodeWWithCtx(selectNode, b.resolveCtx)

pkg/planner/core/operator/logicalop/logical_join.go

@@ -146,6 +146,11 @@ type LogicalJoin struct {
 
 	// EqualCondOutCnt indicates the estimated count of joined rows after evaluating `EqualConditions`.
 	EqualCondOutCnt float64
+
+	// FromDecorrelatedApply marks joins that come from decorrelating an Apply in the
+	// first logical round. It is only used to decide whether an equivalent same-order
+	// PhysicalIndexJoin candidate has already been generated.
+	FromDecorrelatedApply bool
 }
 
 // Init initializes LogicalJoin.

pkg/planner/core/physical_plans.go

@@ -1714,6 +1714,9 @@ type PhysicalIndexJoin struct {
 	// InnerHashKeys indicates the inner keys used to build hash table during
 	// execution. InnerJoinKeys is the prefix of InnerHashKeys.
 	InnerHashKeys []*expression.Column
+	// FromDecorrelatedApply is true only when this IndexJoin keeps the original
+	// Apply outer/inner order after decorrelation.
+	FromDecorrelatedApply bool
 }
 
 // Clone implements op.PhysicalPlan interface.
@@ -1737,6 +1740,7 @@ func (p *PhysicalIndexJoin) Clone(newCtx base.PlanContext) (base.PhysicalPlan, e
 	cloned.CompareFilters = p.CompareFilters.Copy()
 	cloned.OuterHashKeys = util.CloneCols(p.OuterHashKeys)
 	cloned.InnerHashKeys = util.CloneCols(p.InnerHashKeys)
+	cloned.FromDecorrelatedApply = p.FromDecorrelatedApply
 	return cloned, nil
 }
 

pkg/planner/core/planbuilder.go

@@ -256,7 +256,10 @@ type PlanBuilder struct {
 	//   finish building the subquery or CTE.
 	handleHelper *handleColHelper
 
+	// read-only meta derived from ast node.
 	hintProcessor *hint.QBHintHandler
+	// mutable state of QBHint when building.
+	hintState *hint.QBHintBuildState
 	// qbOffset is the offsets of current processing select stmts.
 	qbOffset []int
 
@@ -405,6 +408,11 @@ func GetDBTableInfo(visitInfo []visitInfo) []stmtctx.TableEntry {
 	return tables
 }
 
+// GetHintState gets the HintState from the PlanBuilder.
+func (b *PlanBuilder) GetHintState() *hint.QBHintBuildState {
+	return b.hintState
+}
+
 // GetOptFlag gets the OptFlag of the PlanBuilder.
 func (b *PlanBuilder) GetOptFlag() uint64 {
 	if b.isSampling {
@@ -482,6 +490,9 @@ func (b *PlanBuilder) Init(sctx base.PlanContext, is infoschema.InfoSchema, proc
 	b.ctx = sctx
 	b.is = is
 	b.hintProcessor = processor
+	if processor != nil {
+		b.hintState = processor.NewBuildState()
+	}
 	b.isForUpdateRead = sctx.GetSessionVars().IsPessimisticReadConsistency()
 	b.noDecorrelate = sctx.GetSessionVars().EnableNoDecorrelateInSelect
 	if savedBlockNames == nil {
@@ -523,6 +534,14 @@ func (b *PlanBuilder) ResetForReuse() *PlanBuilder {
 	return b
 }
 
+// HandleUnusedViewHints appends warnings for unused view hints in the current build.
+func (b *PlanBuilder) HandleUnusedViewHints() {
+	if b.hintProcessor == nil {
+		return
+	}
+	b.hintProcessor.SetWarns(b.hintProcessor.HandleUnusedViewHints(b.hintState, nil))
+}
+
 // Build builds the ast node to a Plan.
 func (b *PlanBuilder) Build(ctx context.Context, node *resolve.NodeW) (base.Plan, error) {
 	// Build might be called recursively, right now they all share the same resolve

pkg/planner/core/rule_decorrelate.go

@@ -145,6 +145,10 @@ func (s *DecorrelateSolver) Optimize(ctx context.Context, p base.LogicalPlan, op
 			join := &apply.LogicalJoin
 			join.SetSelf(join)
 			join.SetTP(plancodec.TypeJoin)
+			if p.SCtx().GetSessionVars().EnableAlternativeLogicalPlans {
+				p.SCtx().GetSessionVars().StmtCtx.MarkAlternativeLogicalPlanDecorrelatedApply()
+				join.FromDecorrelatedApply = true
+			}
 			p = join
 			appendApplySimplifiedTraceStep(apply, join, opt)
 		} else if apply.NoDecorrelate {

pkg/planner/core/task.go

@@ -167,6 +167,9 @@ func (p *PhysicalIndexHashJoin) Attach2Task(tasks ...base.Task) base.Task {
 // Attach2Task implements PhysicalPlan interface.
 func (p *PhysicalIndexJoin) Attach2Task(tasks ...base.Task) base.Task {
 	outerTask := tasks[1-p.InnerChildIdx].ConvertToRootTask(p.SCtx())
+	if p.FromDecorrelatedApply {
+		p.SCtx().GetSessionVars().StmtCtx.MarkAlternativeLogicalPlanSameOrderIndexJoin()
+	}
 	if p.InnerChildIdx == 1 {
 		p.SetChildren(outerTask.Plan(), p.innerPlan)
 	} else {