Merge #148464

148464: jobs: use builtin to synthesize pending session job rows in vtable r=dt a=dt

This adds a builtin to enumerate jobs pending creation in the current session.
It then adds a UNION with a call to this builtin, rather than directly accessing
the session in the Go code, to the body of the crdb_internal.jobs vtable, bringing
the vtable closer to being just a pure query that can be replaced by
a delegate or view.

Additionally, deprecated columns in the vtable, which were always null, are removed.

Release note: none.
Epic: none.a

Co-authored-by: David Taylor <[email protected]>

Author: craig[bot]

pkg/cli/zip_table_registry.go

@@ -382,25 +382,21 @@ var zipInternalTablesPerCluster = DebugZipTableRegistry{
 	},
 	// `statement` column can contain customer URI params such as
 	// AWS_ACCESS_KEY_ID.
-	// `error`, `execution_errors`, and `execution_events` columns contain
-	// error text that may contain sensitive data.
+	// `error` column contains error text that may contain sensitive data.
 	"crdb_internal.jobs": {
 		nonSensitiveCols: NonSensitiveColumns{
 			"job_id",
 			"job_type",
 			"description",
 			"user_name",
-			"descriptor_ids",
 			"status",
 			"running_status",
 			"created",
-			"started",
 			"finished",
 			"modified",
 			"fraction_completed",
 			"high_water_timestamp",
 			"coordinator_id",
-			"trace_id",
 		},
 	},
 	"crdb_internal.system_jobs": {

pkg/jobs/jobspb/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
     deps = [
         "//pkg/base",
         "//pkg/roachpb",
+        "//pkg/security/username",
         "//pkg/sql/catalog/catpb",
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/protoreflect",

pkg/jobs/jobspb/jobs.go

@@ -11,6 +11,7 @@ import (
 	"slices"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
@@ -23,6 +24,14 @@ type JobID = catpb.JobID
 // InvalidJobID is the zero value for JobID corresponding to no job.
 const InvalidJobID = catpb.InvalidJobID
 
+// PendingJob represents a job that is pending creation (e.g. in a session).
+type PendingJob struct {
+	JobID       JobID
+	Type        Type
+	Description string
+	Username    username.SQLUsername
+}
+
 // ToText implements the ProtobinExecutionDetailFile interface.
 func (t *TraceData) ToText() []byte {
 	rec := tracingpb.Recording(t.CollectedSpans)

pkg/server/admin.go

@@ -2271,7 +2271,6 @@ SELECT
   fraction_completed,
   high_water_timestamp,
   error,
-  execution_events::string,
   coordinator_id
 FROM crdb_internal.jobs
 WHERE true`) // Simplifies filter construction below.
@@ -2368,7 +2367,6 @@ func scanRowIntoJob(scanner resultScanner, row tree.Datums, job *serverpb.JobRes
 	var fractionCompletedOrNil *float32
 	var highwaterOrNil *apd.Decimal
 	var runningStatusOrNil *string
-	var executionFailuresOrNil *string
 	var coordinatorOrNil *int64
 	if err := scanner.ScanAll(
 		row,
@@ -2385,7 +2383,6 @@ func scanRowIntoJob(scanner resultScanner, row tree.Datums, job *serverpb.JobRes
 		&fractionCompletedOrNil,
 		&highwaterOrNil,
 		&job.Error,
-		&executionFailuresOrNil,
 		&coordinatorOrNil,
 	); err != nil {
 		return errors.Wrap(err, "scan")
@@ -2438,7 +2435,7 @@ func jobHelper(
 	const query = `
 	        SELECT job_id, job_type, description, statement, user_name, status,
 	  						 running_status, created, finished, modified,
-	  						 fraction_completed, high_water_timestamp, error, execution_events::string, coordinator_id
+	  						 fraction_completed, high_water_timestamp, error, coordinator_id
 	          FROM crdb_internal.jobs
 	         WHERE job_id = $1`
 	row, cols, err := sqlServer.internalExecutor.QueryRowExWithCols(

pkg/sql/crdb_internal.go

@@ -1152,20 +1152,15 @@ CREATE TABLE crdb_internal.jobs (
   description           STRING,
   statement             STRING,
   user_name             STRING,
-  descriptor_ids        INT[],
   status                STRING,
   running_status        STRING,
   created               TIMESTAMPTZ,
-  started               TIMESTAMPTZ,
   finished              TIMESTAMPTZ,
   modified              TIMESTAMPTZ,
   fraction_completed    FLOAT,
   high_water_timestamp  DECIMAL,
   error                 STRING,
   coordinator_id        INT,
-  trace_id              INT,
-  execution_errors      STRING[],
-  execution_events      JSONB,
   INDEX(job_id),
   INDEX(status),
   INDEX(job_type)
@@ -1223,8 +1218,11 @@ 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
+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...)
@@ -1237,108 +1235,60 @@ LEFT OUTER JOIN system.public.job_status AS s ON j.id = s.job_id
 		}
 	}()
 
-	sessionJobs := make([]*jobs.Record, 0, p.extendedEvalCtx.jobs.numToCreate())
-	uniqueJobs := make(map[*jobs.Record]struct{})
-	if err := p.extendedEvalCtx.jobs.forEachToCreate(func(job *jobs.Record) error {
-		if _, ok := uniqueJobs[job]; ok {
-			return nil
-		}
-		sessionJobs = append(sessionJobs, job)
-		uniqueJobs[job] = struct{}{}
-		return nil
-	}); err != nil {
-		return emitted, err
-	}
-
 	// Loop while we need to skip a row.
 	for {
 		ok, err := it.Next(ctx)
-		if err != nil {
+		if err != nil || !ok {
 			return emitted, err
 		}
-		// We will read the columns from the query on joined jobs tables into a wide
-		// row, and then copy the values from read rows into named variables to then
-		// use when emitting our output row. If we need to synthesize rows for jobs
-		// pending creation in the session, we'll do so in those same named vars to
-		// keep things organized.
-		//   0,      1,    2,        3,     4,      5,       6,        7,        8,        9,       10,       11,         12
-		var id, typStr, desc, ownerStr, state, status, created, finished, modified, fraction, resolved, errorMsg, instanceID tree.Datum
+		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]
 
-		if ok {
-			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))
+		typ, err := jobspb.TypeFromString(string(tree.MustBeDString(typStr)))
+		if err != nil {
+			return emitted, err
+		}
 
-			owner := username.MakeSQLUsernameFromPreNormalizedString(string(tree.MustBeDString(ownerStr)))
-			jobID := jobspb.JobID(tree.MustBeDInt(id))
-			typ, err := jobspb.TypeFromString(string(tree.MustBeDString(typStr)))
+		getLegacyPayloadForAuth := func(ctx context.Context) (*jobspb.Payload, error) {
+			if !enablePerJobDetailedAuthLookups.Get(&p.EvalContext().Settings.SV) {
+				return nil, errLegacyPerJobAuthDisabledSentinel
+			}
+			if p.EvalContext().Settings.Version.IsActive(ctx, clusterversion.V25_1) {
+				log.Warningf(ctx, "extended job access control based on job-specific details is deprecated and can make SHOW JOBS less performant; consider disabling %s",
+					enablePerJobDetailedAuthLookups.Name())
+				p.BufferClientNotice(ctx,
+					pgnotice.Newf("extended job access control based on job-specific details has been deprecated and can make SHOW JOBS less performant; consider disabling %s",
+						enablePerJobDetailedAuthLookups.Name()))
+			}
+			payload := &jobspb.Payload{}
+			infoStorage := jobs.InfoStorageForJob(p.InternalSQLTxn(), jobID)
+			payloadBytes, exists, err := infoStorage.GetLegacyPayload(ctx, "getLegacyPayload-for-custom-auth")
 			if err != nil {
-				return emitted, err
+				return nil, err
 			}
-
-			getLegacyPayloadForAuth := func(ctx context.Context) (*jobspb.Payload, error) {
-				if !enablePerJobDetailedAuthLookups.Get(&p.EvalContext().Settings.SV) {
-					return nil, errLegacyPerJobAuthDisabledSentinel
-				}
-				if p.EvalContext().Settings.Version.IsActive(ctx, clusterversion.V25_1) {
-					log.Warningf(ctx, "extended job access control based on job-specific details is deprecated and can make SHOW JOBS less performant; consider disabling %s",
-						enablePerJobDetailedAuthLookups.Name())
-					p.BufferClientNotice(ctx,
-						pgnotice.Newf("extended job access control based on job-specific details has been deprecated and can make SHOW JOBS less performant; consider disabling %s",
-							enablePerJobDetailedAuthLookups.Name()))
-				}
-				payload := &jobspb.Payload{}
-				infoStorage := jobs.InfoStorageForJob(p.InternalSQLTxn(), jobID)
-				payloadBytes, exists, err := infoStorage.GetLegacyPayload(ctx, "getLegacyPayload-for-custom-auth")
-				if err != nil {
-					return nil, err
-				}
-				if !exists {
-					return nil, errors.New("job payload not found in system.job_info")
-				}
-				if err := protoutil.Unmarshal(payloadBytes, payload); err != nil {
-					return nil, err
-				}
-				return payload, nil
+			if !exists {
+				return nil, errors.New("job payload not found in system.job_info")
 			}
-			if errorMsg == tree.DNull {
-				errorMsg = emptyString
+			if err := protoutil.Unmarshal(payloadBytes, payload); err != nil {
+				return nil, err
 			}
+			return payload, nil
+		}
+		if errorMsg == tree.DNull {
+			errorMsg = emptyString
+		}
 
-			if err := jobsauth.AuthorizeAllowLegacyAuth(
-				ctx, p, jobID, getLegacyPayloadForAuth, owner, typ, jobsauth.ViewAccess, globalPrivileges,
-			); err != nil {
-				// Filter out jobs which the user is not allowed to see.
-				if IsInsufficientPrivilegeError(err) {
-					continue
-				}
-				return emitted, err
-			}
-		} else if !ok {
-			if len(sessionJobs) == 0 {
-				return emitted, nil
+		if err := jobsauth.AuthorizeAllowLegacyAuth(
+			ctx, p, jobID, getLegacyPayloadForAuth, owner, typ, jobsauth.ViewAccess, globalPrivileges,
+		); err != nil {
+			// Filter out jobs which the user is not allowed to see.
+			if IsInsufficientPrivilegeError(err) {
+				continue
 			}
-			job := sessionJobs[len(sessionJobs)-1]
-			sessionJobs = sessionJobs[:len(sessionJobs)-1]
-			payloadType, err := jobspb.DetailsType(jobspb.WrapPayloadDetails(job.Details))
-			if err != nil {
-				return emitted, err
-			}
-			// synthesize the fields we'd read from the jobs table if this job were in it.
-			id, typStr, desc, ownerStr, state, status, created, finished, modified, fraction, resolved, errorMsg, instanceID =
-				tree.NewDInt(tree.DInt(job.JobID)),
-				tree.NewDString(payloadType.String()),
-				tree.NewDString(job.Description),
-				tree.NewDString(job.Username.Normalized()),
-				tree.NewDString(string(jobs.StatePending)),
-				tree.DNull,
-				tree.MustMakeDTimestampTZ(p.txn.ReadTimestamp().GoTime(), time.Microsecond),
-				tree.DNull,
-				tree.MustMakeDTimestampTZ(p.txn.ReadTimestamp().GoTime(), time.Microsecond),
-				tree.NewDFloat(tree.DFloat(0)),
-				tree.DZeroDecimal,
-				tree.DNull,
-				tree.NewDInt(tree.DInt(p.extendedEvalCtx.ExecCfg.JobRegistry.ID()))
+			return emitted, err
 		}
 
 		if err = addRow(
@@ -1347,20 +1297,15 @@ LEFT OUTER JOIN system.public.job_status AS s ON j.id = s.job_id
 			desc,
 			desc,
 			ownerStr,
-			tree.DNull, // deperecated "descriptor_ids"
 			state,
 			status,
 			created,
-			created, // deprecated "started" field.
 			finished,
 			modified,
 			fraction,
 			resolved,
 			errorMsg,
 			instanceID,
-			tree.DNull, // deprecated "trace_id" field.
-			tree.DNull, // deprecated "executionErrors" field.
-			tree.DNull, // deprecated "executionEvents" field.
 		); err != nil {
 			return emitted, err
 		}

pkg/sql/delegate/show_changefeed_jobs.go

@@ -37,7 +37,7 @@ SELECT
   status,
   running_status,
   created,
-  started,
+  created as started,
   finished,
   modified,
   high_water_timestamp,

pkg/sql/delegate/show_jobs.go

@@ -28,14 +28,10 @@ func constructSelectQuery(n *tree.ShowJobs) string {
 		baseQuery.WriteString(`description, statement, `)
 	}
 	baseQuery.WriteString(`user_name, status, running_status, `)
-	baseQuery.WriteString(`date_trunc('second', created) as created, date_trunc('second', started) as started, `)
+	baseQuery.WriteString(`date_trunc('second', created) as created, date_trunc('second', created) as started, `)
 	baseQuery.WriteString(`date_trunc('second', finished) as finished, date_trunc('second', modified) as modified, `)
 	baseQuery.WriteString(`fraction_completed, error, coordinator_id`)
 
-	if n.Jobs != nil {
-		baseQuery.WriteString(`, trace_id, execution_errors`)
-	}
-
 	// Check if there are any SHOW JOBS options that we need to add columns for.
 	if n.Options != nil {
 		if n.Options.ExecutionDetails {

pkg/sql/faketreeeval/evalctx.go

@@ -647,6 +647,13 @@ func (ep *DummySessionAccessor) HasViewActivityOrViewActivityRedactedRole(
 	return false, false, errors.WithStack(errEvalSessionVar)
 }
 
+func (ep *DummySessionAccessor) ForEachSessionPendingJob(
+	_ func(job jobspb.PendingJob) error,
+) error {
+	// This is a no-op in the dummy implementation.
+	return nil
+}
+
 // DummyClientNoticeSender implements the eval.ClientNoticeSender interface.
 type DummyClientNoticeSender struct{}
 

pkg/sql/jobs_collection.go

@@ -82,3 +82,21 @@ func (j *txnJobsCollection) forEachToCreate(fn func(jobRecord *jobs.Record) erro
 	}
 	return nil
 }
+
+func (p *planner) ForEachSessionPendingJob(fn func(job jobspb.PendingJob) error) error {
+	if p.extendedEvalCtx.jobs == nil {
+		return nil
+	}
+	return p.extendedEvalCtx.jobs.forEachToCreate(func(r *jobs.Record) error {
+		payloadType, err := jobspb.DetailsType(jobspb.WrapPayloadDetails(r.Details))
+		if err != nil {
+			return err
+		}
+		return fn(jobspb.PendingJob{
+			JobID:       r.JobID,
+			Description: r.Description,
+			Username:    r.Username,
+			Type:        payloadType,
+		})
+	})
+}

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

@@ -121,7 +121,7 @@ distribution: local
 vectorized: true
 ·
 • sort
-│ order: -column26,-started
+│ order: -column20,-created
 │
 └── • render
     │