Merge #144316 #144568 #144597 #144638 #144640 #144720

144316: sql: move job check back out of txn creating autostats job r=msbutler,mgartner,rytaft a=michae2

In #119383 we moved the find-running-jobs-of-type query into the same transaction that creates the autostats jobs, in order to avoid a race between different nodes trying to start an autostats job. In practice, however, we've seen that this combined read-write transaction suffers from many retries due to RETRY_SERIALIZABLE errors, when racing transactions write to system.jobs and then fail to refresh the find-running-jobs read.

One the one hand, the fact that these transactions fail to commit does mean fewer rows are visible in system.jobs. On the other hand, each aborted transaction leaves behind an intent to clean up, so I don't think the load on system.jobs is any lighter than it was before.

These types of races, where writing a new row depends on the absence of other rows, is kind of the worst case for optimistic Serializable isolation. If we could materialize the conflict to a single existing row that the transaction locks, we'd avoid both the serialization failures and the race to start a job. But we don't have such a row. (Maybe one day we could create a "NEXT AUTOSTATS JOB" sentinel row in system.jobs that the winner locks before claiming it and starting the job? But that seems like a big change.)

For now, this PR partially reverts #119383 so that find-running-jobs is back in a separate read-only transaction before creating the job. Using a read-only transaction to check avoids the serializable errors, at the cost of sometimes creating too many jobs which will then immediately fail the repeated check within the job. We think this is going to reduce the load on system.jobs during times of high contention. Cluster setting `sql.stats.automatic_job_check_before_creating_job` can be used to get back the old behavior.

In the future when we allow more than one autostats job, we hope to improve the situation by changing the mechanism for preventing concurrent stats collections on the same table. Maybe then we can have some kind of sentinel row per table to materialize the conflict.

Informs: #108435

Release note: None

144568: aggmetric: switch to new btree r=RaduBerinde a=RaduBerinde

Note: first commit is #144555

Use `BtreeG[MetricItem]` in the `aggmetric` code.

Informs: #144504
Release note: None

144597: sql/tests: use newer btree in enum test r=RaduBerinde a=RaduBerinde

Informs: #144504
Release note: None

144638: ui: fix various bugs on db console schedules page r=kyle-a-wong a=kyle-a-wong

- Updated max result size of sql query to fetch schedules
  to ensure all schedules are rendered in
  db console
- Fixed a bug where changing the "show" and "status" values
  in the url query params resulted in a "something went wrong"
  page. Now, if an invalid value is set in the "show" or "status"
  query parameters, the corresponding dropdown will reset to the
  default value. ("Show: Latest 50" and "Status: All" respectively)

Fixes: #143925, #143924
Epic: None
Release note (bug fix): Fixes bugs in the db console page:
- Previously, the schedules page only showed a subset of the total
  schedules for a cluster due to a missing parameter in the server api
  call. Now the schedules table should correctly show all schedules in
  a cluster
- Fixed a bug where manually upadting the "show" or "status" query

144640: logical: wrap recoverable errors in savepoints r=jeffswenson a=jeffswenson

Internal executor statements failures are not atomic. They may fail and
leave behind live KV operations. Usually this is okay because the error
is returned up the stack and the caller aborts the transaction. There is
a bug LDR because it recovers condition failed errors to detect LWW
losses. The insert fast path writes indexes before it writes to the
primary key, so these fast path failures could corrupt index entries.

Now, all LDR code paths that commit a transaction containing a
recoverable internal executor error wrap the statement in a savepoint.

