logical: feature flag off UDF support

This adds a logical_replication.deprecated_udf_writer.enabled feature
flag that turns off LDR UDF support. The UDF writer will be permanently
deleted after a full upgrade cycle.

Fixes: 148312
Release note: none

Author: jeffswenson

pkg/crosscluster/logical/create_logical_replication_stmt.go

@@ -117,6 +117,12 @@ func createLogicalReplicationStreamPlanHook(
 
 		hasUDF := len(options.userFunctions) > 0 || options.defaultFunction != nil && options.defaultFunction.FunctionId != 0
 
+		if hasUDF && !crosscluster.LogicalReplicationUDFWriterEnabled.Get(&p.ExecCfg().Settings.SV) {
+			return pgerror.Newf(pgcode.FeatureNotSupported,
+				"UDF-based logical replication is disabled and will be deleted in a future CockroachDB release. "+
+					"Enable with SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+		}
+
 		mode := jobspb.LogicalReplicationDetails_Immediate
 		if m, ok := options.GetMode(); ok {
 			switch m {

pkg/crosscluster/logical/create_logical_replication_stmt_test.go

@@ -11,6 +11,7 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationtestutils"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
@@ -201,3 +202,27 @@ func TestResolveDestinationObjects(t *testing.T) {
 	}
 
 }
+
+func TestUDFLogicalReplicationDisabled(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	tc, s, runnerA, runnerB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
+	defer tc.Stopper().Stop(ctx)
+
+	runnerA.Exec(t, "CREATE TABLE test_tab(pk INT PRIMARY KEY, v INT)")
+	runnerB.Exec(t, "CREATE TABLE test_tab(pk INT PRIMARY KEY, v INT)")
+
+	runnerB.Exec(t, `
+		CREATE FUNCTION repl_apply(action STRING, proposed test_tab, existing test_tab, prev test_tab, existing_mvcc_timestamp DECIMAL, existing_origin_timestamp DECIMAL, proposed_mvcc_timestamp DECIMAL)
+		RETURNS string AS $$ BEGIN RETURN 'accept_proposed'; END $$ LANGUAGE plpgsql
+	`)
+
+	dbAURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("a"))
+
+	// Should fail with UDF writer disabled (default setting)
+	runnerB.ExpectErr(t, "UDF-based logical replication is disabled",
+		"CREATE LOGICAL REPLICATION STREAM FROM TABLE test_tab ON $1 INTO TABLE test_tab WITH FUNCTION repl_apply FOR TABLE test_tab",
+		dbAURL.String())
+}

pkg/crosscluster/logical/logical_replication_job.go

@@ -87,9 +87,31 @@ type logicalReplicationResumer struct {
 
 var _ jobs.Resumer = (*logicalReplicationResumer)(nil)
 
+func (r *logicalReplicationResumer) jobUsesUDF() bool {
+	payload := r.job.Details().(jobspb.LogicalReplicationDetails)
+
+	if payload.DefaultConflictResolution.FunctionId != 0 {
+		return true
+	}
+
+	for _, pair := range payload.ReplicationPairs {
+		if pair.DstFunctionID != 0 {
+			return true
+		}
+	}
+
+	return false
+}
+
 // Resume is part of the jobs.Resumer interface.
 func (r *logicalReplicationResumer) Resume(ctx context.Context, execCtx interface{}) error {
 	jobExecCtx := execCtx.(sql.JobExecContext)
+
+	if r.jobUsesUDF() && !crosscluster.LogicalReplicationUDFWriterEnabled.Get(&jobExecCtx.ExecCfg().Settings.SV) {
+		r.updateStatusMessage(ctx, "job paused because UDF-based logical replication writer is disabled")
+		return jobs.MarkPauseRequestError(errors.Newf("UDF-based logical replication writer is disabled and will be deleted in a future CockroachDB release"))
+	}
+
 	return r.handleResumeError(ctx, jobExecCtx, r.ingestWithRetries(ctx, jobExecCtx))
 }
 

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -1024,6 +1024,9 @@ func TestPreviouslyInterestingTables(t *testing.T) {
 	tc, s, runnerA, runnerB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
 	defer tc.Stopper().Stop(ctx)
 
+	runnerA.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+	runnerB.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+
 	sqlA := s.SQLConn(t, serverutils.DBName("a"))
 
 	type testCase struct {
@@ -1838,6 +1841,9 @@ func TestLogicalStreamIngestionJobWithFallbackUDF(t *testing.T) {
 	server, s, dbA, dbB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
 	defer server.Stopper().Stop(ctx)
 
+	dbA.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+	dbB.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+
 	lwwFunc := `CREATE OR REPLACE FUNCTION repl_apply(action STRING, proposed tab, existing tab, prev tab, existing_mvcc_timestamp DECIMAL, existing_origin_timestamp DECIMAL, proposed_mvcc_timestamp DECIMAL)
 	RETURNS string
 	AS $$

pkg/crosscluster/logical/udf_row_processor_test.go

@@ -53,6 +53,9 @@ func TestUDFWithRandomTables(t *testing.T) {
 	tc, s, runnerA, runnerB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
 	defer tc.Stopper().Stop(ctx)
 
+	runnerA.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+	runnerB.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+
 	tableName := "rand_table"
 	rng, _ := randutil.NewPseudoRand()
 	createStmt := randgen.RandCreateTableWithName(
@@ -108,6 +111,9 @@ func TestUDFInsertOnly(t *testing.T) {
 	tc, s, runnerA, runnerB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
 	defer tc.Stopper().Stop(ctx)
 
+	runnerA.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+	runnerB.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+
 	tableName := "tallies"
 	stmt := "CREATE TABLE tallies(pk INT PRIMARY KEY, v INT)"
 	runnerA.Exec(t, stmt)
@@ -157,6 +163,9 @@ func TestUDFPreviousValue(t *testing.T) {
 	tc, s, runnerA, runnerB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
 	defer tc.Stopper().Stop(ctx)
 
+	runnerA.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+	runnerB.Exec(t, "SET CLUSTER SETTING logical_replication.deprecated_udf_writer.enabled = true")
+
 	tableName := "tallies"
 	stmt := "CREATE TABLE tallies(pk INT PRIMARY KEY, v INT)"
 	runnerA.Exec(t, stmt)

pkg/crosscluster/settings.go

@@ -126,3 +126,13 @@ var LogicalReplanFrequency = settings.RegisterDurationSetting(
 	10*time.Minute,
 	settings.PositiveDuration,
 )
+
+// LogicalReplicationUDFWriterEnabled controls whether the UDF-based logical
+// data replication writer is enabled. When disabled, existing UDF writer jobs
+// will be paused and new UDF LDR jobs cannot be created.
+var LogicalReplicationUDFWriterEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"logical_replication.deprecated_udf_writer.enabled",
+	"enables the UDF-based logical data replication writer (deprecated, will be removed)",
+	false,
+)