changefeedccl: add more logging of spans during DistSQL planning

This patch adds more logging of spans during various stages of DistSQL
planning to help with debugging plans with erroneous spans.

Release note: None

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -37,6 +37,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
 )
 
@@ -240,6 +241,9 @@ func startDistChangefeed(
 	if err != nil {
 		return err
 	}
+	if log.ExpensiveLogEnabled(ctx, 2) {
+		log.Infof(ctx, "tracked spans: %s", trackedSpans)
+	}
 	localState.trackedSpans = trackedSpans
 
 	// Changefeed flows handle transactional consistency themselves.
@@ -385,9 +389,15 @@ func makePlan(
 		if err != nil {
 			return nil, nil, err
 		}
+		if log.ExpensiveLogEnabled(ctx, 2) {
+			log.Infof(ctx, "spans returned by DistSQL: %s", spanPartitions)
+		}
 		switch {
 		case distMode == sql.DistributionTypeNone || rangeDistribution == int64(defaultDistribution):
 		case rangeDistribution == int64(balancedSimpleDistribution):
+			if log.ExpensiveLogEnabled(ctx, 2) {
+				log.Infof(ctx, "rebalancing ranges using balanced simple distribution")
+			}
 			sender := execCtx.ExecCfg().DB.NonTransactionalSender()
 			distSender := sender.(*kv.CrossRangeTxnWrapperSender).Wrapped().(*kvcoord.DistSender)
 
@@ -396,6 +406,9 @@ func makePlan(
 			if err != nil {
 				return nil, nil, err
 			}
+			if log.ExpensiveLogEnabled(ctx, 2) {
+				log.Infof(ctx, "spans after balanced simple distribution rebalancing: %s", spanPartitions)
+			}
 		default:
 			return nil, nil, errors.AssertionFailedf("unsupported dist strategy %d and dist mode %d",
 				rangeDistribution, distMode)
@@ -426,6 +439,9 @@ func makePlan(
 
 		aggregatorSpecs := make([]*execinfrapb.ChangeAggregatorSpec, len(spanPartitions))
 		for i, sp := range spanPartitions {
+			if log.ExpensiveLogEnabled(ctx, 2) {
+				log.Infof(ctx, "watched spans for node %d: %s", sp.SQLInstanceID, sp)
+			}
 			watches := make([]execinfrapb.ChangeAggregatorSpec_Watch, len(sp.Spans))
 			for watchIdx, nodeSpan := range sp.Spans {
 				initialResolved := initialHighWater