crdb_internal: make system_jobs a view

Release note: none.
Epic: CRDB-48791.

Author: dt

pkg/bench/rttanalysis/testdata/benchmark_expectations

@@ -67,7 +67,7 @@ exp,benchmark
 3,Jobs/jobs_page_latest_50
 3,Jobs/jobs_page_type_filtered
 1-3,Jobs/jobs_page_type_filtered_no_matches
-4,Jobs/non_admin_crdb_internal.system_jobs
+2,Jobs/non_admin_crdb_internal.system_jobs
 2,Jobs/non_admin_show_jobs
 12-15,Jobs/pause_job
 12-15,Jobs/resume_job

pkg/sql/crdb_internal.go

@@ -24,7 +24,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
 	"github.com/cockroachdb/cockroach/pkg/gossip"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
-	"github.com/cockroachdb/cockroach/pkg/jobs/jobsauth"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprotectedts"
 	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -957,183 +956,48 @@ CREATE TABLE crdb_internal.leases (
 	},
 }
 
-const (
-	// systemJobsAndJobInfoBaseQuery consults both the `system.jobs` and
-	// `system.job_info` tables to return relevant information about a job.
-	//
-	// NB: Every job on creation writes a row each for its payload and progress to
-	// the `system.job_info` table. For a given job there will always be at most
-	// one row each for its payload and progress. This is because of the
-	// `system.job_info` write semantics described `InfoStorage.Write`.
-	// Theoretically, a job could have no rows corresponding to its progress and
-	// so we perform a LEFT JOIN to get a NULL value when no progress row is
-	// found.
-	systemJobsAndJobInfoBaseQuery = `
-SELECT
-DISTINCT(id), status, created, payload.value AS payload, progress.value AS progress,
-created_by_type, created_by_id, claim_session_id, claim_instance_id, num_runs, last_run, job_type
-FROM
-system.jobs AS j
-LEFT JOIN system.job_info AS progress ON j.id = progress.job_id AND progress.info_key = 'legacy_progress'
-INNER JOIN system.job_info AS payload ON j.id = payload.job_id AND payload.info_key = 'legacy_payload'
-`
-	systemJobsIDPredicate     = ` WHERE id = $1`
-	systemJobsTypePredicate   = ` WHERE job_type = $1`
-	systemJobsStatusPredicate = ` WHERE status = $1`
-)
-
-type systemJobsPredicate int
-
-const (
-	noPredicate systemJobsPredicate = iota
-	jobID
-	jobType
-	jobStatus
-)
-
-func getInternalSystemJobsQuery(predicate systemJobsPredicate) string {
-	switch predicate {
-	case noPredicate:
-		return systemJobsAndJobInfoBaseQuery
-	case jobID:
-		return systemJobsAndJobInfoBaseQuery + systemJobsIDPredicate
-	case jobType:
-		return systemJobsAndJobInfoBaseQuery + systemJobsTypePredicate
-	case jobStatus:
-		return systemJobsAndJobInfoBaseQuery + systemJobsStatusPredicate
-	}
-
-	return ""
-}
-
 // TODO(tbg): prefix with kv_.
