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

Author: mw5h

pkg/sql/bulkmerge/merge_loopback.go

@@ -10,6 +10,7 @@ import (
 
 	"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"
@@ -41,8 +42,11 @@ func (m *mergeLoopback) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMetadat
 		return nil, m.DrainHelper()
 	}
 	m.done = true
+	// Generate a routing key for the current SQL instance (where this processor is running).
+	// This ensures the routing key matches one of the spans in the range router.
+	routingDatum, _ := physicalplan.RoutingDatumsForSQLInstance(m.FlowCtx.NodeID.SQLInstanceID())
 	return rowenc.EncDatumRow{
-		rowenc.EncDatum{Datum: tree.NewDBytes("loopback")},
+		routingDatum,
 		rowenc.EncDatum{Datum: tree.NewDInt(1)},
 	}, nil
 }

pkg/sql/bulkmerge/merge_test.go

@@ -72,5 +72,7 @@ func TestMergeProcessors(t *testing.T) {
 	)
 
 	require.NoError(t, rowWriter.Err())
-	require.Equal(t, result, "loopback->merge->coordinator")
+	// The output includes the routing key (e.g., "node1") from the SQL instance
+	// where the merge loopback processor runs.
+	require.Equal(t, result, "node1->merge->coordinator")
 }

pkg/sql/physicalplan/routing.go

@@ -56,10 +56,13 @@ func MakeInstanceRouter(
 	ids []base.SQLInstanceID,
 ) (execinfrapb.OutputRouterSpec_RangeRouterSpec, error) {
 	var zero execinfrapb.OutputRouterSpec_RangeRouterSpec
-	defaultStream := int32(0)
 	rangeRouterSpec := execinfrapb.OutputRouterSpec_RangeRouterSpec{
-		Spans:       nil,
-		DefaultDest: &defaultStream,
+		Spans: nil,
+		// DefaultDest is nil so that any routing key that doesn't match a span
+		// will produce an error. This ensures we catch coordination bugs where
+		// a routing key is generated for an instance not in the router's span list,
+		// rather than silently routing to an arbitrary instance.
+		DefaultDest: nil,
 		Encodings: []execinfrapb.OutputRouterSpec_RangeRouterSpec_ColumnEncoding{
 			{
 				Column:   0,