builtins: add builtin to show jobs pending creation in session

Previously these were only accessible by materializing the whole crdb_internal.jobs vtable.

If/when we remove that vtable or replace it with a view, we'll still need to read these pending
jobs from memory, but we don't need a whole vtable for that and can just use a builtin instead.

Release note: none.
Epic: none.

Author: dt

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/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
     │

pkg/sql/sem/builtins/fixed_oids.go

@@ -2665,6 +2665,7 @@ var builtinOidsArray = []string{
 	2702: `crdb_internal.force_retry(val: int) -> int`,
 	2703: `crdb_internal.show_create_all_routines(database_name: string) -> string`,
 	2704: `crdb_internal.show_create_all_triggers(database_name: string) -> string`,
+	2705: `crdb_internal.session_pending_jobs() -> tuple{int AS job_id, string AS job_type, string AS description, string AS user_name}`,
 }
 
 var builtinOidsBySignature map[string]oid.Oid

pkg/sql/sem/builtins/generator_builtins.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/build"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
@@ -633,6 +634,19 @@ The output can be used to recreate a database.'
 			volatility.Volatile,
 		),
 	),
+	"crdb_internal.session_pending_jobs": makeBuiltin(
+		tree.FunctionProperties{
+			Category:         builtinconstants.CategorySystemInfo,
+			DistsqlBlocklist: true, // applicable only on the gateway
+		},
+		makeGeneratorOverload(
+			tree.ParamTypes{},
+			sessionPendingJobsType,
+			makeSessionPendingJobsGenerator,
+			`Returns rows of information about all pending jobs created in the session txn.`,
+			volatility.Volatile,
+		),
+	),
 	"crdb_internal.decode_plan_gist": makeBuiltin(
 		tree.FunctionProperties{
 			Category:         builtinconstants.CategorySystemInfo,
@@ -2875,6 +2889,10 @@ var showCreateAllTriggersGeneratorType = types.String
 var showCreateAllTypesGeneratorType = types.String
 var showCreateAllTablesGeneratorType = types.String
 var showCreateAllRoutinesGeneratorType = types.String
+var sessionPendingJobsType = types.MakeLabeledTuple(
+	[]*types.T{types.Int, types.String, types.String, types.String},
+	[]string{"job_id", "job_type", "description", "user_name"},
+)
 
 // Phase is used to determine if CREATE statements or ALTER statements
 // are being generated for showCreateAllTables.
@@ -3372,6 +3390,53 @@ func makeShowCreateAllRoutinesGenerator(
 	}, nil
 }
 
+func makeSessionPendingJobsGenerator(
+	ctx context.Context, evalCtx *eval.Context, args tree.Datums,
+) (eval.ValueGenerator, error) {
+	records := []jobspb.PendingJob{{}} // Next() always pops first, so pad a zero.
+	if err := evalCtx.SessionAccessor.ForEachSessionPendingJob(func(r jobspb.PendingJob) error {
+		records = append(records, r)
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	return &sessionPendingJobsGenerator{
+		jobs: records,
+	}, nil
+}
+
+type sessionPendingJobsGenerator struct {
+	jobs []jobspb.PendingJob
+}
+
+// ResolvedType implements the eval.ValueGenerator interface.
+func (s *sessionPendingJobsGenerator) ResolvedType() *types.T {
+	return sessionPendingJobsType
+}
+
+// Start implements the eval.ValueGenerator interface.
+func (s *sessionPendingJobsGenerator) Start(ctx context.Context, txn *kv.Txn) error {
+	return nil
+}
+
+func (s *sessionPendingJobsGenerator) Next(ctx context.Context) (bool, error) {
+	s.jobs = s.jobs[1:]
+	return len(s.jobs) > 0, nil
+}
+
+// Values implements the eval.ValueGenerator interface.
+func (s *sessionPendingJobsGenerator) Values() (tree.Datums, error) {
+	return tree.Datums{tree.NewDInt(tree.DInt(s.jobs[0].JobID)),
+		tree.NewDString(s.jobs[0].Type.String()),
+		tree.NewDString(s.jobs[0].Description),
+		tree.NewDString(s.jobs[0].Username.Normalized()),
+	}, nil
+}
+
+// Close implements the eval.ValueGenerator interface.
+func (s *sessionPendingJobsGenerator) Close(ctx context.Context) {
+}
+
 // identGenerator supports the execution of
 // crdb_internal.gen_rand_ident().
 type identGenerator struct {

pkg/sql/sem/eval/deps.go

@@ -530,6 +530,11 @@ type SessionAccessor interface {
 	// HasViewActivityOrViewActivityRedactedRole returns true iff the current session user has the
 	// VIEWACTIVITY or VIEWACTIVITYREDACTED permission.
 	HasViewActivityOrViewActivityRedactedRole(ctx context.Context) (bool, bool, error)
+
+	// ForEachSessionPendingJob calls the provided function for each pending job
+	// created in the session (hidden behind the generic interface{} to avoid
+	// circular dependencies, but the caller can cast it to jobs.Record).
+	ForEachSessionPendingJob(fn func(record jobspb.PendingJob) error) error
 }
 
 // PreparedStatementState is a limited interface that exposes metadata about