Merge #144309

144309: jobs: add jobs.avoid_full_scans.enabled r=mgartner a=mgartner

#### jobs: add jobs.avoid_full_scans.enabled

The `jobs.avoid_full_scans.enabled` cluster setting has been added which
adds `AVOID_FULL_SCAN` hints to two internal, jobs-related queries:
`find-running-jobs-of-type` and `find-all-running-jobs-of-type`. This
hint prevents the optimizer from choosing bad query plans with full
table scans.

The setting is disabled by default. It will be enabled in a future
commit.

Release note: None

#### jobs: enable jobs.avoid_full_scans.enabled by default

Epic: None
Release note (performance improvement): Some internal queries executed
by the jobs system are now less likely to perform full table scans of
the `system.jobs` table, making them more efficient. This change can be
reverted by disabling the `jobs.avoid_full_scans.enabled` cluster
setting.


Co-authored-by: Marcus Gartner <[email protected]>

Author: craig[bot]

pkg/jobs/BUILD.bazel

@@ -66,6 +66,7 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/log/eventpb",
         "//pkg/util/log/severity",
+        "//pkg/util/metamorphic",
         "//pkg/util/metric",
         "//pkg/util/pprofutil",
         "//pkg/util/protoutil",

pkg/jobs/utils.go

@@ -12,39 +12,56 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/settings"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
 	"github.com/cockroachdb/errors"
 )
 
