release-22.2: *: add restriction to running DDL with internal executors

pkg/ccl/backupccl/backup_metadata_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -86,7 +87,11 @@ func checkMetadata(
 		tc.Servers[0].ClusterSettings(),
 		blobs.TestEmptyBlobClientFactory,
 		username.RootUserName(),
-		tc.Servers[0].InternalExecutor().(*sql.InternalExecutor), tc.Servers[0].DB(), nil)
+		tc.Servers[0].InternalExecutor().(*sql.InternalExecutor),
+		tc.Servers[0].CollectionFactory().(*descs.CollectionFactory),
+		tc.Servers[0].DB(),
+		nil, /* limiters */
+	)
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/ccl/backupccl/backup_test.go

@@ -572,7 +572,11 @@ func TestBackupRestoreAppend(t *testing.T) {
 				tc.Servers[0].ClusterSettings(),
 				blobs.TestEmptyBlobClientFactory,
 				username.RootUserName(),
-				tc.Servers[0].InternalExecutor().(*sql.InternalExecutor), tc.Servers[0].DB(), nil)
+				tc.Servers[0].InternalExecutor().(*sql.InternalExecutor),
+				tc.Servers[0].CollectionFactory().(*descs.CollectionFactory),
+				tc.Servers[0].DB(),
+				nil, /* limiters */
+			)
 			require.NoError(t, err)
 			defer store.Close()
 			var files []string
@@ -8021,7 +8025,12 @@ func TestReadBackupManifestMemoryMonitoring(t *testing.T) {
 		base.ExternalIODirConfig{},
 		st,
 		blobs.TestBlobServiceClient(dir),
-		username.RootUserName(), nil, nil, nil)
+		username.RootUserName(),
+		nil, /* ie */
+		nil, /* cf */
+		nil, /* kvDB */
+		nil, /* limiters */
+	)
 	require.NoError(t, err)
 
 	m := mon.NewMonitor("test-monitor", mon.MemoryResource, nil, nil, 0, 0, st)

pkg/ccl/backupccl/restore_data_processor_test.go

@@ -253,7 +253,12 @@ func runTestIngest(t *testing.T, init func(*cluster.Settings)) {
 			DB: kvDB,
 			ExternalStorage: func(ctx context.Context, dest cloudpb.ExternalStorage, opts ...cloud.ExternalStorageOption) (cloud.ExternalStorage, error) {
 				return cloud.MakeExternalStorage(ctx, dest, base.ExternalIODirConfig{},
-					s.ClusterSettings(), blobs.TestBlobServiceClient(s.ClusterSettings().ExternalIODir), nil, nil, nil, opts...)
+					s.ClusterSettings(), blobs.TestBlobServiceClient(s.ClusterSettings().ExternalIODir),
+					nil, /* ie */
+					nil, /* cf */
+					nil, /* kvDB */
+					nil, /* limiters */
+					opts...)
 			},
 			Settings:          s.ClusterSettings(),
 			Codec:             keys.SystemSQLCodec,

pkg/ccl/backupccl/restore_job.go

@@ -1669,7 +1669,7 @@ func (r *restoreResumer) doResume(ctx context.Context, execCtx interface{}) erro
 		// Reload the details as we may have updated the job.
 		details = r.job.Details().(jobspb.RestoreDetails)
 
-		if err := r.cleanupTempSystemTables(ctx, nil /* txn */); err != nil {
+		if err := r.cleanupTempSystemTables(ctx); err != nil {
 			return err
 		}
 	} else if isSystemUserRestore(details) {
@@ -1678,7 +1678,7 @@ func (r *restoreResumer) doResume(ctx context.Context, execCtx interface{}) erro
 		}
 		details = r.job.Details().(jobspb.RestoreDetails)
 
-		if err := r.cleanupTempSystemTables(ctx, nil /* txn */); err != nil {
+		if err := r.cleanupTempSystemTables(ctx); err != nil {
 			return err
 		}
 	}