-var crdbInternalSystemJobsTable = virtualSchemaTable{
+var crdbInternalSystemJobsTable = virtualSchemaView{
 	schema: `
-CREATE TABLE crdb_internal.system_jobs (
-  id                INT8      NOT NULL,
-  status            STRING    NOT NULL,
-  created           TIMESTAMP NOT NULL,
-  payload           BYTES     NOT NULL,
-  progress          BYTES,
-  created_by_type   STRING,
-  created_by_id     INT,
-  claim_session_id  BYTES,
-  claim_instance_id INT8,
-  num_runs          INT8,
-  last_run          TIMESTAMP,
-  job_type          STRING,
-  INDEX (id),
-  INDEX (job_type),
-  INDEX (status)
-)`,
+CREATE VIEW crdb_internal.system_jobs (
+  id,
+  status,
+  created,
+  payload,
+  progress,
+  created_by_type,
+  created_by_id,
+  claim_session_id,
+  claim_instance_id,
+  num_runs,
+  last_run,
+  job_type
+) AS (SELECT j.id, j.status, j.created, payload.value, progress.value,
+	j.created_by_type, j.created_by_id, j.claim_session_id, j.claim_instance_id,
+	j.num_runs, j.last_run, j.job_type
+	FROM system.jobs AS j
+	LEFT JOIN system.job_info AS progress ON j.id = progress.job_id AND progress.info_key = 'legacy_progress'
+	INNER JOIN system.job_info AS payload ON j.id = payload.job_id AND payload.info_key = 'legacy_payload'
+	WHERE crdb_internal.can_view_job(j.owner)
+)
+`,
 	comment: `wrapper over system.jobs with row access control (KV scan)`,
-	indexes: []virtualIndex{
-		{
-			populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-				q := getInternalSystemJobsQuery(jobID)
-				targetType := tree.MustBeDInt(unwrappedConstraint)
-				return populateSystemJobsTableRows(ctx, p, addRow, q, targetType)
-			},
-		},
-		{
-			populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-				q := getInternalSystemJobsQuery(jobType)
-				targetType := tree.MustBeDString(unwrappedConstraint)
-				return populateSystemJobsTableRows(ctx, p, addRow, q, targetType)
-			},
-		},
-		{
-			populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-				q := getInternalSystemJobsQuery(jobStatus)
-				targetType := tree.MustBeDString(unwrappedConstraint)
-				return populateSystemJobsTableRows(ctx, p, addRow, q, targetType)
-			},
-		},
-	},
-	populate: func(ctx context.Context, p *planner, db catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) error {
-		_, err := populateSystemJobsTableRows(ctx, p, addRow, getInternalSystemJobsQuery(noPredicate))
-		return err
+	resultColumns: colinfo.ResultColumns{
+		{Name: "id", Typ: types.Int},
+		{Name: "status", Typ: types.String},
+		{Name: "created", Typ: types.TimestampTZ},
+		{Name: "payload", Typ: types.Bytes},
+		{Name: "progress", Typ: types.Bytes},
+		{Name: "created_by_type", Typ: types.String},
+		{Name: "created_by_id", Typ: types.Int},
+		{Name: "claim_session_id", Typ: types.Int},
+		{Name: "claim_instance_id", Typ: types.Int},
+		{Name: "num_runs", Typ: types.Int},
+		{Name: "last_run", Typ: types.TimestampTZ},
+		{Name: "job_type", Typ: types.String},
 	},
 }
 
-// populateSystemJobsTableRows calls addRow for all rows of the system.jobs table
-// except for rows that the user does not have access to. It returns true
-// if at least one row was generated.
-func populateSystemJobsTableRows(
-	ctx context.Context,
-	p *planner,
-	addRow func(...tree.Datum) error,
-	query string,
-	params ...interface{},
-) (result bool, retErr error) {
-	const jobIdIdx = 0
-	const jobPayloadIdx = 3
-
-	matched := false
-
-	// Note: we query system.jobs as root, so we must be careful about which rows we return.
-	it, err := p.InternalSQLTxn().QueryIteratorEx(ctx,
-		"system-jobs-scan",
-		p.Txn(),
-		sessiondata.NodeUserSessionDataOverride,
-		query,
-		params...,
-	)
-	if err != nil {
-		return matched, err
-	}
-
-	cleanup := func(ctx context.Context) {
-		if err := it.Close(); err != nil {
-			retErr = errors.CombineErrors(retErr, err)
-		}
-	}
-	defer cleanup(ctx)
-
-	globalPrivileges, err := jobsauth.GetGlobalJobPrivileges(ctx, p)
-	if err != nil {
-		return matched, err
-	}
-
-	for {
-		hasNext, err := it.Next(ctx)
-		if !hasNext || err != nil {
-			return matched, err
-		}
-
-		currentRow := it.Cur()
-		jobID, err := strconv.Atoi(currentRow[jobIdIdx].String())
-		if err != nil {
-			return matched, err
-		}
-		payloadBytes := currentRow[jobPayloadIdx]
-		payload, err := jobs.UnmarshalPayload(payloadBytes)
-		if err != nil {
-			return matched, wrapPayloadUnMarshalError(err, currentRow[jobIdIdx])
-		}
-		err = jobsauth.Authorize(
-			ctx, p, jobspb.JobID(jobID), payload.UsernameProto.Decode(), jobsauth.ViewAccess, globalPrivileges,
-		)
-		if err != nil {
-			// Filter out jobs which the user is not allowed to see.
-			if IsInsufficientPrivilegeError(err) {
-				continue
-			}
-			return matched, err
-		}
-
-		if err := addRow(currentRow...); err != nil {
-			return matched, err
-		}
-		matched = true
-	}
-}
-
-func wrapPayloadUnMarshalError(err error, jobID tree.Datum) error {
-	return errors.WithHintf(err, "could not decode the payload for job %s."+
-		" consider deleting this job from system.jobs", jobID)
-}
-
 var crdbInternalJobsView = virtualSchemaView{
 	// TODO(dt): the left-outer joins here in theory mean that if there are more
 	// than one row per job in status or progress the job row would need to be

pkg/sql/crdb_internal_test.go

@@ -1327,7 +1327,7 @@ func TestInternalSystemJobsTableMirrorsSystemJobsTable(t *testing.T) {
 	assert.NoError(t, err)
 
 	tdb.Exec(t,
-		"INSERT INTO system.jobs (id, status, created) values ($1, $2, $3)",
+		"INSERT INTO system.jobs (id, status, created, owner) values ($1, $2, $3, 'root')",
 		1, jobs.StateRunning, timeutil.Now(),
 	)
 	tdb.Exec(t,
@@ -1336,9 +1336,9 @@ func TestInternalSystemJobsTableMirrorsSystemJobsTable(t *testing.T) {
 	)
 
 	tdb.Exec(t,
-		`INSERT INTO system.jobs (id, status, created, created_by_type, created_by_id, 
-                         claim_session_id, claim_instance_id, num_runs, last_run, job_type) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
-		2, jobs.StateRunning, timeutil.Now(), "created by", 2, []byte("claim session id"),
+		`INSERT INTO system.jobs (id, status, created, owner, created_by_type, created_by_id,
+                         claim_session_id, claim_instance_id, num_runs, last_run, job_type) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`,
+		2, jobs.StateRunning, timeutil.Now(), "root", "created by", 2, []byte("claim session id"),
 		2, 2, timeutil.Now(), jobspb.TypeImport.String(),
 	)
 	tdb.Exec(t,

pkg/sql/opt/exec/execbuilder/testdata/explain_analyze

@@ -176,7 +176,7 @@ quality of service: regular
         │       │               └── • virtual table
         │       │                     sql nodes: <hidden>
         │       │                     regions: <hidden>
-        │       │                     actual row count: 363
+        │       │                     actual row count: 360
         │       │                     execution time: 0µs
         │       │                     table: pg_class@primary
         │       │

pkg/sql/opt/exec/execbuilder/testdata/virtual

@@ -482,45 +482,6 @@ vectorized: true
     └── • virtual table
           table: pg_type@primary
 
-# Validate that the virtual index on 'status' works.
-# Vectorized execution may be on or off during tests, so exclude it from the output.
-query T
-SELECT info FROM [EXPLAIN SELECT count(*) FROM crdb_internal.system_jobs WHERE status = 'paused'] WHERE info NOT LIKE 'vectorized%'
-----
-    distribution: local
-    ·
-    • group (scalar)
-    │
-    └── • virtual table
-          table: system_jobs@system_jobs_status_idx
-          spans: [/'paused' - /'paused']
-
-# Validate that the virtual index on 'status' works.
-# Vectorized execution may be on or off during tests, so exclude it from the output.
-query T
-SELECT info FROM [EXPLAIN SELECT count(*) FROM crdb_internal.system_jobs WHERE job_type = 'changefeed'] WHERE info NOT LIKE 'vectorized%'
-----
-    distribution: local
-    ·
-    • group (scalar)
-    │
-    └── • virtual table
-          table: system_jobs@system_jobs_job_type_idx
-          spans: [/'changefeed' - /'changefeed']
-
-# Validate that the virtual index on 'status' works.
-# Vectorized execution may be on or off during tests, so exclude it from the output.
-query T
-SELECT info FROM [EXPLAIN SELECT count(*) FROM crdb_internal.system_jobs WHERE id = 1] WHERE info NOT LIKE 'vectorized%'
-----
-    distribution: local
-    ·
-    • group (scalar)
-    │
-    └── • virtual table
-          table: system_jobs@system_jobs_id_idx
-          spans: [/1 - /1]
-
 # Regression test for PruneCols fix (#94890)
 statement ok
 SET testing_optimizer_disable_rule_probability = 1.0;
@@ -579,14 +540,3 @@ vectorized: true
 statement ok
 RESET disallow_full_table_scans;
 
-# Verify that we're not fooling ourselves by showing that we can use a virtual
-# index when we actually can't (#108317).
-query T
-EXPLAIN SELECT * FROM crdb_internal.system_jobs WHERE id > 100;
-----
-distribution: local
-vectorized: true
-·
-• virtual table
-  table: system_jobs@primary
-  virtual table filter