sql, featureflag: add schema change feature flag

This change adds a feature flag via cluster setting for all
designated features that perform schema changes or DDLs. The
feature is being introduced to address a Cockroach Cloud SRE use
case: needing to disable certain categories of features, such as
schema changes, in case of cluster failure.

Release note (sql change): Adds a feature flag via cluster
setting for all schema change-related features. 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.schemachange.enabled = FALSE;
SET CLUSTER SETTING feature.schemachange.enabled = TRUE;

docs/generated/settings/settings.html

@@ -18,6 +18,7 @@
 <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>feature.schema_change.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to true to enable schema changes, false to disable; default is true</td></tr>
 <tr><td><code>feature.stats.enabled</code></td><td>boolean</td><td><code>true</code></td><td>set to true to enable CREATE STATISTICS/ANALYZE, 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/testdata/backup-restore/feature-flags

@@ -12,7 +12,7 @@ exec-sql
 SET CLUSTER SETTING feature.backup.enabled = FALSE;
 BACKUP TO 'nodelocal://0/test-root/';
 ----
-pq: BACKUP feature was disabled by the database administrator
+pq: feature BACKUP was disabled by the database administrator
 
 # Test running backup when feature flag is enabled.
 exec-sql
@@ -29,7 +29,7 @@ exec-sql
 SET CLUSTER SETTING feature.restore.enabled = FALSE;
 RESTORE TABLE d.t FROM 'nodelocal://0/test-root/';
 ----
-pq: RESTORE feature was disabled by the database administrator
+pq: feature RESTORE was disabled by the database administrator
 
 # Test running restore when feature flag is enabled.
 exec-sql

pkg/ccl/changefeedccl/changefeed_test.go

@@ -1885,9 +1885,9 @@ func TestChangefeedErrors(t *testing.T) {
 	// 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`,
+	sqlDB.ExpectErr(t, `feature CHANGEFEED was disabled by the database administrator`,
 		`CREATE CHANGEFEED FOR foo`)
-	sqlDB.ExpectErr(t, `CHANGEFEED feature was disabled by the database administrator`,
+	sqlDB.ExpectErr(t, `feature CHANGEFEED was disabled by the database administrator`,
 		`EXPERIMENTAL CHANGEFEED FOR foo`)
 
 	sqlDB.Exec(t, `SET CLUSTER SETTING feature.changefeed.enabled = true`)

pkg/ccl/importccl/exportcsv_test.go

@@ -424,11 +424,11 @@ func TestExportFeatureFlag(t *testing.T) {
 
 	// 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`)
+	sqlDB.Exec(t, `CREATE TABLE feature_flags (a INT PRIMARY KEY)`)
+	sqlDB.ExpectErr(t, `feature EXPORT was disabled by the database administrator`,
+		`EXPORT INTO CSV 'nodelocal://0/%s/' FROM TABLE feature_flags`)
 
 	// 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`)
+	sqlDB.Exec(t, `EXPORT INTO CSV 'nodelocal://0/%s/' FROM TABLE feature_flags`)
 }

pkg/ccl/importccl/import_stmt_test.go

@@ -2162,17 +2162,17 @@ func TestImportFeatureFlag(t *testing.T) {
 
 	// 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`,
+	sqlDB.ExpectErr(t, `feature IMPORT 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]))
+	sqlDB.Exec(t, `CREATE TABLE feature_flags (a INT8 PRIMARY KEY, b STRING)`)
+	sqlDB.ExpectErr(t, `feature IMPORT was disabled by the database administrator`,
+		fmt.Sprintf(`IMPORT INTO feature_flags (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]))
+	sqlDB.Exec(t, fmt.Sprintf(`IMPORT INTO feature_flags (a, b) CSV DATA (%s)`, testFiles.files[0]))
 }
 
 func TestImportObjectLevelRBAC(t *testing.T) {

pkg/featureflag/feature_flags.go

@@ -25,7 +25,7 @@ const FeatureFlagEnabledDefault = true
 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",
+			"feature %s was disabled by the database administrator",
 			featureName)
 	}
 	return nil

pkg/sql/BUILD.bazel

@@ -153,6 +153,7 @@ go_library(
         "scan.go",
         "scatter.go",
         "schema.go",
+        "schema_change_cluster_setting.go",
         "schema_changer.go",
         "schema_changer_metrics.go",
         "scrub.go",

pkg/sql/alter_database.go

@@ -30,6 +30,13 @@ type alterDatabaseOwnerNode struct {
 func (p *planner) AlterDatabaseOwner(
 	ctx context.Context, n *tree.AlterDatabaseOwner,
 ) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER DATABASE",
+	); err != nil {
+		return nil, err
+	}
+
 	dbDesc, err := p.ResolveMutableDatabaseDescriptor(ctx, n.Name.String(), true /* required */)
 	if err != nil {
 		return nil, err
@@ -102,13 +109,25 @@ func (n *alterDatabaseOwnerNode) Close(context.Context)        {}
 func (p *planner) AlterDatabaseAddRegion(
 	ctx context.Context, n *tree.AlterDatabaseAddRegion,
 ) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER DATABASE",
+	); err != nil {
+		return nil, err
+	}
 	return nil, unimplemented.New("alter database add region", "implementation pending")
 }
 
 // AlterDatabaseDropRegion transforms a tree.AlterDatabaseDropRegion into a plan node.
 func (p *planner) AlterDatabaseDropRegion(
 	ctx context.Context, n *tree.AlterDatabaseDropRegion,
 ) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER DATABASE",
