crdb_internal: replace crdb_internal.jobs vtable with view

Release note: none.
Epic: https://cockroachlabs.atlassian.net/browse/CRDB-48791.

Author: dt

pkg/bench/rttanalysis/testdata/benchmark_expectations

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

pkg/server/application_api/jobs_test.go

@@ -355,8 +355,8 @@ func TestAdminAPIJobsDetails(t *testing.T) {
 			t.Fatal(err)
 		}
 		sqlDB.Exec(t,
-			`INSERT INTO system.jobs (id, status, num_runs, last_run, job_type) VALUES ($1, $2, $3, $4, $5)`,
-			job.id, job.status, job.numRuns, job.lastRun, payload.Type().String(),
+			`INSERT INTO system.jobs (id, status, num_runs, last_run, job_type, owner) VALUES ($1, $2, $3, $4, $5, $6)`,
+			job.id, job.status, job.numRuns, job.lastRun, payload.Type().String(), job.username.Normalized(),
 		)
 		sqlDB.Exec(t,
 			`INSERT INTO system.job_info (job_id, info_key, value) VALUES ($1, $2, $3)`,

pkg/sql/crdb_internal.go

@@ -162,7 +162,7 @@ var crdbInternal = virtualSchema{
 		catconstants.CrdbInternalIndexSpansTableID:                  crdbInternalIndexSpansTable,
 		catconstants.CrdbInternalIndexUsageStatisticsTableID:        crdbInternalIndexUsageStatistics,
 		catconstants.CrdbInternalInflightTraceSpanTableID:           crdbInternalInflightTraceSpanTable,
-		catconstants.CrdbInternalJobsTableID:                        crdbInternalJobsTable,
+		catconstants.CrdbInternalJobsTableID:                        crdbInternalJobsView,
 		catconstants.CrdbInternalSystemJobsTableID:                  crdbInternalSystemJobsTable,
 		catconstants.CrdbInternalKVNodeStatusTableID:                crdbInternalKVNodeStatusTable,
 		catconstants.CrdbInternalKVStoreStatusTableID:               crdbInternalKVStoreStatusTable,
@@ -1134,142 +1134,82 @@ func wrapPayloadUnMarshalError(err error, jobID tree.Datum) error {
 		" consider deleting this job from system.jobs", jobID)
 }
 
-const (
-	jobIDFilter      = ` WHERE j.id = $1`
-	jobsStatusFilter = ` WHERE j.status = $1`
-	jobsTypeFilter   = ` WHERE j.job_type = $1`
-)
-
-// TODO(tbg): prefix with kv_.
-var crdbInternalJobsTable = virtualSchemaTable{
-	schema: `
-CREATE TABLE crdb_internal.jobs (
-  job_id                INT,
-  job_type              STRING,
-  description           STRING,
-  statement             STRING,
-  user_name             STRING,
-  status                STRING,
-  running_status        STRING,
-  created               TIMESTAMPTZ,
-  finished              TIMESTAMPTZ,
-  modified              TIMESTAMPTZ,
-  fraction_completed    FLOAT,
-  high_water_timestamp  DECIMAL,
-  error                 STRING,
-  coordinator_id        INT,
-  INDEX(job_id),
-  INDEX(status),
-  INDEX(job_type)
+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
+	// repeated for each. While this is never the case in practice, it does mean
+	// the optimizer cannot elide the joins even if the progress or status is not
+	// actually being read from a query on the view. We could change the joins to
+	// be `LATERAL (... WHERE job_id = id LiMIT 1) ON true` which would promise to
+	// the optimizer that the join cannot change the number of rows, allowing it
+	// to elide it entirely if the joined columns are not used, however SHOW JOBS
+	// and other callers typically _are_ looking for progress and status so we
+	// likely will need the joins anyway, queries experts expressed some concern
+	// that it could be possible "lateral join prevents some other optimizations".
+	// Given we typically expect to need the joined columns anyway, these are left
+	// as left joins for now. NB: the `union`, needs to be `union all` to avoid
+	// materializing all of the columns even when they're not used, since a
+	// non-`all` union has to be prepared to compare them all to eliminate
+	// duplicates, even when in the vast majority of calls we expect the session
+	// jobs generator to be empty.
+	schema: `CREATE VIEW crdb_internal.jobs (
+  job_id,
+  job_type,
+  description,
+  statement,
+  user_name,
+  status,
+  running_status,
+  created,
+  finished,
+  modified,
+  fraction_completed,
+  high_water_timestamp,
+  error,
+  coordinator_id
+) AS (
+	SELECT
+		j.id,
+		j.job_type,
+		coalesce(j.description, '') as description,
+		coalesce(j.description, '') as statement,
+		coalesce(j.owner, '') as user_name,
+		j.status as status,
+		s.status as running_status,
+		j.created::timestamptz,
+		j.finished,
+		greatest(j.created, j.finished, p.written, s.written)::timestamptz AS modified,
+		p.fraction as fraction_completed,
+		p.resolved as high_water_timestamp,
+		coalesce(j.error_msg, '') as error,
+		j.claim_instance_id as coordinator_id
+	FROM system.public.jobs AS j
+	LEFT OUTER JOIN system.public.job_progress AS p ON j.id = p.job_id
+	LEFT OUTER JOIN system.public.job_status AS s ON j.id = s.job_id
+	WHERE crdb_internal.can_view_job(j.owner)
+	UNION ALL
+		(SELECT job_id, job_type, description, description, user_name, 'pending',
+					NULL, now(), NULL, now(), NULL, NULL, NULL, NULL
+			FROM crdb_internal.session_pending_jobs()
+		)
 )`,
-	comment: `decoded job metadata from crdb_internal.system_jobs (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) {
-			targetID := tree.MustBeDInt(unwrappedConstraint)
-			return makeJobsTableRows(ctx, p, addRow, jobIDFilter, targetID)
-		},
-	}, {
-		populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-			targetStatus := tree.MustBeDString(unwrappedConstraint)
-			return makeJobsTableRows(ctx, p, addRow, jobsStatusFilter, targetStatus)
-		},
-	}, {
-		populate: func(ctx context.Context, unwrappedConstraint tree.Datum, p *planner, _ catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) (matched bool, err error) {
-			targetType := tree.MustBeDString(unwrappedConstraint)
-			return makeJobsTableRows(ctx, p, addRow, jobsTypeFilter, targetType)
-		},
-	}},
-	populate: func(ctx context.Context, p *planner, db catalog.DatabaseDescriptor, addRow func(...tree.Datum) error) error {
-		_, err := makeJobsTableRows(ctx, p, addRow, "")
-		return err
+	resultColumns: colinfo.ResultColumns{
+		{Name: "job_id", Typ: types.Int},
+		{Name: "job_type", Typ: types.String},
+		{Name: "description", Typ: types.String},
+		{Name: "statement", Typ: types.String},
+		{Name: "user_name", Typ: types.String},
+		{Name: "status", Typ: types.String},
+		{Name: "running_status", Typ: types.String},
+		{Name: "created", Typ: types.TimestampTZ},
+		{Name: "finished", Typ: types.TimestampTZ},
+		{Name: "modified", Typ: types.TimestampTZ},
+		{Name: "fraction_completed", Typ: types.Float},
+		{Name: "high_water_timestamp", Typ: types.Decimal},
+		{Name: "error", Typ: types.String},
+		{Name: "coordinator_id", Typ: types.Int},
 	},
-}
-
-func makeJobsTableRows(
-	ctx context.Context,
-	p *planner,
-	addRow func(...tree.Datum) error,
-	whereClause string,
-	params ...interface{},
-) (emitted bool, retErr error) {
-	globalPrivileges, err := jobsauth.GetGlobalJobPrivileges(ctx, p)
-	if err != nil {
-		return false, err
-	}
-
-	query := `SELECT 
-j.id, j.job_type, coalesce(j.description, ''), coalesce(j.owner, ''), j.status as state,
-s.status, j.created::timestamptz, j.finished, greatest(j.created, j.finished, p.written, s.written)::timestamptz AS last_modified,
-p.fraction,
-p.resolved,
-j.error_msg,
-j.claim_instance_id
-FROM system.public.jobs AS j
-LEFT OUTER JOIN system.public.job_progress AS p ON j.id = p.job_id
-LEFT OUTER JOIN system.public.job_status AS s ON j.id = s.job_id
-	` + whereClause + `UNION
-	(SELECT job_id, job_type, description, user_name, 'pending',
-					NULL, now(), NULL, now(), NULL, NULL, NULL, NULL
-	FROM crdb_internal.session_pending_jobs())`
-
-	it, err := p.InternalSQLTxn().QueryIteratorEx(
-		ctx, "system-jobs-join", p.txn, sessiondata.NodeUserSessionDataOverride, query, params...)
-	if err != nil {
-		return emitted, err
-	}
-	defer func() {
-		if err := it.Close(); err != nil {
-			retErr = errors.CombineErrors(retErr, err)
-		}
-	}()
-
-	// Loop while we need to skip a row.
-	for {
-		ok, err := it.Next(ctx)
-		if err != nil || !ok {
-			return emitted, err
-		}
-		r := it.Cur()
-		id, typStr, desc, ownerStr, state, status, created, finished, modified, fraction, resolved, errorMsg, instanceID :=
-			r[0], r[1], r[2], r[3], r[4], r[5], r[6], r[7], r[8], r[9], r[10], r[11], r[12]
-
-		owner := username.MakeSQLUsernameFromPreNormalizedString(string(tree.MustBeDString(ownerStr)))
-		jobID := jobspb.JobID(tree.MustBeDInt(id))
-
-		if errorMsg == tree.DNull {
-			errorMsg = emptyString
-		}
-
-		if err := jobsauth.Authorize(
-			ctx, p, jobID, owner, jobsauth.ViewAccess, globalPrivileges,
-		); err != nil {
-			// Filter out jobs which the user is not allowed to see.
-			if IsInsufficientPrivilegeError(err) {
-				continue
-			}
-			return emitted, err
-		}
-
-		if err = addRow(
-			id,
-			typStr,
-			desc,
-			desc,
-			ownerStr,
-			state,
-			status,
-			created,
-			finished,
-			modified,
-			fraction,
-			resolved,
-			errorMsg,
-			instanceID,
-		); err != nil {
-			return emitted, err
-		}
-		emitted = true
-	}
+	comment: `decoded job metadata from various jobs tables`,
 }
 
 const crdbInternalKVProtectedTSTableQuery = `

pkg/sql/crdb_internal_test.go

@@ -1453,13 +1453,13 @@ func TestInternalSystemJobsAccess(t *testing.T) {
 // worker goroutine returns "query canceled error".
 //
 // In particular, the following setup is used:
-//   - issue SHOW JOBS query which internally issues a query against
-//     crdb_internal.system_jobs virtual table
-//   - that virtual table is generated by issuing "system-jobs-scan" internal
+//   - issue SHOW ZONE CONFIGURATIONS query which internally issues a query against
+//     crdb_internal.zones virtual table
+//   - that virtual table is generated by issuing "crdb-internal-zones-table" internal
 //     query
-//   - during that "system-jobs-scan" query we're injecting the query canceled
+//   - during that "crdb-internal-zones-table" query we're injecting the query canceled
 //     error (in other words, the error is injected during the generation of
-//     crdb_internal.system_jobs virtual table).
+//     crdb_internal.zones virtual table).
 //
 // The injection is achieved by adding a callback to DistSQLReceiver.Push which
 // replaces the first piece of metadata it sees with the error.
@@ -1481,7 +1481,7 @@ func TestVirtualTableDoesntHangOnQueryCanceledError(t *testing.T) {
 							return nil
 						}
 						opName, ok := sql.GetInternalOpName(ctx)
-						if !ok || !(opName == "system-jobs-scan" || opName == "system-jobs-join") {
+						if !ok || !(opName == "crdb-internal-zones-table") {
 							return nil
 						}
 						numCallbacksAdded.Add(1)
@@ -1503,7 +1503,7 @@ func TestVirtualTableDoesntHangOnQueryCanceledError(t *testing.T) {
 	sqlDB := sqlutils.MakeSQLRunner(db)
 
 	addCallback.Store(true)
-	sqlDB.ExpectErr(t, err.Error(), "SHOW JOBS")
+	sqlDB.ExpectErr(t, err.Error(), "SHOW ZONE CONFIGURATIONS")
 	addCallback.Store(false)
 
 	// Sanity check that the callback was added at least once.

pkg/sql/logictest/testdata/logic_test/select

@@ -831,7 +831,7 @@ statement ok
 SELECT * FROM pg_class
 
 statement ok
-SELECT * FROM crdb_internal.jobs
+SELECT * FROM crdb_internal.node_build_info
 
 # Tests for 'large_full_scan_rows'.
 statement ok

pkg/sql/opt/exec/execbuilder/testdata/explain

@@ -121,13 +121,45 @@ distribution: local
 vectorized: true
 ·
 • sort
-│ order: -column20,-created
+│ order: -column70,-created
 │
 └── • render
     │
+    └── • union all
         │
-        └── • virtual table
-              table: jobs@primary
+        ├── • render
+        │   │
+        │   └── • merge join (left outer)
+        │       │ equality: (id) = (job_id)
+        │       │
+        │       ├── • merge join (right outer)
+        │       │   │ equality: (job_id) = (id)
+        │       │   │ right cols are key
+        │       │   │
+        │       │   ├── • scan
+        │       │   │     missing stats
+        │       │   │     table: job_progress@primary
+        │       │   │     spans: FULL SCAN
+        │       │   │
+        │       │       │
+        │       │       └── • scan
+        │       │             missing stats
+        │       │             table: jobs@primary
+        │       │             spans: FULL SCAN
+        │       │
+        │       └── • scan
+        │             missing stats
+        │             table: job_status@primary
+        │             spans: FULL SCAN
+        │
+        └── • render
+            │
+                │ estimated row count: 3
+                │
+                └── • project set
+                    │ estimated row count: 10
+                    │
+                    └── • emptyrow
 
 statement ok
 CREATE INDEX a ON foo(x)

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: 366
+        │       │                     actual row count: 363
         │       │                     execution time: 0µs
         │       │                     table: pg_class@primary
         │       │