importer: unskip all tests with test tenants

This mostly involved ensuring that the right ApplicationLayerInterface
is being used, or the test just worked. Also a couple of tests explicitly
control the tenants, so mark those accordingly.

My hypothesis is that some tests have been fixed long time ago in
4fd9f70 (`tc.ServerConn` now returns
the right connection), but I didn't verify it.

Only `TestImportIntoCSV` needs a special callout. In that test we have
some expected errors, and some of them when encountered by the tenants
can be retried. In order to speed up the test we reduce the retry
duration from 2 minutes to 2 seconds.

Additionally remove some SQLMemoryPoolSize overrides that now equal the
default of 256MiB (which increased from 128MiB a couple years ago).

Release note: None

Author: yuzefovich

pkg/ccl/importerccl/ccl_test.go

@@ -416,7 +416,7 @@ func TestImportInTenant(t *testing.T) {
 		ExternalIODir: baseDir,
 		// Test is designed to run inside a tenant so no need to
 		// probabilistically run it inside the default test tenant.
-		DefaultTestTenant: base.TODOTestTenantDisabled,
+		DefaultTestTenant: base.TestControlsTenantsExplicitly,
 	}
 	tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{ServerArgs: args})
 	defer tc.Stopper().Stop(ctx)
@@ -470,7 +470,7 @@ func TestImportInMultiServerTenant(t *testing.T) {
 	args := base.TestServerArgs{
 		// Test is designed to run inside a tenant so no need to
 		// probabilistically run it inside the default test tenant.
-		DefaultTestTenant: base.TODOTestTenantDisabled,
+		DefaultTestTenant: base.TestControlsTenantsExplicitly,
 		ExternalIODir:     baseDir,
 	}
 	tc := serverutils.StartCluster(t, 1, base.TestClusterArgs{ServerArgs: args})

pkg/sql/importer/BUILD.bazel

@@ -187,7 +187,6 @@ go_test(
         "//pkg/testutils/datapathutils",
         "//pkg/testutils/jobutils",
         "//pkg/testutils/kvclientutils",
-        "//pkg/testutils/pgurlutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",

pkg/sql/importer/import_into_test.go

@@ -54,12 +54,7 @@ func TestProtectedTimestampsDuringImportInto(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	args := base.TestClusterArgs{
-		ServerArgs: base.TestServerArgs{
-			DefaultTestTenant: base.TestDoesNotWorkWithSecondaryTenantsButWeDontKnowWhyYet(107141),
-		},
-	}
-	tc := testcluster.StartTestCluster(t, 1, args)
+	tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{})
 	defer tc.Stopper().Stop(ctx)
 	s := tc.Server(0).ApplicationLayer()
 	tenantSettings := s.ClusterSettings()

pkg/sql/importer/import_processor_test.go

@@ -682,21 +682,19 @@ func TestCSVImportCanBeResumed(t *testing.T) {
 	defer TestingSetParallelImporterReaderBatchSize(batchSize)()
 	defer row.TestingSetDatumRowConverterBatchSize(2 * batchSize)()
 
-	s, db, _ := serverutils.StartServer(t,
+	srv, db, _ := serverutils.StartServer(t,
 		base.TestServerArgs{
-			// Hangs when run from a test tenant. More investigation is
-			// required here. Tracked with #76378.
-			DefaultTestTenant: base.TODOTestTenantDisabled,
 			Knobs: base.TestingKnobs{
 				JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
 				DistSQL: &execinfra.TestingKnobs{
 					BulkAdderFlushesEveryBatch: true,
 				},
 			},
 		})
-	registry := s.JobRegistry().(*jobs.Registry)
 	ctx := context.Background()
-	defer s.Stopper().Stop(ctx)
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
+	registry := s.JobRegistry().(*jobs.Registry)
 
 	sqlDB := sqlutils.MakeSQLRunner(db)
 	setSmallIngestBufferSizes(t, sqlDB)
@@ -794,21 +792,19 @@ func TestCSVImportMarksFilesFullyProcessed(t *testing.T) {
 	defer TestingSetParallelImporterReaderBatchSize(batchSize)()
 	defer row.TestingSetDatumRowConverterBatchSize(2 * batchSize)()
 
-	s, db, _ := serverutils.StartServer(t,
+	srv, db, _ := serverutils.StartServer(t,
 		base.TestServerArgs{
-			// Test hangs when run within a test tenant. More investigation
-			// is required here. Tracked with #76378.
-			DefaultTestTenant: base.TODOTestTenantDisabled,
 			Knobs: base.TestingKnobs{
 				JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
 				DistSQL: &execinfra.TestingKnobs{
 					BulkAdderFlushesEveryBatch: true,
 				},
 			},
 		})
-	registry := s.JobRegistry().(*jobs.Registry)
 	ctx := context.Background()
-	defer s.Stopper().Stop(ctx)
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
+	registry := s.JobRegistry().(*jobs.Registry)
 
 	sqlDB := sqlutils.MakeSQLRunner(db)
 	sqlDB.Exec(t, `CREATE DATABASE d`)

pkg/sql/importer/import_stmt_test.go

@@ -56,7 +56,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/jobutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/kvclientutils"
-	"github.com/cockroachdb/cockroach/pkg/testutils/pgurlutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
@@ -1296,19 +1295,14 @@ func TestFailedImport(t *testing.T) {
 	ctx := context.Background()
 	baseDir := datapathutils.TestDataPath(t, "csv")
 	tc := serverutils.StartCluster(t, nodes, base.TestClusterArgs{ServerArgs: base.TestServerArgs{
-		// Test fails within a test tenant. This may be because we're trying
-		// to access files in nodelocal://1, which is off node. More
-		// investigation is required. Tracked with #76378.
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-		SQLMemoryPoolSize: 256 << 20,
-		ExternalIODir:     baseDir,
+		ExternalIODir: baseDir,
 	}})
 	defer tc.Stopper().Stop(ctx)
 	conn := tc.ServerConn(0)
 
 	var forceFailure bool
 	for i := 0; i < tc.NumServers(); i++ {
-		tc.Server(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
+		tc.ApplicationLayer(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
 			jobspb.TypeImport,
 			func(raw jobs.Resumer) jobs.Resumer {
 				r := raw.(*importResumer)
@@ -1371,15 +1365,14 @@ func TestImportIntoCSVCancel(t *testing.T) {
 			},
 			JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
 		},
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-		ExternalIODir:     baseDir,
+		ExternalIODir: baseDir,
 	}})
 	defer tc.Stopper().Stop(ctx)
 	conn := tc.ServerConn(0)
 
 	setupDoneCh := make(chan struct{})
 	for i := 0; i < tc.NumServers(); i++ {
-		tc.Server(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
+		tc.ApplicationLayer(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
 			jobspb.TypeImport,
 			func(raw jobs.Resumer) jobs.Resumer {
 				r := raw.(*importResumer)
@@ -1422,17 +1415,13 @@ func TestImportCSVStmt(t *testing.T) {
 	ctx := context.Background()
 	baseDir := datapathutils.TestDataPath(t, "csv")
 	tc := serverutils.StartCluster(t, nodes, base.TestClusterArgs{ServerArgs: base.TestServerArgs{
-		// Test fails when run within a test tenant. More
-		// investigation is required. Tracked with #76378.
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-		SQLMemoryPoolSize: 256 << 20,
-		ExternalIODir:     baseDir,
+		ExternalIODir: baseDir,
 	}})
 	defer tc.Stopper().Stop(ctx)
 	conn := tc.ServerConn(0)
 
 	for i := 0; i < tc.NumServers(); i++ {
-		tc.Server(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
+		tc.ApplicationLayer(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
 			jobspb.TypeImport,
 			func(raw jobs.Resumer) jobs.Resumer {
 				r := raw.(*importResumer)
@@ -1752,10 +1741,7 @@ func TestImportCSVStmt(t *testing.T) {
 	t.Run("RBAC-SuperUser", func(t *testing.T) {
 		sqlDB.Exec(t, `CREATE USER testuser`)
 		sqlDB.Exec(t, `GRANT admin TO testuser`)
-		pgURL, cleanupFunc := pgurlutils.PGUrl(
-			t, tc.ApplicationLayer(0).AdvSQLAddr(), "TestImportPrivileges-testuser",
-			url.User("testuser"),
-		)
+		pgURL, cleanupFunc := tc.ApplicationLayer(0).PGUrl(t, serverutils.User(username.TestUser))
 		defer cleanupFunc()
 		testuser, err := gosql.Open("postgres", pgURL.String())
 		if err != nil {
@@ -1868,7 +1854,7 @@ func TestImportCSVStmt(t *testing.T) {
 	// Test userfile import CSV.
 	t.Run("userfile-simple", func(t *testing.T) {
 		userfileURI := "userfile://defaultdb.public.root/test.csv"
-		userfileStorage, err := tc.Server(0).ExecutorConfig().(sql.ExecutorConfig).DistSQLSrv.
+		userfileStorage, err := tc.ApplicationLayer(0).ExecutorConfig().(sql.ExecutorConfig).DistSQLSrv.
 			ExternalStorageFromURI(ctx, userfileURI, username.RootUserName())
 		require.NoError(t, err)
 
@@ -1884,7 +1870,7 @@ func TestImportCSVStmt(t *testing.T) {
 
 	t.Run("userfile-relative-file-path", func(t *testing.T) {
 		userfileURI := "userfile:///import-test/employees.csv"
-		userfileStorage, err := tc.Server(0).ExecutorConfig().(sql.ExecutorConfig).DistSQLSrv.
+		userfileStorage, err := tc.ApplicationLayer(0).ExecutorConfig().(sql.ExecutorConfig).DistSQLSrv.
 			ExternalStorageFromURI(ctx, userfileURI, username.RootUserName())
 		require.NoError(t, err)
 
@@ -1980,21 +1966,13 @@ func TestImportObjectLevelRBAC(t *testing.T) {
 	const nodes = 3
 
 	ctx := context.Background()
-	tc := serverutils.StartCluster(t, nodes, base.TestClusterArgs{ServerArgs: base.TestServerArgs{
-		// Test fails when run within a test tenant. More investigation
-		// is required. Tracked with #76378.
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-		SQLMemoryPoolSize: 256 << 20,
-	}})
+	tc := serverutils.StartCluster(t, nodes, base.TestClusterArgs{})
 	defer tc.Stopper().Stop(ctx)
 	conn := tc.ServerConn(0)
 	rootDB := sqlutils.MakeSQLRunner(conn)
 
 	rootDB.Exec(t, `CREATE USER testuser`)
-	pgURL, cleanupFunc := pgurlutils.PGUrl(
-		t, tc.ApplicationLayer(0).AdvSQLAddr(), "TestImportPrivileges-testuser",
-		url.User("testuser"),
-	)
+	pgURL, cleanupFunc := tc.ApplicationLayer(0).PGUrl(t, serverutils.User(username.TestUser))
 	defer cleanupFunc()
 
 	startTestUser := func(t *testing.T) *gosql.DB {
@@ -2009,7 +1987,7 @@ func TestImportObjectLevelRBAC(t *testing.T) {
 
 	writeToUserfile := func(t *testing.T, filename, data string) {
 		// Write to userfile storage now that testuser has CREATE privileges.
-		ief := tc.Server(0).InternalDB().(isql.DB)
+		ief := tc.ApplicationLayer(0).InternalDB().(isql.DB)
 		fileTableSystem1, err := cloud.ExternalStorageFromURI(
 			ctx, dest, base.ExternalIODirConfig{},
 			cluster.NoSettings, blobs.TestEmptyBlobClientFactory,
@@ -2139,10 +2117,8 @@ func TestImportIntoCSV(t *testing.T) {
 		Knobs: base.TestingKnobs{
 			JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
 		},
-		// Test fails when run within a test tenant. More investigation
-		// is required. Tracked with #76378.
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-		ExternalIODir:     baseDir}})
+		ExternalIODir: baseDir,
+	}})
 	defer tc.Stopper().Stop(ctx)
 	conn := tc.ServerConn(0)
 
@@ -2154,7 +2130,7 @@ func TestImportIntoCSV(t *testing.T) {
 	var delayImportFinish chan struct{}
 
 	for i := 0; i < tc.NumServers(); i++ {
-		tc.Server(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
+		tc.ApplicationLayer(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
 			jobspb.TypeImport,
 			func(raw jobs.Resumer) jobs.Resumer {
 				r := raw.(*importResumer)
@@ -2186,6 +2162,13 @@ func TestImportIntoCSV(t *testing.T) {
 	sqlDB := sqlutils.MakeSQLRunner(conn)
 
 	sqlDB.Exec(t, `SET CLUSTER SETTING kv.bulk_ingest.batch_size = '10KB'`)
+	// 'import-into-no-glob-wildcard' hits an error that the file doesn't exist
+	// in external storage. When run with test tenants, that error also happens
+	// to be annotated with 'rpc error' marker, which we treat as a retryable
+	// one (see joberror.IsPermanentBulkJobError which calls
+	// sqlerrors.IsDistSQLRetryableError). In order to speed up the test, we'll
+	// reduce the retry duration from 2 minutes to 2 seconds.
+	sqlDB.Exec(t, `SET CLUSTER SETTING bulkio.import.retry_duration = '2s'`)
 
 	testFiles := makeCSVData(t, numFiles, rowsPerFile, nodes, rowsPerRaceFile)
 	if util.RaceEnabled {
@@ -2902,7 +2885,7 @@ func TestImportIntoCSV(t *testing.T) {
 		)
 
 		sqlDB.ExpectErr(
-			t, `ingested key collides with an existing one: /Table/\d+/1/0/0`,
+			t, `ingested key collides with an existing one: (\/Tenant\/10)?/Table/\d+/1/0/0`,
 			fmt.Sprintf(`IMPORT INTO t (a, b) CSV DATA (%s)`, testFiles.files[0]),
 		)
 	})
@@ -2944,7 +2927,7 @@ func TestImportIntoCSV(t *testing.T) {
 
 		preCollisionData := sqlDB.QueryStr(t, `SELECT * FROM t`)
 		sqlDB.ExpectErr(
-			t, `ingested key collides with an existing one: /Table/\d+/1/0/0`,
+			t, `ingested key collides with an existing one: (\/Tenant\/10)?/Table/\d+/1/0/0`,
 			fmt.Sprintf(`IMPORT INTO t (a, b) CSV DATA (%s)`, testFiles.fileWithShadowKeys[0]),
 		)
 		sqlDB.CheckQueryResults(t, `SELECT * FROM t`, preCollisionData)
@@ -2980,7 +2963,7 @@ func TestImportIntoCSV(t *testing.T) {
 	// Test userfile IMPORT INTO CSV.
 	t.Run("import-into-userfile-simple", func(t *testing.T) {
 		userfileURI := "userfile://defaultdb.public.root/test.csv"
-		userfileStorage, err := tc.Server(0).ExecutorConfig().(sql.ExecutorConfig).DistSQLSrv.
+		userfileStorage, err := tc.ApplicationLayer(0).ExecutorConfig().(sql.ExecutorConfig).DistSQLSrv.
 			ExternalStorageFromURI(ctx, userfileURI, username.RootUserName())
 		require.NoError(t, err)
 
@@ -3840,11 +3823,8 @@ func TestImportDefaultWithResume(t *testing.T) {
 
 	ctx := context.Background()
 	filterFunc, verifyFunc := kvclientutils.PrefixTransactionRetryFilter(t, importProgressDebugName, 1)
-	s, db, _ := serverutils.StartServer(t,
+	srv, db, _ := serverutils.StartServer(t,
 		base.TestServerArgs{
-			// Test hangs when run within a test tenant. More investigation
-			// is required. Tracked with #76378.
-			DefaultTestTenant: base.TODOTestTenantDisabled,
 			Knobs: base.TestingKnobs{
 				JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
 				DistSQL: &execinfra.TestingKnobs{
@@ -3856,8 +3836,9 @@ func TestImportDefaultWithResume(t *testing.T) {
 			},
 			SQLMemoryPoolSize: 1 << 30, // 1 GiB
 		})
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
 	registry := s.JobRegistry().(*jobs.Registry)
-	defer s.Stopper().Stop(ctx)
 
 	sqlDB := sqlutils.MakeSQLRunner(db)
 	testCases := []struct {
@@ -4369,13 +4350,7 @@ func TestImportControlJobRBAC(t *testing.T) {
 	defer jobs.ResetConstructors()()
 
 	ctx := context.Background()
-	tc := serverutils.StartCluster(t, 1, base.TestClusterArgs{
-		ServerArgs: base.TestServerArgs{
-			// Test fails when run within a test tenant. More investigation
-			// is required. Tracked with #76378.
-			DefaultTestTenant: base.TODOTestTenantDisabled,
-		},
-	})
+	tc := serverutils.StartCluster(t, 1, base.TestClusterArgs{})
 	defer tc.Stopper().Stop(ctx)
 	rootDB := sqlutils.MakeSQLRunner(tc.ServerConn(0))
 
@@ -4384,10 +4359,7 @@ func TestImportControlJobRBAC(t *testing.T) {
 	// Create non-root user.
 	rootDB.Exec(t, `CREATE USER testuser`)
 	rootDB.Exec(t, `ALTER ROLE testuser CONTROLJOB`)
-	pgURL, cleanupFunc := pgurlutils.PGUrl(
-		t, tc.ApplicationLayer(0).AdvSQLAddr(), "TestImportPrivileges-testuser",
-		url.User("testuser"),
-	)
+	pgURL, cleanupFunc := tc.ApplicationLayer(0).PGUrl(t, serverutils.User(username.TestUser))
 	defer cleanupFunc()
 	testuser, err := gosql.Open("postgres", pgURL.String())
 	if err != nil {
@@ -4413,7 +4385,7 @@ func TestImportControlJobRBAC(t *testing.T) {
 
 	startLeasedJob := func(t *testing.T, record jobs.Record) *jobs.StartableJob {
 		job, err := jobs.TestingCreateAndStartJob(
-			ctx, registry, tc.Server(0).InternalDB().(isql.DB), record,
+			ctx, registry, tc.ApplicationLayer(0).InternalDB().(isql.DB), record,
 		)
 		require.NoError(t, err)
 		return job
@@ -4795,9 +4767,8 @@ func TestCreateStatsAfterImport(t *testing.T) {
 	st := cluster.MakeClusterSettings()
 	stats.AutomaticStatisticsOnSystemTables.Override(context.Background(), &st.SV, false)
 	args := base.TestServerArgs{
-		Settings:          st,
-		DefaultTestTenant: base.TODOTestTenantDisabled,
-		ExternalIODir:     baseDir,
+		Settings:      st,
+		ExternalIODir: baseDir,
 	}
 	tc := serverutils.StartCluster(t, nodes, base.TestClusterArgs{ServerArgs: args})
 	defer tc.Stopper().Stop(ctx)
@@ -5155,7 +5126,7 @@ func waitForJobResult(
 	t *testing.T, tc serverutils.TestClusterInterface, id jobspb.JobID, expected jobs.State,
 ) {
 	// Force newly created job to be adopted and verify its result.
-	tc.Server(0).JobRegistry().(*jobs.Registry).TestingNudgeAdoptionQueue()
+	tc.ApplicationLayer(0).JobRegistry().(*jobs.Registry).TestingNudgeAdoptionQueue()
 	testutils.SucceedsSoon(t, func() error {
 		var unused int64
 		return tc.ServerConn(0).QueryRow(
@@ -5276,17 +5247,14 @@ func TestImportJobEventLogging(t *testing.T) {
 	ctx := context.Background()
 	baseDir := datapathutils.TestDataPath(t, "avro")
 	args := base.TestServerArgs{ExternalIODir: baseDir}
-	// Test fails within a test tenant. More investigation is required.
-	// Tracked with #76378.
-	args.DefaultTestTenant = base.TODOTestTenantDisabled
 	args.Knobs = base.TestingKnobs{JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals()}
 	params := base.TestClusterArgs{ServerArgs: args}
 	tc := serverutils.StartCluster(t, nodes, params)
 	defer tc.Stopper().Stop(ctx)
 
 	var forceFailure bool
 	for i := 0; i < tc.NumServers(); i++ {
-		tc.Server(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
+		tc.ApplicationLayer(i).JobRegistry().(*jobs.Registry).TestingWrapResumerConstructor(
 			jobspb.TypeImport,
 			func(raw jobs.Resumer) jobs.Resumer {
 				r := raw.(*importResumer)