bulkmerge: add processors and DistSQL physical planning

This commit introduces the initial implementation of bulkmerge processors
for distributed merge operations. It adds three new processor types:
MergeLoopback, BulkMerge, and MergeCoordinator, along with the DistSQL
physical planning infrastructure to orchestrate them.

The MergeLoopback processor runs on the coordinator node and generates
initial tasks. The BulkMerge processors run on each SQL instance to
perform merge operations. The MergeCoordinator collects results from
all merge processors.

This commit also updates the vectorized execution engine (execplan.go)
to recognize the new processor cores, preventing panics when these
processors are encountered.

Informs #156580
Release note: None

Author: jeffswenson

pkg/BUILD.bazel

@@ -397,6 +397,7 @@ ALL_TESTS = [
     "//pkg/sql/auditlogging:auditlogging_test",
     "//pkg/sql/backfill:backfill_test",
     "//pkg/sql/bulkingest:bulkingest_test",
+    "//pkg/sql/bulkmerge:bulkmerge_test",
     "//pkg/sql/bulksst:bulksst_test",
     "//pkg/sql/bulkutil:bulkutil_test",
     "//pkg/sql/cacheutil:cacheutil_test",
@@ -1885,6 +1886,8 @@ GO_TARGETS = [
     "//pkg/sql/backfill:backfill_test",
     "//pkg/sql/bulkingest:bulkingest",
     "//pkg/sql/bulkingest:bulkingest_test",
+    "//pkg/sql/bulkmerge:bulkmerge",
+    "//pkg/sql/bulkmerge:bulkmerge_test",
     "//pkg/sql/bulksst:bulksst",
     "//pkg/sql/bulksst:bulksst_test",
     "//pkg/sql/bulkutil:bulkutil",

pkg/sql/bulkmerge/BUILD.bazel

@@ -0,0 +1,57 @@
+# Copyright 2025 The Cockroach Authors.
+#
+# Use of this software is governed by the CockroachDB Software License
+# included in the /LICENSE file.
+
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "bulkmerge",
+    srcs = [
+        "merge_coordinator.go",
+        "merge_loopback.go",
+        "merge_planning.go",
+        "merge_processor.go",
+    ],
+    importpath = "github.com/cockroachdb/cockroach/pkg/sql/bulkmerge",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/base",
+        "//pkg/sql",
+        "//pkg/sql/execinfra",
+        "//pkg/sql/execinfrapb",
+        "//pkg/sql/physicalplan",
+        "//pkg/sql/rowenc",
+        "//pkg/sql/rowexec",
+        "//pkg/sql/sem/tree",
+        "//pkg/sql/types",
+        "@com_github_cockroachdb_errors//:errors",
+    ],
+)
+
+go_test(
+    name = "bulkmerge_test",
+    srcs = [
+        "main_test.go",
+        "merge_processor_test.go",
+        "merge_test.go",
+    ],
+    embed = [":bulkmerge"],
+    deps = [
+        "//pkg/base",
+        "//pkg/kv/kvclient/kvtenant",
+        "//pkg/security/securityassets",
+        "//pkg/security/securitytest",
+        "//pkg/security/username",
+        "//pkg/server",
+        "//pkg/sql",
+        "//pkg/sql/sem/tree",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/randutil",
+        "@com_github_stretchr_testify//require",
+    ],
+)

pkg/sql/bulkmerge/main_test.go

@@ -0,0 +1,30 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package bulkmerge
+
+import (
+	"os"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvtenant"
+	"github.com/cockroachdb/cockroach/pkg/security/securityassets"
+	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
+	"github.com/cockroachdb/cockroach/pkg/server"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+)
+
+//go:generate ../util/leaktest/add-leaktest.sh *_test.go
+
+func TestMain(m *testing.M) {
+	securityassets.SetLoader(securitytest.EmbeddedAssets)
+	randutil.SeedForTests()
+	serverutils.InitTestServerFactory(server.TestServerFactory)
+	serverutils.InitTestClusterFactory(testcluster.TestClusterFactory)
+	kvtenant.InitTestConnectorFactory()
+	os.Exit(m.Run())
+}