+	); err != nil {
+		return nil, err
+	}
 	return nil, unimplemented.New("alter database drop region", "implementation pending")
 }
 
@@ -123,5 +142,11 @@ func (p *planner) AlterDatabasePrimaryRegion(
 func (p *planner) AlterDatabaseSurvivalGoal(
 	ctx context.Context, n *tree.AlterDatabaseSurvivalGoal,
 ) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER DATABASE",
+	); err != nil {
+		return nil, err
+	}
 	return nil, unimplemented.New("alter database survive", "implementation pending")
 }

pkg/sql/alter_index.go

@@ -31,6 +31,13 @@ type alterIndexNode struct {
 // AlterIndex applies a schema change on an index.
 // Privileges: CREATE on table.
 func (p *planner) AlterIndex(ctx context.Context, n *tree.AlterIndex) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER INDEX",
+	); err != nil {
+		return nil, err
+	}
+
 	tableDesc, indexDesc, err := p.getTableAndIndex(ctx, &n.Index, privilege.CREATE)
 	if err != nil {
 		return nil, err

pkg/sql/alter_schema.go

@@ -39,6 +39,13 @@ type alterSchemaNode struct {
 var _ planNode = &alterSchemaNode{n: nil}
 
 func (p *planner) AlterSchema(ctx context.Context, n *tree.AlterSchema) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER SCHEMA",
+	); err != nil {
+		return nil, err
+	}
+
 	dbName := p.CurrentDatabase()
 	if n.Schema.ExplicitCatalog {
 		dbName = n.Schema.Catalog()

pkg/sql/alter_sequence.go

@@ -28,6 +28,13 @@ type alterSequenceNode struct {
 
 // AlterSequence transforms a tree.AlterSequence into a plan node.
 func (p *planner) AlterSequence(ctx context.Context, n *tree.AlterSequence) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER SEQUENCE",
+	); err != nil {
+		return nil, err
+	}
+
 	seqDesc, err := p.ResolveMutableTableDescriptorEx(
 		ctx, n.Name, !n.IfExists, tree.ResolveRequireSequenceDesc,
 	)

pkg/sql/alter_table.go

@@ -53,6 +53,13 @@ type alterTableNode struct {
 //   notes: postgres requires CREATE on the table.
 //          mysql requires ALTER, CREATE, INSERT on the table.
 func (p *planner) AlterTable(ctx context.Context, n *tree.AlterTable) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER TABLE",
+	); err != nil {
+		return nil, err
+	}
+
 	tableDesc, err := p.ResolveMutableTableDescriptorEx(
 		ctx, n.Table, !n.IfExists, tree.ResolveRequireTableDesc,
 	)

pkg/sql/alter_table_regional_affinity.go

@@ -21,5 +21,11 @@ import (
 func (p *planner) AlterTableRegionalAffinity(
 	ctx context.Context, n *tree.AlterTableRegionalAffinity,
 ) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER TABLE",
+	); err != nil {
+		return nil, err
+	}
 	return nil, unimplemented.New("alter table locality", "implementation pending")
 }

pkg/sql/alter_table_set_schema.go

@@ -33,6 +33,13 @@ type alterTableSetSchemaNode struct {
 func (p *planner) AlterTableSetSchema(
 	ctx context.Context, n *tree.AlterTableSetSchema,
 ) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER TABLE/VIEW/SEQUENCE SET SCHEMA",
+	); err != nil {
+		return nil, err
+	}
+
 	tn := n.Name.ToTableName()
 	requiredTableKind := tree.ResolveAnyTableKind
 	if n.IsView {

pkg/sql/alter_type.go

@@ -36,6 +36,13 @@ type alterTypeNode struct {
 var _ planNode = &alterTypeNode{n: nil}
 
 func (p *planner) AlterType(ctx context.Context, n *tree.AlterType) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"ALTER TYPE",
+	); err != nil {
+		return nil, err
+	}
+
 	// Resolve the type.
 	desc, err := p.ResolveMutableTypeDescriptor(ctx, n.Type, true /* required */)
 	if err != nil {

pkg/sql/comment_on_column.go

@@ -29,6 +29,13 @@ type commentOnColumnNode struct {
 // CommentOnColumn add comment on a column.
 // Privileges: CREATE on table.
 func (p *planner) CommentOnColumn(ctx context.Context, n *tree.CommentOnColumn) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"COMMENT ON COLUMN",
+	); err != nil {
+		return nil, err
+	}
+
 	var tableName tree.TableName
 	if n.ColumnItem.TableName != nil {
 		tableName = n.ColumnItem.TableName.ToTableName()

pkg/sql/comment_on_database.go

@@ -32,6 +32,13 @@ type commentOnDatabaseNode struct {
 func (p *planner) CommentOnDatabase(
 	ctx context.Context, n *tree.CommentOnDatabase,
 ) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"COMMENT ON DATABASE",
+	); err != nil {
+		return nil, err
+	}
+
 	dbDesc, err := p.ResolveUncachedDatabaseByName(ctx, string(n.Name), true)
 	if err != nil {
 		return nil, err

pkg/sql/comment_on_index.go

@@ -31,6 +31,13 @@ type commentOnIndexNode struct {
 // CommentOnIndex adds a comment on an index.
 // Privileges: CREATE on table.
 func (p *planner) CommentOnIndex(ctx context.Context, n *tree.CommentOnIndex) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"COMMENT ON INDEX",
+	); err != nil {
+		return nil, err
+	}
+
 	tableDesc, indexDesc, err := p.getTableAndIndex(ctx, &n.Index, privilege.CREATE)
 	if err != nil {
 		return nil, err

pkg/sql/comment_on_table.go

@@ -31,6 +31,13 @@ type commentOnTableNode struct {
 //   notes: postgres requires CREATE on the table.
 //          mysql requires ALTER, CREATE, INSERT on the table.
 func (p *planner) CommentOnTable(ctx context.Context, n *tree.CommentOnTable) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"COMMENT ON TABLE",
+	); err != nil {
+		return nil, err
+	}
+
 	tableDesc, err := p.ResolveUncachedTableDescriptorEx(ctx, n.Table, true, tree.ResolveRequireTableDesc)
 	if err != nil {
 		return nil, err

pkg/sql/create_database.go

@@ -30,6 +30,13 @@ type createDatabaseNode struct {
 // CreateDatabase creates a database.
 // Privileges: superuser or CREATEDB
 func (p *planner) CreateDatabase(ctx context.Context, n *tree.CreateDatabase) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"CREATE DATABASE",
+	); err != nil {
+		return nil, err
+	}
+
 	if n.Name == "" {
 		return nil, errEmptyDatabaseName
 	}

pkg/sql/create_index.go

@@ -44,6 +44,12 @@ type createIndexNode struct {
 //   notes: postgres requires CREATE on the table.
 //          mysql requires INDEX on the table.
 func (p *planner) CreateIndex(ctx context.Context, n *tree.CreateIndex) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"CREATE INDEX",
+	); err != nil {
+		return nil, err
+	}
 	tableDesc, err := p.ResolveMutableTableDescriptor(
 		ctx, &n.Table, true /*required*/, tree.ResolveRequireTableOrViewDesc,
 	)

pkg/sql/create_schema.go

@@ -37,6 +37,13 @@ func (n *createSchemaNode) startExec(params runParams) error {
 }
 
 func (p *planner) createUserDefinedSchema(params runParams, n *tree.CreateSchema) error {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"CREATE SCHEMA",
+	); err != nil {
+		return err
+	}
+
 	// Users can't create a schema without being connected to a DB.
 	if p.CurrentDatabase() == "" {
 		return pgerror.New(pgcode.UndefinedDatabase,
@@ -177,6 +184,13 @@ func (n *createSchemaNode) Close(ctx context.Context)  {}
 
 // CreateSchema creates a schema.
 func (p *planner) CreateSchema(ctx context.Context, n *tree.CreateSchema) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"CREATE SCHEMA",
+	); err != nil {
+		return nil, err
+	}
+
 	return &createSchemaNode{
 		n: n,
 	}, nil

pkg/sql/create_sequence.go

@@ -34,6 +34,13 @@ type createSequenceNode struct {
 }
 
 func (p *planner) CreateSequence(ctx context.Context, n *tree.CreateSequence) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"CREATE SEQUENCE",
+	); err != nil {
+		return nil, err
+	}
+
 	un := n.Name.ToUnresolvedObjectName()
 	dbDesc, _, prefix, err := p.ResolveTargetObject(ctx, un)
 	if err != nil {

pkg/sql/create_type.go

@@ -43,6 +43,13 @@ type createTypeNode struct {
 var _ planNode = &createTypeNode{n: nil}
 
 func (p *planner) CreateType(ctx context.Context, n *tree.CreateType) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"CREATE TYPE",
+	); err != nil {
+		return nil, err
+	}
+
 	return &createTypeNode{n: n}, nil
 }
 

pkg/sql/drop_database.go

@@ -44,6 +44,13 @@ type dropDatabaseNode struct {
 //   Notes: postgres allows only the database owner to DROP a database.
 //          mysql requires the DROP privileges on the database.
 func (p *planner) DropDatabase(ctx context.Context, n *tree.DropDatabase) (planNode, error) {
+	if err := checkSchemaChangeEnabled(
+		&p.ExecCfg().Settings.SV,
+		"DROP DATABASE",
+	); err != nil {
+		return nil, err
+	}
+
 	if n.Name == "" {
 		return nil, errEmptyDatabaseName
 	}