logical: switch back to kv writer

25.2 was going to use the sql writer by default. But #146117 requires us
to roll back #143100, which means the sql writer no longer implements
LWW correctly in the presence of tombstones.

This also includes a change to fix the cput origin time validation. If
the cput is allowed to include origin time in the batch header and the
cput request iff the times match. This behavior will be required by the
sql writer to fix tombstone handling.

This removes the version check in `getWriterType` since the SQL writer
no longer changes it behavior based on the version of the KV server.

Release note: none
Part of: #146117
Informs: #146217

Author: jeffswenson

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -2790,7 +2790,7 @@ func TestGetWriterType(t *testing.T) {
 		)
 		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Immediate, st)
 		require.NoError(t, err)
-		require.Equal(t, writerTypeSQL, wt)
+		require.Equal(t, writerTypeLegacyKV, wt)
 	})
 
 	t.Run("immediate-mode-post-25.2", func(t *testing.T) {

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -114,7 +114,9 @@ var immediateModeWriter = settings.RegisterStringSetting(
 	settings.ApplicationLevel,
 	"logical_replication.consumer.immediate_mode_writer",
 	"the writer to use when in immediate mode",
-	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(writerTypeSQL), string(writerTypeLegacyKV), string(writerTypeCRUD)),
+	// TODO(jeffswenson): re-enable the SQL writer once tombstone handling is fixed
+	// metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(writerTypeSQL), string(writerTypeLegacyKV), string(writerTypeCRUD)),
+	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(writerTypeLegacyKV)),
 	settings.WithValidateString(func(sv *settings.Values, val string) error {
 		if val != string(writerTypeSQL) && val != string(writerTypeLegacyKV) && val != string(writerTypeCRUD) {
 			return errors.Newf("immediate mode writer must be either 'sql', 'legacy-kv', or 'crud', got '%s'", val)
@@ -783,12 +785,7 @@ func getWriterType(
 ) (writerType, error) {
 	switch mode {
 	case jobspb.LogicalReplicationDetails_Immediate:
-		// Require v25.2 to use the sql writer by default to ensure that the
-		// KV origin timestamp validation is available on all nodes.
-		if settings.Version.IsActive(ctx, clusterversion.V25_2) {
-			return writerType(immediateModeWriter.Get(&settings.SV)), nil
-		}
-		return writerTypeSQL, nil
+		return writerType(immediateModeWriter.Get(&settings.SV)), nil
 	case jobspb.LogicalReplicationDetails_Validated:
 		return writerTypeSQL, nil
 	default:

pkg/crosscluster/logical/lww_row_processor_test.go

@@ -467,6 +467,11 @@ 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 {
+					// TODO(jeffswenson): re-enable this test
+					skip.WithIssue(t, 146217, "crud writer does not correctly implement lww")
+				}
+
 				tableNameDst, rp, encoder := setup(t, useKVProc)
 
 				runner.Exec(t, fmt.Sprintf("INSERT INTO %s VALUES ($1, $2)", tableNameDst), row1...)

pkg/crosscluster/logical/table_batch_handler_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"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"
@@ -199,6 +200,9 @@ func TestTableHandlerExhaustive(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
+	// TODO(jeffswenson): re-enable this test
+	skip.WithIssue(t, 146217, "crud writer does not correctly implement lww")
+
 	// This test is an "exhaustive" test of the table handler. It tries to test
 	// cross product of every possible (replication event type, local value,
 	// previous value, lww win).

pkg/kv/kvserver/batcheval/cmd_conditional_put.go

@@ -61,9 +61,9 @@ func ConditionalPut(
 	originTimestampForValueHeader := h.WriteOptions.GetOriginTimestamp()
 	if args.OriginTimestamp.IsSet() {
 		originTimestampForValueHeader = args.OriginTimestamp
-	}
-	if args.OriginTimestamp.IsSet() && h.WriteOptions.GetOriginTimestamp().IsSet() {
-		return result.Result{}, errors.AssertionFailedf("OriginTimestamp cannot be passed via CPut arg and in request header")
+		if h.WriteOptions.GetOriginTimestamp().IsSet() && args.OriginTimestamp != h.WriteOptions.GetOriginTimestamp() {
+			return result.Result{}, errors.AssertionFailedf("OriginTimestamp passed via CPut arg must match the origin timestamp in the request header")
+		}
 	}
 
 	opts := storage.ConditionalPutWriteOptions{