Informs: #144645
Epic: [CRDB-48647](https://cockroachlabs.atlassian.net/browse/CRDB-48647)
Release note: None

144720: vecindex: remove search_beam_size cluster setting r=rafiss a=andy-kimball

The `sql.vecindex.search_beam_size` cluster setting only exists to specify the default value for the vector_search_beam_size session setting. However, this is not necessary; session settings can specify a default value on their own. This commit removes the cluster setting and sets the global default to be "32" for the session setting.

Epic: CRDB-42943
Release note: None

Co-authored-by: Michael Erickson <[email protected]>
Co-authored-by: Radu Berinde <[email protected]>
Co-authored-by: Kyle Wong <[email protected]>
Co-authored-by: Jeff Swenson <[email protected]>
Co-authored-by: Andrew Kimball <[email protected]>

Author: 6 people

pkg/crosscluster/logical/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "purgatory.go",
         "range_stats.go",
         "replication_statements.go",
+        "savepoint.go",
         "sql_crud_writer.go",
         "sql_row_reader.go",
         "sql_row_writer.go",
@@ -127,6 +128,7 @@ go_test(
         "purgatory_test.go",
         "range_stats_test.go",
         "replication_statements_test.go",
+        "savepoint_test.go",
         "sql_row_reader_test.go",
         "sql_row_writer_test.go",
         "table_batch_handler_test.go",

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -181,6 +181,61 @@ func TestLogicalStreamIngestionJobNameResolution(t *testing.T) {
 	}
 }
 
+func TestOptimsitcInsertCorruption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	skip.UnderDeadlock(t)
+	defer log.Scope(t).Close(t)
+
+	// This is a regression test for #144645. Running the optimistic insert code
+	// path could corrupt indexes if the insert is not wrapped in a savepoint.
+
+	ctx := context.Background()
+
+	server, s, dbSource, dbDest := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
+	defer server.Stopper().Stop(ctx)
+
+	dbSource.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.try_optimistic_insert.enabled = true")
+
+	createTableStmt := `CREATE TABLE computed_cols (
+		a INT,
+		b INT,
+		c INT,
+		PRIMARY KEY (a, b)
+	)`
+	dbSource.Exec(t, createTableStmt)
+	dbDest.Exec(t, createTableStmt)
+
+	createIdxStmt := `CREATE INDEX c ON computed_cols (c)`
+	dbSource.Exec(t, createIdxStmt)
+	dbDest.Exec(t, createIdxStmt)
+
+	// Insert initial data into destination that should be overwritten. This
+	// gives more opportunities for corruption.
+	dbDest.Exec(t, "INSERT INTO computed_cols (a, b, c) VALUES (1, 2, 3), (3, 4, 5)")
+	dbSource.Exec(t, "INSERT INTO computed_cols (a, b, c) VALUES (1, 2, 6), (3, 4, 7)")
+
+	// Create logical replication stream from source to destination
+	sourceURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("a"))
+	var jobID jobspb.JobID
+	dbDest.QueryRow(t,
+		"CREATE LOGICAL REPLICATION STREAM FROM TABLE computed_cols ON $1 INTO TABLE computed_cols",
+		sourceURL.String(),
+	).Scan(&jobID)
+
+	WaitUntilReplicatedTime(t, s.Clock().Now(), dbDest, jobID)
+
+	dbSource.CheckQueryResults(t, "SELECT * FROM computed_cols", [][]string{
+		{"1", "2", "6"},
+		{"3", "4", "7"},
+	})
+	dbDest.CheckQueryResults(t, "SELECT * FROM computed_cols", [][]string{
+		{"1", "2", "6"},
+		{"3", "4", "7"},
+	})
+
+	compareReplicatedTables(t, s, "a", "b", "computed_cols", dbSource, dbDest)
+}
+
 type fatalDLQ struct{ *testing.T }
 
 func (fatalDLQ) Create(ctx context.Context) error { return nil }

pkg/crosscluster/logical/lww_row_processor.go

