logical: use KVSavepoint for tombstone updates in table writer (#169360)

logical: use KVSavepoint for tombstone updates in table writer

Author: trunk-io[bot]

pkg/crosscluster/logical/table_batch_handler.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/logical/sqlwriter"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
@@ -30,7 +31,6 @@ type tableHandler struct {
 	sqlReader        sqlwriter.RowReader
 	sqlWriter        *sqlwriter.RowWriter
 	session          isql.Session
-	db               descs.DB
 	tombstoneUpdater *tombstoneUpdater
 }
 
@@ -132,7 +132,6 @@ func newTableHandler(
 	return &tableHandler{
 		sqlReader:        reader,
 		sqlWriter:        writer,
-		db:               db,
 		tombstoneUpdater: tombstoneUpdater,
 		session:          session,
 	}, nil
@@ -170,7 +169,6 @@ func (t *tableHandler) attemptBatch(
 ) (tableBatchStats, error) {
 	var stats tableBatchStats
 
-	var hasTombstoneUpdates bool
 	err := t.session.Txn(ctx, func(ctx context.Context) error {
 		for _, event := range batch {
 			switch {
@@ -181,8 +179,24 @@ func (t *tableHandler) attemptBatch(
 					return err
 				}
 			case event.IsTombstoneUpdate():
-				hasTombstoneUpdates = true
-				// Skip: handled in its own transaction.
+				stats.tombstoneUpdates++
+				// Use a KV savepoint so that a LWW-loser ConditionFailedError
+				// on one tombstone does not abort the surrounding transaction.
+				err := t.session.KVSavepoint(ctx, func(ctx context.Context, txn *kv.Txn) error {
+					batch := txn.NewBatch()
+					batch.Header.WriteOptions = originID1Options
+					if err := t.tombstoneUpdater.addToBatch(ctx, txn, batch, event.RowTimestamp, event.Row); err != nil {
+						return err
+					}
+					return txn.Run(ctx, batch)
+				})
+				if err != nil {
+					if isLwwLoser(err) {
+						stats.kvLwwLosers++
+						continue
+					}
+					return err
+				}
 			case event.IsInsertRow():
 				stats.inserts++
 				err := t.sqlWriter.InsertRow(ctx, event.RowTimestamp, event.Row)
@@ -211,29 +225,6 @@ func (t *tableHandler) attemptBatch(
 		return tableBatchStats{}, err
 	}
 
-	if hasTombstoneUpdates {
-		// TODO(jeffswenson): once we have a way to expose the transaction used by
-		// the Session, we should bundle this with the other txn. The purpose of
-		// these transactions is batching writes in a transaction increases
-		// efficiency. The transactions are not needed for correctness.
-		err = t.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-			for _, event := range batch {
-				if event.IsTombstoneUpdate() {
-					stats.tombstoneUpdates++
-					tombstoneUpdateStats, err := t.tombstoneUpdater.updateTombstone(ctx, txn, event.RowTimestamp, event.Row)
-					if err != nil {
-						return err
-					}
-					stats.kvLwwLosers += tombstoneUpdateStats.kvWriteTooOld
-				}
-			}
-			return nil
-		})
-		if err != nil {
-			return tableBatchStats{}, err
-		}
-	}
-
 	return stats, nil
 }