crosscluster: use new origin header values

This change aims to correct the SQL writer's LWW semantics, enabling the
retirement of the KV writer.

The PR updates the implementation of put, cput, delete, and range delete
to return an error if a replicated write with an origin timestamp would
overwrite a value with a more recent origin or MVCC timestamp. As part
of this update, tie-handling logic has been removed since the current KV
writer does not use it, and a simple "bool wins ties" hint is
insufficient for proper LWW implementation.

No explicit version checks are included. In a mixed-version environment,
a SQL writer running 25.1 and writing to a 25.2 KV server will encounter
"condition failed" errors and retry the write until it is upgraded or
the LDR PTS ages out. This is acceptable, as LDR remains in preview and
the SQL writer is neither the default nor the recommended writer.

Release note: none
Epic: CRDB-48647

Author: jeffswenson

pkg/crosscluster/logical/lww_row_processor.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
@@ -80,6 +81,15 @@ type querier interface {
 	RequiresParsedBeforeRow(catid.DescID) bool
 }
 
+// isLwwLoser returns true if the error is a ConditionFailedError with an
+// OriginTimestampOlderThan set.
+func isLwwLoser(err error) bool {
+	if condErr := (*kvpb.ConditionFailedError)(nil); errors.As(err, &condErr) {
+		return condErr.OriginTimestampOlderThan.IsSet()
+	}
+	return false
+}
+
 type queryBuilder struct {
 	// stmts are parsed SQL statements. They should have the same number
 	// of inputs.
@@ -592,6 +602,9 @@ func (lww *lwwQuerier) InsertRow(
 			sess.QualityOfService = nil
 		}
 		if _, err = ie.ExecParsed(ctx, replicatedOptimisticInsertOpName, kvTxn, sess, stmt, datums...); err != nil {
+			if isLwwLoser(err) {
+				return batchStats{}, nil
+			}
 			// If the optimistic insert failed with unique violation, we have to
 			// fall back to the pessimistic path. If we got a different error,
 			// then we bail completely.
@@ -615,7 +628,11 @@ func (lww *lwwQuerier) InsertRow(
 		sess.QualityOfService = nil
 	}
 	sess.OriginTimestampForLogicalDataReplication = row.MvccTimestamp
-	if _, err = ie.ExecParsed(ctx, replicatedInsertOpName, kvTxn, sess, stmt, datums...); err != nil {
+	_, err = ie.ExecParsed(ctx, replicatedInsertOpName, kvTxn, sess, stmt, datums...)
+	if isLwwLoser(err) {
+		return batchStats{}, nil
+	}
+	if err != nil {
 		log.Warningf(ctx, "replicated insert failed (query: %s): %s", stmt.SQL, err.Error())
 		return batchStats{}, err
 	}

pkg/crosscluster/logical/lww_row_processor_test.go

@@ -459,9 +459,6 @@ func TestLWWConflictResolution(t *testing.T) {
 				runner.CheckQueryResults(t, fmt.Sprintf("SELECT * from %s", tableNameDst), expectedRows)
 			})
 			t.Run("cross-cluster-local-delete", func(t *testing.T) {
-				if !useKVProc {
-					skip.IgnoreLint(t, "local delete ordering is not handled correctly by the SQL processor")
-				}
 				tableNameDst, rp, encoder := setup(t, useKVProc)
 
 				runner.Exec(t, fmt.Sprintf("INSERT INTO %s VALUES ($1, $2)", tableNameDst), row1...)
@@ -507,9 +504,6 @@ func TestLWWConflictResolution(t *testing.T) {
 				runner.CheckQueryResults(t, fmt.Sprintf("SELECT * from %s", tableNameDst), expectedRows)
 			})
 			t.Run("remote-delete-after-local-delete", func(t *testing.T) {
-				if !useKVProc {
-					skip.IgnoreLint(t, "local delete ordering is not handled correctly by the SQL processor")
-				}
 				tableNameDst, rp, encoder := setup(t, useKVProc)
 
 				runner.Exec(t, fmt.Sprintf("INSERT INTO %s VALUES ($1, $2)", tableNameDst), row1...)

pkg/kv/BUILD.bazel

@@ -51,6 +51,7 @@ go_test(
         "admission_test.go",
         "client_test.go",
         "db_test.go",
+        "kv_client_lww_test.go",
         "main_test.go",
         "range_lookup_test.go",
         "txn_external_test.go",

pkg/kv/batch.go

@@ -578,9 +578,7 @@ func (b *Batch) CPutAllowingIfNotExists(key, value interface{}, expValue []byte)
 // This is used by logical data replication and other uses of this API
 // are discouraged since the semantics are subject to change as
 // required by that feature.
-func (b *Batch) CPutWithOriginTimestamp(
-	key, value interface{}, expValue []byte, ts hlc.Timestamp, shouldWinTie bool,
-) {
+func (b *Batch) CPutWithOriginTimestamp(key, value interface{}, expValue []byte, ts hlc.Timestamp) {
 	k, err := marshalKey(key)
 	if err != nil {
 		b.initResult(0, 1, notRaw, err)
@@ -594,7 +592,6 @@ func (b *Batch) CPutWithOriginTimestamp(
 	}
 	r := kvpb.NewConditionalPut(k, v, expValue, false)
 	r.(*kvpb.ConditionalPutRequest).OriginTimestamp = ts
-	r.(*kvpb.ConditionalPutRequest).ShouldWinOriginTimestampTie = shouldWinTie
 	b.appendReqs(r)
 	b.approxMutationReqBytes += len(k) + len(v.RawBytes)
 	b.initResult(1, 1, notRaw, nil)