[INF-1722] Accept absent primary-key deletes during replay

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

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, &params, 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,
 	})
 }
 

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)
 		})
 	}