Merge #154436 #156112 #156309

154436: rangefeed: defer unregistration until stream is removed from manager r=wenyihu6 a=stevendanna

Unregistering the registration from the processor has the side-effect of closing the underlying memory budget. We don't want that to happen until all of the events have been cleared from the queue.

Here, we pass the unregistration callback up via the Disconnector and then the stream manager calls it when removing the stream.

Epic: none
Release note: None

156112: sql/rowexec: enable tests under secondary tenants r=yuzefovich a=yuzefovich

Given the modifications in recent commits, this mostly required plumbing the right codec into `eval.Context`. Additionally, a few tests needed some extra adjustments for the tenant prefix present in the spans. Only `TestWriteResumeSpan` is still skipped since it wasn't immediately clear what's wrong there.

Epic: CRDB-48945
Release note: None

156309: *: miscellaneous test-tenant-related improvements r=yuzefovich a=yuzefovich

Fixes: #142797.
Fixes: #142801.
Fixes: #142802.
Fixes: #142804.
Fixes: #142805.

Co-authored-by: Steven Danna <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>

Author: 3 people

pkg/ccl/multiregionccl/BUILD.bazel

@@ -105,7 +105,6 @@ go_test(
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
-        "//pkg/util",
         "//pkg/util/ctxgroup",
         "//pkg/util/encoding",
         "//pkg/util/leaktest",

pkg/ccl/multiregionccl/cold_start_latency_test.go

@@ -20,7 +20,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
 	"github.com/cockroachdb/cockroach/pkg/server"
-	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/spanconfig"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/pgurlutils"
@@ -117,13 +116,11 @@ func TestColdStartLatency(t *testing.T) {
 		args.Knobs.Server = serverKnobs
 		perServerArgs[i] = args
 	}
-	cs := cluster.MakeTestingClusterSettings()
 	tc := testcluster.NewTestCluster(t, numNodes, base.TestClusterArgs{
 		ParallelStart:     true,
 		ServerArgsPerNode: perServerArgs,
 		ServerArgs: base.TestServerArgs{
-			DefaultTestTenant: base.TODOTestTenantDisabled,
-			Settings:          cs,
+			DefaultTestTenant: base.TestControlsTenantsExplicitly,
 		},
 	})
 	go func() {
@@ -261,7 +258,6 @@ COMMIT;`}
 	const password = "asdf"
 	{
 		tenant, tenantDB := serverutils.StartTenant(t, tc.Server(0), base.TestTenantArgs{
-			Settings: cs,
 			TenantID: serverutils.TestTenantID(),
 			TestingKnobs: base.TestingKnobs{
 				Server: tenantServerKnobs(0),
@@ -310,7 +306,6 @@ COMMIT;`}
 		start := timeutil.Now()
 		sn := tenantServerKnobs(i)
 		tenant, err := tc.Server(i).TenantController().StartTenant(ctx, base.TestTenantArgs{
-			Settings:            cs,
 			TenantID:            serverutils.TestTenantID(),
 			DisableCreateTenant: true,
 			SkipTenantCheck:     true,

pkg/ccl/multiregionccl/datadriven_test.go

@@ -208,20 +208,12 @@ func testMultiRegionDataDriven(t *testing.T, testPath string) {
 				tc := testcluster.StartTestCluster(t, numServers, base.TestClusterArgs{
 					ServerArgsPerNode: serverArgs,
 					ServerArgs: base.TestServerArgs{
-						// We need to disable the default test tenant here
-						// because it appears as though operations like
-						// "wait-for-zone-config-changes" only work correctly
-						// when called from the system tenant. More
-						// investigation is required (tracked with #76378).
-						DefaultTestTenant: base.TODOTestTenantDisabled,
+						DefaultTestTenant: base.TestDoesNotWorkWithSecondaryTenantsButWeDontKnowWhyYet(156305),
 					},
 				})
 				ds.tc = tc
 
-				sqlConn, err := ds.getSQLConn(0)
-				if err != nil {
-					return err.Error()
-				}
+				sysConn := tc.SystemLayer(0).SQLConn(t)
 				// Speed up closing of timestamps, in order to sleep less below before
 				// we can use follower_read_timestamp(). follower_read_timestamp() uses
 				// sum of the following settings. Also, disable all kvserver lease
@@ -237,7 +229,7 @@ SET CLUSTER SETTING kv.allocator.load_based_lease_rebalancing.enabled = false;
 SET CLUSTER SETTING kv.allocator.min_lease_transfer_interval = '5m'
 `,
 					";") {
-					_, err = sqlConn.Exec(stmt)
+					_, err := sysConn.Exec(stmt)
 					if err != nil {
 						return err.Error()
 					}
@@ -341,8 +333,9 @@ SET CLUSTER SETTING kv.allocator.min_lease_transfer_interval = '5m'
 				if err != nil {
 					return err.Error()
 				}
-				cache := ds.tc.Server(idx).DistSenderI().(*kvcoord.DistSender).RangeDescriptorCache()
-				tablePrefix := keys.MustAddr(keys.SystemSQLCodec.TablePrefix(tableID))
+				s := ds.tc.ApplicationLayer(idx)
+				cache := s.DistSenderI().(*kvcoord.DistSender).RangeDescriptorCache()
+				tablePrefix := keys.MustAddr(s.Codec().TablePrefix(tableID))
 				entry, err := cache.TestingGetCached(ctx, tablePrefix, false, roachpb.LAG_BY_CLUSTER_SETTING)
 				if err != nil {
 					return err.Error()
@@ -738,22 +731,22 @@ func (r *replicaPlacement) satisfiesExpectedPlacement(expected *replicaPlacement
 				continue
 			} else {
 				return errors.Newf(
-					"expected node %s to not be present but had replica type %s",
+					"expected node %d to not be present but had replica type %s",
 					node,
 					actualRt.String())
 			}
 		}
 
 		if !found {
 			return errors.Newf(
-				"expected node %s to have replica type %s but was not found",
+				"expected node %d to have replica type %s but was not found",
 				node,
 				expectedRt.String())
 		}
 
 		if expectedRt != actualRt {
 			return errors.Newf(
-				"expected node %s to have replica type %s but was %s",
+				"expected node %d to have replica type %s but was %s",
 				node,
 				expectedRt.String(),
 				actualRt.String())
@@ -793,15 +786,16 @@ func getRangeKeyForInput(
 	var db string
 	d.ScanArgs(t, dbName, &db)
 
-	execCfg := tc.Server(0).ExecutorConfig().(sql.ExecutorConfig)
+	s := tc.ApplicationLayer(0)
+	execCfg := s.ExecutorConfig().(sql.ExecutorConfig)
 
 	tableDesc, err := lookupTable(&execCfg, db, tbName)
 	if err != nil {
 		return nil, err
 	}
 
 	if !d.HasArg(partitionName) {
-		return tableDesc.TableSpan(keys.SystemSQLCodec).Key, nil
+		return tableDesc.TableSpan(s.Codec()).Key, nil
 	}
 
 	var partition string
@@ -831,7 +825,7 @@ func getRangeKeyForInput(
 
 	_, keyPrefix, err := rowenc.DecodePartitionTuple(
 		&tree.DatumAlloc{},
-		keys.SystemSQLCodec,
+		s.Codec(),
 		tableDesc,
 		primaryInd,
 		part,

pkg/ccl/multiregionccl/multiregionccltestutils/testutils.go

@@ -135,8 +135,8 @@ func TestingCreateMultiRegionClusterWithRegionList(
 			// is used to create the MR cluster for all test cases. For
 			// bonus points, the code to re-enable this should also provide more
 			// flexibility in disabling the default test tenant by callers of this
-			// function. Re-enablement is tracked with #76378.
-			DefaultTestTenant: base.TODOTestTenantDisabled,
+			// function.
+			DefaultTestTenant: base.TestDoesNotWorkWithSecondaryTenantsButWeDontKnowWhyYet(156308),
 		},
 	})
 

pkg/ccl/multiregionccl/region_test.go

@@ -28,11 +28,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// startTestCluster starts a 3 node cluster.
-//
-// Note, if a testfeed depends on particular testing knobs, those may
-// need to be applied to each of the servers in the test cluster
-// returned from this function.
 func TestMultiRegionDatabaseStats(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/multiregionccl/regional_by_row_test.go

@@ -34,7 +34,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
-	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -173,16 +172,11 @@ func TestAlterTableLocalityRegionalByRowCorrectZoneConfigBeforeBackfill(t *testi
 func TestAlterTableLocalityRegionalByRowError(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	skip.UnderRace(t, "takes >400s under race")
-	skip.UnderDeadlock(t, "skipping as per issue #146428")
+
+	skip.UnderDuress(t, "too slow")
 
 	var chunkSize int64 = 100
 	var maxValue = 4000
-	if util.RaceEnabled {
-		// Race builds are a lot slower, so use a smaller number of rows.
-		maxValue = 200
-		chunkSize = 5
-	}
 	// BulkInsertIntoTable adds testCase 0 to maxValue inclusive, so
 	// we round (maxValue + 1) / chunkSize to the nearest int.
 	// To round up x / y using integers, we do (x + y - 1) / y.
@@ -350,11 +344,6 @@ func TestAlterTableLocalityRegionalByRowError(t *testing.T) {
 							params.Locality.Tiers = []roachpb.Tier{
 								{Key: "region", Value: "ajstorm-1"},
 							}
-							// Need to disable the test tenant here because
-							// when running inside a tenant, for some reason
-							// this test doesn't error when expected. More
-							// investigation is required. Tracked with #76378.
-							params.DefaultTestTenant = base.TODOTestTenantDisabled
 							var sqlDB *gosql.DB
 							params.Knobs = base.TestingKnobs{
 								SQLSchemaChanger: &sql.SchemaChangerTestingKnobs{
@@ -378,11 +367,12 @@ func TestAlterTableLocalityRegionalByRowError(t *testing.T) {
 								// Decrease the adopt loop interval so that retries happen quickly.
 								JobsTestingKnobs: jobs.NewTestingKnobsWithShortIntervals(),
 							}
-							var s serverutils.TestServerInterface
+							var srv serverutils.TestServerInterface
 							var kvDB *kv.DB
-							s, sqlDB, kvDB = serverutils.StartServer(t, params)
+							srv, sqlDB, kvDB = serverutils.StartServer(t, params)
 							db = sqlDB
-							defer s.Stopper().Stop(ctx)
+							defer srv.Stopper().Stop(ctx)
+							s := srv.ApplicationLayer()
 
 							// Disable strict GC TTL enforcement because we're going to shove a zero-value
 							// TTL into the system with AddImmediateGCZoneConfig.
@@ -437,7 +427,7 @@ USE t;
 							// that the job did not succeed even though it was canceled.
 							testutils.SucceedsSoon(t, func() error {
 								tableDesc := desctestutils.TestingGetPublicTableDescriptor(
-									kvDB, keys.SystemSQLCodec, "t", "test",
+									kvDB, s.Codec(), "t", "test",
 								)
 								if len(tableDesc.AllMutations()) != 0 {
 									return errors.Errorf(
@@ -493,13 +483,13 @@ USE t;
 								return nil
 							})
 
-							tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, keys.SystemSQLCodec, "t", "test")
+							tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, s.Codec(), "t", "test")
 							if _, err := sqltestutils.AddImmediateGCZoneConfig(db, tableDesc.GetID()); err != nil {
 								t.Fatal(err)
 							}
 							// Ensure that the writes from the partial new indexes are cleaned up.
 							testutils.SucceedsSoon(t, func() error {
-								return sqltestutils.CheckTableKeyCount(ctx, kvDB, keys.SystemSQLCodec, 1, maxValue)
+								return sqltestutils.CheckTableKeyCount(ctx, kvDB, s.Codec(), 1, maxValue)
 							})
 						})
 					}

pkg/ccl/multiregionccl/unique_test.go

@@ -10,7 +10,6 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
-	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowenc"
@@ -28,10 +27,10 @@ import (
 func TestValidateUniqueConstraints(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	// This test fails when run within a tenant. More investigation is
-	// required. Tracked with #76378.
-	s, db, kvDB := serverutils.StartServer(t, base.TestServerArgs{DefaultTestTenant: base.TODOTestTenantDisabled})
-	defer s.Stopper().Stop(context.Background())
+
+	srv, db, kvDB := serverutils.StartServer(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(context.Background())
+	s := srv.ApplicationLayer()
 	r := sqlutils.MakeSQLRunner(db)
 
 	// Create two tables and a view.
@@ -57,7 +56,7 @@ CREATE TABLE u (
 	insertValuesT := func(values []tree.Datum) {
 		// Get the table descriptor and primary index of t.
 		tableDesc := desctestutils.TestingGetTableDescriptor(
-			kvDB, keys.SystemSQLCodec, "test", "public", "t",
+			kvDB, s.Codec(), "test", "public", "t",
 		)
 		primaryIndex := tableDesc.GetPrimaryIndex()
 
@@ -67,7 +66,7 @@ CREATE TABLE u (
 
 		// Construct the primary index entry to insert.
 		primaryIndexEntry, err := rowenc.EncodePrimaryIndex(
-			keys.SystemSQLCodec, tableDesc, primaryIndex, colIDtoRowIndex, values, true, /* includeEmpty */
+			s.Codec(), tableDesc, primaryIndex, colIDtoRowIndex, values, true, /* includeEmpty */
 		)
 		require.NoError(t, err)
 		require.Equal(t, 1, len(primaryIndexEntry))
@@ -82,7 +81,7 @@ CREATE TABLE u (
 	insertValuesU := func(values []tree.Datum) {
 		// Get the table descriptor and indexes of u.
 		tableDesc := desctestutils.TestingGetTableDescriptor(
-			kvDB, keys.SystemSQLCodec, "test", "public", "u",
+			kvDB, s.Codec(), "test", "public", "u",
 		)
 		primaryIndex := tableDesc.GetPrimaryIndex()
 		secondaryIndex := tableDesc.PublicNonPrimaryIndexes()[0]
@@ -94,12 +93,12 @@ CREATE TABLE u (
 
 		// Construct the primary and secondary index entries to insert.
 		primaryIndexEntry, err := rowenc.EncodePrimaryIndex(
-			keys.SystemSQLCodec, tableDesc, primaryIndex, colIDtoRowIndex, values, true, /* includeEmpty */
+			s.Codec(), tableDesc, primaryIndex, colIDtoRowIndex, values, true, /* includeEmpty */
 		)
 		require.NoError(t, err)
 		require.Equal(t, 1, len(primaryIndexEntry))
 		secondaryIndexEntry, err := rowenc.EncodeSecondaryIndex(
-			context.Background(), keys.SystemSQLCodec, tableDesc, secondaryIndex,
+			context.Background(), s.Codec(), tableDesc, secondaryIndex,
 			colIDtoRowIndex, values, rowenc.EmptyVectorIndexEncodingHelper, true, /* includeEmpty */
 		)
 		require.NoError(t, err)

pkg/ccl/partitionccl/drop_test.go

@@ -49,7 +49,6 @@ func TestDropIndexWithZoneConfigCCL(t *testing.T) {
 	asyncNotification := make(chan struct{})
 
 	var params base.TestServerArgs
-	params.DefaultTestTenant = base.TODOTestTenantDisabled
 	params.Knobs = base.TestingKnobs{
 		GCJob: &sql.GCJobTestingKnobs{
 			RunBeforeResume: func(_ jobspb.JobID) error {
@@ -202,8 +201,7 @@ SELECT job_id
 		ctx := context.Background()
 		tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{
 			ServerArgs: base.TestServerArgs{
-				DefaultTestTenant: base.TODOTestTenantDisabled,
-				Knobs:             knobs,
+				Knobs: knobs,
 			},
 		})
 		defer tc.Stopper().Stop(ctx)
@@ -248,8 +246,7 @@ range_max_bytes = 654321000`)
 		ctx := context.Background()
 		tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{
 			ServerArgs: base.TestServerArgs{
-				DefaultTestTenant: base.TODOTestTenantDisabled,
-				Knobs:             knobs,
+				Knobs: knobs,
 			},
 		})
 		defer tc.Stopper().Stop(ctx)

pkg/ccl/partitionccl/partition_test.go

@@ -31,11 +31,10 @@ func TestRemovePartitioningExpiredLicense(t *testing.T) {
 	defer utilccl.TestingEnableEnterprise()()
 
 	ctx := context.Background()
-	s, sqlDBRaw, _ := serverutils.StartServer(t, base.TestServerArgs{
-		UseDatabase:       "d",
-		DefaultTestTenant: base.TODOTestTenantDisabled,
+	srv, sqlDBRaw, _ := serverutils.StartServer(t, base.TestServerArgs{
+		UseDatabase: "d",
 	})
-	defer s.Stopper().Stop(ctx)
+	defer srv.Stopper().Stop(ctx)
 
 	// Create a partitioned table and index.
 	sqlDB := sqlutils.MakeSQLRunner(sqlDBRaw)

pkg/ccl/workloadccl/allccl/BUILD.bazel

@@ -60,7 +60,6 @@ go_test(
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
-        "//pkg/util",
         "//pkg/util/bufalloc",
         "//pkg/util/leaktest",
         "//pkg/util/log",