crosscluster: add batch handler test support for sql writers

This extends the batch handler tests so that they can be used to
validate the SQL writers. The new tests are skipped because the SQL
writers do not correctly handle tombstones that should win LWW.

Release note: none
Informs: 146117

Author: jeffswenson

pkg/crosscluster/logical/batch_handler_test.go

@@ -13,21 +13,25 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdctest"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"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/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"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"
@@ -116,7 +120,7 @@ func (b *EventBuilder) deleteEvent(
 // newKvBatchHandler creates a new batch handler for testing.
 func newKvBatchHandler(
 	t *testing.T, s serverutils.ApplicationLayerInterface, tableName string,
-) (*kvRowProcessor, catalog.TableDescriptor) {
+) (BatchHandler, catalog.TableDescriptor) {
 	ctx := context.Background()
 	desc := cdctest.GetHydratedTableDescriptor(t, s.ExecutorConfig(), tree.Name(tableName))
 	sd := sql.NewInternalSessionData(ctx, s.ClusterSettings(), "" /* opName */)
@@ -143,6 +147,38 @@ func newKvBatchHandler(
 	return handler, desc
 }
 
+// newSqlBatchHandler creates a new SQL-based batch handler for testing.
+func newSqlBatchHandler(
+	t *testing.T, s serverutils.ApplicationLayerInterface, tableName string,
+) (BatchHandler, catalog.TableDescriptor) {
+	ctx := context.Background()
+	desc := cdctest.GetHydratedTableDescriptor(t, s.ExecutorConfig(), tree.Name(tableName))
+	sd := sql.NewInternalSessionData(ctx, s.ClusterSettings(), "" /* opName */)
+	codec := s.Codec()
+	handler, err := makeSQLProcessor(
+		ctx,
+		s.ClusterSettings(),
+		map[descpb.ID]sqlProcessorTableConfig{
+			desc.GetID(): {
+				srcDesc: desc,
+			},
+		},
+		0, // jobID
+		s.InternalDB().(descs.DB),
+		s.InternalDB().(isql.DB).Executor(isql.WithSessionData(sd)),
+		sd,
+		execinfrapb.LogicalReplicationWriterSpec{
+			Mode: jobspb.LogicalReplicationDetails_Immediate,
+		},
+		codec,
+		s.LeaseManager().(*lease.Manager),
+	)
+	require.NoError(t, err)
+	return handler, desc
+}
+
+type batchHandlerFactory func(t *testing.T, s serverutils.ApplicationLayerInterface, tableName string) (BatchHandler, catalog.TableDescriptor)
+
 // addAndRemoveColumn is a metamorphic test option that randomly decides whether
 // to add and remove a column to trigger a rebuild of the primary key.
 var addAndRemoveColumn = metamorphic.ConstantWithTestBool("batch-handler-exhaustive-rebuild-primary-index", false)
@@ -151,10 +187,34 @@ var addAndRemoveColumn = metamorphic.ConstantWithTestBool("batch-handler-exhaust
 // to add a unique constraint to the table.
 var uniqueConstraint = metamorphic.ConstantWithTestBool("batch-handler-exhaustive-unique-constraint", false)
 
-func TestBatchHandlerExhaustive(t *testing.T) {
+func TestBatchHandlerExhaustiveKV(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testBatchHandlerExhaustive(t, newKvBatchHandler)
+}
+
+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)
+}
+
+func testBatchHandlerExhaustive(t *testing.T, factory batchHandlerFactory) {
+	t.Helper()
+
 	// This test is an "exhaustive" test of the batch handler. It tries to test
 	// cross product of every possible (replication event type, local value,
 	// previous value, lww win).
@@ -188,7 +248,7 @@ func TestBatchHandlerExhaustive(t *testing.T) {
 		runner.Exec(t, `ALTER TABLE test_table DROP COLUMN temp_col`)
 	}
 
-	handler, desc := newKvBatchHandler(t, s, "test_table")
+	handler, desc := factory(t, s, "test_table")
 	defer handler.ReleaseLeases(ctx)
 
 	// TODO(jeffswenson): test the other handler types.

pkg/crosscluster/logical/sql_crud_writer.go

@@ -45,7 +45,7 @@ func newCrudSqlWriter(
 	discard jobspb.LogicalReplicationDetails_Discard,
 	procConfigByDestID map[descpb.ID]sqlProcessorTableConfig,
 	jobID jobspb.JobID,
-) (*sqlCrudWriter, error) {
+) (BatchHandler, error) {
 	decoder, err := newEventDecoder(ctx, cfg.DB, evalCtx.Settings, procConfigByDestID)
 	if err != nil {
 		return nil, err

pkg/crosscluster/logical/table_batch_handler_test.go

@@ -30,7 +30,7 @@ import (
 
 func newCrudBatchHandler(
 	t *testing.T, s serverutils.ApplicationLayerInterface, tableName string,
-) (*sqlCrudWriter, catalog.TableDescriptor) {
+) (BatchHandler, catalog.TableDescriptor) {
 	ctx := context.Background()
 	desc := cdctest.GetHydratedTableDescriptor(t, s.ExecutorConfig(), tree.Name(tableName))
 	sd := sql.NewInternalSessionData(ctx, s.ClusterSettings(), "" /* opName */)