sql: block DROP TENANT based on a session var

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

@@ -318,7 +318,6 @@ message LocalOnlySessionData {
   // optimizer should use improved statistics calculations for disjunctive
   // filters.
   bool optimizer_use_improved_disjunction_stats = 86;
-
   // OptimizerUseLimitOrderingForStreamingGroupBy enables the exploration rule
   // which optimizes 'SELECT ... GROUP BY ... ORDER BY ... LIMIT n' queries.
   // The rule uses the required ordering in the limit expression to inform an
@@ -367,6 +366,9 @@ message LocalOnlySessionData {
   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  //

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 formatBoolAsPostgresSetting(!enableDropTenant.Get(sv))
+		},
+	},
+
 	// CockroachDB extension.
 	`allow_prepare_as_opt_plan`: {
 		Hidden: true,