logical: use KVSavepoint for tombstone updates in table writer

Move tombstone updates from a separate transaction into the session's
SQL transaction using KVSavepoint. Previously, attemptBatch ran
tombstone updates in their own db.Txn after the session transaction
committed. Now they run inline within the session transaction, each
wrapped in a KV savepoint so that a LWW-loser ConditionFailedError
on one tombstone does not abort the surrounding transaction. This
reduces the number of transactions per batch from two to one.

Part of: #169239
Epic: CRDB-60163
Release note: None

Author: jeffswenson

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
 }