qrep: derive pull columns from source instead of catalog schema

flow/connectors/postgres/postgres_source.go

@@ -410,7 +410,7 @@ func (c *PostgresConnector) GetSelectedColumns(
 	}
 	excludedColumnsSQL := ""
 	if len(excludedColumns) > 0 {
-		excludedColumnsSQL = "AND a.attname NOT IN (" + strings.Join(quotedExcludedColumns, ",") + ")"
+		excludedColumnsSQL = " AND a.attname NOT IN (" + strings.Join(quotedExcludedColumns, ",") + ")"
 	}
 
 	getColumnsSQL := `
@@ -421,7 +421,7 @@ func (c *PostgresConnector) GetSelectedColumns(
 		WHERE n.nspname = $1
 		AND c.relname = $2
 		AND a.attnum > 0
-		AND NOT a.attisdropped ` + excludedColumnsSQL
+		AND NOT a.attisdropped` + excludedColumnsSQL + ` ORDER BY a.attnum`
 
 	rows, err := c.conn.Query(ctx, getColumnsSQL, sourceTable.Namespace, sourceTable.Table)
 	if err != nil {

flow/connectors/postgres/qrep_source.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"log/slog"
 	"math"
-	"slices"
 	"strconv"
 	"strings"
 
@@ -325,62 +324,64 @@ func (c *PostgresConnector) GetMaxValue(
 
 func (c *PostgresConnector) PullQRepRecords(
 	ctx context.Context,
-	catalogPool shared.CatalogPool,
+	_catalogPool shared.CatalogPool,
 	_otelManager *otel_metrics.OtelManager,
 	config *protos.QRepConfig,
 	dstType protos.DBType,
 	partition *protos.QRepPartition,
 	stream *model.QRecordStream,
 ) (int64, int64, error) {
-	return corePullQRepRecords(c, ctx, catalogPool, config, partition, &RecordStreamSink{
+	return corePullQRepRecords(c, ctx, config, partition, &RecordStreamSink{
 		QRecordStream:   stream,
 		DestinationType: dstType,
 	})
 }
 
 func (c *PostgresConnector) PullPgQRepRecords(
 	ctx context.Context,
-	catalogPool shared.CatalogPool,
+	_catalogPool shared.CatalogPool,
 	_otelManager *otel_metrics.OtelManager,
 	config *protos.QRepConfig,
 	_dstType protos.DBType,
 	partition *protos.QRepPartition,
 	stream PgCopyWriter,
 ) (int64, int64, error) {
-	return corePullQRepRecords(c, ctx, catalogPool, config, partition, stream)
+	return corePullQRepRecords(c, ctx, config, partition, stream)
 }
 
 func corePullQRepRecords(
 	c *PostgresConnector,
 	ctx context.Context,
-	catalogPool shared.CatalogPool,
 	config *protos.QRepConfig,
 	partition *protos.QRepPartition,
 	sink QRepPullSink,
 ) (int64, int64, error) {
 	partitionIdLog := slog.String(string(shared.PartitionIDKey), partition.PartitionId)
 
+	parsedSrcTable, err := common.ParseTableIdentifier(config.WatermarkTable)
+	if err != nil {
+		c.logger.Error("unable to parse source table", slog.Any("error", err))
+		return 0, 0, fmt.Errorf("unable to parse source table: %w", err)
+	}
+
 	selectedColumns := "*"
 	if len(config.Exclude) != 0 || len(partition.ChildTableRanges) > 0 {
-		tableSchema, err := internal.LoadTableSchemaFromCatalog(ctx, catalogPool, config.ParentMirrorName, config.DestinationTableIdentifier)
+		// derive columns from the source directly; the catalog schema may be absent
+		// (e.g. resync/table-addition renames the destination) and isn't needed here
+		columns, err := c.GetSelectedColumns(ctx, parsedSrcTable, config.Exclude)
 		if err != nil {
-			return 0, 0, fmt.Errorf("failed to load table schema: %w", err)
+			return 0, 0, fmt.Errorf("failed to get selected columns: %w", err)
 		}
-		quotedColumns := make([]string, 0, len(tableSchema.Columns))
-		for _, col := range tableSchema.Columns {
-			if !slices.Contains(config.Exclude, col.Name) {
-				quotedColumns = append(quotedColumns, common.QuoteIdentifier(col.Name))
-			}
+		if len(columns) == 0 {
+			return 0, 0, fmt.Errorf("table %s doesn't have queriable columns", parsedSrcTable)
+		}
+		quotedColumns := make([]string, 0, len(columns))
+		for _, col := range columns {
+			quotedColumns = append(quotedColumns, common.QuoteIdentifier(col))
 		}
 		selectedColumns = strings.Join(quotedColumns, ",")
 	}
 
-	parsedSrcTable, err := common.ParseTableIdentifier(config.WatermarkTable)
-	if err != nil {
-		c.logger.Error("unable to parse source table", slog.Any("error", err))
-		return 0, 0, fmt.Errorf("unable to parse source table: %w", err)
-	}
-
 	if partition.FullTablePartition {
 		c.logger.Info("pulling full table partition", partitionIdLog)
 		executor, err := c.NewQRepQueryExecutorSnapshot(ctx, config.Env, config.Version, config.SnapshotName,