Merge #56872

56872: changefeedccl, importccl, sql: IMPORT/EXPORT/changefeed feature flags r=otan a=angelapwen

(Duplicates #56833 which was not updating my most recent commits)

Adds a cluster setting to toggle a feature flag for the IMPORT,
EXPORT, and changefeed commands off and on. The feature is being
introduced to address a Cockroach Cloud SRE use case: needing to
disable certain categories of features in case of cluster failure.
See #51643

Release note (enterprise change): Adds cluster settings to enable/
disable the IMPORT, EXPORT, and changefeed  commands. If a user
attempts to use these features while they are disabled, an error
indicating that the database administrator has disabled the
feature is surfaced.

Example usage for the database administrator:
SET CLUSTER SETTING feature.import.enabled = FALSE;
SET CLUSTER SETTING feature.import.enabled = TRUE;
SET CLUSTER SETTING feature.export.enabled = FALSE;
SET CLUSTER SETTING feature.export.enabled = TRUE;
SET CLUSTER SETTING feature.changefeed.enabled = FALSE;
SET CLUSTER SETTING feature.changefeed.enabled = TRUE;

Co-authored-by: angelapwen <[email protected]>

docs/generated/settings/settings.html

@@ -14,6 +14,9 @@
 <tr><td><code>external.graphite.endpoint</code></td><td>string</td><td><code></code></td><td>if nonempty, push server metrics to the Graphite or Carbon server at the specified host:port</td></tr>
 <tr><td><code>external.graphite.interval</code></td><td>duration</td><td><code>10s</code></td><td>the interval at which metrics are pushed to Graphite (if enabled)</td></tr>
 <tr><td><code>feature.backup.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to true to enable backups, false to disable; default is true</td></tr>
+<tr><td><code>feature.changefeed.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to true to enable changefeeds, false to disable; default is true</td></tr>
+<tr><td><code>feature.export.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to true to enable exports, false to disable; default is true</td></tr>
+<tr><td><code>feature.import.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to true to enable imports, false to disable; default is true</td></tr>
 <tr><td><code>feature.restore.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to true to enable restore, false to disable; default is true</td></tr>
 <tr><td><code>jobs.retention_time</code></td><td>duration</td><td><code>336h0m0s</code></td><td>the amount of time to retain records for completed jobs before</td></tr>
 <tr><td><code>kv.allocator.load_based_lease_rebalancing.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to enable rebalancing of range leases based on load and latency</td></tr>

pkg/ccl/backupccl/backup_planning.go

@@ -509,10 +509,11 @@ func backupPlanHook(
 		return nil, nil, nil, false, nil
 	}
 
-	// Check whether feature backup is enabled or not.
-	if !featureBackupEnabled.Get(&p.ExecCfg().Settings.SV) {
-		return nil, nil, nil, false,
-			pgerror.Newf(pgcode.OperatorIntervention, "BACKUP feature was disabled by the database administrator")
+	if err := featureflag.CheckEnabled(featureBackupEnabled,
+		&p.ExecCfg().Settings.SV,
+		"BACKUP",
+	); err != nil {
+		return nil, nil, nil, false, err
 	}
 
 	var err error
@@ -604,7 +605,7 @@ func backupPlanHook(
 			return err
 		}
 		if len(to) > 1 {
-			if err := requireEnterprise("partitoned destinations"); err != nil {
+			if err := requireEnterprise("partitioned destinations"); err != nil {
 				return err
 			}
 		}

pkg/ccl/backupccl/restore_planning.go

@@ -1184,10 +1184,11 @@ func restorePlanHook(
 		return nil, nil, nil, false, nil
 	}
 
-	// Check whether feature restore is enabled or not.
-	if !featureRestoreEnabled.Get(&p.ExecCfg().Settings.SV) {
-		return nil, nil, nil, false,
-			pgerror.Newf(pgcode.OperatorIntervention, "RESTORE feature was disabled by the database administrator")
+	if err := featureflag.CheckEnabled(featureRestoreEnabled,
+		&p.ExecCfg().Settings.SV,
+		"RESTORE",
+	); err != nil {
+		return nil, nil, nil, false, err
 	}
 
 	fromFns := make([]func() ([]string, error), len(restoreStmt.From))

pkg/ccl/changefeedccl/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//pkg/ccl/changefeedccl/kvfeed",
         "//pkg/ccl/utilccl",
         "//pkg/docs",
+        "//pkg/featureflag",
         "//pkg/geo",
         "//pkg/geo/geopb",
         "//pkg/jobs",
@@ -39,6 +40,7 @@ go_library(
         "//pkg/roachpb",
         "//pkg/security",
         "//pkg/server/telemetry",
+        "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog",

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedbase"
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
 	"github.com/cockroachdb/cockroach/pkg/docs"
+	"github.com/cockroachdb/cockroach/pkg/featureflag"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprotectedts"
@@ -30,6 +31,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
@@ -51,6 +53,12 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
+// featureChangefeedEnabled is used to enable and disable the CHANGEFEED feature.
+var featureChangefeedEnabled = settings.RegisterPublicBoolSetting(
+	"feature.changefeed.enabled",
+	"set to true to enable changefeeds, false to disable; default is true",
+	featureflag.FeatureFlagEnabledDefault)
+
 func init() {
 	sql.AddPlanHook(changefeedPlanHook)
 	jobs.RegisterConstructor(
@@ -70,6 +78,13 @@ func changefeedPlanHook(
 		return nil, nil, nil, false, nil
 	}
 
+	if err := featureflag.CheckEnabled(featureChangefeedEnabled,
+		&p.ExecCfg().Settings.SV,
+		"CHANGEFEED",
+	); err != nil {
+		return nil, nil, nil, false, err
+	}
+
 	var sinkURIFn func() (string, error)
 	var header colinfo.ResultColumns
 	unspecifiedSink := changefeedStmt.SinkURI == nil

pkg/ccl/changefeedccl/changefeed_test.go

@@ -1881,6 +1881,16 @@ func TestChangefeedErrors(t *testing.T) {
 	)
 	sqlDB.Exec(t, `SET CLUSTER SETTING kv.rangefeed.enabled = true`)
 
+	// Feature flag for changefeeds is off — test that CREATE CHANGEFEED and
+	// EXPERIMENTAL CHANGEFEED FOR surface error.
+	sqlDB.Exec(t, `SET CLUSTER SETTING feature.changefeed.enabled = false`)
+	sqlDB.ExpectErr(t, `CHANGEFEED feature was disabled by the database administrator`,
+		`CREATE CHANGEFEED FOR foo`)
+	sqlDB.ExpectErr(t, `CHANGEFEED feature was disabled by the database administrator`,
+		`EXPERIMENTAL CHANGEFEED FOR foo`)
+
+	sqlDB.Exec(t, `SET CLUSTER SETTING feature.changefeed.enabled = true`)
+
 	sqlDB.ExpectErr(
 		t, `unknown format: nope`,
 		`EXPERIMENTAL CHANGEFEED FOR foo WITH format=nope`,

pkg/ccl/importccl/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "//pkg/ccl/utilccl",
         "//pkg/clusterversion",
         "//pkg/col/coldata",
+        "//pkg/featureflag",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobsprotectedts",
@@ -34,6 +35,7 @@ go_library(
         "//pkg/roachpb",
         "//pkg/security",
         "//pkg/server/telemetry",
+        "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog",

pkg/ccl/importccl/exportcsv_test.go

@@ -409,3 +409,26 @@ func TestExportVectorized(t *testing.T) {
 	sqlDB.Exec(t, `SET vectorize_row_count_threshold=0`)
 	sqlDB.Exec(t, `EXPORT INTO CSV 'http://0.1:37957/exp_1' FROM TABLE t`)
 }
+
+// TestExportFeatureFlag tests the feature flag logic that allows the EXPORT
+// command to be toggled off via cluster settings.
+func TestExportFeatureFlag(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	dir, cleanupDir := testutils.TempDir(t)
+	defer cleanupDir()
+
+	srv, db, _ := serverutils.StartServer(t, base.TestServerArgs{ExternalIODir: dir})
+	defer srv.Stopper().Stop(context.Background())
+	sqlDB := sqlutils.MakeSQLRunner(db)
+
+	// Feature flag is off — test that EXPORT surfaces error.
+	sqlDB.Exec(t, `SET CLUSTER SETTING feature.export.enabled = FALSE`)
+	sqlDB.Exec(t, `CREATE TABLE feature_flag (a INT PRIMARY KEY)`)
+	sqlDB.ExpectErr(t, `EXPORT feature was disabled by the database administrator`,
+		`EXPORT INTO CSV 'nodelocal://0/%s/' FROM TABLE feature_flag`)
+
+	// Feature flag is on — test that EXPORT does not error.
+	sqlDB.Exec(t, `SET CLUSTER SETTING feature.export.enabled = TRUE`)
+	sqlDB.Exec(t, `EXPORT INTO CSV 'nodelocal://0/%s/' FROM TABLE feature_flag`)
+}

pkg/ccl/importccl/import_stmt.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl"
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/featureflag"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprotectedts"
@@ -30,6 +31,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
@@ -174,6 +176,12 @@ var allowedIntoFormats = map[string]struct{}{
 	"PGCOPY":    {},
 }
 
+// featureImportEnabled is used to enable and disable the IMPORT feature.
+var featureImportEnabled = settings.RegisterPublicBoolSetting(
+	"feature.import.enabled",
+	"set to true to enable imports, false to disable; default is true",
+	featureflag.FeatureFlagEnabledDefault)
+
 func validateFormatOptions(
 	format string, specified map[string]string, formatAllowed map[string]struct{},
 ) error {
@@ -260,6 +268,13 @@ func importPlanHook(
 
 	addToFileFormatTelemetry(importStmt.FileFormat, "attempted")
 
+	if err := featureflag.CheckEnabled(featureImportEnabled,
+		&p.ExecCfg().Settings.SV,
+		"IMPORT",
+	); err != nil {
+		return nil, nil, nil, false, err
+	}
+
 	filesFn, err := p.TypeAsStringArray(ctx, importStmt.Files, "IMPORT")
 	if err != nil {
 		return nil, nil, nil, false, err

pkg/ccl/importccl/import_stmt_test.go

@@ -2138,6 +2138,41 @@ b STRING) CSV DATA (%s)`, testFiles.files[0])); err != nil {
 	})
 }
 
+// TestImportFeatureFlag tests the feature flag logic that allows the IMPORT and
+// IMPORT INTO commands to be toggled off via cluster settings.
+func TestImportFeatureFlag(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	defer jobs.ResetConstructors()()
+
+	const nodes = 1
+	numFiles := nodes + 2
+	rowsPerFile := 1000
+	rowsPerRaceFile := 16
+
+	ctx := context.Background()
+	baseDir := filepath.Join("testdata", "csv")
+	tc := testcluster.StartTestCluster(t, nodes, base.TestClusterArgs{ServerArgs: base.TestServerArgs{ExternalIODir: baseDir}})
+	defer tc.Stopper().Stop(ctx)
+	sqlDB := sqlutils.MakeSQLRunner(tc.Conns[0])
+
+	testFiles := makeCSVData(t, numFiles, rowsPerFile, nodes, rowsPerRaceFile)
+
+	// Feature flag is off — test that IMPORT and IMPORT INTO surface error.
+	sqlDB.Exec(t, `SET CLUSTER SETTING feature.import.enabled = FALSE`)
+	sqlDB.ExpectErr(t, `IMPORT feature was disabled by the database administrator`,
+		fmt.Sprintf(`IMPORT TABLE t (a INT8 PRIMARY KEY, b STRING) CSV DATA (%s)`, testFiles.files[0]))
+	sqlDB.Exec(t, `CREATE TABLE feature_flag (a INT8 PRIMARY KEY, b STRING)`)
+	sqlDB.ExpectErr(t, `IMPORT feature was disabled by the database administrator`,
+		fmt.Sprintf(`IMPORT INTO feature_flag (a, b) CSV DATA (%s)`, testFiles.files[0]))
+
+	// Feature flag is on — test that IMPORT and IMPORT INTO do not error.
+	sqlDB.Exec(t, `SET CLUSTER SETTING feature.import.enabled = TRUE`)
+	sqlDB.Exec(t, fmt.Sprintf(`IMPORT TABLE t (a INT8 PRIMARY KEY, b STRING) CSV DATA (%s)`,
+		testFiles.files[0]))
+	sqlDB.Exec(t, fmt.Sprintf(`IMPORT INTO feature_flag (a, b) CSV DATA (%s)`, testFiles.files[0]))
+}
+
 func TestImportObjectLevelRBAC(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/featureflag/BUILD.bazel

@@ -5,4 +5,9 @@ go_library(
     srcs = ["feature_flags.go"],
     importpath = "github.com/cockroachdb/cockroach/pkg/featureflag",
     visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/settings",
+        "//pkg/sql/pgwire/pgcode",
+        "//pkg/sql/pgwire/pgerror",
+    ],
 )

pkg/featureflag/feature_flags.go

@@ -10,5 +10,23 @@
 
 package featureflag
 
-// FeatureFlagEnabledDefault is used for the default value of all feature flag cluster settings.
+import (
+	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
+)
+
+// FeatureFlagEnabledDefault is used for the default value of all feature flag
+// cluster settings.
 const FeatureFlagEnabledDefault = true
+
+// CheckEnabled checks whether a specific feature has been enabled or disabled
+// via its cluster settings, and returns an error if it has been disabled.
+func CheckEnabled(s *settings.BoolSetting, sv *settings.Values, featureName string) error {
+	if enabled := s.Get(sv); !enabled {
+		return pgerror.Newf(pgcode.OperatorIntervention,
+			"%s feature was disabled by the database administrator",
+			featureName)
+	}
+	return nil
+}

pkg/sql/BUILD.bazel

@@ -227,6 +227,7 @@ go_library(
         "//pkg/config",
         "//pkg/config/zonepb",
         "//pkg/docs",
+        "//pkg/featureflag",
         "//pkg/geo/geoindex",
         "//pkg/geo/geopb",
         "//pkg/geo/geoprojbase",

pkg/sql/export.go

@@ -16,7 +16,9 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/cockroachdb/cockroach/pkg/featureflag"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/opt/exec"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
@@ -79,10 +81,28 @@ const exportFilePatternPart = "%part%"
 const exportFilePatternDefault = exportFilePatternPart + ".csv"
 const exportCompressionCodec = "gzip"
 
+// featureExportEnabled is used to enable and disable the EXPORT feature.
+var featureExportEnabled = settings.RegisterPublicBoolSetting(
+	"feature.export.enabled",
+	"set to true to enable exports, false to disable; default is true",
+	featureflag.FeatureFlagEnabledDefault)
+
 // ConstructExport is part of the exec.Factory interface.
 func (ef *execFactory) ConstructExport(
 	input exec.Node, fileName tree.TypedExpr, fileFormat string, options []exec.KVOption,
 ) (exec.Node, error) {
+	if !featureExportEnabled.Get(&ef.planner.ExecCfg().Settings.SV) {
+		return nil, pgerror.Newf(pgcode.OperatorIntervention,
+			"EXPORT feature was disabled by the database administrator")
+	}
+
+	if err := featureflag.CheckEnabled(featureExportEnabled,
+		&ef.planner.ExecCfg().Settings.SV,
+		"EXPORT",
+	); err != nil {
+		return nil, err
+	}
+
 	if !ef.planner.ExtendedEvalContext().TxnImplicit {
 		return nil, errors.Errorf("EXPORT cannot be used inside a transaction")
 	}