Merge cockroachdb#104228 cockroachdb#107765 cockroachdb#107889 cockroachdb#108052 cockroachdb#108066

104228: dev: replace `stress` with `--runs_per_test` r=rail a=rickystewart

`--runs_per_test` does the same thing that `stress` does (runs a test
many times to weed out flakes), but better, in that:

1. Better Bazel support -- we had to hack Bazel support into `stress`
   to keep it working
2. Bazel gives us statistics and control over how the tests are
   scheduled in a way that is standardized as opposed to the ad-hoc
   arguments one can pass to `stress`
3. Bazel can collect logs and artifacts for different test runs and
   expose them to the user in a way that `stress` cannot
4. Drop-in support for remote execution when we have access to it,
   obviating the need for `roachprod-stress`

For now, the default implementation of `dev test --stress` is to
run the test 1,000 times, stopping the build if any test fails.
This is meant to replicate how people use `stress` (i.e., just start it
running and stop if there are any failures). The 1,000 figure is
arbitrary and meant to just be a "very high number" of times to run unit
tests. The default figure of 1,000 can be adjusted with `--count`. Also
update documentation with some new recipes and add extra logging to
warn about the change in behavior.

Closes cockroachdb#102879

Epic: none
Release note: None

107765: sql: remove calls to CreateTestServerParams r=yuzefovich a=yuzefovich

This PR removes calls to `tests.CreateTestServerParams` outside of `sql_test` tests as well as does some miscellaneous cleanups along the way. The rationale for doing this is that vast majority of callers didn't actually need the setup performed by the helper but inherited the disabling of test tenant. It seems more prudent for each test to be explicit about the kind of testing knobs and settings it needs.

Informs: cockroachdb#76378.
Epic: CRDB-18499.

Release note: None

107889: concurrency: enable joint claims on the request scan path r=nvanbenschoten a=arulajmani

This patch addresses a TODO that was blocking joint claims on a request's scan path and adds a test to show joint claims work. In particular, previously the lock mode of a queued request was hardcoded to use locking strength `lock.Intent`. We now use the correct lock mode by snapshotting an immutable copy of the request's lock strength on to the queuedGuard when inserting a it into a lock's wait queue.

Informs cockroachdb#102275

Release note: None

108052: changefeedccl: Release allocation when skipping events r=miretskiy a=miretskiy

The changefeed (or KV feed to be precise) may skip some events when "scan boundary" is reached.
Scan boundary is a timestamp when certain event occurs -- usually a schema change.  But, it may also occur
when the `end_time` option is set.

The KV feed ignores events that have MVCC timestamp greater or equal to the scan boundary event.

Unfortunately, due to a long outstanding bug, the memory allocation associated with the event would not be released when KV feed decides to skip the event.

Because of this, allocated memory was "leaked" and not reclaimed. If enough additional events arrive, those leaked events may account for all of the memory budget, thus leading to inability for additional events to be added.

This bug impacts any changefeeds running with the `end_time` option set.  It might also impact changefeeds that observe normal schema change event, though this situation is highly unlikely(the same transaction that perform schema change had to have modified sufficient number of rows in the table to fill up all of the memory budget).

Fixes cockroachdb#108040

Release note (enterprise change): Fix a potential "deadlock" when running changefeed with `end_time` option set.

108066: dev: set `COCKROACH_DEV_LICENSE` for `compose` r=rail a=rickystewart

This test won't run without the license key set.

Epic: CRDB-17171
Release note: None

Co-authored-by: Ricky Stewart <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: Arul Ajmani <[email protected]>
Co-authored-by: Yevgeniy Miretskiy <[email protected]>

Author: 5 people

dev

@@ -8,7 +8,7 @@ fi
 set -euo pipefail
 
 # Bump this counter to force rebuilding `dev` on all machines.
-DEV_VERSION=82
+DEV_VERSION=83
 
 THIS_DIR=$(cd "$(dirname "$0")" && pwd)
 BINARY_DIR=$THIS_DIR/bin/dev-versions

