add resolved tables to specs

Author: andyyang890

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -13,6 +13,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/changefeedpb"
 	"github.com/cockroachdb/cockroach/pkg/ccl/kvccl/kvfollowerreadsccl"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
@@ -470,6 +471,20 @@ func makePlan(
 			}
 		}
 
+		var resolvedTables *changefeedpb.ResolvedTables
+		if progressConfig != nil && progressConfig.PerTableTracking {
+			var rt changefeedpb.ResolvedTables
+			if err := execCtx.ExecCfg().InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+				return readChangefeedJobInfo(ctx, resolvedTablesFilename, &rt, txn, jobID)
+			}); err != nil {
+				return nil, nil, err
+			}
+			resolvedTables = &rt
+			if log.ExpensiveLogEnabled(ctx, 2) {
+				log.Dev.Infof(ctx, "read resolved tables: %v", resolvedTables)
+			}
+		}
+
 		aggregatorSpecs := make([]*execinfrapb.ChangeAggregatorSpec, len(spanPartitions))
 		for i, sp := range spanPartitions {
 			if log.ExpensiveLogEnabled(ctx, 2) {
@@ -493,6 +508,7 @@ func makePlan(
 				Select:              execinfrapb.Expression{Expr: details.Select},
 				Description:         description,
 				ProgressConfig:      progressConfig,
+				ResolvedTables:      resolvedTables,
 			}
 		}
 
@@ -508,6 +524,7 @@ func makePlan(
 			UserProto:           execCtx.User().EncodeProto(),
 			Description:         description,
 			ProgressConfig:      progressConfig,
+			ResolvedTables:      resolvedTables,
 		}
 
 		if haveKnobs && maybeCfKnobs.OnDistflowSpec != nil {

pkg/ccl/changefeedccl/changefeed_job_info.go

@@ -49,7 +49,6 @@ func writeChangefeedJobInfo(
 
 // readChangefeedJobInfo reads a changefeed job info protobuf from the
 // job_info table. A changefeed-specific filename is required.
-// TODO(#148119): Use this function to read.
 func readChangefeedJobInfo(
 	ctx context.Context, filename string, info protoutil.Message, txn isql.Txn, jobID jobspb.JobID,
 ) error {