pkg/sql/bulkmerge/merge_coordinator.go

@@ -0,0 +1,90 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package bulkmerge
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
+	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/errors"
+)
+
+var (
+	_ execinfra.Processor = &mergeCoordinator{}
+	_ execinfra.RowSource = &mergeCoordinator{}
+)
+
+// Emits a single row on completion which is a protobuf containing the details
+// of the merged SSTs.
+// TODO(jeffswenson): define the protobuf
+var mergeCoordinatorOutputTypes = []*types.T{
+	types.Bytes,
+}
+
+type mergeCoordinator struct {
+	execinfra.ProcessorBase
+	input execinfra.RowSource
+}
+
+// Next implements execinfra.RowSource.
+func (m *mergeCoordinator) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMetadata) {
+	for m.State == execinfra.StateRunning {
+		row, meta := m.input.Next()
+		switch {
+		case row == nil && meta == nil:
+			m.MoveToDraining(nil /* err */)
+			break
+		case meta != nil && meta.Err != nil:
+			m.MoveToDraining(meta.Err)
+			break
+		case meta != nil:
+			m.MoveToDraining(errors.Newf("unexpected meta: %v", meta))
+			break
+		case row != nil:
+			base := *row[2].Datum.(*tree.DBytes)
+			return rowenc.EncDatumRow{
+				rowenc.EncDatum{Datum: tree.NewDBytes(base + "->coordinator")},
+			}, nil
+		}
+	}
+	return nil, m.DrainHelper()
+}
+
+// Start implements execinfra.RowSource.
+func (m *mergeCoordinator) Start(ctx context.Context) {
+	m.StartInternal(ctx, "mergeCoordinator")
+	m.input.Start(ctx)
+}
+
+func init() {
+	rowexec.NewMergeCoordinatorProcessor = func(
+		ctx context.Context,
+		flow *execinfra.FlowCtx,
+		flowID int32,
+		spec execinfrapb.MergeCoordinatorSpec,
+		postSpec *execinfrapb.PostProcessSpec,
+		input execinfra.RowSource,
+	) (execinfra.Processor, error) {
+		mc := &mergeCoordinator{
+			input: input,
+		}
+		err := mc.Init(
+			ctx, mc, postSpec, mergeCoordinatorOutputTypes, flow, flowID, nil,
+			execinfra.ProcStateOpts{
+				InputsToDrain: []execinfra.RowSource{input},
+			},
+		)
+		if err != nil {
+			return nil, err
+		}
+		return mc, nil
+	}
+}

pkg/sql/bulkmerge/merge_loopback.go

@@ -0,0 +1,74 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package bulkmerge
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
+	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
+)
+
+var (
+	_ execinfra.Processor = &mergeLoopback{}
+	_ execinfra.RowSource = &mergeLoopback{}
+)
+
+var mergeLoopbackOutputTypes = []*types.T{
+	// Span key for the range router. It encodes the destination
+	// processor's SQL instance ID.
+	types.Bytes,
+	// Task ID
+	types.Int4,
+}
+
+type mergeLoopback struct {
+	execinfra.ProcessorBase
+	done bool
+}
+
+// Next implements execinfra.RowSource.
+func (m *mergeLoopback) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMetadata) {
+	if m.done {
+		m.MoveToDraining(nil)
+		return nil, m.DrainHelper()
+	}
+	m.done = true
+	return rowenc.EncDatumRow{
+		rowenc.EncDatum{Datum: tree.NewDBytes("loopback")},
+		rowenc.EncDatum{Datum: tree.NewDInt(1)},
+	}, nil
+}
+
+// Start implements execinfra.RowSource.
+func (m *mergeLoopback) Start(ctx context.Context) {
+	m.StartInternal(ctx, "mergeLoopback")
+	// TODO(jeffswenson): create the initial set of tasks
+}
+
+func init() {
+	rowexec.NewMergeLoopbackProcessor = func(
+		ctx context.Context,
+		flow *execinfra.FlowCtx,
+		flowID int32,
+		spec execinfrapb.MergeLoopbackSpec,
+		postSpec *execinfrapb.PostProcessSpec,
+	) (execinfra.Processor, error) {
+		ml := &mergeLoopback{}
+		err := ml.Init(
+			ctx, ml, postSpec, mergeLoopbackOutputTypes, flow, flowID, nil,
+			execinfra.ProcStateOpts{},
+		)
+		if err != nil {
+			return nil, err
+		}
+		return ml, nil
+	}
+}