@@ -2214,8 +2214,8 @@ func (r *restoreResumer) OnFailOrCancel(
 	logJobCompletion(ctx, restoreJobEventType, r.job.ID(), false, jobErr)
 
 	execCfg := execCtx.(sql.JobExecContext).ExecCfg()
-	if err := sql.DescsTxn(ctx, execCfg, func(
-		ctx context.Context, txn *kv.Txn, descsCol *descs.Collection,
+	if err := execCfg.CollectionFactory.TxnWithExecutor(ctx, execCfg.DB, p.SessionData(), func(
+		ctx context.Context, txn *kv.Txn, descsCol *descs.Collection, ie sqlutil.InternalExecutor,
 	) error {
 		for _, tenant := range details.Tenants {
 			tenant.State = descpb.TenantInfo_DROP
@@ -2233,7 +2233,6 @@ func (r *restoreResumer) OnFailOrCancel(
 		if details.DescriptorCoverage == tree.AllDescriptors {
 			// We've dropped defaultdb and postgres in the planning phase, we must
 			// recreate them now if the full cluster restore failed.
-			ie := p.ExecCfg().InternalExecutor
 			_, err := ie.Exec(ctx, "recreate-defaultdb", txn, "CREATE DATABASE IF NOT EXISTS defaultdb")
 			if err != nil {
 				return err
@@ -2252,7 +2251,7 @@ func (r *restoreResumer) OnFailOrCancel(
 	if details.DescriptorCoverage == tree.AllDescriptors {
 		// The temporary system table descriptors should already have been dropped
 		// in `dropDescriptors` but we still need to drop the temporary system db.
-		if err := execCfg.DB.Txn(ctx, r.cleanupTempSystemTables); err != nil {
+		if err := r.cleanupTempSystemTables(ctx); err != nil {
 			return err
 		}
 	}
@@ -2801,12 +2800,12 @@ func (r *restoreResumer) restoreSystemTables(
 	return nil
 }
 
-func (r *restoreResumer) cleanupTempSystemTables(ctx context.Context, txn *kv.Txn) error {
+func (r *restoreResumer) cleanupTempSystemTables(ctx context.Context) error {
 	executor := r.execCfg.InternalExecutor
 	// Check if the temp system database has already been dropped. This can happen
 	// if the restore job fails after the system database has cleaned up.
 	checkIfDatabaseExists := "SELECT database_name FROM [SHOW DATABASES] WHERE database_name=$1"
-	if row, err := executor.QueryRow(ctx, "checking-for-temp-system-db" /* opName */, txn, checkIfDatabaseExists, restoreTempSystemDB); err != nil {
+	if row, err := executor.QueryRow(ctx, "checking-for-temp-system-db" /* opName */, nil /* txn */, checkIfDatabaseExists, restoreTempSystemDB); err != nil {
 		return errors.Wrap(err, "checking for temporary system db")
 	} else if row == nil {
 		// Temporary system DB might already have been dropped by the restore job.
@@ -2816,11 +2815,11 @@ func (r *restoreResumer) cleanupTempSystemTables(ctx context.Context, txn *kv.Tx
 	// After restoring the system tables, drop the temporary database holding the
 	// system tables.
 	gcTTLQuery := fmt.Sprintf("ALTER DATABASE %s CONFIGURE ZONE USING gc.ttlseconds=1", restoreTempSystemDB)
-	if _, err := executor.Exec(ctx, "altering-gc-ttl-temp-system" /* opName */, txn, gcTTLQuery); err != nil {
+	if _, err := executor.Exec(ctx, "altering-gc-ttl-temp-system" /* opName */, nil /* txn */, gcTTLQuery); err != nil {
 		log.Errorf(ctx, "failed to update the GC TTL of %q: %+v", restoreTempSystemDB, err)
 	}
 	dropTableQuery := fmt.Sprintf("DROP DATABASE %s CASCADE", restoreTempSystemDB)
-	if _, err := executor.Exec(ctx, "drop-temp-system-db" /* opName */, txn, dropTableQuery); err != nil {
+	if _, err := executor.Exec(ctx, "drop-temp-system-db" /* opName */, nil /* txn */, dropTableQuery); err != nil {
 		return errors.Wrap(err, "dropping temporary system db")
 	}
 	return nil

pkg/ccl/backupccl/restore_planning.go

@@ -58,6 +58,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catconstants"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/sql/syntheticprivilege"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -705,8 +706,9 @@ func getDatabaseIDAndDesc(
 // as regular databases, we drop them before restoring them again in the
 // restore.
 func dropDefaultUserDBs(ctx context.Context, execCfg *sql.ExecutorConfig) error {
-	return sql.DescsTxn(ctx, execCfg, func(ctx context.Context, txn *kv.Txn, col *descs.Collection) error {
-		ie := execCfg.InternalExecutor
+	return execCfg.CollectionFactory.TxnWithExecutor(ctx, execCfg.DB, nil /* session data */, func(
+		ctx context.Context, txn *kv.Txn, _ *descs.Collection, ie sqlutil.InternalExecutor,
+	) error {
 		_, err := ie.Exec(ctx, "drop-defaultdb", txn, "DROP DATABASE IF EXISTS defaultdb")
 		if err != nil {
 			return err

pkg/ccl/backupccl/testdata/backup-restore/file_table_read_write

@@ -0,0 +1,43 @@
+subtest backup_file_table
+
+new-server name=s1
+----
+
+exec-sql
+CREATE DATABASE to_backup;
+----
+
+exec-sql
+CREATE DATABASE backups;
+----
+
+exec-sql
+BACKUP DATABASE to_backup INTO 'userfile://backups.public.userfiles_$user/data';
+----
+
+query-sql
+SELECT * FROM backups.crdb_internal.invalid_objects;
+----
+
+exec-sql
+USE backups;
+----
+
+query-sql
+SELECT * FROM pg_catalog.pg_tables where schemaname='public';
+----
+public userfiles_$user_upload_files root <nil> true false false false
+public userfiles_$user_upload_payload root <nil> true false false false
+
+query-sql
+SELECT conname FROM pg_catalog.pg_constraint con
+INNER JOIN pg_catalog.pg_class rel   ON rel.oid = con.conrelid
+INNER JOIN pg_catalog.pg_namespace nsp
+ON nsp.oid = connamespace
+WHERE rel.relname='userfiles_$user_upload_payload'
+ORDER BY conname;
+----
+file_id_fk
+userfiles_$user_upload_payload_pkey
+
+subtest end

pkg/ccl/changefeedccl/sink_cloudstorage_test.go

@@ -169,7 +169,13 @@ func TestCloudStorageSink(t *testing.T) {
 	externalStorageFromURI := func(ctx context.Context, uri string, user username.SQLUsername, opts ...cloud.ExternalStorageOption) (cloud.ExternalStorage,
 		error) {
 		return cloud.ExternalStorageFromURI(ctx, uri, base.ExternalIODirConfig{}, settings,
-			clientFactory, user, nil, nil, nil, opts...)
+			clientFactory,
+			user,
+			nil, /* ie */
+			nil, /* cf */
+			nil, /* kvDB */
+			nil, /* limiters */
+			opts...)
 	}
 
 	user := username.RootUserName()

pkg/ccl/cliccl/debug_backup.go

@@ -291,7 +291,12 @@ func externalStorageFromURIFactory(
 	defaultSettings := &cluster.Settings{}
 	defaultSettings.SV.Init(ctx, nil /* opaque */)
 	return cloud.ExternalStorageFromURI(ctx, uri, base.ExternalIODirConfig{},
-		defaultSettings, newBlobFactory, user, nil /*Internal Executor*/, nil /*kvDB*/, nil, opts...)
+		defaultSettings, newBlobFactory, user,
+		nil, /* ie */
+		nil, /* cf */
+		nil, /* kvDB */
+		nil, /* limiters */
+		opts...)
 }
 
 func getManifestFromURI(ctx context.Context, path string) (backuppb.BackupManifest, error) {
@@ -588,7 +593,13 @@ func makeIters(
 		var err error
 		clusterSettings := cluster.MakeClusterSettings()
 		dirStorage[i], err = cloud.MakeExternalStorage(ctx, file.Dir, base.ExternalIODirConfig{},
-			clusterSettings, newBlobFactory, nil /*internal executor*/, nil /*kvDB*/, nil)
+			clusterSettings,
+			newBlobFactory,
+			nil, /* ie */
+			nil, /* cf */
+			nil, /* kvDB */
+			nil, /* limiters */
+		)
 		if err != nil {
 			return nil, nil, errors.Wrapf(err, "making external storage")
 		}

pkg/ccl/storageccl/BUILD.bazel

@@ -46,6 +46,7 @@ go_test(
         "//pkg/security/username",
         "//pkg/server",
         "//pkg/sql",
+        "//pkg/sql/catalog/descs",
         "//pkg/storage",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",

pkg/ccl/storageccl/external_sst_reader_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/storageutils"
@@ -122,7 +123,11 @@ func TestNewExternalSSTReader(t *testing.T) {
 			clusterSettings,
 			blobs.TestBlobServiceClient(tempDir),
 			username.RootUserName(),
-			tc.Servers[0].InternalExecutor().(*sql.InternalExecutor), tc.Servers[0].DB(), nil)
+			tc.Servers[0].InternalExecutor().(*sql.InternalExecutor),
+			tc.Servers[0].CollectionFactory().(*descs.CollectionFactory),
+			tc.Servers[0].DB(),
+			nil, /* limiters */
+		)
 		require.NoError(t, err)
 		fileStores[i].Store = store
 

pkg/ccl/workloadccl/storage.go

@@ -35,9 +35,18 @@ func GetStorage(ctx context.Context, cfg FixtureConfig) (cloud.ExternalStorage,
 		return nil, errors.AssertionFailedf("unsupported external storage provider; valid providers are gs, s3, and azure")
 	}
 
-	s, err := cloud.ExternalStorageFromURI(ctx, cfg.ObjectPathToURI(),
-		base.ExternalIODirConfig{}, clustersettings.MakeClusterSettings(),
-		nil, username.SQLUsername{}, nil, nil, nil)
+	s, err := cloud.ExternalStorageFromURI(
+		ctx,
+		cfg.ObjectPathToURI(),
+		base.ExternalIODirConfig{},
+		clustersettings.MakeClusterSettings(),
+		nil, /* blobClientFactory */
+		username.SQLUsername{},
+		nil, /* ie */
+		nil, /* cf */
+		nil, /* kvDB */
+		nil, /* limiters */
+	)
 	if err != nil {
 		return nil, errors.Wrap(err, storageError)
 	}

pkg/cloud/BUILD.bazel

@@ -23,6 +23,7 @@ go_library(
         "//pkg/security/username",
         "//pkg/settings",
         "//pkg/settings/cluster",
+        "//pkg/sql/catalog/descs",
         "//pkg/sql/sqlutil",
         "//pkg/util/ctxgroup",
         "//pkg/util/ioctx",