sql: always use cput for ldr primary key writes

This change reworks the SQL layer so that it always uses a CPUT with an
origin timestamp if the LDR origin timestamp option is set. This change
allows the classic and crud SQL writers to correctly implement LWW in
the presence of tombstones.

Note: the classic SQL writer only depends on the CPUT when inserting or
upserting over a tombstone. The crud SQL writer relies on the CPut of
inserts, updates, and deletes.

Release note: none
Fixes: cockroachdb#146117

Author: jeffswenson

pkg/crosscluster/logical/batch_handler_test.go

@@ -31,7 +31,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
-	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -198,17 +197,13 @@ func TestBatchHandlerExhaustiveSQL(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 146117)
-
 	testBatchHandlerExhaustive(t, newSqlBatchHandler)
 }
 
 func TestBatchHandlerExhaustiveCrud(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 146117)
-
 	testBatchHandlerExhaustive(t, newCrudBatchHandler)
 }
 
@@ -369,7 +364,6 @@ func testBatchHandlerExhaustive(t *testing.T, factory batchHandlerFactory) {
 		for _, localValue := range localValues {
 			for _, replicationType := range replicationTypes {
 				for _, winLww := range []bool{true, false} {
-
 					if !winLww {
 						if localValue == localValueNull {
 							// If there is no tombstone or local row, then its

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -2790,18 +2790,7 @@ func TestGetWriterType(t *testing.T) {
 		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 	})
 
-	t.Run("immediate-mode-pre-25.2", func(t *testing.T) {
-		st := cluster.MakeTestingClusterSettingsWithVersions(
-			clusterversion.V25_1.Version(),
-			clusterversion.V25_1.Version(),
-			true, /* initializeVersion */
-		)
-		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
-		require.NoError(t, err)
-		require.Equal(t, sqlclustersettings.LDRWriterTypeLegacyKV, wt)
-	})
-
-	t.Run("immediate-mode-post-25.2", func(t *testing.T) {
+	t.Run("immediate-mode", func(t *testing.T) {
 		st := cluster.MakeTestingClusterSettingsWithVersions(
 			clusterversion.V25_2.Version(),
 			clusterversion.PreviousRelease.Version(),
@@ -2812,9 +2801,9 @@ func TestGetWriterType(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 
-		sqlclustersettings.LDRImmediateModeWriter.Override(ctx, &st.SV, string(sqlclustersettings.LDRWriterTypeSQL))
+		sqlclustersettings.LDRImmediateModeWriter.Override(ctx, &st.SV, string(sqlclustersettings.LDRWriterTypeCRUD))
 		wt, err = getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
 		require.NoError(t, err)
-		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
+		require.Equal(t, sqlclustersettings.LDRWriterTypeCRUD, wt)
 	})
 }

pkg/crosscluster/logical/lww_row_processor.go

@@ -897,7 +897,8 @@ DELETE FROM [%d as t] WHERE %s
    AND ((t.crdb_internal_mvcc_timestamp < $%[3]d
         AND t.crdb_internal_origin_timestamp IS NULL)
     OR (t.crdb_internal_origin_timestamp < $%[3]d
-        AND t.crdb_internal_origin_timestamp IS NOT NULL))`
+        AND t.crdb_internal_origin_timestamp IS NOT NULL))
+RETURNING *`
 	stmt, err := parser.ParseOne(
 		fmt.Sprintf(baseQuery, dstTableDescID, whereClause.String(), originTSIdx))
 	if err != nil {

pkg/crosscluster/logical/replication_statements.go

@@ -254,8 +254,11 @@ func newDeleteStatement(
 			TableID: int64(table.GetID()),
 			As:      tree.AliasClause{Alias: "replication_target"},
 		},
-		Where:     &tree.Where{Type: tree.AstWhere, Expr: whereClause},
-		Returning: tree.AbsentReturningClause,
+		Where: &tree.Where{Type: tree.AstWhere, Expr: whereClause},
+		// NOTE: we use RETURNING * to ensure that every column in the table is decoded.
+		// This ensures that the Deleter can reconstruct the previous value when generating
+		// the cput to update the primary key.
+		Returning: &tree.ReturningExprs{tree.StarSelectExpr()},
 	}
 
 	return toParsedStatement(delete)

pkg/crosscluster/logical/testdata/ldr_statements

@@ -18,7 +18,7 @@ UPDATE [104 AS replication_target] SET id = $5::INT8, name = $6::STRING, value =
 
 show-delete table=basic_table
 ----
-DELETE FROM [104 AS replication_target] WHERE (((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (value IS NOT DISTINCT FROM $3::INT8)) AND (data IS NOT DISTINCT FROM $4::BYTES)
+DELETE FROM [104 AS replication_target] WHERE (((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (value IS NOT DISTINCT FROM $3::INT8)) AND (data IS NOT DISTINCT FROM $4::BYTES) RETURNING *
 
 show-select table=basic_table
 ----
@@ -52,7 +52,7 @@ UPDATE [107 AS replication_target] SET id = $7::INT8, title = $8::STRING, descri
 
 show-delete table=tasks
 ----
-DELETE FROM [107 AS replication_target] WHERE (((((id = $1::INT8) AND (title IS NOT DISTINCT FROM $2::STRING)) AND (description IS NOT DISTINCT FROM $3::STRING)) AND (status IS NOT DISTINCT FROM $4::@100105)) AND (priority IS NOT DISTINCT FROM $5::INT8)) AND (created_at IS NOT DISTINCT FROM $6::TIMESTAMP)
+DELETE FROM [107 AS replication_target] WHERE (((((id = $1::INT8) AND (title IS NOT DISTINCT FROM $2::STRING)) AND (description IS NOT DISTINCT FROM $3::STRING)) AND (status IS NOT DISTINCT FROM $4::@100105)) AND (priority IS NOT DISTINCT FROM $5::INT8)) AND (created_at IS NOT DISTINCT FROM $6::TIMESTAMP) RETURNING *
 
 show-select table=tasks
 ----
@@ -86,7 +86,7 @@ UPDATE [108 AS replication_target] SET id = $7::INT8, name = $8::STRING, unit_pr
 # NOTE: total_price and discount_price are not included since they are computed.[
 show-delete table=products
 ----
-DELETE FROM [108 AS replication_target] WHERE ((((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (unit_price IS NOT DISTINCT FROM $3::DECIMAL(10,2))) AND (quantity IS NOT DISTINCT FROM $4::INT8)) AND (last_updated IS NOT DISTINCT FROM $6::TIMESTAMP)
+DELETE FROM [108 AS replication_target] WHERE ((((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (unit_price IS NOT DISTINCT FROM $3::DECIMAL(10,2))) AND (quantity IS NOT DISTINCT FROM $4::INT8)) AND (last_updated IS NOT DISTINCT FROM $6::TIMESTAMP) RETURNING *
 
 # NOTE: total_price is not included because it is a computed column, but
 # discount_price is included because its part of the primary key.
@@ -137,7 +137,7 @@ UPDATE [109 AS replication_target] SET id = $8::INT8, first_name = $9::STRING, l
 
 show-delete table=employees
 ----
-DELETE FROM [109 AS replication_target] WHERE ((((((id = $1::INT8) AND (first_name IS NOT DISTINCT FROM $2::STRING)) AND (last_name IS NOT DISTINCT FROM $3::STRING)) AND (email IS NOT DISTINCT FROM $4::STRING)) AND (salary IS NOT DISTINCT FROM $5::DECIMAL(12,2))) AND (department IS NOT DISTINCT FROM $6::STRING)) AND (hire_date IS NOT DISTINCT FROM $7::DATE)
+DELETE FROM [109 AS replication_target] WHERE ((((((id = $1::INT8) AND (first_name IS NOT DISTINCT FROM $2::STRING)) AND (last_name IS NOT DISTINCT FROM $3::STRING)) AND (email IS NOT DISTINCT FROM $4::STRING)) AND (salary IS NOT DISTINCT FROM $5::DECIMAL(12,2))) AND (department IS NOT DISTINCT FROM $6::STRING)) AND (hire_date IS NOT DISTINCT FROM $7::DATE) RETURNING *
 
 show-select table=employees
 ----
@@ -171,7 +171,7 @@ UPDATE [112 AS replication_target] SET id = $7::UUID, user_id = $8::INT8, event_
 
 show-delete table=user_events
 ----
-DELETE FROM [112 AS replication_target] WHERE (((((id = $1::UUID) AND (user_id IS NOT DISTINCT FROM $2::INT8)) AND (event_type IS NOT DISTINCT FROM $3::STRING)) AND (event_data IS NOT DISTINCT FROM $4::JSONB)) AND (created_at IS NOT DISTINCT FROM $5::TIMESTAMP)) AND (region = $6::@100110)
+DELETE FROM [112 AS replication_target] WHERE (((((id = $1::UUID) AND (user_id IS NOT DISTINCT FROM $2::INT8)) AND (event_type IS NOT DISTINCT FROM $3::STRING)) AND (event_data IS NOT DISTINCT FROM $4::JSONB)) AND (created_at IS NOT DISTINCT FROM $5::TIMESTAMP)) AND (region = $6::@100110) RETURNING *
 
 show-select table=user_events
 ----

pkg/sql/create_table.go

@@ -614,7 +614,7 @@ func (n *createTableNode) startExec(params runParams) error {
 				// indexes, partial, vector, or otherwise, to update.
 				var pm row.PartialIndexUpdateHelper
 				var vh row.VectorIndexUpdateHelper
-				if err := ti.row(params.ctx, rowBuffer, pm, vh, params.extendedEvalCtx.Tracing.KVTracingEnabled()); err != nil {
+				if err := ti.row(params.ctx, rowBuffer, pm, vh, nil, params.extendedEvalCtx.Tracing.KVTracingEnabled()); err != nil {
 					return err
 				}
 			}

pkg/sql/insert.go

@@ -242,8 +242,15 @@ func (r *insertRun) processSourceRow(params runParams, rowVals tree.Datums) erro
 		return err
 	}
 
+	var oth *row.OriginTimestampCPutHelper
+	if params.p.SessionData().OriginTimestampForLogicalDataReplication.IsSet() {
+		oth = &row.OriginTimestampCPutHelper{
+			OriginTimestamp: params.p.SessionData().OriginTimestampForLogicalDataReplication,
+		}
+	}
+
 	// Queue the insert in the KV batch.
-	if err := r.ti.row(params.ctx, insertVals, pm, vh, r.traceKV); err != nil {
+	if err := r.ti.row(params.ctx, insertVals, pm, vh, oth, r.traceKV); err != nil {
 		return err
 	}
 

pkg/sql/sqlclustersettings/clustersettings.go

@@ -129,9 +129,9 @@ var LDRImmediateModeWriter = settings.RegisterStringSetting(
 	settings.ApplicationLevel,
 	"logical_replication.consumer.immediate_mode_writer",
 	"the writer to use when in immediate mode",
-	// TODO(jeffswenson): make the sql writer the default.
-	//metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeSQL), string(LDRWriterTypeLegacyKV), string(LDRWriterTypeCRUD)),
-	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeLegacyKV)),
+	metamorphic.ConstantWithTestChoice(
+		"logical_replication.consumer.immediate_mode_writer",
+		string(LDRWriterTypeSQL), string(LDRWriterTypeLegacyKV), string(LDRWriterTypeCRUD)),
 	settings.WithValidateString(func(sv *settings.Values, val string) error {
 		if val != string(LDRWriterTypeSQL) && val != string(LDRWriterTypeLegacyKV) && val != string(LDRWriterTypeCRUD) {
 			return errors.Newf("immediate mode writer must be either 'sql', 'legacy-kv', or 'crud', got '%s'", val)

pkg/sql/tablewriter_delete.go

@@ -21,12 +21,13 @@ import (
 // tableDeleter handles writing kvs and forming table rows for deletes.
 type tableDeleter struct {
 	tableWriterBase
-
-	rd row.Deleter
+	rd      row.Deleter
+	evalCtx *eval.Context
 }
 
 // init initializes the tableDeleter with a Txn.
 func (td *tableDeleter) init(_ context.Context, txn *kv.Txn, evalCtx *eval.Context) error {
+	td.evalCtx = evalCtx
 	return td.tableWriterBase.init(txn, td.tableDesc(), evalCtx)
 }
 
@@ -59,7 +60,13 @@ func (td *tableDeleter) row(
 	traceKV bool,
 ) error {
 	td.currentBatchSize++
-	return td.rd.DeleteRow(ctx, td.b, values, pm, vh, nil, mustValidateOldPKValues, traceKV)
+	var oth *row.OriginTimestampCPutHelper
+	if td.evalCtx.SessionData().OriginTimestampForLogicalDataReplication.IsSet() {
+		oth = &row.OriginTimestampCPutHelper{
+			OriginTimestamp: td.evalCtx.SessionData().OriginTimestampForLogicalDataReplication,
+		}
+	}
+	return td.rd.DeleteRow(ctx, td.b, values, pm, vh, oth, mustValidateOldPKValues, traceKV)
 }
 
 // deleteIndex runs the kv operations necessary to delete all kv entries in the

pkg/sql/tablewriter_insert.go

@@ -48,10 +48,11 @@ func (ti *tableInserter) row(
 	values tree.Datums,
 	pm row.PartialIndexUpdateHelper,
 	vh row.VectorIndexUpdateHelper,
+	oth *row.OriginTimestampCPutHelper,
 	traceKV bool,
 ) error {
 	ti.currentBatchSize++
-	return ti.ri.InsertRow(ctx, &ti.putter, values, pm, vh, nil, row.CPutOp, traceKV)
+	return ti.ri.InsertRow(ctx, &ti.putter, values, pm, vh, oth, row.CPutOp, traceKV)
 }
 
 // tableDesc returns the TableDescriptor for the table that the tableInserter

pkg/sql/tablewriter_update.go

@@ -18,11 +18,13 @@ import (
 // tableUpdater handles writing kvs and forming table rows for updates.
 type tableUpdater struct {
 	tableWriterBase
-	ru row.Updater
+	ru      row.Updater
+	evalCtx *eval.Context
 }
 
 // init initializes the tableUpdater with a Txn.
 func (tu *tableUpdater) init(_ context.Context, txn *kv.Txn, evalCtx *eval.Context) error {
+	tu.evalCtx = evalCtx
 	return tu.tableWriterBase.init(txn, tu.tableDesc(), evalCtx)
 }
 
@@ -55,8 +57,14 @@ func (tu *tableUpdater) rowForUpdate(
 	traceKV bool,
 ) (tree.Datums, error) {
 	tu.currentBatchSize++
+	var oth *row.OriginTimestampCPutHelper
+	if tu.evalCtx.SessionData().OriginTimestampForLogicalDataReplication.IsSet() {
+		oth = &row.OriginTimestampCPutHelper{
+			OriginTimestamp: tu.evalCtx.SessionData().OriginTimestampForLogicalDataReplication,
+		}
+	}
 	return tu.ru.UpdateRow(
-		ctx, tu.b, oldValues, updateValues, pm, vh, nil, mustValidateOldPKValues, traceKV,
+		ctx, tu.b, oldValues, updateValues, pm, vh, oth, mustValidateOldPKValues, traceKV,
 	)
 }
 

pkg/sql/tablewriter_upsert.go

@@ -171,6 +171,13 @@ func (tu *tableUpserter) row(
 ) error {
 	tu.currentBatchSize++
 
+	var oth *row.OriginTimestampCPutHelper
+	if tu.originTimestamp.IsSet() {
+		oth = &row.OriginTimestampCPutHelper{
+			OriginTimestamp: tu.originTimestamp,
+		}
+	}
+
 	// Consult the canary column to determine whether to insert or update. For
 	// more details on how canary columns work, see the block comment on
 	// Builder.buildInsert in opt/optbuilder/insert.go.
@@ -186,11 +193,11 @@ func (tu *tableUpserter) row(
 		// - if buffered writes are disabled, then the KV layer will write an
 		// intent which acts as a lock.
 		kvOp := row.PutMustAcquireExclusiveLockOp
-		return tu.insertNonConflictingRow(ctx, datums[:insertEnd], pm, vh, kvOp, traceKV)
+		return tu.insertNonConflictingRow(ctx, datums[:insertEnd], pm, vh, oth, kvOp, traceKV)
 	}
 	if datums[tu.canaryOrdinal] == tree.DNull {
 		// No conflict, so insert a new row.
-		return tu.insertNonConflictingRow(ctx, datums[:insertEnd], pm, vh, row.CPutOp, traceKV)
+		return tu.insertNonConflictingRow(ctx, datums[:insertEnd], pm, vh, oth, row.CPutOp, traceKV)
 	}
 
 	// If no columns need to be updated, then possibly collect the unchanged row.
@@ -229,11 +236,12 @@ func (tu *tableUpserter) insertNonConflictingRow(
 	insertRow tree.Datums,
 	pm row.PartialIndexUpdateHelper,
 	vh row.VectorIndexUpdateHelper,
+	oh *row.OriginTimestampCPutHelper,
 	kvOp row.KVInsertOp,
 	traceKV bool,
 ) error {
 	// Perform the insert proper.
-	if err := tu.ri.InsertRow(ctx, &tu.putter, insertRow, pm, vh, nil /* oth */, kvOp, traceKV); err != nil {
+	if err := tu.ri.InsertRow(ctx, &tu.putter, insertRow, pm, vh, oh, kvOp, traceKV); err != nil {
 		return err
 	}