From efd504a8ff153a745fa32f3159ee3fd8e3d91847 Mon Sep 17 00:00:00 2001 From: thesyncim Date: Fri, 28 Aug 2026 09:54:07 +0100 Subject: [PATCH] fix(cdc): accept absent primary-key deletes during replay --- README.md | 4 +- internal/cdc/applier.go | 19 +- internal/cdc/cdc_integration_test.go | 20 +- internal/cdc/delete_noop_integration_test.go | 284 ++++++++++++++++++ internal/cdc/replay_claim_integration_test.go | 31 +- 5 files changed, 337 insertions(+), 21 deletions(-) create mode 100644 internal/cdc/delete_noop_integration_test.go diff --git a/README.md b/README.md index 9c0afaf..9ee1f71 100644 --- a/README.md +++ b/README.md @@ -988,7 +988,9 @@ controlled by `PGMIGRATE_CDC_BENCH_TRANSACTIONS` and preflight-blocked; foreign and unlogged tables and materialized views produce findings and need an operator plan. - The target is assumed not to receive independent application traffic before - cutover. Replay divergence stops the run. + cutover. A DELETE using a catalog-validated primary key is a no-op if its target + row is already absent; its source transaction and progress still commit normally. + Other replay divergence stops the run. This is not bidirectional conflict resolution. - Replay parallelism is conservative. Transactions without a safely comparable primary key, or with target behavior that can couple otherwise distinct rows, use the ordered serial path. A workload dominated by one hot-key component or diff --git a/internal/cdc/applier.go b/internal/cdc/applier.go index 76d7a0c..b1149dc 100644 --- a/internal/cdc/applier.go +++ b/internal/cdc/applier.go @@ -2161,6 +2161,7 @@ type applyExpectation struct { description string expectedRows int64 expectedOrdinals int + allowMissingRows bool // Only for DELETEs using a catalog-validated primary key. expectedTag string progressGuard bool statement string @@ -2362,7 +2363,8 @@ func (p *applyPipeline) sync() error { if ordinalErr != nil && firstErr == nil { firstErr = ordinalErr } - if expectation.expectedRows >= 0 && tag.RowsAffected() != expectation.expectedRows && firstErr == nil { + if expectation.expectedRows >= 0 && tag.RowsAffected() != expectation.expectedRows && + (!expectation.allowMissingRows || tag.RowsAffected() > expectation.expectedRows) && firstErr == nil { firstErr = divergenceFor(expectation.relation, expectation.kind, fmt.Sprintf( "affected %d rows, expected %d", tag.RowsAffected(), expectation.expectedRows, )) @@ -2446,7 +2448,7 @@ func (expectation applyExpectation) validateOrdinals(reader *pgconn.ResultReader return result } for ordinal, matched := range seen { - if !matched { + if !matched && !expectation.allowMissingRows { return divergenceFor(expectation.relation, expectation.kind, fmt.Sprintf( "batched replay did not match identity ordinal %d", ordinal, )) @@ -4919,6 +4921,9 @@ func appendPrimaryKeyDeletePredicate( ) error { for i, column := range primary { datum := tuple[column.sourceIndex] + if datum.Kind == DatumNull { + return divergenceFor(relation, ChangeDelete, "primary key contains NULL") + } if datum.Kind == DatumUnchangedToast { return divergenceFor(relation, ChangeDelete, "primary key contains unchanged TOAST") } @@ -4947,6 +4952,9 @@ func batchDeleteIdentityKey( var key strings.Builder for _, column := range identityColumns { datum := (*change.Old)[column.sourceIndex] + if column.primary && datum.Kind == DatumNull { + return "", divergenceFor(relation, ChangeDelete, "primary key contains NULL") + } if datum.Kind == DatumUnchangedToast { return "", divergenceFor(relation, ChangeDelete, "replica identity contains unchanged TOAST") } @@ -5029,6 +5037,7 @@ func applyDeleteTextStage( relation: relation, kind: ChangeDelete, description: "staged delete from " + relation.quoted, expectedRows: int64(len(changes)), expectedOrdinals: len(changes), + allowMissingRows: deleteUsesTargetPrimaryKey(relation, identityColumns), }) } @@ -5082,6 +5091,7 @@ func applyDeleteValueChunk( relation: relation, kind: ChangeDelete, description: "batch delete from " + relation.quoted, expectedRows: int64(len(changes)), expectedOrdinals: len(changes), + allowMissingRows: deleteUsesTargetPrimaryKey(relation, identityColumns), }) } @@ -5150,6 +5160,7 @@ func applyDeleteArrayChunk( relation: relation, kind: ChangeDelete, description: "array batch delete from " + relation.quoted, expectedRows: int64(len(changes)), expectedOrdinals: len(changes), + allowMissingRows: deleteUsesTargetPrimaryKey(relation, identityColumns), }) } @@ -5162,7 +5173,8 @@ func applyDelete(replay *applyPipeline, relation *targetRelation, change *Change sql.WriteString(relation.quoted) sql.WriteString(" WHERE ") params := make([]rawParam, 0, len(relation.columns)) - if primary, safe := primaryKeyDeleteColumns(relation); safe { + primary, safe := primaryKeyDeleteColumns(relation) + if safe { if err := appendPrimaryKeyDeletePredicate( &sql, ¶ms, relation, primary, *change.Old, ); err != nil { @@ -5176,6 +5188,7 @@ func applyDelete(replay *applyPipeline, relation *targetRelation, change *Change return replay.queue(sql.String(), params, applyExpectation{ relation: relation, kind: ChangeDelete, description: "delete from " + relation.quoted, expectedRows: 1, + allowMissingRows: safe, }) } diff --git a/internal/cdc/cdc_integration_test.go b/internal/cdc/cdc_integration_test.go index f399605..0b32ccf 100644 --- a/internal/cdc/cdc_integration_test.go +++ b/internal/cdc/cdc_integration_test.go @@ -96,6 +96,10 @@ func TestPG17LiveWALStageApplyCrashRetry(t *testing.T) { if _, err := sourceSQL.Exec(ctx, "CREATE PUBLICATION pgmigrate_cdc_test FOR TABLE cdc_items, cdc_truncated, cdc_truncated_child, cdc_custom, cdc_generated_only, cdc_empty"); err != nil { t.Fatal(err) } + // Its later pgoutput DELETE must also replay when the target is already absent. + if _, err := sourceSQL.Exec(ctx, "INSERT INTO cdc_items (id, note) VALUES (999, 'source-only')"); err != nil { + t.Fatal(err) + } replication := source.ReplicationConnect(t) slot, err := pglogrepl.CreateReplicationSlot( @@ -183,6 +187,7 @@ func TestPG17LiveWALStageApplyCrashRetry(t *testing.T) { {"UPDATE cdc_empty SET absent = 'set' WHERE id = 1", nil}, {"INSERT INTO cdc_empty (id, optional) VALUES (3, '')", nil}, {"DELETE FROM cdc_empty WHERE id = 3", nil}, + {"DELETE FROM cdc_items WHERE id = 999", nil}, } for index, statement := range statements { if _, err := sourceSQL.Exec(ctx, statement.sql, statement.args...); err != nil { @@ -308,6 +313,7 @@ func TestPG17LiveWALStageApplyCrashRetry(t *testing.T) { {"cdc_items", ChangeUpdate, "1"}, {"cdc_items", ChangeInsert, "2"}, {"cdc_items", ChangeDelete, "2"}, + {"cdc_items", ChangeDelete, "999"}, {"cdc_empty", ChangeDelete, "3"}, } { if !samples.saw(want.table, want.kind, want.key) { @@ -2818,18 +2824,10 @@ func TestPG17PipelinedApplyPreservesAtomicOrderedReplay(t *testing.T) { CommitLSN: 30 + LSN(kind), EndLSN: endLSN, Relations: []Relation{source}, Changes: []Change{change}, }) - if kind == ChangeUpdate { - if err != nil { - t.Fatalf("missing-row update upsert: %v", err) - } - assertProgress(t, stream, endLSN) - return - } - var divergence *DivergenceError - if !errors.As(err, &divergence) { - t.Fatalf("zero-row delete error=%v, want divergence", err) + if err != nil { + t.Fatalf("missing-row %s: %v", changeKindName(kind), err) } - assertProgress(t, stream, 0) + assertProgress(t, stream, endLSN) }) } diff --git a/internal/cdc/delete_noop_integration_test.go b/internal/cdc/delete_noop_integration_test.go new file mode 100644 index 0000000..1346fd8 --- /dev/null +++ b/internal/cdc/delete_noop_integration_test.go @@ -0,0 +1,284 @@ +//go:build integration + +package cdc + +import ( + "errors" + "fmt" + "strings" + "testing" + + "github.com/GetStream/pgmigrate/internal/pgtest" + "github.com/GetStream/pgmigrate/internal/postgres" +) + +func TestPrimaryKeyDeleteNoop(t *testing.T) { + for _, major := range pgtest.Majors(t) { + t.Run(fmt.Sprint(major), func(t *testing.T) { + target := pgtest.Start(t, major) + conn := target.Connect(t) + ctx := t.Context() + if _, err := conn.Exec(ctx, ` + CREATE DOMAIN public.delete_key AS bigint CHECK (VALUE > 0); + CREATE TABLE public.delete_uuid (id uuid PRIMARY KEY, value text); + CREATE TABLE public.delete_stage (id public.delete_key PRIMARY KEY, value text); + CREATE TABLE public.delete_composite (id text, value text, PRIMARY KEY (value, id)); + CREATE TABLE public.delete_full (id text, value text); + CREATE TABLE public.delete_trigger (id text PRIMARY KEY, value text); + CREATE FUNCTION public.suppress_delete() RETURNS trigger LANGUAGE plpgsql AS + $$ BEGIN RETURN NULL; END $$; + CREATE TRIGGER suppress_delete BEFORE DELETE ON public.delete_trigger + FOR EACH ROW EXECUTE FUNCTION public.suppress_delete(); + ALTER TABLE public.delete_trigger ENABLE ALWAYS TRIGGER suppress_delete; + INSERT INTO public.delete_trigger VALUES ('1', 'keep'); + `); err != nil { + t.Fatal(err) + } + if err := configureApplySession(ctx, conn); err != nil { + t.Fatal(err) + } + statements := newApplyStatementCache(applyStatementCacheCapacity) + for _, tc := range []struct { + name, table, path string + typeOID uint32 + rows int + strict bool + }{ + {name: "single UUID", table: "delete_uuid", path: "single", typeOID: 2950, rows: 1}, + {name: "VALUES UUID", table: "delete_uuid", path: "values", typeOID: 2950, rows: 4}, + {name: "array UUID", table: "delete_uuid", path: "array", typeOID: 2950, rows: 4}, + {name: "typed stage", table: "delete_stage", path: "stage", typeOID: 90001, rows: minimumTextCopyStageRows}, + {name: "composite catalog order", table: "delete_composite", path: "array", typeOID: 25, rows: 4}, + {name: "full identity stays strict", table: "delete_full", path: "single", typeOID: 25, rows: 1, strict: true}, + {name: "full identity batch stays strict", table: "delete_full", path: "values", typeOID: 25, rows: 4, strict: true}, + {name: "suppressed delete stays strict", table: "delete_trigger", path: "single", typeOID: 25, rows: 1, strict: true}, + } { + t.Run(tc.name, func(t *testing.T) { + source := Relation{ + OID: 9001, Namespace: "public", Name: tc.table, ReplicaIdentity: 'd', + Columns: []Column{{Name: "id", Type: tc.typeOID, Flags: 1}, {Name: "value", Type: 25}}, + } + if tc.table == "delete_composite" { + source.Columns[1].Flags = 1 + } + if tc.table == "delete_full" { + source.ReplicaIdentity = 'f' + source.Columns[1].Flags = 1 + } + relation, err := loadTargetRelation(ctx, conn, &source) + if err != nil { + t.Fatal(err) + } + primary, safe := primaryKeyDeleteColumns(relation) + if safe == tc.strict { + t.Fatalf("primary-key delete safety=%t, strict=%t", safe, tc.strict) + } + if !safe { + primary = batchUpdateIdentityColumns(relation) + if tc.table == "delete_full" { + primary = relation.columns + } + } + changes := make([]Change, tc.rows) + for i := range changes { + id := fmt.Sprint(i + 1) + if tc.typeOID == 2950 { + id = fmt.Sprintf("00000000-0000-0000-0000-%012d", i+1) + } + changes[i] = Change{RelationOID: source.OID, Kind: ChangeDelete, Old: replayTuple(id, "keep")} + } + // Run an entirely absent batch, then a mixed existing/absent batch. + for _, populate := range []bool{false, true} { + if populate && !tc.strict { + for i := 0; i < len(changes); i += 2 { + if _, err := conn.Exec(ctx, "INSERT INTO "+relation.quoted+" VALUES ($1, 'keep')", + string((*changes[i].Old)[0].Data)); err != nil { + t.Fatal(err) + } + } + } + replay := newApplyPipeline(ctx, conn.PgConn(), statements) + defer replay.abort() + replay.begin() + switch tc.path { + case "single": + err = applyDelete(replay, relation, &changes[0]) + case "values": + err = applyDeleteValueChunk(replay, relation, primary, changes) + case "array": + var applied bool + applied, err = applyDeleteArrayChunk(replay, relation, primary, changes) + if !applied && err == nil { + t.Fatal("array path was not exercised") + } + case "stage": + var applied bool + applied, err = applyDeleteTextStage(replay, relation, primary, changes) + if !applied && err == nil { + t.Fatal("typed stage path was not exercised") + } + } + if err == nil { + err = replay.sync() + } + if tc.strict { + var divergence *DivergenceError + if !errors.As(err, &divergence) { + t.Fatalf("strict delete error=%v, want divergence", err) + } + if !strings.Contains(err.Error(), "affected 0 rows") && !strings.Contains(err.Error(), "did not match identity ordinal") { + t.Fatalf("strict delete failed for the wrong reason: %v", err) + } + if err := replay.abort(); err != nil { + t.Fatal(err) + } + return + } + if err != nil { + replay.abort() + t.Fatal(err) + } + replay.commit() + if err := replay.sync(); err != nil { + t.Fatal(err) + } + if err := replay.close(); err != nil { + t.Fatal(err) + } + var rows int + if err := conn.QueryRow(ctx, "SELECT count(*) FROM "+relation.quoted).Scan(&rows); err != nil { + t.Fatal(err) + } + if rows != 0 { + t.Fatalf("delete left %d rows", rows) + } + } + }) + } + }) + } +} + +func TestMissingDeletePreservesTransactionAtomicity(t *testing.T) { + target := pgtest.Start(t, 18) + conn := target.Connect(t) + ctx := t.Context() + if _, err := conn.Exec(ctx, `CREATE TABLE public.noop_atomic (id text PRIMARY KEY, value text NOT NULL)`); err != nil { + t.Fatal(err) + } + relation := replayTestRelation(9002, "noop_atomic") + statements := newApplyStatementCache(applyStatementCacheCapacity) + for _, fail := range []bool{true, false} { + stream := fmt.Sprintf("noop-atomic-%t", fail) + if err := EnsureStreamProgressIdentity(ctx, conn, StreamIdentityConfig{ + StreamID: stream, Generation: stream, FreshSetup: true, TargetHasCopiedData: true, + }); err != nil { + t.Fatal(err) + } + changes := []Change{ + {RelationOID: relation.source.OID, Kind: ChangeDelete, Old: replayTuple("same-key", "old")}, + {RelationOID: relation.source.OID, Kind: ChangeInsert, New: replayTuple("same-key", "source")}, + } + if fail { + changes = append(changes, Change{ + RelationOID: relation.source.OID, Kind: ChangeInsert, + New: &Tuple{{Kind: DatumText, Data: []byte("invalid")}, {Kind: DatumNull}}, + }) + } + transaction := replayTestTransaction(1000, relation, changes...) + applier := &Applier{config: ApplierConfig{StreamID: stream, StreamGeneration: stream}} + err := applier.applyTransaction(ctx, conn, newTargetRelationCache(), statements, 0, &transaction) + if fail { + var divergence *DivergenceError + if !errors.As(err, &divergence) { + t.Fatalf("SQL error=%v, want divergence", err) + } + if !strings.Contains(err.Error(), "SQLSTATE 23502") { + t.Fatalf("transaction failed before the NOT NULL violation: %v", err) + } + if _, exists, err := postgres.ReadReplicationProgress(ctx, conn, stream); err != nil || exists { + t.Fatalf("failed transaction published progress: exists=%t, err=%v", exists, err) + } + var rows int + if err := conn.QueryRow(ctx, "SELECT count(*) FROM public.noop_atomic").Scan(&rows); err != nil || rows != 0 { + t.Fatalf("failed transaction retained rows=%d, err=%v", rows, err) + } + continue + } + if err != nil { + t.Fatal(err) + } + assertReplayProgress(t, conn, stream, transaction.EndLSN, 1, 2) + var value string + if err := conn.QueryRow(ctx, "SELECT value FROM public.noop_atomic WHERE id='same-key'").Scan(&value); err != nil || value != "source" { + t.Fatalf("delete then insert value=%q, err=%v", value, err) + } + } + for _, count := range []int{1, 2} { + t.Run(fmt.Sprintf("NULL identity with %d deletes", count), func(t *testing.T) { + stream := fmt.Sprintf("null-delete-%d", count) + if err := EnsureStreamProgressIdentity(ctx, conn, StreamIdentityConfig{ + StreamID: stream, Generation: stream, FreshSetup: true, TargetHasCopiedData: true, + }); err != nil { + t.Fatal(err) + } + changes := make([]Change, count) + for i := range changes { + changes[i] = Change{RelationOID: relation.source.OID, Kind: ChangeDelete, Old: &Tuple{{Kind: DatumNull}, {Kind: DatumNull}}} + } + transaction := replayTestTransaction(2000, relation, changes...) + applier := &Applier{config: ApplierConfig{StreamID: stream, StreamGeneration: stream}} + err := applier.applyTransaction(ctx, conn, newTargetRelationCache(), statements, 0, &transaction) + var divergence *DivergenceError + if !errors.As(err, &divergence) || !strings.Contains(err.Error(), "primary key contains NULL") { + t.Fatalf("invalid DELETE identity error=%v", err) + } + if _, exists, err := postgres.ReadReplicationProgress(ctx, conn, stream); err != nil || exists { + t.Fatalf("invalid identity published progress: exists=%t, err=%v", exists, err) + } + }) + } +} + +func TestDeleteNoopKeepsResultValidation(t *testing.T) { + target := pgtest.Start(t, 18) + conn := target.Connect(t) + statements := newApplyStatementCache(applyStatementCacheCapacity) + for _, tc := range []struct { + name, query, want string + ordinals int + allowMissing bool + }{ + {name: "absent ordinal", query: "SELECT 0 WHERE false", ordinals: 1, allowMissing: true}, + {name: "duplicate ordinal", query: "SELECT 0 UNION ALL SELECT 0", ordinals: 1, allowMissing: true, want: "more than once"}, + {name: "negative ordinal", query: "SELECT -1", ordinals: 1, allowMissing: true, want: "invalid identity ordinal"}, + {name: "out of range ordinal", query: "SELECT 1", ordinals: 1, allowMissing: true, want: "invalid identity ordinal"}, + {name: "malformed ordinal", query: "SELECT 'bad'", ordinals: 1, allowMissing: true, want: "invalid identity ordinal"}, + {name: "extra column", query: "SELECT 0, 0", ordinals: 1, allowMissing: true, want: "identity columns, expected 1"}, + {name: "too many affected rows", query: "SELECT 0 UNION ALL SELECT 1", allowMissing: true, want: "affected 2 rows, expected 1"}, + {name: "strict row count", query: "SELECT 0 WHERE false", want: "affected 0 rows, expected 1"}, + {name: "strict ordinal", query: "SELECT 0 WHERE false", ordinals: 1, want: "did not match identity ordinal"}, + } { + t.Run(tc.name, func(t *testing.T) { + replay := newApplyPipeline(t.Context(), conn.PgConn(), statements) + defer replay.abort() + replay.begin() + err := replay.queue(tc.query, nil, applyExpectation{ + kind: ChangeDelete, expectedRows: 1, expectedOrdinals: tc.ordinals, allowMissingRows: tc.allowMissing, + }) + if err == nil { + err = replay.sync() + } + if tc.want == "" { + if err != nil { + t.Fatal(err) + } + } else { + var divergence *DivergenceError + if !errors.As(err, &divergence) || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("validation error=%v, want divergence containing %q", err, tc.want) + } + } + }) + } +} diff --git a/internal/cdc/replay_claim_integration_test.go b/internal/cdc/replay_claim_integration_test.go index 9e64ff4..5e1caf9 100644 --- a/internal/cdc/replay_claim_integration_test.go +++ b/internal/cdc/replay_claim_integration_test.go @@ -19,6 +19,14 @@ import ( ) func TestPG17ReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t *testing.T) { + for _, version := range []int{2, 5} { + t.Run(fmt.Sprint(version), func(t *testing.T) { + testReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t, version) + }) + } +} + +func testReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t *testing.T, version int) { target := pgtest.Start(t, 17) control := target.Connect(t) ctx := context.Background() @@ -54,8 +62,19 @@ func TestPG17ReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t *testing.T) { t.Fatal(err) } for i := range transactions { + if _, err := control.Exec(ctx, "INSERT INTO public.claim_items VALUES ($1, 'old')", fmt.Sprintf("old-%03d", i)); err != nil { + t.Fatal(err) + } transactions[i] = replayTestTransaction( LSN(1_000+i*2), relation, + Change{ + RelationOID: relation.source.OID, Kind: ChangeDelete, + Old: replayTuple(fmt.Sprintf("old-%03d", i), "old"), + }, + Change{ + RelationOID: relation.source.OID, Kind: ChangeDelete, + Old: replayTuple(fmt.Sprintf("missing-%03d", i), "old"), + }, Change{ RelationOID: relation.source.OID, Kind: ChangeInsert, New: replayTuple(fmt.Sprintf("id-%03d-a", i), fmt.Sprintf("value-%03d-a", i)), @@ -68,13 +87,13 @@ func TestPG17ReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t *testing.T) { resolved[i] = map[uint32]*targetRelation{relation.source.OID: loaded} } plan, err := buildReplayPlanForGenerationVersion( - streamID, generation, generation, 0, 8, transactions, resolved, 2, + streamID, generation, generation, 0, 8, transactions, resolved, version, ) if err != nil { t.Fatal(err) } - if plan.Claim.PlanVersion != 2 { - t.Fatalf("legacy resume fixture plan version=%d, want 2", plan.Claim.PlanVersion) + if plan.Claim.PlanVersion != version { + t.Fatalf("resume fixture plan version=%d, want %d", plan.Claim.PlanVersion, version) } if !plan.HasParallel || len(plan.Works) < 2 { t.Fatalf("fixture did not produce parallel work: %#v", plan.Steps) @@ -153,14 +172,14 @@ func TestPG17ReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t *testing.T) { } } reconstructed, err := buildReplayPlanForGenerationVersion( - streamID, generation, generation, 0, 8, transactions, resolved, 2, + streamID, generation, generation, 0, 8, transactions, resolved, version, ) if err != nil { t.Fatal(err) } if !replayClaimsEqual(reconstructed.Claim, claim) || !slices.Equal(reconstructed.Works, plan.Works) { - t.Fatal("new executor did not reconstruct the exact active plan-version-2 claim") + t.Fatal("new executor did not reconstruct the exact active claim") } reconstructed.Claim = claim plan = reconstructed @@ -221,7 +240,7 @@ func TestPG17ReplayClaimResumesExactLaneReceiptsAndFinalizesOnce(t *testing.T) { assertReplayProgress( t, control, streamID, plan.Claim.EndLSN, - int64(len(transactions)), int64(len(transactions)*2), + int64(len(transactions)), int64(len(transactions)*4), ) if _, exists, err := readReplayClaim(ctx, control, streamID); err != nil || exists { t.Fatalf("finalized claim exists=%t err=%v", exists, err)