@@ -640,7 +640,11 @@ func (lww *lwwQuerier) InsertRow(
 		if !useLowPriority.Get(&lww.settings.SV) {
 			sess.QualityOfService = nil
 		}
-		if _, err = ie.ExecParsed(ctx, replicatedOptimisticInsertOpName, kvTxn, sess, stmt, datums...); err != nil {
+		err = withSavepoint(ctx, kvTxn, func() error {
+			_, err = ie.ExecParsed(ctx, replicatedOptimisticInsertOpName, kvTxn, sess, stmt, datums...)
+			return err
+		})
+		if err != nil {
 			if isLwwLoser(err) {
 				return batchStats{}, nil
 			}
@@ -667,7 +671,10 @@ func (lww *lwwQuerier) InsertRow(
 		sess.QualityOfService = nil
 	}
 	sess.OriginTimestampForLogicalDataReplication = row.MvccTimestamp
-	_, err = ie.ExecParsed(ctx, replicatedInsertOpName, kvTxn, sess, stmt, datums...)
+	err = withSavepoint(ctx, kvTxn, func() error {
+		_, err = ie.ExecParsed(ctx, replicatedInsertOpName, kvTxn, sess, stmt, datums...)
+		return err
+	})
 	if isLwwLoser(err) {
 		return batchStats{}, nil
 	}

pkg/crosscluster/logical/savepoint.go

@@ -0,0 +1,49 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package logical
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/errors"
+)
+
+// withSavepoint is a utility function that runs the provided function within
+// the context of the savepoint.
+//
+// If the caller wishes to recover from an error returned by an internal
+// executor and keep the transaction, then the call to the internal executor
+// must be scoped within a savepoint. Failed internal executor calls are not
+// atomic and may have left behind some KV operations. Usually that is okay
+// because the error is passed up the call stack and the transaction is rolled
+// back.
+func withSavepoint(ctx context.Context, txn *kv.Txn, fn func() error) error {
+	// In the classic sql writer the txn is optional.
+	if txn == nil {
+		return fn()
+	}
+	// TODO(jeffswenson): consider changing the internal executor so all calls
+	// implicitly create and apply/rollback savepoints.
+	savepoint, err := txn.CreateSavepoint(ctx)
+	if err != nil {
+		return err
+	}
+	err = fn()
+	if err != nil {
+		// NOTE: we return the save point error if rollback fails because we do not
+		// want something checking error types to attempt to handle the inner
+		// error.
+		if savePointErr := txn.RollbackToSavepoint(ctx, savepoint); savePointErr != nil {
+			return errors.WithSecondaryError(savePointErr, err)
+		}
+		return err
+	}
+	if err := txn.ReleaseSavepoint(ctx, savepoint); err != nil {
+		return err
+	}
+	return nil
+}

pkg/crosscluster/logical/savepoint_test.go

@@ -0,0 +1,70 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package logical
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWithSavepoint(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	srv, rawDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(context.Background())
+
+	ctx := context.Background()
+	sqlDB := sqlutils.MakeSQLRunner(rawDB)
+	sqlDB.Exec(t, "CREATE TABLE test (id STRING PRIMARY KEY, value STRING)")
+
+	require.NoError(t, srv.DB().Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+		err := withSavepoint(ctx, txn, func() error {
+			_, err := srv.InternalExecutor().(*sql.InternalExecutor).ExecEx(
+				ctx,
+				"test-insert",
+				txn,
+				sessiondata.NodeUserSessionDataOverride,
+				"INSERT INTO defaultdb.test VALUES ('ok', 'is-persisted')",
+			)
+			return err
+		})
+		require.NoError(t, err)
+
+		err = withSavepoint(ctx, txn, func() error {
+			_, err := srv.InternalExecutor().(*sql.InternalExecutor).ExecEx(
+				ctx,
+				"test-insert",
+				txn,
+				sessiondata.NodeUserSessionDataOverride,
+				"INSERT INTO defaultdb.test VALUES ('fails', 'is-rolled-back')",
+			)
+			require.NoError(t, err)
+			// NOTE: the query above is okay, which means it wrote things to KV,
+			// but we're going to return an error which rolls back the
+			// savepoint.
+			return errors.New("something to rollback")
+		})
+		require.ErrorContains(t, err, "something to rollback")
+
+		return nil
+	}))
+
+	sqlDB.CheckQueryResults(t, "SELECT id, value FROM test", [][]string{
+		{"ok", "is-persisted"},
+	})
+}

pkg/crosscluster/logical/table_batch_handler.go

