sql: fix CREATE AS [SHOW CREATE FUNCTION] job failure

Fixes cockroachdb#106268

Previously `CREATE TABLE AS`/`CREATE MATERIALIZED VIEW AS` sourcing from
`SHOW CREATE FUNCTION <function>` generated a failing schema change job with a
`unknown function: f(): function undefined` error because the job runs in the
system database and functions cannot be referenced cross database.

This PR fixes this by using the original user session (which includes the
database) in the schema change job.

Release note (bug fix): Fixes the schema changer job when CREATE AS sources
from SHOW CREATE FUNCTION:
CREATE TABLE t AS SELECT * FROM [SHOW CREATE FUNCTION f];

Author: ecwall

pkg/jobs/jobspb/jobs.proto

@@ -887,7 +887,9 @@ message SchemaChangeDetails {
     (gogoproto.customtype) = "github.com/cockroachdb/cockroach/pkg/util/uuid.UUID"
   ];
 
-  // Next id 13.
+  sessiondatapb.SessionData session_data = 13;
+
+  // Next id 14.
 }
 
 message SchemaChangeProgress {

pkg/sql/create_as_test.go

@@ -129,7 +129,6 @@ func TestCreateAsShow(t *testing.T) {
 	testCases := []struct {
 		sql   string
 		setup string
-		skip  bool
 	}{
 		{
 			sql: "SHOW CLUSTER SETTINGS",
@@ -165,9 +164,6 @@ func TestCreateAsShow(t *testing.T) {
 		{
 			sql:   "SHOW CREATE FUNCTION show_create_fn",
 			setup: "CREATE FUNCTION show_create_fn(i int) RETURNS INT AS 'SELECT i' LANGUAGE SQL",
-			// TODO(sql-foundations): Fix `unknown function: show_create_fn(): function undefined` error in job.
-			//  See https://github.com/cockroachdb/cockroach/issues/106268.
-			skip: true,
 		},
 		{
 			sql: "SHOW CREATE ALL TYPES",
@@ -294,9 +290,6 @@ func TestCreateAsShow(t *testing.T) {
 
 	for i, testCase := range testCases {
 		t.Run(testCase.sql, func(t *testing.T) {
-			if testCase.skip {
-				return
-			}
 			if testCase.setup != "" {
 				if s.StartedDefaultTestTenant() && strings.Contains(testCase.setup, "create_tenant") {
 					// Only the system tenant has the ability to create other

pkg/sql/delegate/show_function.go

@@ -27,12 +27,20 @@ FROM crdb_internal.create_function_statements
 WHERE schema_name = %[1]s
 AND function_name = %[2]s
 `
-	un, ok := n.Name.FunctionReference.(*tree.UnresolvedName)
+	resolvableFunctionReference := &n.Name
+	un, ok := resolvableFunctionReference.FunctionReference.(*tree.UnresolvedName)
 	if !ok {
 		return nil, errors.AssertionFailedf("not a valid function name")
 	}
 
-	fn, err := d.catalog.ResolveFunction(d.ctx, un, &d.evalCtx.SessionData().SearchPath)
+	searchPath := &d.evalCtx.SessionData().SearchPath
+	var fn *tree.ResolvedFunctionDefinition
+	var err error
+	if d.qualifyDataSourceNamesInAST {
+		fn, err = resolvableFunctionReference.Resolve(d.ctx, searchPath, d.catalog)
+	} else {
+		fn, err = d.catalog.ResolveFunction(d.ctx, un, searchPath)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -41,12 +49,26 @@ AND function_name = %[2]s
 	for _, o := range fn.Overloads {
 		if o.IsUDF {
 			udfSchema = o.Schema
+			break
 		}
 	}
 	if udfSchema == "" {
 		return nil, errors.Errorf("function %s does not exist", tree.AsString(un))
 	}
 
+	if d.qualifyDataSourceNamesInAST {
+		referenceByName := resolvableFunctionReference.ReferenceByName
+		if !referenceByName.HasExplicitSchema() {
+			referenceByName.Parts[1] = udfSchema
+		}
+		if !referenceByName.HasExplicitCatalog() {
+			referenceByName.Parts[2] = d.evalCtx.SessionData().Database
+		}
+		if referenceByName.NumParts < 3 {
+			referenceByName.NumParts = 3
+		}
+	}
+
 	fullQuery := fmt.Sprintf(query, lexbase.EscapeSQLString(udfSchema), lexbase.EscapeSQLString(un.Parts[0]))
 	return d.parse(fullQuery)
 }

pkg/sql/drop_function.go

@@ -279,6 +279,7 @@ func (p *planner) writeDropFuncSchemaChange(ctx context.Context, funcDesc *funcd
 		DescriptorIDs: descpb.IDs{funcDesc.ID},
 		Details: jobspb.SchemaChangeDetails{
 			DroppedFunctions: descpb.IDs{funcDesc.ID},
+			SessionData:      &p.SessionData().SessionData,
 		},
 		Progress: jobspb.TypeSchemaChangeProgress{},
 	}

pkg/sql/drop_schema.go

@@ -258,6 +258,7 @@ func (p *planner) createDropSchemaJob(
 			// The version distinction for database jobs doesn't matter for jobs that
 			// drop schemas.
 			FormatVersion: jobspb.DatabaseJobFormatVersion,
+			SessionData:   &p.SessionData().SessionData,
 		},
 		Progress:      jobspb.SchemaChangeProgress{},
 		NonCancelable: true,

pkg/sql/logictest/testdata/logic_test/vectorize

@@ -3,6 +3,8 @@
 # Note that there is no much benefit from running this file with other logic
 # test configurations because we override vectorize setting.
 
+subtest todo1
+
 # NOTE: all queries in this file should run with vectorize=experimental_always
 # unless they are known to be unsupported (like generate_series, etc). If you do
 # need to execute an unsupported query, follow the pattern of
@@ -605,6 +607,14 @@ SELECT EXTRACT(YEAR FROM x) FROM extract_test
 ----
 2017
 
+statement ok
+RESET vectorize
+
+subtest end
+
+subtest 38937
+
+
 # Regression test for #38937
 statement ok
 CREATE TABLE t38937 (_int2) AS SELECT 1::INT2;
@@ -614,6 +624,13 @@ SELECT sum_int(_int2) FROM t38937
 ----
 1
 
+subtest end
+
+subtest todo2
+
+statement ok
+SET vectorize=experimental_always
+
 # Regression tests for #38959
 
 statement ok
@@ -1167,6 +1184,13 @@ SELECT crdb_internal_mvcc_timestamp IS NOT NULL FROM mvcc@i
 ----
 true
 
+statement ok
+RESET vectorize
+
+subtest end
+
+subtest 51841
+
 # Regression test for builtin funcs that take no arguments, combined with
 # data that can't be safely decoded as input.
 
@@ -1176,6 +1200,10 @@ CREATE TABLE t51841 (a) AS SELECT gen_random_uuid();
 statement ok
 SELECT random() from t51841
 
+subtest end
+
+subtest todo3
+
 # Regression test to ensure that we can EXPLAIN (VEC) plans that contain user
 # defined types. We aren't interested in the output here, but are just ensuring
 # that we don't panic.
@@ -1247,6 +1275,13 @@ SELECT b FROM t66706@u WHERE NOT (b = 'foo')
 bar
 bar
 
+statement ok
+RESET vectorize
+
+subtest end
+
+subtest 68040
+
 # Regression test for ignoring the escaping in the LIKE pattern (#68040).
 statement ok
 CREATE TABLE t68040 (c) AS SELECT 'string with \ backslash'
@@ -1256,6 +1291,13 @@ SELECT c FROM t68040 WHERE c LIKE '%\\%'
 ----
 string with \ backslash
 
+subtest end
+
+subtest todo4
+
+statement ok
+SET vectorize=experimental_always
+
 # Regression test for #68979. The IN operator should evaluate correctly when the
 # tuple contents are not sorted by the optimizer.
 statement ok
@@ -1296,3 +1338,8 @@ query I
 SELECT _int2 * _int2 FROM ints WHERE _int4 + _int4 = _int8 + 2
 ----
 4
+
+statement ok
+RESET vectorize
+
+subtest end

pkg/sql/schema.go

@@ -73,6 +73,7 @@ func (p *planner) writeSchemaDescChange(
 				// The version distinction for database jobs doesn't matter for schema
 				// jobs.
 				FormatVersion: jobspb.DatabaseJobFormatVersion,
+				SessionData:   &p.SessionData().SessionData,
 			},
 			Progress:      jobspb.SchemaChangeProgress{},
 			NonCancelable: true,

pkg/sql/schema_changer.go

@@ -51,6 +51,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqltelemetry"
 	"github.com/cockroachdb/cockroach/pkg/storage"
@@ -156,6 +157,7 @@ type SchemaChanger struct {
 	clock                *hlc.Clock
 	settings             *cluster.Settings
 	execCfg              *ExecutorConfig
+	sessionData          *sessiondatapb.SessionData
 }
 
 // NewSchemaChangerForTesting only for tests.
@@ -305,6 +307,8 @@ func (sc *SchemaChanger) backfillQueryIntoTable(
 			return err
 		}
 
+		sd := NewInternalSessionData(ctx, sc.execCfg.Settings, "backfillQueryIntoTable")
+		sd.SessionData = *sc.sessionData
 		// Create an internal planner as the planner used to serve the user query
 		// would have committed by this point.
 		p, cleanup := NewInternalPlanner(
@@ -313,7 +317,7 @@ func (sc *SchemaChanger) backfillQueryIntoTable(
 			username.RootUserName(),
 			&MemoryMetrics{},
 			sc.execCfg,
-			NewInternalSessionData(ctx, sc.execCfg.Settings, "backfillQueryIntoTable"),
+			sd,
 		)
 
 		defer cleanup()
@@ -2155,6 +2159,7 @@ func (sc *SchemaChanger) updateJobForRollback(
 			TableMutationID: sc.mutationID,
 			ResumeSpanList:  spanList,
 			FormatVersion:   oldDetails.FormatVersion,
+			SessionData:     sc.sessionData,
 		},
 	); err != nil {
 		return err
@@ -2611,6 +2616,7 @@ func (r schemaChangeResumer) Resume(ctx context.Context, execCtx interface{}) er
 			settings:             p.ExecCfg().Settings,
 			execCfg:              p.ExecCfg(),
 			metrics:              p.ExecCfg().SchemaChangerMetrics,
+			sessionData:          details.SessionData,
 		}
 		opts := retry.Options{
 			InitialBackoff: 20 * time.Millisecond,
@@ -2828,6 +2834,7 @@ func (r schemaChangeResumer) OnFailOrCancel(
 		clock:                p.ExecCfg().Clock,
 		settings:             p.ExecCfg().Settings,
 		execCfg:              p.ExecCfg(),
+		sessionData:          details.SessionData,
 	}
 
 	if r.job.Payload().FinalResumeError == nil {
@@ -2931,6 +2938,7 @@ func (sc *SchemaChanger) queueCleanupJob(
 				// The version distinction for database jobs doesn't matter for jobs on
 				// tables.
 				FormatVersion: jobspb.DatabaseJobFormatVersion,
+				SessionData:   sc.sessionData,
 			},
 			Progress:      jobspb.SchemaChangeProgress{},
 			NonCancelable: true,

pkg/sql/table.go

@@ -71,6 +71,7 @@ func (p *planner) createDropDatabaseJob(
 			DroppedFunctions:  funcIDs,
 			DroppedDatabaseID: databaseID,
 			FormatVersion:     jobspb.DatabaseJobFormatVersion,
+			SessionData:       &p.SessionData().SessionData,
 		},
 		Progress:      jobspb.SchemaChangeProgress{},
 		NonCancelable: true,
@@ -92,6 +93,7 @@ func (p *planner) createNonDropDatabaseChangeJob(
 		Details: jobspb.SchemaChangeDetails{
 			DescID:        databaseID,
 			FormatVersion: jobspb.DatabaseJobFormatVersion,
+			SessionData:   &p.SessionData().SessionData,
 		},
 		Progress:      jobspb.SchemaChangeProgress{},
 		NonCancelable: true,
@@ -163,6 +165,7 @@ func (p *planner) createOrUpdateSchemaChangeJob(
 				// The version distinction for database jobs doesn't matter for jobs on
 				// tables.
 				FormatVersion: jobspb.DatabaseJobFormatVersion,
+				SessionData:   &p.SessionData().SessionData,
 			},
 			Progress: jobspb.SchemaChangeProgress{},
 			// Mark jobs without a mutation ID as non-cancellable,
@@ -193,6 +196,7 @@ func (p *planner) createOrUpdateSchemaChangeJob(
 		// The version distinction for database jobs doesn't matter for jobs on
 		// tables.
 		FormatVersion: jobspb.DatabaseJobFormatVersion,
+		SessionData:   &p.SessionData().SessionData,
 	}
 	if oldDetails.TableMutationID != descpb.InvalidMutationID {
 		// The previous queued schema change job was associated with a mutation,