logical: create table handler

The table handler is a `BatchHandler` that operates on rows from a
single table. Each batch is attempted twice. Once assuming the values
supplied are accurate and a second time with values read from the local
database.

Release note: none
Epic: CRDB-48647

Author: jeffswenson

pkg/crosscluster/logical/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     srcs = [
         "create_logical_replication_stmt.go",
         "dead_letter_queue.go",
+        "event_decoder.go",
         "logical_replication_dist.go",
         "logical_replication_job.go",
         "logical_replication_writer_processor.go",
@@ -18,6 +19,7 @@ go_library(
         "replication_statements.go",
         "sql_row_reader.go",
         "sql_row_writer.go",
+        "table_batch_handler.go",
         "tombstone_updater.go",
         "udf_row_processor.go",
     ],
@@ -125,6 +127,7 @@ go_test(
         "replication_statements_test.go",
         "sql_row_reader_test.go",
         "sql_row_writer_test.go",
+        "table_batch_handler_test.go",
         "udf_row_processor_test.go",
     ],
     data = glob(["testdata/**"]) + [

pkg/crosscluster/logical/event_decoder.go

@@ -0,0 +1,39 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package logical
+
+import (
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+)
+
+// decodedEvent is constructed from a replication stream event. The replication
+// stream event is encoded using the source table descriptor. The decoded must
+// contain datums that are compatible with the destination table descriptor.
+type decodedEvent struct {
+	// dstDescID is the descriptor ID for the table in the destination cluster.
+	dstDescID descpb.ID
+
+	// isDelete is true if the event is a delete. This implies values in the row
+	// may be NULL that are not allowed to be NULL in the source table's schema.
+	// Only the primary key columns are expected to have values.
+	isDelete bool
+
+	// originTimestamp is the mvcc timestamp of the row in the source cluster.
+	originTimestamp hlc.Timestamp
+
+	// row is the decoded row from the replication stream. The datums in row are
+	// in col id order for the destination table.
+	row tree.Datums
+
+	// prevRow is either the previous row from the replication stream or it is
+	// the local version of the row if there was a read refresh.
+	//
+	// The datums in prevRow are in col id order for the destination table. nil
+	// prevRow may still lose LWW if there is a recent tombstone.
+	prevRow tree.Datums
+}

pkg/crosscluster/logical/table_batch_handler.go

@@ -0,0 +1,240 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package logical
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"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/isql"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/errors"
+)
+
+// tableHandler applies batches of replication events that are destined for a
+// sinlgle table.
+type tableHandler struct {
+	sqlReader        *sqlRowReader
+	sqlWriter        *sqlRowWriter
+	db               descs.DB
+	tombstoneUpdater *tombstoneUpdater
+}
+
+type tableBatchStats struct {
+	// refreshedRows are the number of rows that were re-read from the local
+	// database.
+	refreshedRows int64
+	// inserts is the number of rows that were inserted.
+	inserts int64
+	// updates is the number of rows that were updated.
+	updates int64
+	// deletes is the number of rows that were deleted.
+	deletes int64
+	// tombstoneUpdates is the number of tombstones that were updated. This case
+	// only occurs if the event is a replicated delete and the row did not exist
+	// locally.
+	tombstoneUpdates int64
+	// refreshLwwLosers is the number of rows that were dropped as lww losers
+	// after reading them locally.
+	refreshLwwLosers int64
+	// kvLwwLosers is the number of rows that were dropped as lww losers after
+	// attempting to write to the KV layer. This case only occurs if there is a
+	// tombstone that is more recent than the replicated action.
+	kvLwwLosers int64
+}
+
+func (t *tableBatchStats) Add(o tableBatchStats) {
+	t.refreshedRows += o.refreshedRows
+	t.inserts += o.inserts
+	t.updates += o.updates
+	t.deletes += o.deletes
+	t.tombstoneUpdates += o.tombstoneUpdates
+	t.refreshLwwLosers += o.refreshLwwLosers
+	t.kvLwwLosers += o.kvLwwLosers
+}
+
+// newTableHandler creates a new tableHandler for the given table descriptor ID.
+// It internally constructs the sqlReader and sqlWriter components.
+func newTableHandler(
+	tableID descpb.ID,
+	db descs.DB,
+	codec keys.SQLCodec,
+	sd *sessiondata.SessionData,
+	leaseMgr *lease.Manager,
+	settings *cluster.Settings,
+) (*tableHandler, error) {
+	var table catalog.TableDescriptor
+
+	// NOTE: we don't hold a lease on the table descriptor, but validation
+	// prevents users from changing the set of columns or the primary key of an
+	// LDR replicated table.
+	err := db.DescsTxn(context.Background(), func(ctx context.Context, txn descs.Txn) error {
+		var err error
+		table, err = txn.Descriptors().GetLeasedImmutableTableByID(ctx, txn.KV(), tableID)
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	reader, err := newSQLRowReader(table)
+	if err != nil {
+		return nil, err
+	}
+
+	writer, err := newSQLRowWriter(table)
+	if err != nil {
+		return nil, err
+	}
+
+	tombstoneUpdater := newTombstoneUpdater(codec, db.KV(), leaseMgr, tableID, sd, settings)
+
+	return &tableHandler{
+		sqlReader:        reader,
+		sqlWriter:        writer,
+		db:               db,
+		tombstoneUpdater: tombstoneUpdater,
+	}, nil
+}
+
+func (t *tableHandler) handleDecodedBatch(
+	ctx context.Context, batch []decodedEvent,
+) (tableBatchStats, error) {
+	stats, err := t.attemptBatch(ctx, batch)
+	if err == nil {
+		return stats, nil
+	}
+
+	refreshedBatch, refreshStats, err := t.refreshPrevRows(ctx, batch)
+	if err != nil {
+		return stats, err
+	}
+
+	stats, err = t.attemptBatch(ctx, refreshedBatch)
+	if err != nil {
+		return tableBatchStats{}, err
+	}
+
+	stats.Add(refreshStats)
+
+	return stats, nil
+}
+
+func (t *tableHandler) attemptBatch(
+	ctx context.Context, batch []decodedEvent,
+) (tableBatchStats, error) {
+	var stats tableBatchStats
+	err := t.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		for _, event := range batch {
+			switch {
+			case event.isDelete && len(event.prevRow) != 0:
+				stats.deletes++
+				err := t.sqlWriter.DeleteRow(ctx, txn, event.originTimestamp, event.prevRow)
+				if err != nil {
+					return err
+				}
+			case event.isDelete && len(event.prevRow) == 0:
+				stats.tombstoneUpdates++
+				tombstoneUpdateStats, err := t.tombstoneUpdater.updateTombstone(ctx, txn, event.originTimestamp, event.row)
+				if err != nil {
+					return err
+				}
+				stats.kvLwwLosers += tombstoneUpdateStats.kvWriteTooOld
+			case event.prevRow == nil:
+				stats.inserts++
+				err := t.sqlWriter.InsertRow(ctx, txn, event.originTimestamp, event.row)
+				if isLwwLoser(err) {
+					// Insert may observe a LWW failure if it attempts to write over a tombstone.
+					stats.kvLwwLosers++
+					continue
+				}
+				if err != nil {
+					return err
+				}
+			case event.prevRow != nil:
+				stats.updates++
+				err := t.sqlWriter.UpdateRow(ctx, txn, event.originTimestamp, event.prevRow, event.row)
+				if err != nil {
+					return err
+				}
+			default:
+				return errors.AssertionFailedf("unhandled event type: %v", event)
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return tableBatchStats{}, err
+	}
+	return stats, nil
+}
+
+// refreshPrevRows refreshes the prevRow field for each event in the batch. If
+// any event is known to be a lww loser based on the read, its dropped from the
+// batch.
+func (t *tableHandler) refreshPrevRows(
+	ctx context.Context, batch []decodedEvent,
+) ([]decodedEvent, tableBatchStats, error) {
+	var stats tableBatchStats
+	stats.refreshedRows = int64(len(batch))
+
+	rows := make([]tree.Datums, 0, len(batch))
+	for _, event := range batch {
+		rows = append(rows, event.row)
+	}
+
+	var refreshedRows map[int]priorRow
+	err := t.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		var err error
+		// TODO(jeffswenson): should we apply the batch in the same transaction
+		// that we perform the read refresh? We could maybe even use locking reads.
+		refreshedRows, err = t.sqlReader.ReadRows(ctx, txn, rows)
+		return err
+	})
+	if err != nil {
+		return nil, tableBatchStats{}, err
+	}
+
+	refreshedBatch := make([]decodedEvent, 0, len(batch))
+	for i, event := range batch {
+		var prevRow tree.Datums
+		if refreshed, found := refreshedRows[i]; found {
+			if !refreshed.logicalTimestamp.Less(event.originTimestamp) {
+				// TODO(jeffswenson): update this logic when its time to handle
+				// ties.
+				// Skip the row because it is a lww loser. Note: we can only identify LWW
+				// losers during the read refresh if the row exists. If its a tombstone,
+				// the local value may win LWW, but we have to attempt the
+				// insert/tombstone update to find out. We even have to do this if the
+				// replicated event is a delete because the local tombstone may have an
+				// older logical timestamp.
+				stats.refreshLwwLosers++
+				continue
+			}
+			prevRow = refreshed.row
+		}
+		refreshedEvent := decodedEvent{
+			dstDescID:       event.dstDescID,
+			isDelete:        event.isDelete,
+			row:             event.row,
+			originTimestamp: event.originTimestamp,
+			prevRow:         prevRow,
+		}
+		refreshedBatch = append(refreshedBatch, refreshedEvent)
+	}
+
+	return refreshedBatch, stats, nil
+}
+
+func (t *tableHandler) ReleaseLeases(ctx context.Context) {
+	t.tombstoneUpdater.ReleaseLeases(ctx)
+}