diff --git a/internal/cdc/applier.go b/internal/cdc/applier.go
index 5a32ae0..15181e0 100644
--- a/internal/cdc/applier.go
+++ b/internal/cdc/applier.go
@@ -318,17 +318,70 @@ func (a *Applier) applyAvailable(ctx context.Context, conn *pgx.Conn, progress L
}
const (
- // Catch-up batches are bounded independently by source transaction count,
- // row changes, and decoded payload size. A resident source transaction is
- // never split: when it alone exceeds a bound it becomes a one-transaction
- // replay claim, preserving its atomicity while retaining set-based DML and
- // durable claim receipts. Only disk-spilled transactions keep the streaming
- // standalone path.
+ // Catch-up scheduling slices are bounded independently by source transaction
+ // count, row changes, and decoded payload size. A four-slice durable window
+ // gives the key-affine executor enough work to keep every target session busy.
+ // A resident source transaction is never split: when it alone exceeds the
+ // window it becomes a one-transaction replay claim. Only disk-spilled
+ // transactions keep the streaming standalone path.
applyBatchMaxTransactions = 16384
applyBatchDefaultChanges = 131072
applyBatchDefaultDataBytes = 32 << 20
+
+ // Eight sessions is one replay scheduling slice. Four configured slices form
+ // the default durable target wave; higher session counts extend the wave so
+ // per-session work does not shrink. The claim format does not change, so an
+ // older binary can reconstruct and finish an active window from its stored
+ // EndLSN, lane count, manifest, and receipts.
+ replayWorkersPerClaimSlice = 8
)
+func replayClaimWindowSlices(workers int) int {
+ if workers < replayWorkersPerClaimSlice {
+ return 1
+ }
+ // Match crdb-to-pg's four-claim target wave even at the conservative default
+ // of eight sessions. More than 32 sessions add one slice per additional
+ // eight sessions so per-session work does not shrink as concurrency grows.
+ return max(4, (workers+replayWorkersPerClaimSlice-1)/replayWorkersPerClaimSlice)
+}
+
+func replayClaimLaneCount(workers int) int {
+ if workers <= 1 {
+ return 1
+ }
+ // Extra logical lanes smooth hash-skew tails while worker counts are small.
+ // Once there are 32 real sessions, retain one deterministic lane per session
+ // instead of multiplying receipt transactions and synchronous commits.
+ return min(max(workers, workers*4), max(32, workers))
+}
+
+func replayClaimWindowLimits(workers, changes int, dataBytes int64) (int, int, int64) {
+ slices := replayClaimWindowSlices(workers)
+ maxInt := int(^uint(0) >> 1)
+ transactionsLimit := applyBatchMaxTransactions
+ changesLimit := changes
+ dataBytesLimit := dataBytes
+ if slices > 1 {
+ if transactionsLimit > maxInt/slices {
+ transactionsLimit = maxInt
+ } else {
+ transactionsLimit *= slices
+ }
+ if changesLimit > maxInt/slices {
+ changesLimit = maxInt
+ } else {
+ changesLimit *= slices
+ }
+ if dataBytesLimit > int64(^uint64(0)>>1)/int64(slices) {
+ dataBytesLimit = int64(^uint64(0) >> 1)
+ } else {
+ dataBytesLimit *= int64(slices)
+ }
+ }
+ return transactionsLimit, changesLimit, dataBytesLimit
+}
+
func (a *Applier) applyFromReader(
ctx context.Context,
conn *pgx.Conn,
@@ -361,7 +414,10 @@ func (a *Applier) applyFromReader(
ctx, conn, relationCache, statementCache, workers, batch, progress, claim,
)
}
- batch := make([]Transaction, 0, applyBatchMaxTransactions)
+ transactionsLimit, changesLimit, dataBytesLimit := replayClaimWindowLimits(
+ a.config.ReplayWorkers, a.config.BatchMaxChanges, a.config.BatchMaxDataBytes,
+ )
+ batch := make([]Transaction, 0, transactionsLimit)
batchChanges := 0
var batchDataBytes int64
for {
@@ -452,9 +508,9 @@ func (a *Applier) applyFromReader(
transactionChanges := int(transaction.ChangeCount())
transactionDataBytes := int64(transactionApplyDataBytes(&transaction))
if len(batch) != 0 &&
- (batchChanges+transactionChanges > a.config.BatchMaxChanges ||
- batchDataBytes+transactionDataBytes > a.config.BatchMaxDataBytes) {
- // Keep the configured claim bound without splitting this source
+ (batchChanges+transactionChanges > changesLimit ||
+ batchDataBytes+transactionDataBytes > dataBytesLimit) {
+ // Keep the configured wave bound without splitting this source
// transaction. The next pass will claim it alone.
reader.pending = &transaction
return applyBatch(batch, nil)
@@ -462,9 +518,9 @@ func (a *Applier) applyFromReader(
batchChanges += transactionChanges
batchDataBytes += transactionDataBytes
batch = append(batch, transaction)
- if len(batch) >= applyBatchMaxTransactions ||
- batchChanges >= a.config.BatchMaxChanges ||
- batchDataBytes >= a.config.BatchMaxDataBytes {
+ if len(batch) >= transactionsLimit ||
+ batchChanges >= changesLimit ||
+ batchDataBytes >= dataBytesLimit {
return applyBatch(batch, nil)
}
}
@@ -747,10 +803,9 @@ func (a *Applier) applyTransactionBatchWithWorkers(
if resume != nil {
laneCount = resume.LaneCount
} else if laneCount > 1 {
- // More logical lanes than sessions reduce hash-skew tails without opening
- // more target connections. Receipts remain keyed by the durable lane, and
- // the executor size-balances those independent lanes across the workers.
- laneCount = min(laneCount*4, migrationconfig.ReplayWorkersMax)
+ // Extra logical lanes reduce hash-skew tails at small worker counts. The
+ // executor size-balances those deterministic lanes across real sessions.
+ laneCount = replayClaimLaneCount(laneCount)
}
if laneCount > 1 && (resume != nil || len(workers) > 1) {
startGeneration := a.effectiveStreamGeneration()
@@ -872,7 +927,11 @@ func (a *Applier) applyTransactionBatchWithWorkers(
}
func shouldUseConcurrentReplayPlan(resume *replayClaim, plan replayPlan) bool {
- return resume != nil || (plan.HasParallel && !replayPlanHasSerialWork(plan))
+ // executeReplayPlan treats every unsafe source transaction as an ordered
+ // barrier between parallel epochs. Keeping those barriers inside the durable
+ // window lets safe work on either side use all target sessions without moving
+ // the frontier past the barrier or splitting a source transaction.
+ return resume != nil || plan.HasParallel
}
type relationBatchedChange struct {
@@ -2488,6 +2547,118 @@ func (p *applyPipeline) copyFrom(
const minimumTextCopyStageRows = 64
+const minimumBinaryCopyStageRows = 64
+
+// loadBinaryCopyStage is the built-in-type counterpart of loadTextCopyStage.
+// It mirrors crdb-to-pg's fast path: COPY a lane group into a transaction-local
+// typed stage, then consume it with one set-based statement. The stage and DML
+// live in the same replay-work transaction as the durable receipt.
+func (p *applyPipeline) loadBinaryCopyStage(
+ relation *targetRelation,
+ kind ChangeKind,
+ columns []targetColumn,
+ values []TupleDatum,
+ rowCount int,
+) (string, bool, error) {
+ if rowCount < minimumBinaryCopyStageRows || len(columns) == 0 ||
+ !relation.capabilities.binaryCopy {
+ return "", false, nil
+ }
+ data, supported, err := binaryCopyStageData(relation, kind, columns, values, rowCount)
+ if err != nil || !supported {
+ return "", supported, err
+ }
+ stage := textCopyStageName(relation, kind, columns)
+ var create strings.Builder
+ create.WriteString("CREATE TEMP TABLE IF NOT EXISTS ")
+ create.WriteString(stage)
+ create.WriteString(" ON COMMIT DELETE ROWS AS SELECT 0::bigint AS ordinal")
+ for i, column := range columns {
+ create.WriteByte(',')
+ create.WriteString("pgmigrate_target.")
+ create.WriteString(column.quoted)
+ fmt.Fprintf(&create, " AS column_%d", i)
+ }
+ create.WriteString(" FROM ")
+ create.WriteString(relation.quoted)
+ create.WriteString(" AS pgmigrate_target WITH NO DATA")
+ p.queueUnprepared(create.String(), nil, applyExpectation{
+ relation: relation, kind: kind,
+ description: "create binary replay stage for " + relation.quoted, expectedRows: -1,
+ })
+ p.queueUnprepared("TRUNCATE "+stage, nil, applyExpectation{
+ relation: relation, kind: kind,
+ description: "clear binary replay stage for " + relation.quoted, expectedRows: -1,
+ })
+
+ var copySQL strings.Builder
+ copySQL.WriteString("COPY ")
+ copySQL.WriteString(stage)
+ copySQL.WriteString(" (ordinal")
+ for i := range columns {
+ fmt.Fprintf(©SQL, ",column_%d", i)
+ }
+ copySQL.WriteString(") FROM STDIN BINARY")
+ if err := p.copyFrom(
+ relation, kind, "binary copy into replay stage for "+relation.quoted,
+ copySQL.String(), data, rowCount,
+ ); err != nil {
+ return "", true, err
+ }
+ return stage, true, nil
+}
+
+func binaryCopyStageData(
+ relation *targetRelation,
+ kind ChangeKind,
+ columns []targetColumn,
+ values []TupleDatum,
+ rowCount int,
+) ([]byte, bool, error) {
+ if rowCount < 0 || len(values) != rowCount*len(columns) {
+ return nil, true, divergenceFor(relation, kind, fmt.Sprintf(
+ "binary stage has %d values for %d rows of %d columns",
+ len(values), rowCount, len(columns),
+ ))
+ }
+ estimatedBytes := 21 + rowCount*(14+len(columns)*4)
+ for row := 0; row < rowCount; row++ {
+ for columnIndex, column := range columns {
+ datum := values[row*len(columns)+columnIndex]
+ switch datum.Kind {
+ case DatumNull:
+ case DatumBinary:
+ if _, err := datumParamForColumn(relation, column, datum, kind); err != nil {
+ return nil, true, err
+ }
+ estimatedBytes += len(datum.Data)
+ default:
+ return nil, false, nil
+ }
+ }
+ }
+ data := make([]byte, 0, estimatedBytes)
+ data = append(data, []byte("PGCOPY\n\xff\r\n\x00")...)
+ data = binary.BigEndian.AppendUint32(data, 0)
+ data = binary.BigEndian.AppendUint32(data, 0)
+ for row := 0; row < rowCount; row++ {
+ data = binary.BigEndian.AppendUint16(data, uint16(len(columns)+1))
+ data = binary.BigEndian.AppendUint32(data, 8)
+ data = binary.BigEndian.AppendUint64(data, uint64(row))
+ for columnIndex := range columns {
+ datum := values[row*len(columns)+columnIndex]
+ if datum.Kind == DatumNull {
+ data = binary.BigEndian.AppendUint32(data, ^uint32(0))
+ continue
+ }
+ data = binary.BigEndian.AppendUint32(data, uint32(len(datum.Data)))
+ data = append(data, datum.Data...)
+ }
+ }
+ data = binary.BigEndian.AppendUint16(data, ^uint16(0))
+ return data, true, nil
+}
+
// loadTextCopyStage copies text pgoutput values into a target-typed temporary
// relation. This is the escape hatch that parameter arrays cannot provide
// efficiently for user-defined types: PostgreSQL's COPY input functions do the
@@ -3133,6 +3304,9 @@ func applyPrimaryKeyUpsertChunk(
if len(changes) == 0 {
return nil
}
+ if applied, err := applyPrimaryKeyUpsertBinaryStage(replay, relation, changes); applied || err != nil {
+ return err
+ }
if applied, err := applyPrimaryKeyUpsertTextStage(replay, relation, changes); applied || err != nil {
return err
}
@@ -3149,6 +3323,46 @@ func applyPrimaryKeyUpsertChunk(
return nil
}
+func applyPrimaryKeyUpsertBinaryStage(
+ replay *applyPipeline,
+ relation *targetRelation,
+ changes []Change,
+) (bool, error) {
+ values := make([]TupleDatum, 0, len(changes)*len(relation.columns))
+ for row := range changes {
+ if err := validateTuple(relation, changes[row].New, ChangeUpdate); err != nil {
+ return true, err
+ }
+ for _, column := range relation.columns {
+ values = append(values, (*changes[row].New)[column.sourceIndex])
+ }
+ }
+ stage, applied, err := replay.loadBinaryCopyStage(
+ relation, ChangeUpdate, relation.columns, values, len(changes),
+ )
+ if err != nil || !applied {
+ return applied, err
+ }
+ var sql strings.Builder
+ writePrimaryKeyUpsertPrefix(&sql, relation)
+ sql.WriteString(" SELECT ")
+ for i := range relation.columns {
+ if i != 0 {
+ sql.WriteByte(',')
+ }
+ fmt.Fprintf(&sql, "column_%d", i)
+ }
+ sql.WriteString(" FROM ")
+ sql.WriteString(stage)
+ sql.WriteString(" ORDER BY ordinal")
+ appendPrimaryKeyConflictClause(&sql, relation)
+ return true, replay.queue(sql.String(), nil, applyExpectation{
+ relation: relation, kind: ChangeUpdate,
+ description: "binary-staged primary-key upsert into " + relation.quoted,
+ expectedRows: int64(len(changes)),
+ })
+}
+
func applyPrimaryKeyUpsertTextStage(
replay *applyPipeline,
relation *targetRelation,
@@ -4110,6 +4324,35 @@ func writeCompositeIdentityCTIDPredicate(
identityColumns []targetColumn,
batchColumnPrefix string,
batchColumnOffset int,
+) {
+ writeCompositeIdentityCTIDPredicateMode(
+ sql, relation, identityColumns, batchColumnPrefix, batchColumnOffset, false,
+ )
+}
+
+// writeCompositePrimaryKeyCTIDPredicate adds equal lower and upper bounds in
+// the target primary-key order. The scalar equalities remain for exactness,
+// while the row bounds stop PostgreSQL from choosing a smaller prefix index
+// and filtering the remaining key columns after a potentially huge scan.
+func writeCompositePrimaryKeyCTIDPredicate(
+ sql *strings.Builder,
+ relation *targetRelation,
+ identityColumns []targetColumn,
+ batchColumnPrefix string,
+ batchColumnOffset int,
+) {
+ writeCompositeIdentityCTIDPredicateMode(
+ sql, relation, identityColumns, batchColumnPrefix, batchColumnOffset, true,
+ )
+}
+
+func writeCompositeIdentityCTIDPredicateMode(
+ sql *strings.Builder,
+ relation *targetRelation,
+ identityColumns []targetColumn,
+ batchColumnPrefix string,
+ batchColumnOffset int,
+ primaryKeyBounds bool,
) {
sql.WriteString("pgmigrate_target.ctid=(SELECT pgmigrate_lookup.ctid FROM ")
sql.WriteString(relation.quoted)
@@ -4125,6 +4368,33 @@ func writeCompositeIdentityCTIDPredicate(
batchColumnPrefix, batchColumnOffset+i,
)
}
+ if primaryKeyBounds && len(identityColumns) > 1 {
+ writeLookupPrimaryKeyBound := func(operator string) {
+ sql.WriteString(" AND ROW(")
+ for i, column := range identityColumns {
+ if i != 0 {
+ sql.WriteByte(',')
+ }
+ sql.WriteString("pgmigrate_lookup.")
+ sql.WriteString(column.quoted)
+ }
+ sql.WriteByte(')')
+ sql.WriteString(operator)
+ sql.WriteString("ROW(")
+ for i := range identityColumns {
+ if i != 0 {
+ sql.WriteByte(',')
+ }
+ fmt.Fprintf(
+ sql, "pgmigrate_batch.%s%d",
+ batchColumnPrefix, batchColumnOffset+i,
+ )
+ }
+ sql.WriteByte(')')
+ }
+ writeLookupPrimaryKeyBound(">=")
+ writeLookupPrimaryKeyBound("<=")
+ }
sql.WriteString(" OFFSET 0)")
}
@@ -4634,12 +4904,64 @@ func writeDeleteIdentityPredicate(
offset int,
) {
if deleteUsesTargetPrimaryKey(relation, identityColumns) {
- writeCompositeIdentityCTIDPredicate(sql, relation, identityColumns, prefix, offset)
+ writeCompositePrimaryKeyCTIDPredicate(sql, relation, identityColumns, prefix, offset)
return
}
writeBatchIdentityPredicate(sql, identityColumns, prefix, offset)
}
+func appendPrimaryKeyDeletePredicate(
+ sql *strings.Builder,
+ params *[]rawParam,
+ relation *targetRelation,
+ primary []targetColumn,
+ tuple Tuple,
+) error {
+ positions := make([]int, len(primary))
+ for i, column := range primary {
+ datum := tuple[column.sourceIndex]
+ if datum.Kind == DatumUnchangedToast {
+ return divergenceFor(relation, ChangeDelete, "primary key contains unchanged TOAST")
+ }
+ param, err := datumParamForColumn(relation, column, datum, ChangeDelete)
+ if err != nil {
+ return err
+ }
+ *params = append(*params, param)
+ positions[i] = len(*params)
+ if i != 0 {
+ sql.WriteString(" AND ")
+ }
+ sql.WriteString(column.quoted)
+ fmt.Fprintf(sql, " = $%d", positions[i])
+ }
+ if len(primary) < 2 {
+ return nil
+ }
+ writeBound := func(operator string) {
+ sql.WriteString(" AND ROW(")
+ for i, column := range primary {
+ if i != 0 {
+ sql.WriteByte(',')
+ }
+ sql.WriteString(column.quoted)
+ }
+ sql.WriteByte(')')
+ sql.WriteString(operator)
+ sql.WriteString("ROW(")
+ for i, position := range positions {
+ if i != 0 {
+ sql.WriteByte(',')
+ }
+ fmt.Fprintf(sql, "$%d", position)
+ }
+ sql.WriteByte(')')
+ }
+ writeBound(">=")
+ writeBound("<=")
+ return nil
+}
+
func batchDeleteIdentityKey(
relation *targetRelation,
identityColumns []targetColumn,
@@ -4866,8 +5188,16 @@ func applyDelete(replay *applyPipeline, relation *targetRelation, change *Change
sql.WriteString(relation.quoted)
sql.WriteString(" WHERE ")
params := make([]rawParam, 0, len(relation.columns))
- if err := appendPredicate(&sql, ¶ms, relation, *change.Old, ChangeDelete); err != nil {
- return err
+ if primary, safe := primaryKeyDeleteColumns(relation); safe {
+ if err := appendPrimaryKeyDeletePredicate(
+ &sql, ¶ms, relation, primary, *change.Old,
+ ); err != nil {
+ return err
+ }
+ } else {
+ if err := appendPredicate(&sql, ¶ms, relation, *change.Old, ChangeDelete); err != nil {
+ return err
+ }
}
return replay.queue(sql.String(), params, applyExpectation{
relation: relation, kind: ChangeDelete,
diff --git a/internal/cdc/pipeline_test.go b/internal/cdc/pipeline_test.go
index 894e2de..bf8f51f 100644
--- a/internal/cdc/pipeline_test.go
+++ b/internal/cdc/pipeline_test.go
@@ -4,6 +4,7 @@ import (
"context"
"encoding/binary"
"errors"
+ "fmt"
"slices"
"strings"
"testing"
@@ -129,6 +130,49 @@ func TestApplierReplayBatchLimitsDefaultAndAllowOverrides(t *testing.T) {
}
}
+func TestReplayClaimWindowScalesOneConfiguredSlicePerEightSessions(t *testing.T) {
+ t.Parallel()
+ tests := []struct {
+ workers int
+ wantSlices int
+ wantTransactions int
+ wantChanges int
+ wantBytes int64
+ }{
+ {workers: 1, wantSlices: 1, wantTransactions: 16_384, wantChanges: 32_768, wantBytes: 8 << 20},
+ {workers: 8, wantSlices: 4, wantTransactions: 65_536, wantChanges: 131_072, wantBytes: 32 << 20},
+ {workers: 9, wantSlices: 4, wantTransactions: 65_536, wantChanges: 131_072, wantBytes: 32 << 20},
+ {workers: 32, wantSlices: 4, wantTransactions: 65_536, wantChanges: 131_072, wantBytes: 32 << 20},
+ {workers: 64, wantSlices: 8, wantTransactions: 131_072, wantChanges: 262_144, wantBytes: 64 << 20},
+ }
+ for _, test := range tests {
+ t.Run(fmt.Sprint(test.workers), func(t *testing.T) {
+ t.Parallel()
+ if got := replayClaimWindowSlices(test.workers); got != test.wantSlices {
+ t.Fatalf("replayClaimWindowSlices(%d) = %d, want %d", test.workers, got, test.wantSlices)
+ }
+ transactions, changes, dataBytes := replayClaimWindowLimits(test.workers, 32_768, 8<<20)
+ if transactions != test.wantTransactions || changes != test.wantChanges || dataBytes != test.wantBytes {
+ t.Fatalf(
+ "window limits for %d workers = %d tx / %d changes / %d bytes, want %d / %d / %d",
+ test.workers, transactions, changes, dataBytes,
+ test.wantTransactions, test.wantChanges, test.wantBytes,
+ )
+ }
+ })
+ }
+}
+
+func TestReplayClaimLaneCountUsesThirtyTwoDeterministicLanesBeforeMoreSessions(t *testing.T) {
+ t.Parallel()
+ tests := map[int]int{1: 1, 2: 8, 4: 16, 8: 32, 16: 32, 32: 32, 64: 64}
+ for workers, want := range tests {
+ if got := replayClaimLaneCount(workers); got != want {
+ t.Errorf("replayClaimLaneCount(%d) = %d, want %d", workers, got, want)
+ }
+ }
+}
+
func TestTargetRelationCacheReloadsOnlyForChangedSourceDefinition(t *testing.T) {
t.Parallel()
cache := newTargetRelationCache()
@@ -395,7 +439,11 @@ func TestPrimaryKeyDeleteUsesCatalogIndexOrder(t *testing.T) {
writeDeleteIdentityPredicate(&sql, relation, primary, "identity_", 0)
want := `pgmigrate_target.ctid=(SELECT pgmigrate_lookup.ctid FROM "shard_schema"."messages" AS pgmigrate_lookup ` +
`WHERE pgmigrate_lookup."app_pk"=pgmigrate_batch.identity_0 AND ` +
- `pgmigrate_lookup."id"=pgmigrate_batch.identity_1 OFFSET 0)`
+ `pgmigrate_lookup."id"=pgmigrate_batch.identity_1 AND ` +
+ `ROW(pgmigrate_lookup."app_pk",pgmigrate_lookup."id")>=` +
+ `ROW(pgmigrate_batch.identity_0,pgmigrate_batch.identity_1) AND ` +
+ `ROW(pgmigrate_lookup."app_pk",pgmigrate_lookup."id")<=` +
+ `ROW(pgmigrate_batch.identity_0,pgmigrate_batch.identity_1) OFFSET 0)`
if got := sql.String(); got != want {
t.Fatalf("delete predicate = %q, want forced target primary key %q", got, want)
}
@@ -406,6 +454,45 @@ func TestPrimaryKeyDeleteUsesCatalogIndexOrder(t *testing.T) {
}
}
+func TestSinglePrimaryKeyDeleteAddsExactCatalogOrderedBounds(t *testing.T) {
+ t.Parallel()
+ relation := &targetRelation{
+ quoted: `"shard_schema"."read_state"`,
+ capabilities: targetRelationCapabilities{keyedSetDML: true},
+ source: Relation{Columns: []Column{
+ {Name: "channel_cid"}, {Name: "app_pk"}, {Name: "user_id"},
+ }},
+ columns: []targetColumn{
+ {name: "channel_cid", quoted: `"channel_cid"`, sourceIndex: 0, key: true, primary: true, primaryPos: 3, notNull: true},
+ {name: "app_pk", quoted: `"app_pk"`, sourceIndex: 1, key: true, primary: true, primaryPos: 1, notNull: true},
+ {name: "user_id", quoted: `"user_id"`, sourceIndex: 2, key: true, primary: true, primaryPos: 2, notNull: true},
+ },
+ }
+ primary, safe := primaryKeyDeleteColumns(relation)
+ if !safe {
+ t.Fatal("complete primary key was not eligible for exact delete")
+ }
+ var sql strings.Builder
+ params := make([]rawParam, 0, len(primary))
+ tuple := Tuple{
+ {Kind: DatumText, Data: []byte("channel")},
+ {Kind: DatumText, Data: []byte("7")},
+ {Kind: DatumText, Data: []byte("user")},
+ }
+ if err := appendPrimaryKeyDeletePredicate(&sql, ¶ms, relation, primary, tuple); err != nil {
+ t.Fatal(err)
+ }
+ want := `"app_pk" = $1 AND "user_id" = $2 AND "channel_cid" = $3` +
+ ` AND ROW("app_pk","user_id","channel_cid")>=ROW($1,$2,$3)` +
+ ` AND ROW("app_pk","user_id","channel_cid")<=ROW($1,$2,$3)`
+ if got := sql.String(); got != want {
+ t.Fatalf("single delete predicate = %q, want %q", got, want)
+ }
+ if got := []string{string(params[0].data), string(params[1].data), string(params[2].data)}; !slices.Equal(got, []string{"7", "user", "channel"}) {
+ t.Fatalf("single delete params = %v, want catalog primary-key order", got)
+ }
+}
+
func TestPrimaryKeyUpsertRequiresCompleteStableRow(t *testing.T) {
t.Parallel()
relation := &targetRelation{
@@ -597,6 +684,53 @@ func TestTextCopyStageDataEscapesNullEmptyAndControlBytes(t *testing.T) {
}
}
+func TestBinaryCopyStageDataIncludesOrdinalAndExactBinaryValues(t *testing.T) {
+ t.Parallel()
+ relation := preparationRelation()
+ column := relation.columns[0]
+ data, supported, err := binaryCopyStageData(
+ relation,
+ ChangeUpdate,
+ []targetColumn{column},
+ []TupleDatum{{Kind: DatumBinary, Data: []byte{0, 0, 0, 7}}},
+ 1,
+ )
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !supported {
+ t.Fatal("binary stage unexpectedly unsupported")
+ }
+ if len(data) < 43 || string(data[:11]) != "PGCOPY\n\xff\r\n\x00" {
+ t.Fatalf("binary stage header is invalid: %x", data)
+ }
+ if fields := binary.BigEndian.Uint16(data[19:21]); fields != 2 {
+ t.Fatalf("binary stage fields = %d, want ordinal plus one value", fields)
+ }
+ if ordinalBytes := binary.BigEndian.Uint32(data[21:25]); ordinalBytes != 8 ||
+ binary.BigEndian.Uint64(data[25:33]) != 0 {
+ t.Fatalf("binary stage ordinal is invalid: %x", data[21:33])
+ }
+ if valueBytes := binary.BigEndian.Uint32(data[33:37]); valueBytes != 4 ||
+ !slices.Equal(data[37:41], []byte{0, 0, 0, 7}) {
+ t.Fatalf("binary stage value is invalid: %x", data[33:])
+ }
+
+ _, supported, err = binaryCopyStageData(
+ relation,
+ ChangeUpdate,
+ []targetColumn{column},
+ []TupleDatum{{Kind: DatumText, Data: []byte("7")}},
+ 1,
+ )
+ if err != nil {
+ t.Fatal(err)
+ }
+ if supported {
+ t.Fatal("text datum unexpectedly supported by binary stage")
+ }
+}
+
func TestTextCopyStageNameTracksShapeAndOperation(t *testing.T) {
t.Parallel()
relation := preparationRelation()
diff --git a/internal/cdc/replay_benchmark_integration_test.go b/internal/cdc/replay_benchmark_integration_test.go
index e2aac5c..cbfda5b 100644
--- a/internal/cdc/replay_benchmark_integration_test.go
+++ b/internal/cdc/replay_benchmark_integration_test.go
@@ -333,6 +333,18 @@ func TestPG17CDCReplayThroughput(t *testing.T) {
if err := <-applyDone; err != nil && !errors.Is(err, context.Canceled) {
t.Fatalf("stop benchmark applier: %v", err)
}
+ replayProgress, exists, err := postgres.ReadReplicationProgress(ctx, targetSQL, streamID)
+ if err != nil {
+ t.Fatalf("read final benchmark replay counters: %v", err)
+ }
+ if !exists || replayProgress.Transactions != int64(transactionCount) ||
+ replayProgress.Rows != int64(expectedChanges) {
+ t.Fatalf(
+ "final replay counters exist=%t transactions=%d/%d rows=%d/%d",
+ exists, replayProgress.Transactions, transactionCount,
+ replayProgress.Rows, expectedChanges,
+ )
+ }
for _, table := range []string{"accounts", "events", "sessions", "guarded"} {
sourceCount, sourceDigest := benchmarkTableDigest(t, sourceSQL, table)
diff --git a/internal/cdc/replay_claim_integration_test.go b/internal/cdc/replay_claim_integration_test.go
index caa46b7..9e64ff4 100644
--- a/internal/cdc/replay_claim_integration_test.go
+++ b/internal/cdc/replay_claim_integration_test.go
@@ -4,6 +4,7 @@ package cdc
import (
"context"
+ "encoding/binary"
"errors"
"fmt"
"slices"
@@ -256,6 +257,109 @@ func TestPG17ReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t *testing.T) {
}
}
+func TestPG17ReplayClaimResumesCoalescedBinaryStageWithoutRepeatingDML(t *testing.T) {
+ target := pgtest.Start(t, 17)
+ control := target.Connect(t)
+ ctx := context.Background()
+ if _, err := control.Exec(ctx, `
+ CREATE TABLE public.coalesced_binary_items (
+ id bigint PRIMARY KEY,
+ value bigint NOT NULL
+ );
+ INSERT INTO public.coalesced_binary_items
+ SELECT id, 0 FROM generate_series(1, 128) AS id
+ `); err != nil {
+ t.Fatal(err)
+ }
+
+ const streamID = "coalesced-binary-resume"
+ const generation = "coalesced-binary-resume-v1"
+ if err := EnsureStreamProgressIdentity(ctx, control, StreamIdentityConfig{
+ StreamID: streamID, Generation: generation,
+ FreshSetup: true, TargetHasCopiedData: true,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ source := Relation{
+ OID: 9_301, Namespace: "public", Name: "coalesced_binary_items", ReplicaIdentity: 'd',
+ Columns: []Column{
+ {Name: "id", Type: pgtype.Int8OID, Flags: 1},
+ {Name: "value", Type: pgtype.Int8OID},
+ },
+ }
+ loaded, err := loadTargetRelation(ctx, control, &source)
+ if err != nil {
+ t.Fatal(err)
+ }
+ binaryInt8 := func(value int64) TupleDatum {
+ return TupleDatum{Kind: DatumBinary, Data: binary.BigEndian.AppendUint64(nil, uint64(value))}
+ }
+ transactions := make([]Transaction, 0, 256)
+ resolved := make([]map[uint32]*targetRelation, 0, 256)
+ for round := int64(0); round < 2; round++ {
+ for id := int64(1); id <= 128; id++ {
+ oldTuple := Tuple{binaryInt8(id), binaryInt8(round)}
+ newTuple := Tuple{binaryInt8(id), binaryInt8(round + 1)}
+ endLSN := LSN(len(transactions) + 1)
+ transactions = append(transactions, Transaction{
+ CommitLSN: endLSN, EndLSN: endLSN,
+ Relations: []Relation{source},
+ Changes: []Change{{
+ RelationOID: source.OID, Kind: ChangeUpdate,
+ Old: &oldTuple, New: &newTuple,
+ }},
+ })
+ resolved = append(resolved, map[uint32]*targetRelation{source.OID: loaded})
+ }
+ }
+ plan, err := buildReplayPlan(streamID, generation, 0, 1, transactions, resolved)
+ if err != nil {
+ t.Fatal(err)
+ }
+ claim, err := ensureReplayClaim(ctx, control, plan.Claim, plan.Works)
+ if err != nil {
+ t.Fatal(err)
+ }
+ plan.Claim = claim
+ interrupted := errors.New("test: stop after coalesced binary lane commit")
+ applier := &Applier{config: ApplierConfig{
+ StreamID: streamID, StreamGeneration: generation,
+ afterReplayWork: func(replayClaim, replayClaimWork) error { return interrupted },
+ }}
+ cache := newApplyStatementCache(applyStatementCacheCapacity)
+ if err := configureApplySession(ctx, control); err != nil {
+ t.Fatal(err)
+ }
+ workers := []*applyWorker{{conn: control, statements: cache}}
+ if err := applier.executeReplayPlan(ctx, workers, plan, transactions, resolved); !errors.Is(err, interrupted) {
+ t.Fatalf("interrupted replay error = %v, want %v", err, interrupted)
+ }
+ assertReplayProgress(t, control, streamID, 0, 0, 0)
+ var finalRows int
+ if err := control.QueryRow(ctx, `
+ SELECT count(*) FROM public.coalesced_binary_items WHERE value = 2
+ `).Scan(&finalRows); err != nil {
+ t.Fatal(err)
+ }
+ if finalRows != 128 {
+ t.Fatalf("coalesced binary DML produced %d final rows, want 128", finalRows)
+ }
+
+ applier.config.afterReplayWork = nil
+ if err := applier.executeReplayPlan(ctx, workers, plan, transactions, resolved); err != nil {
+ t.Fatal(err)
+ }
+ assertReplayProgress(t, control, streamID, claim.EndLSN, 256, 256)
+ if err := control.QueryRow(ctx, `
+ SELECT count(*) FROM public.coalesced_binary_items WHERE value = 2
+ `).Scan(&finalRows); err != nil {
+ t.Fatal(err)
+ }
+ if finalRows != 128 {
+ t.Fatalf("resumed replay repeated binary DML; final rows = %d", finalRows)
+ }
+}
+
func TestPG17ReplayClaimV3ReconstructsAfterV4CatalogTightening(t *testing.T) {
target := pgtest.Start(t, 17)
control := target.Connect(t)
diff --git a/internal/cdc/replay_execute.go b/internal/cdc/replay_execute.go
index 1e9dc9e..e552f17 100644
--- a/internal/cdc/replay_execute.go
+++ b/internal/cdc/replay_execute.go
@@ -397,6 +397,10 @@ func (a *Applier) executeReplayWork(
}
func queueParallelReplayLane(replay *applyPipeline, items []relationBatchedChange) error {
+ items, err := coalesceReplayLaneUpdates(items)
+ if err != nil {
+ return err
+ }
// Source transactions commonly interleave several relations. Preserve the
// first-seen relation order and exact per-relation change order, but collect
// each homogeneous relation into one lane before invoking the existing set
@@ -421,3 +425,87 @@ func queueParallelReplayLane(replay *applyPipeline, items []relationBatchedChang
}
return nil
}
+
+// coalesceReplayLaneUpdates keeps only the newest complete-row update for a
+// primary key inside one lane transaction. All source transactions covered by
+// the lane still share one target commit and one exact receipt, so no
+// intermediate target state was observable before this optimization either.
+// Inserts, deletes, selective/unchanged-TOAST updates, and relations with
+// cross-key conflicts form hard per-key boundaries and are never coalesced.
+func coalesceReplayLaneUpdates(items []relationBatchedChange) ([]relationBatchedChange, error) {
+ if len(items) < 2 {
+ return items, nil
+ }
+ keep := make([]bool, len(items))
+ for i := range keep {
+ keep[i] = true
+ }
+ latest := make(map[string]int)
+ for i := range items {
+ item := &items[i]
+ if item.relation == nil || item.change == nil {
+ return nil, errors.New("cdc: replay lane contains missing relation or change")
+ }
+ key, keyed, err := coalescibleReplayPrimaryKey(item.relation, item.change)
+ if err != nil {
+ return nil, err
+ }
+ if !keyed {
+ // A PK mutation or another change whose exact key cannot be proven is a
+ // lane-wide ordering boundary. Forget every candidate rather than let
+ // a later update coalesce across that unknown dependency.
+ clear(latest)
+ continue
+ }
+ if item.change.Kind != ChangeUpdate ||
+ !item.relation.capabilities.relationLane ||
+ item.relation.capabilities.crossKeyConflicts ||
+ !canPrimaryKeyUpsert(item.relation, item.change) {
+ delete(latest, key)
+ continue
+ }
+ if previous, exists := latest[key]; exists {
+ keep[previous] = false
+ }
+ latest[key] = i
+ }
+ result := make([]relationBatchedChange, 0, len(items))
+ for i := range items {
+ if keep[i] {
+ result = append(result, items[i])
+ }
+ }
+ return result, nil
+}
+
+func coalescibleReplayPrimaryKey(relation *targetRelation, change *Change) (string, bool, error) {
+ if relation == nil || change == nil || len(primaryKeyColumns(relation)) == 0 {
+ return "", false, nil
+ }
+ var tuple *Tuple
+ switch change.Kind {
+ case ChangeInsert:
+ tuple = change.New
+ case ChangeUpdate:
+ if !canShardUpdateByPrimaryKey(relation, change) {
+ return "", false, nil
+ }
+ tuple = change.New
+ case ChangeDelete:
+ columns, safe := primaryKeyDeleteColumns(relation)
+ if !safe || !sameTargetColumns(primaryKeyColumns(relation), columns) {
+ return "", false, nil
+ }
+ tuple = change.Old
+ default:
+ return "", false, nil
+ }
+ if tuple == nil {
+ return "", false, nil
+ }
+ key, err := primaryKeyTupleKey(relation, tuple)
+ if err != nil {
+ return "", false, err
+ }
+ return fmt.Sprintf("%d:%s", relation.source.OID, key), true, nil
+}
diff --git a/internal/cdc/replay_plan_test.go b/internal/cdc/replay_plan_test.go
index 9ae0513..d9e7c99 100644
--- a/internal/cdc/replay_plan_test.go
+++ b/internal/cdc/replay_plan_test.go
@@ -609,7 +609,7 @@ func TestReplayPlanV4RelaxesOnlyRelationLocalOrdering(t *testing.T) {
}
}
-func TestFreshFragmentedReplayPlanUsesBoundedOrderedFallback(t *testing.T) {
+func TestFreshFragmentedReplayPlanKeepsOrderedBarriersInsideConcurrentWindow(t *testing.T) {
t.Parallel()
safe := replayTestRelation(53, "safe_items")
barrier := replayTestRelation(54, "barrier_items")
@@ -636,8 +636,12 @@ func TestFreshFragmentedReplayPlanUsesBoundedOrderedFallback(t *testing.T) {
if len(plan.Steps) != 3 || !replayPlanHasSerialWork(plan) {
t.Fatalf("fixture is not fragmented parallel/serial work: %#v", plan.Steps)
}
- if shouldUseConcurrentReplayPlan(nil, plan) {
- t.Fatal("fresh fragmented plan would create a multi-commit concurrent claim")
+ // The small fixture has only one lane in each safe epoch. Admission depends
+ // on the planner's HasParallel capability bit; larger windows set it when an
+ // epoch contains two or more independent components.
+ plan.HasParallel = true
+ if !shouldUseConcurrentReplayPlan(nil, plan) {
+ t.Fatal("fresh fragmented plan discarded safe parallel epochs around its barrier")
}
resume := plan.Claim
if !shouldUseConcurrentReplayPlan(&resume, plan) {
@@ -807,6 +811,71 @@ func replayTestRelation(oid uint32, name string) *targetRelation {
}
}
+func TestCoalesceReplayLaneUpdatesKeepsNewestCompletePrimaryKeyImage(t *testing.T) {
+ t.Parallel()
+ relation := replayTestRelation(70, "coalesced_items")
+ changes := []Change{
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "old"), New: replayTuple("a", "one")},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("b", "old"), New: replayTuple("b", "other")},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "one"), New: replayTuple("a", "two")},
+ }
+ items := make([]relationBatchedChange, len(changes))
+ for i := range changes {
+ items[i] = relationBatchedChange{change: &changes[i], relation: relation}
+ }
+ got, err := coalesceReplayLaneUpdates(items)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 2 || got[0].change != &changes[1] || got[1].change != &changes[2] {
+ t.Fatalf("coalesced changes = %#v, want other key and newest a image", got)
+ }
+}
+
+func TestCoalesceReplayLaneUpdatesDoesNotCrossDeleteSelectiveOrConflictBoundaries(t *testing.T) {
+ t.Parallel()
+ relation := replayTestRelation(71, "ordered_items")
+ selectiveNew := replayTuple("a", "ignored")
+ (*selectiveNew)[1] = TupleDatum{Kind: DatumUnchangedToast}
+ changes := []Change{
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "old"), New: replayTuple("a", "one")},
+ {RelationOID: relation.source.OID, Kind: ChangeDelete, Old: replayTuple("a", "one")},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "one"), New: replayTuple("a", "two")},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "two"), New: selectiveNew},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "two"), New: replayTuple("a", "three")},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "three"), New: replayTuple("renamed", "three")},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("a", "three"), New: replayTuple("a", "four")},
+ }
+ items := make([]relationBatchedChange, len(changes))
+ for i := range changes {
+ items[i] = relationBatchedChange{change: &changes[i], relation: relation}
+ }
+ got, err := coalesceReplayLaneUpdates(items)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != len(items) {
+ t.Fatalf("boundary sequence coalesced %d changes to %d", len(items), len(got))
+ }
+
+ relation.capabilities.crossKeyConflicts = true
+ conflicting := []Change{
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("b", "old"), New: replayTuple("b", "one")},
+ {RelationOID: relation.source.OID, Kind: ChangeUpdate, Old: replayTuple("b", "one"), New: replayTuple("b", "two")},
+ }
+ conflictItems := []relationBatchedChange{
+ {change: &conflicting[0], relation: relation},
+ {change: &conflicting[1], relation: relation},
+ }
+ got, err = coalesceReplayLaneUpdates(conflictItems)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != len(conflictItems) {
+ t.Fatalf("cross-key relation coalesced %d changes to %d", len(conflictItems), len(got))
+ }
+}
+
func replayTestTransaction(lsn LSN, relation *targetRelation, changes ...Change) Transaction {
return Transaction{
CommitLSN: lsn, EndLSN: lsn + 1, CommitTime: time.Unix(int64(lsn), 0).UTC(),
diff --git a/internal/cli/cli.go b/internal/cli/cli.go
index 11eab6b..b2039ce 100644
--- a/internal/cli/cli.go
+++ b/internal/cli/cli.go
@@ -62,8 +62,8 @@ func NewRootCommand() *cobra.Command {
flags.DurationVar(&cfg.WALSampleDuration, "wal-sample-duration", cfg.WALSampleDuration, "source WAL-rate sample duration")
flags.DurationVar(&cfg.SegmentPruneInterval, "segment-prune-interval", cfg.SegmentPruneInterval, "minimum interval between applied CDC segment pruning")
flags.IntVar(&cfg.ReplayWorkers, "replay-workers", cfg.ReplayWorkers, "parallel target workers for independent transaction components in each durable replay claim")
- flags.Int64Var(&cfg.ReplayBatchBytes, "replay-batch-bytes", cfg.ReplayBatchBytes, "maximum decoded CDC payload covered by one durable replay claim")
- flags.IntVar(&cfg.ReplayBatchChanges, "replay-batch-changes", cfg.ReplayBatchChanges, "maximum row changes covered by one durable replay claim")
+ flags.Int64Var(&cfg.ReplayBatchBytes, "replay-batch-bytes", cfg.ReplayBatchBytes, "decoded CDC payload per replay scheduling slice; four slices form the default durable wave")
+ flags.IntVar(&cfg.ReplayBatchChanges, "replay-batch-changes", cfg.ReplayBatchChanges, "row changes per replay scheduling slice; four slices form the default durable wave")
flags.BoolVar(&cfg.RetryBaseCopy, "retry-base-copy", false, "restart the base copy even though the last attempts failed the same way")
flags.BoolVar(&cfg.SkipTargetTuning, "skip-target-tuning", false, "leave target settings alone during the bulk load")
flags.BoolVar(&cfg.WarnOnTuningErrors, "warn-on-tuning-errors", false, "continue when a target setting cannot be tuned instead of stopping")
diff --git a/internal/controller/ui.html b/internal/controller/ui.html
index 2b4f349..88e580d 100644
--- a/internal/controller/ui.html
+++ b/internal/controller/ui.html
@@ -112,10 +112,10 @@ Migration configuration
WAL sample duration
Segment prune interval
Replay workers
- Replay batch bytes
- Replay batch changes
+ Replay slice bytes
+ Replay slice changes
CDC sample rows
- Replay workers apply small primary-key lanes concurrently. Each lane commits with its exact durable receipt; the resume LSN advances only after every receipt in the claim exists.
+ Four scheduling slices form one durable replay wave. Workers apply key-affine lanes concurrently; safe repeated full-row updates are coalesced, large built-in groups use binary COPY staging, and the resume LSN advances only after every receipt exists and matches exactly.