Merge pull request #146279 from jeffswenson/backport25.2-146220

release-25.2: revert "kv: use origin timestamp to validate lww semantics"

Author: msbutler

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/BUILD.bazel

@@ -52,7 +52,6 @@ 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",