pkg/ccl/changefeedccl/changefeed_test.go

@@ -7342,6 +7342,25 @@ func (s *memoryHoggingSink) Close() error {
 	return nil
 }
 
+type countEmittedRowsSink struct {
+	memoryHoggingSink
+	numRows int64 // Accessed atomically; not using atomic.Int64 to make backports possible.
+}
+
+func (s *countEmittedRowsSink) EmitRow(
+	ctx context.Context,
+	topic TopicDescriptor,
+	key, value []byte,
+	updated, mvcc hlc.Timestamp,
+	alloc kvevent.Alloc,
+) error {
+	alloc.Release(ctx)
+	atomic.AddInt64(&s.numRows, 1)
+	return nil
+}
+
+var _ Sink = (*countEmittedRowsSink)(nil)
+
 func TestChangefeedFlushesSinkToReleaseMemory(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -7392,6 +7411,54 @@ func TestChangefeedFlushesSinkToReleaseMemory(t *testing.T) {
 	require.Greater(t, sink.numFlushes(), 0)
 }
 
+// Test verifies that KV feed does not leak event memory allocation
+// when it reaches end_time or scan boundary.
+func TestKVFeedDoesNotLeakMemoryWhenSkippingEvents(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	s, stopServer := makeServer(t)
+	defer stopServer()
+
+	sqlDB := sqlutils.MakeSQLRunner(s.DB)
+	knobs := s.TestingKnobs.
+		DistSQL.(*execinfra.TestingKnobs).
+		Changefeed.(*TestingKnobs)
+
+	// Arrange for a small memory budget.
+	knobs.MemMonitor = startMonitorWithBudget(4096)
+
+	// Arrange for custom sink to be used -- a sink that counts emitted rows.
+	sink := &countEmittedRowsSink{}
+	knobs.WrapSink = func(_ Sink, _ jobspb.JobID) Sink {
+		return sink
+	}
+	sqlDB.Exec(t, `CREATE TABLE foo(key INT PRIMARY KEY DEFAULT unique_rowid(), val INT)`)
+
+	startTime := s.Server.Clock().Now().AsOfSystemTime()
+
+	// Insert 123 rows -- this fills up our tiny memory buffer (~26 rows do)
+	// Collect statement timestamp -- this will become our end time.
+	var insertTimeStr string
+	sqlDB.QueryRow(t,
+		`INSERT INTO foo (val) SELECT * FROM generate_series(1, 123) RETURNING cluster_logical_timestamp();`,
+	).Scan(&insertTimeStr)
+	endTime := parseTimeToHLC(t, insertTimeStr).AsOfSystemTime()
+
+	// Start the changefeed, with end_time set to be equal to the insert time.
+	// KVFeed should ignore all events.
+	var jobID jobspb.JobID
+	sqlDB.QueryRow(t, `CREATE CHANGEFEED FOR foo INTO 'null:' WITH cursor = $1, end_time = $2`,
+		startTime, endTime).Scan(&jobID)
+
+	// If everything is fine (events are ignored, but their memory allocation is released),
+	// the changefeed should terminate.  If not, we'll time out waiting for job.
+	waitForJobStatus(sqlDB, t, jobID, jobs.StatusSucceeded)
+
+	// No rows should have been emitted (all should have been filtered out due to end_time).
+	require.EqualValues(t, 0, atomic.LoadInt64(&sink.numRows))
+}
+
 func TestChangefeedMultiPodTenantPlanning(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/kvfeed/kv_feed.go

@@ -584,6 +584,7 @@ func copyFromSourceToDestUntilTableEvent(
 ) error {
 	var (
 		scanBoundary errBoundaryReached
+		endTimeIsSet = !endTime.IsEmpty()
 
 		// checkForScanBoundary takes in a new event's timestamp (event generated
 		// from rangefeed), and asks "Is some type of 'boundary' reached
@@ -595,7 +596,11 @@ func copyFromSourceToDestUntilTableEvent(
 			// If the scanBoundary is not nil, it either means that there is a table
 			// event boundary set or a boundary for the end time. If the boundary is
 			// for the end time, we should keep looking for table events.
-			_, isEndTimeBoundary := scanBoundary.(*errEndTimeReached)
+			isEndTimeBoundary := false
+			if endTimeIsSet {
+				_, isEndTimeBoundary = scanBoundary.(*errEndTimeReached)
+			}
+
 			if scanBoundary != nil && !isEndTimeBoundary {
 				return nil
 			}
@@ -610,7 +615,7 @@ func copyFromSourceToDestUntilTableEvent(
 			// precedence to table events.
 			if len(nextEvents) > 0 {
 				scanBoundary = &errTableEventReached{nextEvents[0]}
-			} else if !endTime.IsEmpty() && scanBoundary == nil {
+			} else if endTimeIsSet && scanBoundary == nil {
 				scanBoundary = &errEndTimeReached{
 					endTime: endTime,
 				}
@@ -691,6 +696,17 @@ func copyFromSourceToDestUntilTableEvent(
 			if err != nil {
 				return err
 			}
+
+			if skipEntry || scanBoundaryReached {
+				// We will skip this entry or outright terminate kvfeed (if boundary reached).
+				// Regardless of the reason, we must release this event memory allocation
+				// since other ranges might not have reached scan boundary yet.
+				// Failure to release this event allocation may prevent other events from being
+				// enqueued in the blocking buffer due to memory limit.
+				a := e.DetachAlloc()
+				a.Release(ctx)
+			}
+
 			if scanBoundaryReached {
 				// All component rangefeeds are now at the boundary.
 				// Break out of the ctxgroup by returning the sentinel error.
@@ -702,7 +718,6 @@ func copyFromSourceToDestUntilTableEvent(
 			return addEntry(e)
 		}
 	)
-
 	for {
 		e, err := source.Get(ctx)
 		if err != nil {

pkg/ccl/serverccl/BUILD.bazel

@@ -17,7 +17,6 @@ go_library(
         "//pkg/sql",
         "//pkg/sql/contention",
         "//pkg/sql/sqlstats/persistedsqlstats",
-        "//pkg/sql/tests",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/sqlutils",
         "//pkg/util/httputil",

pkg/ccl/serverccl/statusccl/tenant_grpc_test.go

@@ -46,9 +46,7 @@ func TestTenantGRPCServices(t *testing.T) {
 
 	tenantID := serverutils.TestTenantID()
 	testingKnobs := base.TestingKnobs{
-		SQLStatsKnobs: &sqlstats.TestingKnobs{
-			AOSTClause: "AS OF SYSTEM TIME '-1us'",
-		},
+		SQLStatsKnobs: sqlstats.CreateTestingKnobs(),
 	}
 	tenant, connTenant := serverutils.StartTenant(t, server, base.TestTenantArgs{
 		TenantID:     tenantID,

pkg/ccl/serverccl/statusccl/tenant_status_test.go

@@ -391,9 +391,7 @@ func TestTenantCannotSeeNonTenantStats(t *testing.T) {
 	tenant, sqlDB := serverutils.StartTenant(t, server, base.TestTenantArgs{
 		TenantID: roachpb.MustMakeTenantID(10 /* id */),
 		TestingKnobs: base.TestingKnobs{
-			SQLStatsKnobs: &sqlstats.TestingKnobs{
-				AOSTClause: "AS OF SYSTEM TIME '-1us'",
-			},
+			SQLStatsKnobs: sqlstats.CreateTestingKnobs(),
 		},
 	})
 

pkg/ccl/serverccl/tenant_test_utils.go

@@ -23,7 +23,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/contention"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats"
-	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/httputil"
@@ -221,9 +220,10 @@ func newTenantClusterHelper(
 
 	var cluster tenantCluster = make([]TestTenant, tenantClusterSize)
 	for i := 0; i < tenantClusterSize; i++ {
-		args := tests.CreateTestTenantParams(roachpb.MustMakeTenantID(tenantID))
-		args.TestingKnobs = knobs
-		cluster[i] = newTestTenant(t, server, args)
+		cluster[i] = newTestTenant(t, server, base.TestTenantArgs{
+			TenantID:     roachpb.MustMakeTenantID(tenantID),
+			TestingKnobs: knobs,
+		})
 	}
 
 	return cluster

pkg/ccl/sqlproxyccl/BUILD.bazel

@@ -95,7 +95,6 @@ go_test(
         "//pkg/sql/pgwire",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/pgwire/pgwirebase",
-        "//pkg/sql/tests",
         "//pkg/testutils",
         "//pkg/testutils/datapathutils",
         "//pkg/testutils/serverutils",

pkg/ccl/sqlproxyccl/proxy_handler_test.go

@@ -37,7 +37,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
-	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
@@ -1931,15 +1930,16 @@ func TestConnectionMigration(t *testing.T) {
 	require.NoError(t, err)
 
 	// Start first SQL pod.
-	tenant1, tenantDB1 := serverutils.StartTenant(t, s, tests.CreateTestTenantParams(tenantID))
+	tenant1, tenantDB1 := serverutils.StartTenant(t, s, base.TestTenantArgs{TenantID: tenantID})
 	tenant1.PGPreServer().(*pgwire.PreServeConnHandler).TestingSetTrustClientProvidedRemoteAddr(true)
 	defer tenant1.Stopper().Stop(ctx)
 	defer tenantDB1.Close()
 
 	// Start second SQL pod.
-	params2 := tests.CreateTestTenantParams(tenantID)
-	params2.DisableCreateTenant = true
-	tenant2, tenantDB2 := serverutils.StartTenant(t, s, params2)
+	tenant2, tenantDB2 := serverutils.StartTenant(t, s, base.TestTenantArgs{
+		TenantID:            tenantID,
+		DisableCreateTenant: true,
+	})
 	tenant2.PGPreServer().(*pgwire.PreServeConnHandler).TestingSetTrustClientProvidedRemoteAddr(true)
 	defer tenant2.Stopper().Stop(ctx)
 	defer tenantDB2.Close()
@@ -2984,10 +2984,11 @@ func startTestTenantPodsWithStopper(
 
 	var tenants []serverutils.ApplicationLayerInterface
 	for i := 0; i < count; i++ {
-		params := tests.CreateTestTenantParams(tenantID)
-		params.TestingKnobs = knobs
-		params.Stopper = stopper
-		tenant, tenantDB := serverutils.StartTenant(t, ts, params)
+		tenant, tenantDB := serverutils.StartTenant(t, ts, base.TestTenantArgs{
+			TenantID:     tenantID,
+			TestingKnobs: knobs,
+			Stopper:      stopper,
+		})
 		tenant.PGPreServer().(*pgwire.PreServeConnHandler).TestingSetTrustClientProvidedRemoteAddr(true)
 
 		// Create a test user. We only need to do it once.

pkg/ccl/testccl/sqlccl/gc_job_test.go

@@ -16,7 +16,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqltestutils"
-	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
@@ -36,7 +35,9 @@ func TestGCJobGetsMarkedIdle(t *testing.T) {
 	})
 	sqltestutils.SetShortRangeFeedIntervals(t, mainDB)
 	defer s.Stopper().Stop(ctx)
-	tenant, tenantDB := serverutils.StartTenant(t, s, tests.CreateTestTenantParams(serverutils.TestTenantID()))
+	tenant, tenantDB := serverutils.StartTenant(t, s, base.TestTenantArgs{
+		TenantID: serverutils.TestTenantID(),
+	})
 	defer tenant.Stopper().Stop(ctx)
 	defer tenantDB.Close()