pkg/sql/bulkmerge/merge_planning.go

@@ -0,0 +1,94 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package bulkmerge
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
+	"github.com/cockroachdb/errors"
+)
+
+func newBulkMergePlan(
+	ctx context.Context, execCtx sql.JobExecContext,
+) (*sql.PhysicalPlan, *sql.PlanningCtx, error) {
+	// NOTE: This implementation is inspired by the physical plan created by
+	// restore in `pkg/backup/restore_processor_planning.go`
+	planCtx, sqlInstanceIDs, err := execCtx.DistSQLPlanner().SetupAllNodesPlanning(
+		ctx, execCtx.ExtendedEvalContext(), execCtx.ExecCfg())
+	if err != nil {
+		return nil, nil, err
+	}
+
+	plan := planCtx.NewPhysicalPlan()
+	// Use the gateway node as the coordinator, which is where the job was initiated.
+	coordinatorID := plan.GatewaySQLInstanceID
+
+	router, err := physicalplan.MakeInstanceRouter(sqlInstanceIDs)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "unable to make instance router")
+	}
+
+	loopbackID := plan.AddProcessor(physicalplan.Processor{
+		SQLInstanceID: coordinatorID,
+		Spec: execinfrapb.ProcessorSpec{
+			Core: execinfrapb.ProcessorCoreUnion{
+				MergeLoopback: &execinfrapb.MergeLoopbackSpec{},
+			},
+			Post: execinfrapb.PostProcessSpec{},
+			Output: []execinfrapb.OutputRouterSpec{{
+				Type:            execinfrapb.OutputRouterSpec_BY_RANGE,
+				RangeRouterSpec: router,
+			}},
+			StageID:     plan.NewStageOnNodes([]base.SQLInstanceID{coordinatorID}),
+			ResultTypes: mergeLoopbackOutputTypes,
+		},
+	})
+
+	mergeStage := plan.NewStageOnNodes(sqlInstanceIDs)
+	for streamID, sqlInstanceID := range sqlInstanceIDs {
+		pIdx := plan.AddProcessor(physicalplan.Processor{
+			SQLInstanceID: sqlInstanceID,
+			Spec: execinfrapb.ProcessorSpec{
+				Input: []execinfrapb.InputSyncSpec{{
+					ColumnTypes: mergeLoopbackOutputTypes,
+				}},
+				Core: execinfrapb.ProcessorCoreUnion{
+					BulkMerge: &execinfrapb.BulkMergeSpec{
+						// TODO(jeffswenson): fill in the rest of the spec
+					},
+				},
+				Post: execinfrapb.PostProcessSpec{},
+				Output: []execinfrapb.OutputRouterSpec{{
+					Type: execinfrapb.OutputRouterSpec_PASS_THROUGH,
+				}},
+				StageID:     mergeStage,
+				ResultTypes: bulkMergeProcessorOutputTypes,
+			},
+		})
+		plan.Streams = append(plan.Streams, physicalplan.Stream{
+			SourceProcessor:  loopbackID,
+			SourceRouterSlot: streamID,
+			DestProcessor:    pIdx,
+			DestInput:        0,
+		})
+		plan.ResultRouters = append(plan.ResultRouters, pIdx)
+	}
+
+	plan.AddSingleGroupStage(ctx, coordinatorID, execinfrapb.ProcessorCoreUnion{
+		MergeCoordinator: &execinfrapb.MergeCoordinatorSpec{
+			// TODO fill in the rest of the spec
+		},
+	}, execinfrapb.PostProcessSpec{}, mergeCoordinatorOutputTypes, nil /* finalizeLastStageCb */)
+
+	plan.PlanToStreamColMap = []int{0} // Needed for FinalizePlan to populate ResultTypes
+	sql.FinalizePlan(ctx, planCtx, plan)
+
+	return plan, planCtx, nil
+}