executor: put sort files into one folder

pkg/executor/BUILD.bazel

@@ -77,7 +77,6 @@ go_library(
         "shuffle.go",
         "simple.go",
         "slow_query.go",
-        "sort.go",
         "split.go",
         "stmtsummary.go",
         "table_reader.go",
@@ -127,6 +126,7 @@ go_library(
         "//pkg/executor/lockstats",
         "//pkg/executor/metrics",
         "//pkg/executor/mppcoordmanager",
+        "//pkg/executor/sortexec",
         "//pkg/expression",
         "//pkg/expression/aggregation",
         "//pkg/infoschema",
@@ -349,7 +349,6 @@ go_test(
         "shuffle_test.go",
         "slow_query_sql_test.go",
         "slow_query_test.go",
-        "sort_test.go",
         "split_test.go",
         "stale_txn_test.go",
         "stmtsummary_test.go",
@@ -381,6 +380,7 @@ go_test(
         "//pkg/executor/importer",
         "//pkg/executor/internal/builder",
         "//pkg/executor/internal/exec",
+        "//pkg/executor/sortexec",
         "//pkg/expression",
         "//pkg/expression/aggregation",
         "//pkg/infoschema",

pkg/executor/benchmark_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pingcap/tidb/pkg/executor/aggfuncs"
 	"github.com/pingcap/tidb/pkg/executor/aggregate"
 	"github.com/pingcap/tidb/pkg/executor/internal/exec"
+	"github.com/pingcap/tidb/pkg/executor/sortexec"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/expression/aggregation"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -1646,20 +1647,20 @@ func prepare4MergeJoin(tc *mergeJoinTestCase, innerDS, outerDS *mockDataSource,
 
 	var leftExec, rightExec exec.Executor
 	if sorted {
-		leftSortExec := &SortExec{
+		leftSortExec := &sortexec.SortExec{
 			BaseExecutor: exec.NewBaseExecutor(tc.ctx, innerDS.Schema(), 3, innerDS),
 			ByItems:      make([]*util.ByItems, 0, len(tc.innerJoinKeyIdx)),
-			schema:       innerDS.Schema(),
+			ExecSchema:   innerDS.Schema(),
 		}
 		for _, key := range innerJoinKeys {
 			leftSortExec.ByItems = append(leftSortExec.ByItems, &util.ByItems{Expr: key})
 		}
 		leftExec = leftSortExec
 
-		rightSortExec := &SortExec{
+		rightSortExec := &sortexec.SortExec{
 			BaseExecutor: exec.NewBaseExecutor(tc.ctx, outerDS.Schema(), 4, outerDS),
 			ByItems:      make([]*util.ByItems, 0, len(tc.outerJoinKeyIdx)),
-			schema:       outerDS.Schema(),
+			ExecSchema:   outerDS.Schema(),
 		}
 		for _, key := range outerJoinKeys {
 			rightSortExec.ByItems = append(rightSortExec.ByItems, &util.ByItems{Expr: key})
@@ -1917,10 +1918,10 @@ func benchmarkSortExec(b *testing.B, cas *sortCase) {
 		ndvs:   cas.ndvs,
 	}
 	dataSource := buildMockDataSource(opt)
-	executor := &SortExec{
+	executor := &sortexec.SortExec{
 		BaseExecutor: exec.NewBaseExecutor(cas.ctx, dataSource.Schema(), 4, dataSource),
 		ByItems:      make([]*util.ByItems, 0, len(cas.orderByIdx)),
-		schema:       dataSource.Schema(),
+		ExecSchema:   dataSource.Schema(),
 	}
 	for _, idx := range cas.orderByIdx {
 		executor.ByItems = append(executor.ByItems, &util.ByItems{Expr: cas.columns()[idx]})

pkg/executor/builder.go

@@ -47,6 +47,7 @@ import (
 	"github.com/pingcap/tidb/pkg/executor/internal/vecgroupchecker"
 	"github.com/pingcap/tidb/pkg/executor/lockstats"
 	executor_metrics "github.com/pingcap/tidb/pkg/executor/metrics"
+	"github.com/pingcap/tidb/pkg/executor/sortexec"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/expression/aggregation"
 	"github.com/pingcap/tidb/pkg/infoschema"
@@ -1297,15 +1298,15 @@ func (b *executorBuilder) buildTrace(v *plannercore.Trace) exec.Executor {
 		optimizerTraceTarget: v.OptimizerTraceTarget,
 	}
 	if t.format == plannercore.TraceFormatLog && !t.optimizerTrace {
-		return &SortExec{
+		return &sortexec.SortExec{
 			BaseExecutor: exec.NewBaseExecutor(b.ctx, v.Schema(), v.ID(), t),
 			ByItems: []*plannerutil.ByItems{
 				{Expr: &expression.Column{
 					Index:   0,
 					RetType: types.NewFieldType(mysql.TypeTimestamp),
 				}},
 			},
-			schema: v.Schema(),
+			ExecSchema: v.Schema(),
 		}
 	}
 	return t
@@ -2245,10 +2246,10 @@ func (b *executorBuilder) buildSort(v *plannercore.PhysicalSort) exec.Executor {
 	if b.err != nil {
 		return nil
 	}
-	sortExec := SortExec{
+	sortExec := sortexec.SortExec{
 		BaseExecutor: exec.NewBaseExecutor(b.ctx, v.Schema(), v.ID(), childExec),
 		ByItems:      v.ByItems,
-		schema:       v.Schema(),
+		ExecSchema:   v.Schema(),
 	}
 	executor_metrics.ExecutorCounterSortExec.Inc()
 	return &sortExec
@@ -2259,15 +2260,15 @@ func (b *executorBuilder) buildTopN(v *plannercore.PhysicalTopN) exec.Executor {
 	if b.err != nil {
 		return nil
 	}
-	sortExec := SortExec{
+	sortExec := sortexec.SortExec{
 		BaseExecutor: exec.NewBaseExecutor(b.ctx, v.Schema(), v.ID(), childExec),
 		ByItems:      v.ByItems,
-		schema:       v.Schema(),
+		ExecSchema:   v.Schema(),
 	}
 	executor_metrics.ExecutorCounterTopNExec.Inc()
-	return &TopNExec{
+	return &sortexec.TopNExec{
 		SortExec: sortExec,
-		limit:    &plannercore.PhysicalLimit{Count: v.Count, Offset: v.Offset},
+		Limit:    &plannercore.PhysicalLimit{Count: v.Count, Offset: v.Offset},
 	}
 }
 

pkg/executor/executor.go

@@ -39,6 +39,7 @@ import (
 	"github.com/pingcap/tidb/pkg/executor/aggregate"
 	"github.com/pingcap/tidb/pkg/executor/internal/exec"
 	"github.com/pingcap/tidb/pkg/executor/internal/pdhelper"
+	"github.com/pingcap/tidb/pkg/executor/sortexec"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/kv"
@@ -102,12 +103,12 @@ var (
 	_ exec.Executor = &ShowDDLExec{}
 	_ exec.Executor = &ShowDDLJobsExec{}
 	_ exec.Executor = &ShowDDLJobQueriesExec{}
-	_ exec.Executor = &SortExec{}
+	_ exec.Executor = &sortexec.SortExec{}
 	_ exec.Executor = &aggregate.StreamAggExec{}
 	_ exec.Executor = &TableDualExec{}
 	_ exec.Executor = &TableReaderExecutor{}
 	_ exec.Executor = &TableScanExec{}
-	_ exec.Executor = &TopNExec{}
+	_ exec.Executor = &sortexec.TopNExec{}
 	_ exec.Executor = &UnionExec{}
 	_ exec.Executor = &FastCheckTableExec{}
 

pkg/executor/executor_pkg_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tidb/pkg/executor/aggfuncs"
 	"github.com/pingcap/tidb/pkg/executor/aggregate"
 	"github.com/pingcap/tidb/pkg/executor/internal/exec"
+	"github.com/pingcap/tidb/pkg/executor/sortexec"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/kv"
 	plannerutil "github.com/pingcap/tidb/pkg/planner/util"
@@ -290,10 +291,10 @@ func TestSortSpillDisk(t *testing.T) {
 		ndvs:   cas.ndvs,
 	}
 	dataSource := buildMockDataSource(opt)
-	exe := &SortExec{
+	exe := &sortexec.SortExec{
 		BaseExecutor: exec.NewBaseExecutor(cas.ctx, dataSource.Schema(), 0, dataSource),
 		ByItems:      make([]*plannerutil.ByItems, 0, len(cas.orderByIdx)),
-		schema:       dataSource.Schema(),
+		ExecSchema:   dataSource.Schema(),
 	}
 	for _, idx := range cas.orderByIdx {
 		exe.ByItems = append(exe.ByItems, &plannerutil.ByItems{Expr: cas.columns()[idx]})
@@ -311,9 +312,9 @@ func TestSortSpillDisk(t *testing.T) {
 		}
 	}
 	// Test only 1 partition and all data in memory.
-	require.Len(t, exe.partitionList, 1)
-	require.Equal(t, false, exe.partitionList[0].AlreadySpilledSafeForTest())
-	require.Equal(t, 2048, exe.partitionList[0].NumRow())
+	require.Len(t, exe.PartitionList, 1)
+	require.Equal(t, false, exe.PartitionList[0].AlreadySpilledSafeForTest())
+	require.Equal(t, 2048, exe.PartitionList[0].NumRow())
 	err = exe.Close()
 	require.NoError(t, err)
 
@@ -334,16 +335,16 @@ func TestSortSpillDisk(t *testing.T) {
 	// Now spilling is in parallel.
 	// Maybe the second add() will called before spilling, depends on
 	// Golang goroutine scheduling. So the result has two possibilities.
-	if len(exe.partitionList) == 2 {
-		require.Len(t, exe.partitionList, 2)
-		require.Equal(t, true, exe.partitionList[0].AlreadySpilledSafeForTest())
-		require.Equal(t, true, exe.partitionList[1].AlreadySpilledSafeForTest())
-		require.Equal(t, 1024, exe.partitionList[0].NumRow())
-		require.Equal(t, 1024, exe.partitionList[1].NumRow())
+	if len(exe.PartitionList) == 2 {
+		require.Len(t, exe.PartitionList, 2)
+		require.Equal(t, true, exe.PartitionList[0].AlreadySpilledSafeForTest())
+		require.Equal(t, true, exe.PartitionList[1].AlreadySpilledSafeForTest())
+		require.Equal(t, 1024, exe.PartitionList[0].NumRow())
+		require.Equal(t, 1024, exe.PartitionList[1].NumRow())
 	} else {
-		require.Len(t, exe.partitionList, 1)
-		require.Equal(t, true, exe.partitionList[0].AlreadySpilledSafeForTest())
-		require.Equal(t, 2048, exe.partitionList[0].NumRow())
+		require.Len(t, exe.PartitionList, 1)
+		require.Equal(t, true, exe.PartitionList[0].AlreadySpilledSafeForTest())
+		require.Equal(t, 2048, exe.PartitionList[0].NumRow())
 	}
 
 	err = exe.Close()
@@ -363,9 +364,9 @@ func TestSortSpillDisk(t *testing.T) {
 		}
 	}
 	// Test only 1 partition but spill disk.
-	require.Len(t, exe.partitionList, 1)
-	require.Equal(t, true, exe.partitionList[0].AlreadySpilledSafeForTest())
-	require.Equal(t, 2048, exe.partitionList[0].NumRow())
+	require.Len(t, exe.PartitionList, 1)
+	require.Equal(t, true, exe.PartitionList[0].AlreadySpilledSafeForTest())
+	require.Equal(t, 2048, exe.PartitionList[0].NumRow())
 	err = exe.Close()
 	require.NoError(t, err)
 
@@ -385,10 +386,10 @@ func TestSortSpillDisk(t *testing.T) {
 		ndvs:   cas.ndvs,
 	}
 	dataSource = buildMockDataSource(opt)
-	exe = &SortExec{
+	exe = &sortexec.SortExec{
 		BaseExecutor: exec.NewBaseExecutor(cas.ctx, dataSource.Schema(), 0, dataSource),
 		ByItems:      make([]*plannerutil.ByItems, 0, len(cas.orderByIdx)),
-		schema:       dataSource.Schema(),
+		ExecSchema:   dataSource.Schema(),
 	}
 	for _, idx := range cas.orderByIdx {
 		exe.ByItems = append(exe.ByItems, &plannerutil.ByItems{Expr: cas.columns()[idx]})
@@ -406,7 +407,7 @@ func TestSortSpillDisk(t *testing.T) {
 		}
 	}
 	// Don't spill too many partitions.
-	require.True(t, len(exe.partitionList) <= 4)
+	require.True(t, len(exe.PartitionList) <= 4)
 	err = exe.Close()
 	require.NoError(t, err)
 }

pkg/executor/executor_required_rows_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/executor/internal/exec"
+	"github.com/pingcap/tidb/pkg/executor/sortexec"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/expression/aggregation"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -278,10 +279,10 @@ func TestSortRequiredRows(t *testing.T) {
 }
 
 func buildSortExec(sctx sessionctx.Context, byItems []*util.ByItems, src exec.Executor) exec.Executor {
-	sortExec := SortExec{
+	sortExec := sortexec.SortExec{
 		BaseExecutor: exec.NewBaseExecutor(sctx, src.Schema(), 0, src),
 		ByItems:      byItems,
-		schema:       src.Schema(),
+		ExecSchema:   src.Schema(),
 	}
 	return &sortExec
 }
@@ -385,14 +386,14 @@ func TestTopNRequiredRows(t *testing.T) {
 }
 
 func buildTopNExec(ctx sessionctx.Context, offset, count int, byItems []*util.ByItems, src exec.Executor) exec.Executor {
-	sortExec := SortExec{
+	sortExec := sortexec.SortExec{
 		BaseExecutor: exec.NewBaseExecutor(ctx, src.Schema(), 0, src),
 		ByItems:      byItems,
-		schema:       src.Schema(),
+		ExecSchema:   src.Schema(),
 	}
-	return &TopNExec{
+	return &sortexec.TopNExec{
 		SortExec: sortExec,
-		limit:    &plannercore.PhysicalLimit{Count: uint64(count), Offset: uint64(offset)},
+		Limit:    &plannercore.PhysicalLimit{Count: uint64(count), Offset: uint64(offset)},
 	}
 }
 

pkg/executor/sortexec/BUILD.bazel

@@ -0,0 +1,48 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "sortexec",
+    srcs = ["sort.go"],
+    importpath = "github.com/pingcap/tidb/pkg/executor/sortexec",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/executor/internal/exec",
+        "//pkg/expression",
+        "//pkg/planner/core",
+        "//pkg/planner/util",
+        "//pkg/sessionctx/variable",
+        "//pkg/util/chunk",
+        "//pkg/util/disk",
+        "//pkg/util/memory",
+        "//pkg/util/sqlkiller",
+        "@com_github_pingcap_failpoint//:failpoint",
+    ],
+)
+
+go_test(
+    name = "sort_exec_test",
+    timeout = "short",
+    srcs = ["sort_test.go"],
+    flaky = True,
+    deps = [
+        "//pkg/config",
+        "//pkg/sessionctx/variable",
+        "//pkg/testkit",
+        "//pkg/util",
+        "@com_github_pingcap_failpoint//:failpoint",
+        "@com_github_stretchr_testify//require",
+    ],
+)
+
+go_test(
+    name = "sortexec_test",
+    srcs = ["sort_test.go"],
+    deps = [
+        "//pkg/config",
+        "//pkg/sessionctx/variable",
+        "//pkg/testkit",
+        "//pkg/util",
+        "@com_github_pingcap_failpoint//:failpoint",
+        "@com_github_stretchr_testify//require",
+    ],
+)