sql: block DROP TENANT based on a session var

In clusters where we will promote tenant management operations, we
would like to ensure there is one extra step needed for administrators
to drop a tenant (and thus irremedially lose data). Given that
`sql_safe_updates` is not set automatically when users open their
SQL session using their own client, we need another mechanism.

This change introduces the new (hidden) session var,
`disable_drop_tenant`. When set, tenant deletion fails with the
following error message:

```
demo@127.0.0.1:26257/movr> drop tenant foo;
ERROR: rejected (via sql_safe_updates or disable_drop_tenant): DROP TENANT causes irreversible data loss
SQLSTATE: 01000
```

(The session var `sql_safe_updates` is _also_ included as a blocker in
the mechanism so that folk using `cockroach sql` get double
protection).

The default value of this session var is `false` in single-tenant
clusters (set via a cluster setting `sql.drop_tenant.enabled`), for
compatibility with CC Serverless. The default will be set to `true`
via a config profile when suitable.

Release note: None

pkg/sql/exec_util.go

@@ -747,6 +747,25 @@ var errTransactionInProgress = errors.New("there is already a transaction in pro
 const sqlTxnName string = "sql txn"
 const metricsSampleInterval = 10 * time.Second
 
+// enableDropTenant (or rather, its inverted boolean value) defines
+// the default value for the session var "disable_drop_tenant".
+//
+// Note:
+//   - We use a cluster setting here instead of a default role option
+//     because we need this to be settable also for the 'admin' role.
+//   - The cluster setting is named "enable" because boolean cluster
+//     settings are all ".enabled" -- we do not have ".disabled"
+//     settings anywhere.
+//   - The session var is named "disable_" because we want the Go
+//     default value (false) to mean that tenant deletion is enabled.
+//     This is needed for backward-compatibility with Cockroach Cloud.
+var enableDropTenant = settings.RegisterBoolSetting(
+	settings.SystemOnly,
+	"sql.drop_tenant.enabled",
+	"default value (inverted) for the disable_drop_tenant session setting",
+	true,
+)
+
 // Fully-qualified names for metrics.
 var (
 	MetaSQLExecLatency = metric.Metadata{
@@ -3146,6 +3165,10 @@ func (m *sessionDataMutator) SetSafeUpdates(val bool) {
 	m.data.SafeUpdates = val
 }
 
+func (m *sessionDataMutator) SetDisableDropTenant(val bool) {
+	m.data.DisableDropTenant = val
+}
+
 func (m *sessionDataMutator) SetCheckFunctionBodies(val bool) {
 	m.data.CheckFunctionBodies = val
 }

pkg/sql/logictest/testdata/logic_test/information_schema

@@ -5206,6 +5206,7 @@ default_transaction_read_only                         off
 default_transaction_use_follower_reads                off
 default_with_oids                                     off
 descriptor_validation                                 on
+disable_drop_tenant                                   off
 disable_hoist_projection_in_join_limitation           off
 disable_partially_distributed_plans                   off
 disable_plan_gists                                    off

pkg/sql/logictest/testdata/logic_test/tenant

@@ -474,3 +474,21 @@ DROP TENANT tmpl
 
 statement ok
 RESET CLUSTER SETTING sql.create_tenant.default_template
+
+subtest block_drop_tenant
+
+statement ok
+SET disable_drop_tenant = 'true'
+
+statement ok
+CREATE TENANT nodelete
+
+statement error rejected.*irreversible data loss
+DROP TENANT nodelete
+
+statement ok
+RESET disable_drop_tenant
+
+statement ok
+DROP TENANT nodelete
+

pkg/sql/sessiondatapb/local_only_session_data.proto

@@ -365,9 +365,14 @@ message LocalOnlySessionData {
   // Execution of these deallocated prepared statements will fail until they are
   // prepared again.
   int64 prepared_statements_cache_size = 97;
+
   // StreamerEnabled controls whether the Streamer API can be used.
   bool streamer_enabled = 98;
 
+  // DisableDropTenant causes errors when the client
+  // attempts to drop tenants or tenant records.
+  bool disable_drop_tenant = 99;
+
   ///////////////////////////////////////////////////////////////////////////
   // WARNING: consider whether a session parameter you're adding needs to  //
   // be propagated to the remote nodes. If so, that parameter should live  //

pkg/sql/tenant_deletion.go

@@ -32,6 +32,13 @@ import (
 func (p *planner) DropTenantByID(
 	ctx context.Context, tenID uint64, synchronousImmediateDrop, ignoreServiceMode bool,
 ) error {
+	if p.SessionData().DisableDropTenant || p.SessionData().SafeUpdates {
+		err := errors.Newf("DROP TENANT causes irreversible data loss")
+		err = errors.WithMessage(err, "rejected (via sql_safe_updates or disable_drop_tenant)")
+		err = pgerror.WithCandidateCode(err, pgcode.Warning)
+		return err
+	}
+
 	if p.EvalContext().TxnReadOnly {
 		return readOnlyError("DROP TENANT")
 	}

pkg/sql/vars.go

@@ -1520,6 +1520,34 @@ var varGen = map[string]sessionVar{
 		// Setting is done by the SetTracing statement.
 	},
 
+	// CockroachDB extension.
+	`disable_drop_tenant`: {
+		Hidden: true,
+		Get: func(evalCtx *extendedEvalContext, _ *kv.Txn) (string, error) {
+			return formatBoolAsPostgresSetting(evalCtx.SessionData().DisableDropTenant), nil
+		},
+		Set: func(_ context.Context, m sessionDataMutator, s string) error {
+			b, err := paramparse.ParseBoolVar("disable_drop_tenant", s)
+			if err != nil {
+				return err
+			}
+			m.SetDisableDropTenant(b)
+			return nil
+		},
+		GlobalDefault: func(sv *settings.Values) string {
+			// Note:
+			// - We use a cluster setting here instead of a default role option
+			//   because we need this to be settable also for the 'admin' role.
+			// - The cluster setting is named "enable" because boolean cluster
+			//   settings are all ".enabled" -- we do not have ".disabled"
+			//   settings anywhere.
+			// - The session var is named "disable_" because we want the Go
+			//   default value (false) to mean that tenant deletion is enabled.
+			//   This is needed for backward-compatibility with Cockroach Cloud.
+			return fmt.Sprint(!enableDropTenant.Get(sv))
+		},
+	},
+
 	// CockroachDB extension.
 	`allow_prepare_as_opt_plan`: {
 		Hidden: true,