changefeedccl: use new bulk oracle for changefeed planning

This change uses the BulkOracle by default as part of changefeed
planning, instead of the bin packing oracle. This will allow changefeeds
to have plans that randomly assign spans to any replica, including
followers if enabled, following locality filter constraints.

A new cluster setting, `changefeed.balanced_distribution.enabled`,
protects this change. When enabled (by default), changefeeds will use
the new BulkOracle for planning. If disabled, changefeeds will use the
previous bin packing oracle.

Epic: none
Fixes: cockroachdb#119777
Fixes: cockroachdb#114611

Release note (enterprise change): Changefeeds now use the BulkOracle for
planning, which distributes work evenly across all replica in the
locality filter, including followers if enabled. This is enabled by
default with the cluster setting
`changefeed.balanced_distribution.enabled`. If disabled, changefeed
planning reverts to its previous bin packing oracle.

pkg/ccl/changefeedccl/BUILD.bazel

@@ -54,6 +54,7 @@ go_library(
         "//pkg/ccl/changefeedccl/kvevent",
         "//pkg/ccl/changefeedccl/kvfeed",
         "//pkg/ccl/changefeedccl/schemafeed",
+        "//pkg/ccl/kvccl/kvfollowerreadsccl",
         "//pkg/ccl/utilccl",
         "//pkg/cloud",
         "//pkg/cloud/externalconn",

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdceval"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/kvfeed"
+	"github.com/cockroachdb/cockroach/pkg/ccl/kvccl/kvfollowerreadsccl"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprofiler"
 	"github.com/cockroachdb/cockroach/pkg/kv"
@@ -348,6 +349,12 @@ var enableBalancedRangeDistribution = settings.RegisterBoolSetting(
 		"changefeed.balance_range_distribution.enable", false),
 ).WithPublic()
 
+var useBulkOracle = settings.RegisterBoolSetting(
+	settings.TenantWritable,
+	"changefeed.random_replica_selection.enabled",
+	"randomize the selection of which replica backs up each range",
+	false)
+
 func makePlan(
 	execCtx sql.JobExecContext,
 	jobID jobspb.JobID,
@@ -372,8 +379,13 @@ func makePlan(
 			}
 		}
 
-		planCtx := dsp.NewPlanningCtxWithOracle(ctx, execCtx.ExtendedEvalContext(), nil /* planner */, blankTxn,
-			sql.DistributionType(distMode), physicalplan.DefaultReplicaChooser, locFilter)
+		evalCtx := execCtx.ExtendedEvalContext()
+		oracle := physicalplan.DefaultReplicaChooser
+		if useBulkOracle.Get(&evalCtx.Settings.SV) {
+			oracle = kvfollowerreadsccl.NewBulkOracle(dsp.ReplicaOracleConfig(evalCtx.Locality), locFilter)
+		}
+		planCtx := dsp.NewPlanningCtxWithOracle(ctx, execCtx.ExtendedEvalContext(), nil, /* planner */
+			blankTxn, sql.DistributionType(distMode), oracle, locFilter)
 		spanPartitions, err := dsp.PartitionSpans(ctx, planCtx, trackedSpans)
 		if err != nil {
 			return nil, nil, err