Merge #158341 #158596 #158606 #158610

158341: distmerge: add merge processor scaffolding r=mw5h a=mw5h

#### 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

#### backup: deduplicate routing logic with physicalplan package

Previously, the backup package contained duplicate implementations of
routingDatumsForSQLInstance and routingSpanForSQLInstance that were
identical to functions recently added to pkg/sql/physicalplan. This
commit removes these duplicates and uses the shared implementations:
RoutingDatumsForSQLInstance and MakeInstanceRouter.

This reduces code duplication and ensures consistent routing behavior
across DistSQL physical planning for both backup/restore and bulk merge
operations.

Informs #156580
Release note: None

#### physicalplan: set DefaultDest to nil to catch routing bugs

Changes MakeInstanceRouter to set DefaultDest to nil instead of stream 0.
When DefaultDest is nil, any routing key that doesn't match a span will
produce an error rather than silently routing to an arbitrary stream.
This helps catch coordination bugs early.

Updates merge_loopback.go to generate a routing key for its SQL instance
using physicalplan.RoutingDatumsForSQLInstance instead of a hardcoded
"loopback" key. This ensures the routing key matches one of the spans in
the router rather than relying on the DefaultDest fallback.

The test expectation is updated from "loopback->merge->coordinator" to
"node1->merge->coordinator" to reflect the explicit routing behavior.

Fixes #156580
Release note: None


158596: kvnemesis: correctly verify the span config of the system range r=miraradeva a=miraradeva

Previously, kvnemesis verified the correct span config for some critical ranges: meta (r1), liveness (r2) and system (r3). However, as of 8844739, meta1 and meta2 start off as separate ranges, pushing the system range to r4. It's important for this range to be replicated correctly in kvnemesis to ensure it's available in the presence of network partitions. Unavailability of the system range can result in splits not able to allocate new range IDs.

This commit bumps up the max range ID used for span config verification from 3 to 4.

Informs: #158366

Release note: None

158606: mma: always incorporate RangeMsg.RangeLoad r=wenyihu6 a=sumeerbhola

Epic: CRDB-55052

Release note: None

158610: mma: tweak the comment of MakeStoreLoadMsg r=wenyihu6 a=sumeerbhola

This is mainly to clarify that we sum the load and capacity reported per store inside MMA, so tweaking the capacity calculation approach may not be possible in isolation.

Epic: CRDB-55052

Release note: None

Co-authored-by: Jeff Swenson <[email protected]>
Co-authored-by: Matt White <[email protected]>
Co-authored-by: Mira Radeva <[email protected]>
Co-authored-by: sumeerbhola <[email protected]>

Author: 5 people

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/backup/BUILD.bazel