@@ -170,7 +170,9 @@ func (t *tableHandler) attemptBatch(
 				stats.kvLwwLosers += tombstoneUpdateStats.kvWriteTooOld
 			case event.prevRow == nil:
 				stats.inserts++
-				err := t.sqlWriter.InsertRow(ctx, txn, event.originTimestamp, event.row)
+				err := withSavepoint(ctx, txn.KV(), func() error {
+					return t.sqlWriter.InsertRow(ctx, txn, event.originTimestamp, event.row)
+				})
 				if isLwwLoser(err) {
 					// Insert may observe a LWW failure if it attempts to write over a tombstone.
 					stats.kvLwwLosers++

pkg/settings/registry.go

@@ -271,8 +271,13 @@ var retiredSettings = map[InternalKey]struct{}{
 
 // grandfatheredDefaultSettings is the list of "grandfathered" existing sql.defaults
 // cluster settings. In 22.2 and later, new session settings do not need an
-// associated sql.defaults cluster setting. Instead they can have their default
-// changed with ALTER ROLE ... SET.
+// associated sql.defaults cluster setting (see the `vector_search_beam_size`
+// setting in vars.go for an example). A session setting can have its default
+// changed with ALTER ROLE ... SET, similar to this (the example assumes that
+// all roles should use the new default):
+//
+//	ALTER ROLE ALL SET vector_search_beam_size=128;
+//
 // Caveat: in some cases, we may still add new sql.defaults cluster settings,
 // but the new ones *must* be marked as non-public. Undocumented settings are
 // excluded from the check that prevents new sql.defaults settings. The

pkg/sql/create_stats.go

@@ -76,6 +76,13 @@ var nonIndexJSONHistograms = settings.RegisterBoolSetting(
 	false,
 	settings.WithPublic)
 
+var automaticJobCheckBeforeCreatingJob = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"sql.stats.automatic_job_check_before_creating_job.enabled",
+	"set to true to perform the autostats job check before creating the job, instead of in the same "+
+		"transaction as creating the job",
+	true)
+
 const nonIndexColHistogramBuckets = 2
 
 // StubTableStats generates "stub" statistics for a table which are missing
@@ -148,18 +155,27 @@ func (n *createStatsNode) runJob(ctx context.Context) error {
 	}
 	details := record.Details.(jobspb.CreateStatsDetails)
 
-	if n.Name != jobspb.AutoStatsName && n.Name != jobspb.AutoPartialStatsName {
+	jobCheckBefore := automaticJobCheckBeforeCreatingJob.Get(n.p.ExecCfg().SV())
+	if (n.Name == jobspb.AutoStatsName || n.Name == jobspb.AutoPartialStatsName) && jobCheckBefore {
+		// Don't start the job if there is already a CREATE STATISTICS job running.
+		// (To handle race conditions we check this again after the job starts,
+		// but this check is used to prevent creating a large number of jobs that
+		// immediately fail).
+		if err := checkRunningJobs(
+			ctx, nil /* job */, n.p, n.Name == jobspb.AutoPartialStatsName, n.p.ExecCfg().JobRegistry,
+			details.Table.ID,
+		); err != nil {
+			return err
+		}
+	} else {
 		telemetry.Inc(sqltelemetry.CreateStatisticsUseCounter)
 	}
 
 	var job *jobs.StartableJob
 	jobID := n.p.ExecCfg().JobRegistry.MakeJobID()
 	if err := n.p.ExecCfg().InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) (err error) {
-		if n.Name == jobspb.AutoStatsName || n.Name == jobspb.AutoPartialStatsName {
+		if (n.Name == jobspb.AutoStatsName || n.Name == jobspb.AutoPartialStatsName) && !jobCheckBefore {
 			// Don't start the job if there is already a CREATE STATISTICS job running.
-			// (To handle race conditions we check this again after the job starts,
-			// but this check is used to prevent creating a large number of jobs that
-			// immediately fail).
 			if err := checkRunningJobsInTxn(
 				ctx, n.p.EvalContext().Settings, jobspb.InvalidJobID, txn,
 			); err != nil {

pkg/sql/tests/BUILD.bazel

@@ -146,13 +146,13 @@ go_test(
         "@com_github_cockroachdb_cockroach_go_v2//crdb/crdbpgxv5",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_google_btree//:btree",
         "@com_github_google_pprof//profile",
         "@com_github_jackc_pgx_v5//:pgx",
         "@com_github_jackc_pgx_v5//pgconn",
         "@com_github_kr_pretty//:pretty",
         "@com_github_lib_pq//:pq",
         "@com_github_petermattis_goid//:goid",
-        "@com_github_raduberinde_btree//:btree",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@org_golang_x_sync//errgroup",

pkg/sql/tests/enum_test.go

@@ -6,19 +6,20 @@
 package tests
 
 import (
+	"cmp"
 	"context"
 	"fmt"
 	"math/rand"
 	"sort"
 	"strings"
 	"testing"
 
-	"github.com/RaduBerinde/btree" // TODO(#144504): switch to the newer btree
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/google/btree"
 	"github.com/stretchr/testify/require"
 )
 
@@ -46,17 +47,17 @@ func TestLargeEnums(t *testing.T) {
 	// have to wait for lots of versions and it would take a very long time.
 	var createEnumsQuery string
 	{
-		alreadyInserted := btree.New(8)
+		alreadyInserted := btree.NewG[int](8, cmp.Less[int])
 		next := func(n int) (next int, ok bool) {
-			alreadyInserted.AscendGreaterOrEqual(intItem(n), func(i btree.Item) (wantMore bool) {
-				next, ok = int(i.(intItem)), true
+			alreadyInserted.AscendGreaterOrEqual(n, func(i int) (wantMore bool) {
+				next, ok = i, true
 				return false
 			})
 			return next, ok
 		}
 		prev := func(n int) (prev int, ok bool) {
-			alreadyInserted.DescendLessOrEqual(intItem(n), func(i btree.Item) (wantMore bool) {
-				prev, ok = int(i.(intItem)), true
+			alreadyInserted.DescendLessOrEqual(n, func(i int) (wantMore bool) {
+				prev, ok = i, true
 				return false
 			})
 			return prev, ok
@@ -74,7 +75,7 @@ func TestLargeEnums(t *testing.T) {
 				require.Truef(t, ok, "prev %v %v", n, order[:i])
 				fmt.Fprintf(&buf, " AFTER '%d';\n", prev)
 			}
-			alreadyInserted.ReplaceOrInsert(intItem(n))
+			alreadyInserted.ReplaceOrInsert(n)
 		}
 		buf.WriteString("COMMIT;")
 		createEnumsQuery = buf.String()
@@ -101,12 +102,6 @@ func TestLargeEnums(t *testing.T) {
 	require.Truef(t, sort.IntsAreSorted(read), "%v", read)
 }
 
-type intItem int
-
-func (i intItem) Less(o btree.Item) bool {
-	return i < o.(intItem)
-}
-
 // TestEnumPlaceholderWithAsOfSystemTime is a regression test for an edge case
 // with bind where we would not properly deal with leases involving types.
 func TestEnumPlaceholderWithAsOfSystemTime(t *testing.T) {