pkg/planner: add order-aware logical join reorder rule

pkg/planner/core/BUILD.bazel

@@ -60,6 +60,7 @@ go_library(
         "rule_join_reorder.go",
         "rule_join_reorder_dp.go",
         "rule_join_reorder_greedy.go",
+        "rule_order_aware_join_reorder.go",
         "rule_outer_to_inner_join.go",
         "rule_predicate_push_down.go",
         "rule_push_down_sequence.go",

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

@@ -20,7 +20,7 @@ go_test(
     ],
     data = glob(["testdata/**"]),
     flaky = True,
-    shard_count = 28,
+    shard_count = 30,
     deps = [
         "//pkg/config",
         "//pkg/domain",

pkg/planner/core/casetest/rule/main_test.go

@@ -36,6 +36,7 @@ func TestMain(m *testing.M) {
 	testDataMap.LoadTestSuiteData("testdata", "predicate_simplification", true)
 	testDataMap.LoadTestSuiteData("testdata", "outer_to_semi_join_suite", true)
 	testDataMap.LoadTestSuiteData("testdata", "cdc_join_reorder_suite", true)
+	testDataMap.LoadTestSuiteData("testdata", "order_aware_join_reorder_suite", true)
 
 	opts := []goleak.Option{
 		goleak.IgnoreTopFunction("github.com/golang/glog.(*fileSink).flushDaemon"),
@@ -82,3 +83,7 @@ func GetOuterToSemiJoinSuiteData() testdata.TestData {
 func GetCDCJoinReorderSuiteData() testdata.TestData {
 	return testDataMap["cdc_join_reorder_suite"]
 }
+
+func GetOrderAwareJoinReorderSuiteData() testdata.TestData {
+	return testDataMap["order_aware_join_reorder_suite"]
+}

pkg/planner/core/casetest/rule/rule_cdc_join_reorder_test.go

@@ -24,6 +24,56 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func prepareOrderAwareJoinReorderTables(tk *testkit.TestKit) {
+	tk.MustExec("use test")
+	tk.MustExec("set @@tidb_opt_enable_alternative_logical_plans = 1")
+	tk.MustExec("drop table if exists t6, t7, t8, t9")
+	tk.MustExec("create table t6(id int not null, category varchar(20), payload int, key idx_category_id_payload(category, id, payload))")
+	tk.MustExec("create table t7(id int not null primary key, payload int)")
+	tk.MustExec("create table t8(id int not null primary key, payload int)")
+	tk.MustExec("create table t9(id int not null primary key, payload int)")
+	tk.MustExec("insert into t6 values (1,'hot',10),(2,'hot',20),(3,'cold',30),(4,'hot',40)")
+	tk.MustExec("insert into t7 values (1,100),(2,200),(4,400)")
+	tk.MustExec("insert into t8 values (1,1000),(2,2000),(4,4000)")
+	tk.MustExec("insert into t9 values (1,10000),(2,20000),(4,40000)")
+	tk.MustExec("analyze table t6 all columns")
+	tk.MustExec("analyze table t7 all columns")
+	tk.MustExec("analyze table t8 all columns")
+	tk.MustExec("analyze table t9 all columns")
+}
+
+func prepareOrderAwareAlternativeRoundTables(tk *testkit.TestKit) {
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists oa_order_t1, oa_order_t2, oa_order_t3, oa_order_t4")
+	tk.MustExec("create table oa_order_t1(id int not null primary key, category varchar(20), created_at int, key idx_category_created(category, created_at, id))")
+	tk.MustExec("create table oa_order_t2(id int not null primary key, t1_id int not null, key idx_t1_id(t1_id))")
+	tk.MustExec("create table oa_order_t3(id int not null primary key, t2_id int not null, key idx_t2_id(t2_id))")
+	tk.MustExec("create table oa_order_t4(id int not null primary key, t3_id int not null, payload int, key idx_payload_t3(payload, t3_id))")
+
+	oaOrderT1Rows := make([]string, 0, 5000)
+	oaOrderT2Rows := make([]string, 0, 5000)
+	oaOrderT3Rows := make([]string, 0, 5000)
+	oaOrderT4Rows := make([]string, 0, 5000)
+	for i := 1; i <= 5000; i++ {
+		oaOrderT1Rows = append(oaOrderT1Rows, fmt.Sprintf("(%d,'hot',%d)", i, i))
+		oaOrderT2Rows = append(oaOrderT2Rows, fmt.Sprintf("(%d,%d)", i, i))
+		oaOrderT3Rows = append(oaOrderT3Rows, fmt.Sprintf("(%d,%d)", i, i))
+		payload := 0
+		if i%10 == 0 {
+			payload = 1
+		}
+		oaOrderT4Rows = append(oaOrderT4Rows, fmt.Sprintf("(%d,%d,%d)", i, i, payload))
+	}
+	tk.MustExec("insert into oa_order_t1 values " + strings.Join(oaOrderT1Rows, ","))
+	tk.MustExec("insert into oa_order_t2 values " + strings.Join(oaOrderT2Rows, ","))
+	tk.MustExec("insert into oa_order_t3 values " + strings.Join(oaOrderT3Rows, ","))
+	tk.MustExec("insert into oa_order_t4 values " + strings.Join(oaOrderT4Rows, ","))
+	tk.MustExec("analyze table oa_order_t1 all columns")
+	tk.MustExec("analyze table oa_order_t2 all columns")
+	tk.MustExec("analyze table oa_order_t3 all columns")
+	tk.MustExec("analyze table oa_order_t4 all columns")
+}
+
 func TestCDCJoinReorder(tt *testing.T) {
 	testkit.RunTestUnderCascades(tt, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
 		tk.MustExec("use test")
@@ -64,8 +114,6 @@ func TestCDCJoinReorder(tt *testing.T) {
 
 		// Phase 2: Enable CD-C algorithm, then verify both the plan and the
 		// result correctness for every case.
-		tk.MustExec("set @@tidb_opt_enable_advanced_join_reorder = 1")
-
 		for i, sql := range input {
 			testdata.OnRecord(func() {
 				output[i].SQL = sql
@@ -104,8 +152,6 @@ func TestJoinReorderPushSelection(tt *testing.T) {
 		tk.MustExec("analyze table t4 all columns")
 		tk.MustExec("analyze table t5 all columns")
 
-		tk.MustExec("set @@tidb_opt_enable_advanced_join_reorder = 1")
-
 		var input []string
 		var output []struct {
 			SQL  string
@@ -171,7 +217,6 @@ func TestDPJoinReorder(tt *testing.T) {
 		tk.MustExec("analyze table t3 all columns")
 		tk.MustExec("analyze table t4 all columns")
 		tk.MustExec("analyze table t5 all columns")
-
 		var input []string
 		var output []struct {
 			SQL    string
@@ -206,6 +251,104 @@ func TestDPJoinReorder(tt *testing.T) {
 	})
 }
 
+func TestOrderAwareJoinReorderPushSelection(tt *testing.T) {
+	testkit.RunTestUnderCascades(tt, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
+		prepareOrderAwareJoinReorderTables(tk)
+		tk.MustExec("set @@tidb_opt_join_reorder_through_sel = 1")
+
+		var input []string
+		var output []struct {
+			SQL  string
+			Plan []string
+		}
+		suite := GetOrderAwareJoinReorderSuiteData()
+		suite.LoadTestCasesByName("TestOrderAwareJoinReorderPushSelection", t, &input, &output, cascades, caller)
+
+		for i, sql := range input {
+			testdata.OnRecord(func() {
+				if i >= len(output) {
+					output = append(output, struct {
+						SQL  string
+						Plan []string
+					}{})
+				}
+				output[i].SQL = sql
+				output[i].Plan = testdata.ConvertRowsToStrings(tk.MustQuery(sql).Rows())
+			})
+			require.Lessf(t, i, len(output), "missing expected output for case[%d], sql: %s", i, sql)
+			require.Equalf(t, sql, output[i].SQL, "input/output SQL mismatch at case[%d]", i)
+			tk.MustQuery(sql).Check(testkit.Rows(output[i].Plan...))
+			require.NotContains(t, strings.Join(testdata.ConvertRowsToStrings(tk.MustQuery("show warnings").Rows()), "\n"),
+				"leading hint is inapplicable")
+		}
+	})
+}
+
+func TestOrderAwareJoinReorderAlternativeRound(tt *testing.T) {
+	testkit.RunTestUnderCascades(tt, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
+		prepareOrderAwareAlternativeRoundTables(tk)
+
+		var input []string
+		var output []struct {
+			SQL  string
+			Plan []string
+		}
+		suite := GetOrderAwareJoinReorderSuiteData()
+		suite.LoadTestCasesByName("TestOrderAwareJoinReorderAlternativeRound", t, &input, &output, cascades, caller)
+
+		plans := make([][]string, 0, 3)
+		for i, sql := range input {
+			normalized := strings.ToLower(strings.TrimSpace(sql))
+			if strings.HasPrefix(normalized, "set ") {
+				testdata.OnRecord(func() {
+					output[i].SQL = sql
+					output[i].Plan = nil
+				})
+				require.Equalf(t, sql, output[i].SQL, "input/output SQL mismatch at case[%d]", i)
+				tk.MustExec(sql)
+				continue
+			}
+
+			plan := tk.MustQuery(sql)
+			rows := testdata.ConvertRowsToStrings(plan.Rows())
+			testdata.OnRecord(func() {
+				output[i].SQL = sql
+				output[i].Plan = rows
+			})
+
+			require.Equalf(t, sql, output[i].SQL,
+				"input/output SQL mismatch at case[%d]", i)
+			plan.Check(testkit.Rows(output[i].Plan...))
+			require.NotContains(t, strings.Join(testdata.ConvertRowsToStrings(tk.MustQuery("show warnings").Rows()), "\n"),
+				"leading hint is inapplicable")
+			plans = append(plans, rows)
+		}
+		require.Equalf(t, len(input), len(output),
+			"unexpected output case count, input=%d, output=%d", len(input), len(output))
+		require.Len(t, plans, 3, "expected on/off/leading explain plans")
+
+		onPlan := plans[0]
+		offPlan := plans[1]
+		leadingPlan := plans[2]
+		onPlanText := strings.Join(onPlan, "\n")
+		offPlanText := strings.Join(offPlan, "\n")
+		leadingPlanText := strings.Join(leadingPlan, "\n")
+
+		require.NotEqualf(t, offPlan, onPlan,
+			"expected order-aware alternative round to change the chosen plan\noff:\n%s\non:\n%s",
+			offPlanText, onPlanText)
+		require.Equalf(t, leadingPlan, onPlan,
+			"expected order-aware alternative round to match explicit leading plan\noff:\n%s\non:\n%s\nleading:\n%s",
+			offPlanText, onPlanText, leadingPlanText)
+		require.Contains(t, offPlanText, "TopN")
+		require.Contains(t, offPlanText, "IndexHashJoin")
+		require.Contains(t, onPlanText, "Limit")
+		require.NotContains(t, onPlanText, "TopN")
+		require.Contains(t, onPlanText, "IndexJoin")
+		require.Contains(t, onPlanText, "idx_category_created")
+	})
+}
+
 // TestDPJoinReorderLeadingHint verifies that a leading hint produces a warning
 // when the DP algorithm is active, since DP does not support leading hints.
 func TestDPJoinReorderLeadingHint(tt *testing.T) {

pkg/planner/core/casetest/rule/testdata/order_aware_join_reorder_suite_in.json

@@ -0,0 +1,32 @@
+[
+  {
+    "name": "TestOrderAwareCDCJoinReorder",
+    "cases": [
+      "SELECT t6.id, t7.payload, t8.payload FROM t7 JOIN t8 ON t7.id = t8.id JOIN t6 ON t6.id = t7.id WHERE t6.category = 'hot' ORDER BY t6.id LIMIT 2",
+      "SELECT /*+ LEADING(t7, t8, t6) */ t6.id, t7.payload, t8.payload FROM t7 JOIN t8 ON t7.id = t8.id JOIN t6 ON t6.id = t7.id WHERE t6.category = 'hot' ORDER BY t6.id LIMIT 2",
+      "SELECT /*+ TIDB_INLJ(t7, t8, t9) */ t6.id, t7.payload, t8.payload, t9.payload FROM t7 JOIN t8 ON t7.id = t8.id JOIN t9 ON t8.id = t9.id JOIN t6 ON t6.id = t7.id WHERE t6.category = 'hot' ORDER BY t6.id LIMIT 2",
+      "SELECT /*+ TIDB_INLJ(t7, t8, t9) LEADING(t7, t8, t9, t6) */ t6.id, t7.payload, t8.payload, t9.payload FROM t7 JOIN t8 ON t7.id = t8.id JOIN t9 ON t8.id = t9.id JOIN t6 ON t6.id = t7.id WHERE t6.category = 'hot' ORDER BY t6.id LIMIT 2"
+    ]
+  },
+  {
+    "name": "TestOrderAwareJoinReorderPushSelection",
+    "cases": [
+      "explain format = 'plan_tree' select t6.id, t7.payload, t8.payload from t7 join t8 on t7.id = t8.id join t6 on t6.id = t7.id where t6.category = 'hot' order by t6.id limit 2",
+      "explain format = 'plan_tree' select /*+ LEADING(t7, t8, t6) */ t6.id, t7.payload, t8.payload from t7 join t8 on t7.id = t8.id join t6 on t6.id = t7.id where t6.category = 'hot' order by t6.id limit 2",
+      "explain format = 'plan_tree' select /*+ TIDB_INLJ(t7, t8, t9) */ t6.id, t7.payload, t8.payload, t9.payload from t7 join t8 on t7.id = t8.id join t9 on t8.id = t9.id join t6 on t6.id = t7.id where t6.category = 'hot' order by t6.id limit 2",
+      "explain format = 'plan_tree' select /*+ TIDB_INLJ(t7, t8, t9) LEADING(t7, t8, t9, t6) */ t6.id, t7.payload, t8.payload, t9.payload from t7 join t8 on t7.id = t8.id join t9 on t8.id = t9.id join t6 on t6.id = t7.id where t6.category = 'hot' order by t6.id limit 2"
+    ]
+  },
+  {
+    "name": "TestOrderAwareJoinReorderAlternativeRound",
+    "cases": [
+      "set @@tidb_opt_join_reorder_through_sel = 1",
+      "set @@tidb_opt_enable_alternative_logical_plans = 1",
+      "explain format = 'plan_tree' select /*+ TIDB_INLJ(oa_order_t2, oa_order_t3, oa_order_t4) */ oa_order_t1.created_at, oa_order_t4.payload from oa_order_t2 join oa_order_t3 on oa_order_t3.t2_id = oa_order_t2.id join oa_order_t4 on oa_order_t4.t3_id = oa_order_t3.id join oa_order_t1 on oa_order_t1.id = oa_order_t2.t1_id where oa_order_t1.category = 'hot' and oa_order_t4.payload = 1 order by oa_order_t1.created_at limit 2",
+      "set @@tidb_opt_enable_alternative_logical_plans = 0",
+      "explain format = 'plan_tree' select /*+ TIDB_INLJ(oa_order_t2, oa_order_t3, oa_order_t4) */ oa_order_t1.created_at, oa_order_t4.payload from oa_order_t2 join oa_order_t3 on oa_order_t3.t2_id = oa_order_t2.id join oa_order_t4 on oa_order_t4.t3_id = oa_order_t3.id join oa_order_t1 on oa_order_t1.id = oa_order_t2.t1_id where oa_order_t1.category = 'hot' and oa_order_t4.payload = 1 order by oa_order_t1.created_at limit 2",
+      "set @@tidb_opt_enable_alternative_logical_plans = 1",
+      "explain format = 'plan_tree' select /*+ LEADING(oa_order_t1, oa_order_t2, oa_order_t3, oa_order_t4) TIDB_INLJ(oa_order_t2, oa_order_t3, oa_order_t4) */ oa_order_t1.created_at, oa_order_t4.payload from oa_order_t2 join oa_order_t3 on oa_order_t3.t2_id = oa_order_t2.id join oa_order_t4 on oa_order_t4.t3_id = oa_order_t3.id join oa_order_t1 on oa_order_t1.id = oa_order_t2.t1_id where oa_order_t1.category = 'hot' and oa_order_t4.payload = 1 order by oa_order_t1.created_at limit 2"
+    ]
+  }
+]