@@ -83,7 +83,6 @@ go_library(
         "//pkg/sql",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catalogkeys",
-        "//pkg/sql/catalog/catenumpb",
         "//pkg/sql/catalog/catpb",
         "//pkg/sql/catalog/colinfo",
         "//pkg/sql/catalog/dbdesc",

pkg/backup/generative_split_and_scatter_processor.go

@@ -7,7 +7,6 @@ package backup
 
 import (
 	"context"
-	"fmt"
 	"hash/fnv"
 	"math/rand"
 	"strings"
@@ -22,9 +21,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
-	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -249,15 +248,6 @@ func (s dbSplitAndScatterer) findDestination(res *kvpb.AdminScatterResponse) roa
 	return roachpb.NodeID(0)
 }
 
-func routingDatumsForSQLInstance(
-	sqlInstanceID base.SQLInstanceID,
-) (rowenc.EncDatum, rowenc.EncDatum) {
-	routingBytes := roachpb.Key(fmt.Sprintf("node%d", sqlInstanceID))
-	startDatum := rowenc.DatumToEncDatumUnsafe(types.Bytes, tree.NewDBytes(tree.DBytes(routingBytes)))
-	endDatum := rowenc.DatumToEncDatumUnsafe(types.Bytes, tree.NewDBytes(tree.DBytes(routingBytes.Next())))
-	return startDatum, endDatum
-}
-
 type entryNode struct {
 	entry execinfrapb.RestoreSpanEntry
 	node  roachpb.NodeID
@@ -393,7 +383,7 @@ func (gssp *generativeSplitAndScatterProcessor) Next() (
 		// The routing datums informs the router which output stream should be used.
 		routingDatum, ok := gssp.routingDatumCache.getRoutingDatum(scatteredEntry.node)
 		if !ok {
-			routingDatum, _ = routingDatumsForSQLInstance(base.SQLInstanceID(scatteredEntry.node))
+			routingDatum, _ = physicalplan.RoutingDatumsForSQLInstance(base.SQLInstanceID(scatteredEntry.node))
 			gssp.routingDatumCache.putRoutingDatum(scatteredEntry.node, routingDatum)
 		}
 
@@ -745,24 +735,6 @@ var splitAndScatterOutputTypes = []*types.T{
 	types.Bytes, // RestoreDataEntry bytes
 }
 
-// routingSpanForSQLInstance provides the mapping to be used during distsql planning
-// when setting up the output router.
-func routingSpanForSQLInstance(sqlInstanceID base.SQLInstanceID) ([]byte, []byte, error) {
-	var alloc tree.DatumAlloc
-	startDatum, endDatum := routingDatumsForSQLInstance(sqlInstanceID)
-
-	startBytes, endBytes := make([]byte, 0), make([]byte, 0)
-	startBytes, err := startDatum.Encode(splitAndScatterOutputTypes[0], &alloc, catenumpb.DatumEncoding_ASCENDING_KEY, startBytes)
-	if err != nil {
-		return nil, nil, err
-	}
-	endBytes, err = endDatum.Encode(splitAndScatterOutputTypes[0], &alloc, catenumpb.DatumEncoding_ASCENDING_KEY, endBytes)
-	if err != nil {
-		return nil, nil, err
-	}
-	return startBytes, endBytes, nil
-}
-
 func init() {
 	rowexec.NewGenerativeSplitAndScatterProcessor = newGenerativeSplitAndScatterProcessor
 }

pkg/backup/restore_processor_planning.go

@@ -6,10 +6,8 @@
 package backup
 
 import (
-	"bytes"
 	"context"
 	"math"
-	"slices"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
@@ -19,7 +17,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
-	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -123,34 +120,10 @@ func distRestore(
 		// Plan SplitAndScatter on the coordinator node.
 		splitAndScatterStageID := p.NewStageOnNodes(sqlInstanceIDs)
 
-		defaultStream := int32(0)
-		rangeRouterSpec := execinfrapb.OutputRouterSpec_RangeRouterSpec{
-			Spans:       nil,
-			DefaultDest: &defaultStream,
-			Encodings: []execinfrapb.OutputRouterSpec_RangeRouterSpec_ColumnEncoding{
-				{
-					Column:   0,
-					Encoding: catenumpb.DatumEncoding_ASCENDING_KEY,
-				},
-			},
-		}
-		for stream, sqlInstanceID := range sqlInstanceIDs {
-			startBytes, endBytes, err := routingSpanForSQLInstance(sqlInstanceID)
-			if err != nil {
-				return nil, nil, err
-			}
-
-			span := execinfrapb.OutputRouterSpec_RangeRouterSpec_Span{
-				Start:  startBytes,
-				End:    endBytes,
-				Stream: int32(stream),
-			}
-			rangeRouterSpec.Spans = append(rangeRouterSpec.Spans, span)
+		rangeRouterSpec, err := physicalplan.MakeInstanceRouter(sqlInstanceIDs)
+		if err != nil {
+			return nil, nil, err
 		}
-		// The router expects the spans to be sorted.
-		slices.SortFunc(rangeRouterSpec.Spans, func(a, b execinfrapb.OutputRouterSpec_RangeRouterSpec_Span) int {
-			return bytes.Compare(a.Start, b.Start)
-		})
 
 		// TODO(pbardea): This not super principled. I just wanted something that
 		// wasn't a constant and grew slower than linear with the length of

pkg/kv/kvnemesis/kvnemesis_test.go

@@ -851,7 +851,9 @@ func setAndVerifyZoneConfigs(
 							Key:    desc.StartKey.AsRawKey(),
 							EndKey: desc.EndKey.AsRawKey(),
 						}
-						if replicaSpan.Overlaps(dataSpan) || desc.RangeID <= 3 {
+						// Ranges 1-4 are the ranges we constrained above
+						// (1: meta1, 2: meta2, 3: liveness, 4: system).
+						if replicaSpan.Overlaps(dataSpan) || desc.RangeID <= 4 {
 							overlappingReplicas = append(overlappingReplicas, replica)
 						}
 						return true // continue

pkg/kv/kvserver/allocator/mmaprototype/cluster_state.go

@@ -1370,6 +1370,7 @@ func (cs *clusterState) processStoreLeaseholderMsgInternal(
 		} else if rs.localRangeOwner != msg.StoreID {
 			rs.localRangeOwner = msg.StoreID
 		}
+		rs.load = rangeMsg.RangeLoad
 		if !rangeMsg.MaybeSpanConfIsPopulated && len(rs.pendingChanges) == 0 {
 			// Common case: no pending changes, and span config not provided.
 			//
@@ -1397,16 +1398,18 @@ func (cs *clusterState) processStoreLeaseholderMsgInternal(
 				mayHaveDiverged = true
 			}
 			if !mayHaveDiverged {
+				// This is the common case where there were no pending changes, and no
+				// span config provided and no replicas changes. We don't need to do
+				// any more processing of this RangeMsg.
 				continue
 			}
 			// Else fall through and do the expensive work.
 		}
+		// Slow path, which reconstructs everything about the range.
+
 		// Set the range state and store state to match the range message state
 		// initially. The pending changes which are not enacted in the range
 		// message are handled and added back below.
-		if rangeMsg.MaybeSpanConfIsPopulated {
-			rs.load = rangeMsg.RangeLoad
-		}
 		for _, replica := range rs.replicas {
 			ss := cs.stores[replica.StoreID]
 			if ss == nil {
@@ -1548,10 +1551,10 @@ func (cs *clusterState) processStoreLeaseholderMsgInternal(
 			}
 			rs.conf = normSpanConfig
 		}
-		// NB: Always recompute the analyzed range constraints for any range,
-		// assuming the leaseholder wouldn't have sent the message if there was no
-		// change, or we noticed a divergence in membership above and fell through
-		// here.
+		// Ensure (later) recomputation of the analyzed range constraints for the
+		// range, by clearing the existing analyzed constraints. This is done
+		// since in this slow path the span config or the replicas may have
+		// changed.
 		rs.clearAnalyzedConstraints()
 	}
 	// Remove ranges for which this is the localRangeOwner, but for which it is

pkg/kv/kvserver/allocator/mmaprototype/cluster_state_test.go

@@ -187,7 +187,7 @@ func parseStoreLeaseholderMsg(t *testing.T, in string) StoreLeaseholderMsg {
 				case "raft-cpu":
 					rMsg.RangeLoad.RaftCPU = LoadValue(parseInt(t, parts[1]))
 					rMsg.MaybeSpanConfIsPopulated = true
-				case "not-populated":
+				case "span-config-not-populated":
 					notPopulatedOverride = true
 				default:
 					t.Fatalf("unknown argument: %s", parts[0])

pkg/kv/kvserver/allocator/mmaprototype/messages.go

@@ -42,21 +42,21 @@ type StoreLeaseholderMsg struct {
 }
 
 // RangeMsg is generated by the leaseholder store (and part of
-// StoreLeaseholderMsg). If there is any change for that range, the full
-// information for that range is provided (and MaybeSpanConfIsPopulated is set
-// to true). This is also the case for a new leaseholder since it does not know
-// whether something has changed since the last leaseholder informed the
-// allocator. A tiny change to the RangeLoad (decided by the caller) will not
-// cause the fields to be populated.
+// StoreLeaseholderMsg). All fields except for MaybeSpanConf are always
+// populated. The MaybeSpanConf is populated if there is a possible change in
+// the SpanConfig (including when this is a new leaseholder for the range),
+// and is indicated by MaybeSpanConfIsPopulated being true. The optional
+// population is done to avoid the expensive processing of a SpanConfig by
+// MMA, on every message for the range.
 //
-// To ensure that the allocator does not lose synchronization with the current
-// set of replicas, due to spurious changes (we had one undiagnosed example
-// where the allocator was spuriously told that a lease was transferred away),
-// the Replicas field is always populated).
+// The main reason we always populate Replicas is to ensure that the allocator
+// does not lose synchronization with the current set of replicas, due to
+// spurious changes (we had one undiagnosed example where the allocator was
+// spuriously told that a lease was transferred away).
 type RangeMsg struct {
 	roachpb.RangeID
 	Replicas                 []StoreIDAndReplicaState
+	RangeLoad                RangeLoad
 	MaybeSpanConfIsPopulated bool
 	MaybeSpanConf            roachpb.SpanConfig
-	RangeLoad                RangeLoad
 }

pkg/kv/kvserver/allocator/mmaprototype/testdata/cluster_state/multiple_ranges.txt

@@ -51,19 +51,19 @@ store-load-msg
   store-id=2 node-id=2 load=[45,40,50] capacity=[100,100,50] secondary-load=1 load-time=0s
 ----
 
-# StoreLeaseholderMsg, with empty information, since the range states have not changed.
+# StoreLeaseholderMsg, with no span configs for the ranges.
 store-leaseholder-msg
 store-id=1
-  range-id=1 not-populated
+  range-id=1 load=[10,10,20] raft-cpu=5 span-config-not-populated
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
     store-id=2 replica-id=2 type=VOTER_FULL
-  range-id=2 not-populated
+  range-id=2 load=[20,10,15] raft-cpu=10 span-config-not-populated
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
     store-id=2 replica-id=2 type=VOTER_FULL
-  range-id=3 not-populated
+  range-id=3 load=[30,10,10] raft-cpu=25 span-config-not-populated
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
     store-id=2 replica-id=2 type=VOTER_FULL
-  range-id=4 not-populated
+  range-id=4 load=[40,10,5] raft-cpu=5 span-config-not-populated
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
     store-id=2 replica-id=2 type=VOTER_FULL
 ----
@@ -76,28 +76,32 @@ store-id=1 node-id=1 status=ok accepting all reported=[cpu:100, write-bandwidth:
 store-id=2 node-id=2 status=ok accepting all reported=[cpu:45, write-bandwidth:40, byte-size:50] adjusted=[cpu:45, write-bandwidth:40, byte-size:50] node-reported-cpu=45 node-adjusted-cpu=45 seq=2
   top-k-ranges (local-store-id=1) dim=ByteSize: r1 r2
 
-# StoreLeaseholderMsg not containing r1 and r4 since no longer the leaseholder.
+# StoreLeaseholderMsg not containing r1 and r4 since no longer the
+# leaseholder. Change the load of r2, while MaybeSpanConfIsPopulated=false,
+# to ensure that we notice it.
 store-leaseholder-msg
 store-id=1
-  range-id=2 not-populated
+  range-id=2 load=[40,20,5] raft-cpu=10 span-config-not-populated
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
     store-id=2 replica-id=2 type=VOTER_FULL
-  range-id=3 not-populated
+  range-id=3 load=[30,10,10] raft-cpu=25 span-config-not-populated
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
     store-id=2 replica-id=2 type=VOTER_FULL
 ----
 
-# r1 and r4 no longer mentioned in the top-k.
+# r1 and r4 no longer mentioned in the top-k. And the top-k ordering has
+# changed for both stores since r2 is now more than r3 in terms of CPU,
+# but less than r3 in terms of ByteSize.
 get-load-info
 ----
 store-id=1 node-id=1 status=ok accepting all reported=[cpu:100, write-bandwidth:40, byte-size:50] adjusted=[cpu:100, write-bandwidth:40, byte-size:50] node-reported-cpu=100 node-adjusted-cpu=100 seq=1
-  top-k-ranges (local-store-id=1) dim=CPURate: r3 r2
+  top-k-ranges (local-store-id=1) dim=CPURate: r2 r3
 store-id=2 node-id=2 status=ok accepting all reported=[cpu:45, write-bandwidth:40, byte-size:50] adjusted=[cpu:45, write-bandwidth:40, byte-size:50] node-reported-cpu=45 node-adjusted-cpu=45 seq=2
-  top-k-ranges (local-store-id=1) dim=ByteSize: r2 r3
+  top-k-ranges (local-store-id=1) dim=ByteSize: r3 r2
 
 ranges
 ----
-range-id=2 local-store=1 load=[cpu:20, write-bandwidth:10, byte-size:15] raft-cpu=10
+range-id=2 local-store=1 load=[cpu:40, write-bandwidth:20, byte-size:5] raft-cpu=10
   store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
   store-id=2 replica-id=2 type=VOTER_FULL
 range-id=3 local-store=1 load=[cpu:30, write-bandwidth:10, byte-size:10] raft-cpu=25
@@ -109,17 +113,17 @@ range-id=3 local-store=1 load=[cpu:30, write-bandwidth:10, byte-size:10] raft-cp
 # of replicas, and the other the replica-id of one of the store's replicas.
 store-leaseholder-msg
 store-id=1
-  range-id=2 not-populated
+  range-id=2 load=[40,20,30] raft-cpu=10 span-config-not-populated
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
-  range-id=3 not-populated
+  range-id=3 load=[30,10,10] raft-cpu=25 span-config-not-populated
     store-id=2 replica-id=5 type=VOTER_FULL
     store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
 ----
 
 # The ranges reflect the latest state from the StoreLeaseholderMsg.
 ranges
 ----
-range-id=2 local-store=1 load=[cpu:20, write-bandwidth:10, byte-size:15] raft-cpu=10
+range-id=2 local-store=1 load=[cpu:40, write-bandwidth:20, byte-size:30] raft-cpu=10
   store-id=1 replica-id=1 type=VOTER_FULL leaseholder=true
 range-id=3 local-store=1 load=[cpu:30, write-bandwidth:10, byte-size:10] raft-cpu=25
   store-id=2 replica-id=5 type=VOTER_FULL