Merge #143567

143567: crosscluster: fix tombstone updates in sql writer r=jeffswenson a=jeffswenson

This fixes a subtle bug in the sql LDR writer. If a row is deleted in
two clusters replicated via LDR, when the LWW winning delete is
replicated to the losing cluster, the losing cluster does not update the
origin timestamp assigned to its tombstone. This can lead to
non-convergence in three cluster setups.

While this is fixing a bug, the most common case that will take the new
code path is replays during catch up scans or processor replanning. In
that case the replay will see zero rows impacted by the delete, then the
cput will fail LWW against the value that is stored in the local DB.

Release note: none
Epic: [CRDB-48647](https://cockroachlabs.atlassian.net/browse/CRDB-48647)

Co-authored-by: Jeff Swenson <[email protected]>

Author: craig[bot]

pkg/crosscluster/logical/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "purgatory.go",
         "range_stats.go",
         "replication_statements.go",
+        "tombstone_updater.go",
         "udf_row_processor.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/crosscluster/logical",
@@ -154,6 +155,7 @@ go_test(
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/catalog/descs",
         "//pkg/sql/catalog/desctestutils",
+        "//pkg/sql/catalog/lease",
         "//pkg/sql/catalog/resolver",
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -1388,6 +1388,86 @@ func TestThreeWayReplication(t *testing.T) {
 	verifyExpectedRowAllServers(t, runners, expectedRows, dbNames)
 }
 
+func TestTombstoneUpdate(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	skip.UnderDeadlock(t)
+	defer log.Scope(t).Close(t)
+	// This test is a regression test for a bug where replicating a delete does
+	// not update the origin timestamp assigned to a tombstone if the replicated
+	// value is overwriting a tombstone.
+	//
+	// Replicating the bug requires three databases ('src-a', 'src-b', 'dst').
+	//
+	// Here's the timeline of events:
+	//
+	// 1. row is inserted into 'src-a', 'src-b', and 'dst'
+	// 2. row is deleted from 'dst'
+	// 3. row is updated in 'src-b'
+	// 4. row is deleted in 'src-a'
+	//
+	// 5. Replication is started from 'src-a' -> 'dst' and allowed to catch up to now.
+	// 6. Replication is started from 'srb-b' -> 'dst' and allowed to catch up to now.
+	//
+	// In this example, the last delete is the LWW winner. But if tombstones are not updated, then the update in 'src-b' will be
+	// replicated to 'dst' will win LWW since the replicated delete was no-op'd.
+
+	ctx := context.Background()
+
+	clusterArgs := base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			DefaultTestTenant: base.TestControlsTenantsExplicitly,
+			Knobs: base.TestingKnobs{
+				JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
+			},
+		},
+	}
+
+	numDBs := 3
+	server, s, runners, dbNames := setupServerWithNumDBs(t, ctx, clusterArgs, 1, numDBs)
+	defer server.Stopper().Stop(ctx)
+	PGURLs := GetPGURLs(t, s, dbNames)
+
+	// Insert row into all three databases
+	for i := range runners {
+		runners[i].Exec(t, "UPSERT INTO tab VALUES (1, 'initial')")
+	}
+
+	srcA, srcB, dst := runners[0], runners[1], runners[2]
+	urlSrcA, urlSrcB, _ := PGURLs[0].String(), PGURLs[1].String(), PGURLs[2].String()
+
+	// Grab timestamp that will be used as a cursor for starting the logical replication jobs
+	start := s.Clock().Now()
+
+	// Delete the row in the dest, this should lose all LWW conflicts.
+	dst.Exec(t, "DELETE FROM tab WHERE pk = 1")
+
+	// Update the row in one of the sources, this should lose LWW conlficts to
+	// the final delete.
+	srcB.Exec(t, "UPSERT INTO tab VALUES (1, 'updated')")
+
+	// Delete the row in the other source, this should win LWW.
+	srcA.Exec(t, "DELETE FROM tab WHERE pk = 1")
+
+	dst.CheckQueryResults(t, "SELECT * FROM tab", [][]string{})
+	srcA.CheckQueryResults(t, "SELECT * FROM tab", [][]string{})
+	srcB.CheckQueryResults(t, "SELECT * FROM tab", [][]string{{"1", "updated"}})
+
+	// 5. Replicate the delete from 'src-a' -> 'dst'
+	var jobIDSrcA jobspb.JobID
+	dst.QueryRow(t, "CREATE LOGICAL REPLICATION STREAM FROM TABLE tab ON $1 INTO TABLE tab WITH MODE = VALIDATED, CURSOR = $2",
+		urlSrcA, start.AsOfSystemTime()).Scan(&jobIDSrcA)
+	WaitUntilReplicatedTime(t, s.Clock().Now(), dst, jobIDSrcA)
+
+	// 6. Replicate the update from 'src-b' -> 'dst'
+	var jobIDSrcB jobspb.JobID
+	dst.QueryRow(t, "CREATE LOGICAL REPLICATION STREAM FROM TABLE tab ON $1 INTO TABLE tab WITH MODE = VALIDATED, CURSOR = $2",
+		urlSrcB, start.AsOfSystemTime()).Scan(&jobIDSrcB)
+	WaitUntilReplicatedTime(t, s.Clock().Now(), dst, jobIDSrcB)
+
+	// Verify that the delete won LWW since it has the highest mvcc value
+	dst.CheckQueryResults(t, "SELECT * FROM tab WHERE pk = 1", [][]string{})
+}
+
 func TestForeignKeyConstraints(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	skip.UnderDeadlock(t)

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/typedesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
@@ -738,6 +739,8 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 				// avoid creating a new copy on each executor usage.
 				flowCtx.Cfg.DB.Executor(isql.WithSessionData(sql.NewInternalSessionData(ctx, flowCtx.Cfg.Settings, "" /* opName */))),
 				sd, lrw.spec,
+				flowCtx.Codec(),
+				flowCtx.Cfg.LeaseManager.(*lease.Manager),
 			)
 			if err != nil {
 				return err

pkg/crosscluster/logical/lww_row_processor.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/lexbase"
@@ -79,6 +80,7 @@ type querier interface {
 	InsertRow(ctx context.Context, txn isql.Txn, ie isql.Executor, row cdcevent.Row, prevRow *cdcevent.Row, likelyInsert bool) (batchStats, error)
 	DeleteRow(ctx context.Context, txn isql.Txn, ie isql.Executor, row cdcevent.Row, prevRow *cdcevent.Row) (batchStats, error)
 	RequiresParsedBeforeRow(catid.DescID) bool
+	ReleaseLeases(ctx context.Context)
 }
 
 // isLwwLoser returns true if the error is a ConditionFailedError with an
@@ -274,7 +276,9 @@ func (sqlRowProcessor) ReportMutations(_ *stats.Refresher) {}
 
 // ReleaseLeases implements the BatchHandler interface but is a no-op since each
 // query does this itself.
-func (sqlRowProcessor) ReleaseLeases(_ context.Context) {}
+func (srp *sqlRowProcessor) ReleaseLeases(ctx context.Context) {
+	srp.querier.ReleaseLeases(ctx)
+}
 
 func (*sqlRowProcessor) Close(ctx context.Context) {}
 
@@ -458,6 +462,8 @@ func makeSQLProcessor(
 	ie isql.Executor,
 	sd *sessiondata.SessionData,
 	spec execinfrapb.LogicalReplicationWriterSpec,
+	codec keys.SQLCodec,
+	leaseMgr *lease.Manager,
 ) (*sqlRowProcessor, error) {
 
 	needUDFQuerier := false
@@ -469,10 +475,14 @@ func makeSQLProcessor(
 
 	lwwQuerier := &lwwQuerier{
 		settings: settings,
+		codec:    codec,
+		db:       db,
+		leaseMgr: leaseMgr,
 		queryBuffer: queryBuffer{
 			deleteQueries: make(map[catid.DescID]queryBuilder, len(tableConfigByDestID)),
 			insertQueries: make(map[catid.DescID]map[catid.FamilyID]queryBuilder, len(tableConfigByDestID)),
 		},
+		tombstoneUpdaters:          make(map[descpb.ID]*tombstoneUpdater, len(tableConfigByDestID)),
 		ieOverrideOptimisticInsert: getIEOverride(replicatedOptimisticInsertOpName, jobID),
 		ieOverrideInsert:           getIEOverride(replicatedInsertOpName, jobID),
 		ieOverrideDelete:           getIEOverride(replicatedDeleteOpName, jobID),
@@ -532,6 +542,13 @@ func (m *muxQuerier) RequiresParsedBeforeRow(id catid.DescID) bool {
 	return m.shouldUseUDF[id]
 }
 
+func (m *muxQuerier) ReleaseLeases(ctx context.Context) {
+	if m.udfQuerier != nil {
+		m.udfQuerier.ReleaseLeases(ctx)
+	}
+	m.lwwQuerier.ReleaseLeases(ctx)
+}
+
 // lwwQuerier is a querier that implements partial last-write-wins
 // semantics using SQL queries.
 //
@@ -542,12 +559,16 @@ func (m *muxQuerier) RequiresParsedBeforeRow(id catid.DescID) bool {
 //
 // See the design document for possible solutions to these problems.
 type lwwQuerier struct {
-	settings    *cluster.Settings
-	queryBuffer queryBuffer
+	settings          *cluster.Settings
+	codec             keys.SQLCodec
+	db                descs.DB
+	queryBuffer       queryBuffer
+	tombstoneUpdaters map[descpb.ID]*tombstoneUpdater
 
 	ieOverrideOptimisticInsert sessiondata.InternalExecutorOverride
 	ieOverrideInsert           sessiondata.InternalExecutorOverride
 	ieOverrideDelete           sessiondata.InternalExecutorOverride
+	leaseMgr                   *lease.Manager
 }
 
 func (lww *lwwQuerier) AddTable(targetDescID int32, tc sqlProcessorTableConfig) error {
@@ -561,6 +582,15 @@ func (lww *lwwQuerier) AddTable(targetDescID int32, tc sqlProcessorTableConfig)
 	if err != nil {
 		return err
 	}
+
+	lww.tombstoneUpdaters[td.GetID()] = newTombstoneUpdater(
+		lww.codec,
+		lww.db.KV(),
+		lww.leaseMgr,
+		catid.DescID(targetDescID),
+		lww.settings,
+	)
+
 	return nil
 }
 
@@ -665,13 +695,26 @@ func (lww *lwwQuerier) DeleteRow(
 		sess.QualityOfService = nil
 	}
 	sess.OriginTimestampForLogicalDataReplication = row.MvccTimestamp
-	if _, err := ie.ExecParsed(ctx, replicatedDeleteOpName, kvTxn, sess, stmt, datums...); err != nil {
+	rowCount, err := ie.ExecParsed(ctx, replicatedDeleteOpName, kvTxn, sess, stmt, datums...)
+	if err != nil {
 		log.Warningf(ctx, "replicated delete failed (query: %s): %s", stmt.SQL, err.Error())
 		return batchStats{}, err
 	}
+	if rowCount != 1 {
+		// NOTE: at this point we don't know if we are updating a tombstone or if
+		// we are losing LWW. As long as it is a LWW loss or a tombstone update,
+		// updateTombstone will return okay.
+		return lww.tombstoneUpdaters[row.TableID].updateTombstone(ctx, txn, row)
+	}
 	return batchStats{}, nil
 }
 
+func (lww *lwwQuerier) ReleaseLeases(ctx context.Context) {
+	for _, tu := range lww.tombstoneUpdaters {
+		tu.ReleaseLeases(ctx)
+	}
+}
+
 const (
 	insertQueryOptimistic  = `INSERT INTO [%d AS t] (%s) VALUES (%s)`
 	insertQueryPessimistic = `

pkg/crosscluster/logical/lww_row_processor_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
@@ -79,7 +80,7 @@ func TestLWWInsertQueryGeneration(t *testing.T) {
 		return tableName
 	}
 
-	setup := func(t *testing.T, schemaTmpl string) (*sqlRowProcessor, func(...interface{}) roachpb.KeyValue) {
+	setup := func(t *testing.T, schemaTmpl string) (*sqlRowProcessor, func(...interface{}) roachpb.KeyValue, func()) {
 		tableNameSrc := createTable(t, schemaTmpl)
 		tableNameDst := createTable(t, schemaTmpl)
 		srcDesc := desctestutils.TestingGetPublicTableDescriptor(s.DB(), s.Codec(), "defaultdb", tableNameSrc)
@@ -89,20 +90,23 @@ func TestLWWInsertQueryGeneration(t *testing.T) {
 			dstDesc.GetID(): {
 				srcDesc: srcDesc,
 			},
-		}, jobspb.JobID(1), s.InternalDB().(descs.DB), s.InternalExecutor().(isql.Executor), sd, execinfrapb.LogicalReplicationWriterSpec{})
+		}, jobspb.JobID(1), s.InternalDB().(descs.DB), s.InternalExecutor().(isql.Executor), sd, execinfrapb.LogicalReplicationWriterSpec{}, s.Codec(), s.LeaseManager().(*lease.Manager))
 		require.NoError(t, err)
 		return rp, func(datums ...interface{}) roachpb.KeyValue {
-			kv := replicationtestutils.EncodeKV(t, s.Codec(), srcDesc, datums...)
-			kv.Value.Timestamp = hlc.Timestamp{WallTime: timeutil.Now().UnixNano()}
-			return kv
-		}
+				kv := replicationtestutils.EncodeKV(t, s.Codec(), srcDesc, datums...)
+				kv.Value.Timestamp = hlc.Timestamp{WallTime: timeutil.Now().UnixNano()}
+				return kv
+			}, func() {
+				rp.ReleaseLeases(ctx)
+			}
 	}
 
 	for _, tc := range testCases {
 		t.Run(fmt.Sprintf("%s/insert", tc.name), func(t *testing.T) {
 			runner.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.try_optimistic_insert.enabled=true")
 			defer runner.Exec(t, "RESET CLUSTER SETTING logical_replication.consumer.try_optimistic_insert.enabled")
-			rp, encoder := setup(t, tc.schemaTmpl)
+			rp, encoder, cleanup := setup(t, tc.schemaTmpl)
+			defer cleanup()
 			keyValue := encoder(tc.row...)
 			require.NoError(t, s.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 				_, err := rp.ProcessRow(ctx, txn, keyValue, roachpb.Value{})
@@ -112,15 +116,17 @@ func TestLWWInsertQueryGeneration(t *testing.T) {
 		t.Run(fmt.Sprintf("%s/insert-without-optimistic-insert", tc.name), func(t *testing.T) {
 			runner.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.try_optimistic_insert.enabled=false")
 			defer runner.Exec(t, "RESET CLUSTER SETTING logical_replication.consumer.try_optimistic_insert.enabled")
-			rp, encoder := setup(t, tc.schemaTmpl)
+			rp, encoder, cleanup := setup(t, tc.schemaTmpl)
+			defer cleanup()
 			keyValue := encoder(tc.row...)
 			require.NoError(t, s.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
 				_, err := rp.ProcessRow(ctx, txn, keyValue, roachpb.Value{})
 				return err
 			}))
 		})
 		t.Run(fmt.Sprintf("%s/delete", tc.name), func(t *testing.T) {
-			rp, encoder := setup(t, tc.schemaTmpl)
+			rp, encoder, cleanup := setup(t, tc.schemaTmpl)
+			defer cleanup()
 			keyValue := encoder(tc.row...)
 			keyValue.Value.RawBytes = nil
 			require.NoError(t, s.InternalDB().(isql.DB).Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
@@ -157,7 +163,7 @@ func BenchmarkLWWInsertBatch(b *testing.B) {
 		desc.GetID(): {
 			srcDesc: desc,
 		},
-	}, jobspb.JobID(1), s.InternalDB().(descs.DB), s.InternalDB().(isql.DB).Executor(isql.WithSessionData(sd)), sd, execinfrapb.LogicalReplicationWriterSpec{})
+	}, jobspb.JobID(1), s.InternalDB().(descs.DB), s.InternalDB().(isql.DB).Executor(isql.WithSessionData(sd)), sd, execinfrapb.LogicalReplicationWriterSpec{}, s.Codec(), s.LeaseManager().(*lease.Manager))
 	require.NoError(b, err)
 
 	// In some configs, we'll be simulating processing the same INSERT over and
@@ -345,7 +351,7 @@ func TestLWWConflictResolution(t *testing.T) {
 			dstDesc.GetID(): {
 				srcDesc: srcDesc,
 			},
-		}, jobspb.JobID(1), s.InternalDB().(descs.DB), s.InternalExecutor().(isql.Executor), sd, execinfrapb.LogicalReplicationWriterSpec{})
+		}, jobspb.JobID(1), s.InternalDB().(descs.DB), s.InternalExecutor().(isql.Executor), sd, execinfrapb.LogicalReplicationWriterSpec{}, s.Codec(), s.LeaseManager().(*lease.Manager))
 		require.NoError(t, err)
 
 		if useKVProc {