+var testingAvoidFullScans = metamorphic.ConstantWithTestBool(
+	"jobs.avoid_full_scans_in_find_running_jobs",
+	true, /* defaultValue */
+)
+
+var avoidFullScans = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"jobs.avoid_full_scans_in_find_running_jobs.enabled",
+	"when true, enables hints to avoid full scans for internal, jobs-related queries",
+	testingAvoidFullScans)
+
 // RunningJobExists checks that whether there are any job of the given types
 // in the pending, running, or paused state, optionally ignoring the job with
 // the ID specified by ignoreJobID as well as any jobs created after it, if
 // the passed ID is not InvalidJobID.
 func RunningJobExists(
-	ctx context.Context, ignoreJobID jobspb.JobID, txn isql.Txn, jobTypes ...jobspb.Type,
+	ctx context.Context,
+	cs *cluster.Settings,
+	ignoreJobID jobspb.JobID,
+	txn isql.Txn,
+	jobTypes ...jobspb.Type,
 ) (exists bool, retErr error) {
 	typeStrs, err := getJobTypeStrs(jobTypes)
 	if err != nil {
 		return false, err
 	}
 
-	orderBy := " ORDER BY created"
+	orderBy := "ORDER BY created"
 	if ignoreJobID == jobspb.InvalidJobID {
 		// There is no need to order by the created column if there is no job to
 		// ignore.
 		orderBy = ""
 	}
 
-	stmt := `
-SELECT
-  id
-FROM
-  system.jobs@jobs_status_created_idx
-WHERE
-	job_type IN ` + typeStrs + ` AND
-  status IN ` + NonTerminalStateTupleString + orderBy + `
-LIMIT 1`
+	hint := "jobs_status_created_idx"
+	if avoidFullScans.Get(&cs.SV) {
+		hint = "{FORCE_INDEX=jobs_status_created_idx,AVOID_FULL_SCAN}"
+	}
+
+	q := `SELECT id FROM system.jobs@%s WHERE job_type IN %s AND status IN %s %s LIMIT 1`
+	stmt := fmt.Sprintf(q, hint, typeStrs, NonTerminalStateTupleString, orderBy)
+
 	it, err := txn.QueryIterator(
 		ctx,
 		"find-running-jobs-of-type",
@@ -74,28 +91,31 @@ LIMIT 1`
 // by ignoreJobID as well as any jobs created after it, if the passed ID is not
 // InvalidJobID.
 func RunningJobs(
-	ctx context.Context, ignoreJobID jobspb.JobID, txn isql.Txn, jobTypes ...jobspb.Type,
+	ctx context.Context,
+	cs *cluster.Settings,
+	ignoreJobID jobspb.JobID,
+	txn isql.Txn,
+	jobTypes ...jobspb.Type,
 ) (jobIDs []jobspb.JobID, retErr error) {
 	typeStrs, err := getJobTypeStrs(jobTypes)
 	if err != nil {
 		return jobIDs, err
 	}
 
-	orderBy := " ORDER BY created"
+	orderBy := "ORDER BY created"
 	if ignoreJobID == jobspb.InvalidJobID {
 		// There is no need to order by the created column if there is no job to
 		// ignore.
 		orderBy = ""
 	}
 
-	stmt := `
-SELECT
-  id
-FROM
-  system.jobs@jobs_status_created_idx
-WHERE
-	job_type IN ` + typeStrs + ` AND
-  status IN ` + NonTerminalStateTupleString + orderBy
+	hint := "jobs_status_created_idx"
+	if avoidFullScans.Get(&cs.SV) {
+		hint = "{FORCE_INDEX=jobs_status_created_idx,AVOID_FULL_SCAN}"
+	}
+
+	q := `SELECT id FROM system.jobs@%s WHERE job_type IN %s AND status IN %s %s`
+	stmt := fmt.Sprintf(q, hint, typeStrs, NonTerminalStateTupleString, orderBy)
 	it, err := txn.QueryIterator(
 		ctx,
 		"find-all-running-jobs-of-type",

pkg/spanconfig/spanconfigmanager/manager.go

@@ -128,7 +128,7 @@ func (m *Manager) run(ctx context.Context) {
 			return
 		}
 
-		started, err := m.createAndStartJobIfNoneExists(ctx)
+		started, err := m.createAndStartJobIfNoneExists(ctx, m.settings)
 		if err != nil {
 			log.Errorf(ctx, "error starting auto span config reconciliation job: %v", err)
 		}
@@ -162,7 +162,9 @@ func (m *Manager) run(ctx context.Context) {
 // createAndStartJobIfNoneExists creates span config reconciliation job iff it
 // hasn't been created already and notifies the jobs registry to adopt it.
 // Returns a boolean indicating if the job was created.
-func (m *Manager) createAndStartJobIfNoneExists(ctx context.Context) (bool, error) {
+func (m *Manager) createAndStartJobIfNoneExists(
+	ctx context.Context, cs *cluster.Settings,
+) (bool, error) {
 	if m.knobs.ManagerDisableJobCreation {
 		return false, nil
 	}
@@ -177,8 +179,9 @@ func (m *Manager) createAndStartJobIfNoneExists(ctx context.Context) (bool, erro
 
 	var job *jobs.Job
 	if err := m.db.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-		exists, err := jobs.RunningJobExists(ctx, jobspb.InvalidJobID, txn,
-			jobspb.TypeAutoSpanConfigReconciliation)
+		exists, err := jobs.RunningJobExists(
+			ctx, cs, jobspb.InvalidJobID, txn, jobspb.TypeAutoSpanConfigReconciliation,
+		)
 		if err != nil {
 			return err
 		}

pkg/spanconfig/spanconfigmanager/manager_test.go

@@ -104,7 +104,7 @@ func TestManagerConcurrentJobCreation(t *testing.T) {
 
 	var g errgroup.Group
 	g.Go(func() error {
-		started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx)
+		started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx, ts.ClusterSettings())
 		if err != nil {
 			return err
 		}
@@ -117,7 +117,7 @@ func TestManagerConcurrentJobCreation(t *testing.T) {
 		// Only try to start the job if the first goroutine has reached the testing
 		// knob and is blocked.
 		<-isBlocked
-		started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx)
+		started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx, ts.ClusterSettings())
 		if err != nil {
 			return err
 		}
@@ -183,7 +183,7 @@ func TestManagerStartsJobIfFailed(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx)
+	started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx, ts.ClusterSettings())
 	require.NoError(t, err)
 	require.True(t, started)
 }
@@ -331,7 +331,7 @@ func TestReconciliationJobErrorAndRecovery(t *testing.T) {
 		},
 	)
 
-	started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx)
+	started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx, ts.ClusterSettings())
 	require.NoError(t, err)
 	require.True(t, started)
 
@@ -354,7 +354,7 @@ func TestReconciliationJobErrorAndRecovery(t *testing.T) {
 	mu.err = nil
 	mu.Unlock()
 
-	started, err = manager.TestingCreateAndStartJobIfNoneExists(ctx)
+	started, err = manager.TestingCreateAndStartJobIfNoneExists(ctx, ts.ClusterSettings())
 	require.NoError(t, err)
 	require.True(t, started)
 
@@ -421,7 +421,7 @@ func TestReconciliationUsesRightCheckpoint(t *testing.T) {
 		nil,
 	)
 
-	started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx)
+	started, err := manager.TestingCreateAndStartJobIfNoneExists(ctx, ts.ClusterSettings())
 	require.NoError(t, err)
 	require.True(t, started)
 

pkg/spanconfig/spanconfigmanager/test_helpers.go

@@ -5,10 +5,16 @@
 
 package spanconfigmanager
 
-import "context"
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+)
 
 // TestingCreateAndStartJobIfNoneExists is a wrapper around
 // createAndStartJobIfNoneExists for testing it.
-func (m *Manager) TestingCreateAndStartJobIfNoneExists(ctx context.Context) (bool, error) {
-	return m.createAndStartJobIfNoneExists(ctx)
+func (m *Manager) TestingCreateAndStartJobIfNoneExists(
+	ctx context.Context, cs *cluster.Settings,
+) (bool, error) {
+	return m.createAndStartJobIfNoneExists(ctx, cs)
 }

pkg/sql/create_stats.go

@@ -160,13 +160,18 @@ func (n *createStatsNode) runJob(ctx context.Context) error {
 			// (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, jobspb.InvalidJobID, txn); err != nil {
+			if err := checkRunningJobsInTxn(
+				ctx, n.p.EvalContext().Settings, jobspb.InvalidJobID, txn,
+			); err != nil {
 				return err
 			}
 			// Don't start auto partial stats jobs if there is another auto partial
 			// stats job running on the same table.
 			if n.Name == jobspb.AutoPartialStatsName {
-				if err := checkRunningAutoPartialJobsInTxn(ctx, jobspb.InvalidJobID, txn, n.p.ExecCfg().JobRegistry, details.Table.ID); err != nil {
+				if err := checkRunningAutoPartialJobsInTxn(
+					ctx, n.p.EvalContext().Settings, jobspb.InvalidJobID, txn,
+					n.p.ExecCfg().JobRegistry, details.Table.ID,
+				); err != nil {
 					return err
 				}
 			}
@@ -881,11 +886,15 @@ func checkRunningJobs(
 		jobID = job.ID()
 	}
 	return p.ExecCfg().InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) (err error) {
-		if err = checkRunningJobsInTxn(ctx, jobID, txn); err != nil {
+		if err = checkRunningJobsInTxn(
+			ctx, p.ExtendedEvalContext().Settings, jobID, txn,
+		); err != nil {
 			return err
 		}
 		if autoPartial {
-			return checkRunningAutoPartialJobsInTxn(ctx, jobID, txn, jobRegistry, tableID)
+			return checkRunningAutoPartialJobsInTxn(
+				ctx, p.ExtendedEvalContext().Settings, jobID, txn, jobRegistry, tableID,
+			)
 		}
 		return nil
 	})
@@ -896,9 +905,11 @@ func checkRunningJobs(
 // that started earlier than this one. If there are, checkRunningJobsInTxn
 // returns an error. If jobID is jobspb.InvalidJobID, checkRunningJobsInTxn just
 // checks if there are any pending, running, or paused CreateStats jobs.
-func checkRunningJobsInTxn(ctx context.Context, jobID jobspb.JobID, txn isql.Txn) error {
-	exists, err := jobs.RunningJobExists(ctx, jobID, txn,
-		jobspb.TypeCreateStats, jobspb.TypeAutoCreateStats,
+func checkRunningJobsInTxn(
+	ctx context.Context, cs *cluster.Settings, jobID jobspb.JobID, txn isql.Txn,
+) error {
+	exists, err := jobs.RunningJobExists(
+		ctx, cs, jobID, txn, jobspb.TypeCreateStats, jobspb.TypeAutoCreateStats,
 	)
 	if err != nil {
 		return err
@@ -919,13 +930,14 @@ func checkRunningJobsInTxn(ctx context.Context, jobID jobspb.JobID, txn isql.Txn
 // AutoCreatePartialStats jobs for the same table.
 func checkRunningAutoPartialJobsInTxn(
 	ctx context.Context,
+	cs *cluster.Settings,
 	jobID jobspb.JobID,
 	txn isql.Txn,
 	jobRegistry *jobs.Registry,
 	tableID descpb.ID,
 ) error {
-	autoPartialStatJobIDs, err := jobs.RunningJobs(ctx, jobID, txn,
-		jobspb.TypeAutoCreatePartialStats,
+	autoPartialStatJobIDs, err := jobs.RunningJobs(
+		ctx, cs, jobID, txn, jobspb.TypeAutoCreatePartialStats,
 	)
 	if err != nil {
 		return err