streamingccl: refactor SHOW VIRTUAL CLUSTER ... with REPLICATION STATUS

Epic: none

Release note (sql change): This patch refactors the output of SHOW VC WITH
REPLICATION STATUS to remove the replication_job_id return field and to reorder
to following columns to: retained_time, replicated_time, replication_lag,
cutover_time.

pkg/ccl/streamingccl/streamingest/replication_stream_e2e_test.go

@@ -1037,14 +1037,13 @@ func TestTenantStreamingShowTenant(t *testing.T) {
 		serviceMode    string
 		source         string
 		sourceUri      string
-		jobId          int
 		replicationLag string
 		maxReplTime    time.Time
 		protectedTime  time.Time
 		cutoverTime    []byte // should be nil
 	)
 	row := c.DestSysSQL.QueryRow(t, fmt.Sprintf("SHOW TENANT %s WITH REPLICATION STATUS", args.DestTenantName))
-	row.Scan(&id, &dest, &status, &serviceMode, &source, &sourceUri, &jobId, &replicationLag, &maxReplTime, &protectedTime, &cutoverTime)
+	row.Scan(&id, &dest, &status, &serviceMode, &source, &sourceUri, &protectedTime, &maxReplTime, &replicationLag, &cutoverTime)
 	require.Equal(t, 2, id)
 	require.Equal(t, "destination", dest)
 	require.Equal(t, "replicating", status)
@@ -1053,7 +1052,6 @@ func TestTenantStreamingShowTenant(t *testing.T) {
 	expectedURI, err := streamclient.RedactSourceURI(c.SrcURL.String())
 	require.NoError(t, err)
 	require.Equal(t, expectedURI, sourceUri)
-	require.Equal(t, ingestionJobID, jobId)
 	require.Less(t, maxReplTime, timeutil.Now())
 	require.Less(t, protectedTime, timeutil.Now())
 	require.GreaterOrEqual(t, maxReplTime, targetReplicatedTime.GoTime())

pkg/sql/catalog/colinfo/result_columns.go

@@ -304,13 +304,12 @@ var TenantColumns = ResultColumns{
 var TenantColumnsWithReplication = ResultColumns{
 	{Name: "source_tenant_name", Typ: types.String},
 	{Name: "source_cluster_uri", Typ: types.String},
-	{Name: "replication_job_id", Typ: types.Int},
-	{Name: "replication_lag", Typ: types.Interval},
-	// The latest fully replicated time.
-	{Name: "replicated_time", Typ: types.TimestampTZ},
 	// The protected timestamp on the destination cluster, meaning we cannot
 	// cutover to before this time.
 	{Name: "retained_time", Typ: types.TimestampTZ},
+	// The latest fully replicated time.
+	{Name: "replicated_time", Typ: types.TimestampTZ},
+	{Name: "replication_lag", Typ: types.Interval},
 	{Name: "cutover_time", Typ: types.Decimal},
 }
 

pkg/sql/show_tenant.go

@@ -228,15 +228,13 @@ func (n *showTenantNode) Values() tree.Datums {
 		// This is a 'SHOW VIRTUAL CLUSTER name WITH REPLICATION STATUS' command.
 		sourceTenantName := tree.DNull
 		sourceClusterUri := tree.DNull
-		replicationJobId := tree.DNull
 		replicatedTimestamp := tree.DNull
 		retainedTimestamp := tree.DNull
 		cutoverTimestamp := tree.DNull
 		replicationLag := tree.DNull
 
 		replicationInfo := v.replicationInfo
 		if replicationInfo != nil {
-			replicationJobId = tree.NewDInt(tree.DInt(tenantInfo.PhysicalReplicationConsumerJobID))
 			sourceTenantName = tree.NewDString(string(replicationInfo.IngestionDetails.SourceTenantName))
 			sourceClusterUri = tree.NewDString(replicationInfo.IngestionDetails.StreamAddress)
 			if replicationInfo.ReplicationLagInfo != nil {
@@ -269,10 +267,9 @@ func (n *showTenantNode) Values() tree.Datums {
 		result = append(result,
 			sourceTenantName,
 			sourceClusterUri,
-			replicationJobId,
-			replicationLag,
-			replicatedTimestamp,
 			retainedTimestamp,
+			replicatedTimestamp,
+			replicationLag,
 			cutoverTimestamp,
 		)
 	}