fixup! importer: set up FunctionResolver for computed column type-checking

Move FunctionResolver setup from makeInputConverter (shared semaCtx) to
NewDatumRowConverter (per-instance) to eliminate a data race. Multiple
parallel import workers were sharing the same semaCtx.FunctionResolver
backed by a *descs.Collection, causing concurrent access to
leasedDescriptors.maybeInitReadTimestamp.

Each DatumRowConverter now creates its own bare-bones descs.Collection
inside a short-lived txn, mirroring the existing per-instance sequence
resolution pattern. Since bare-bones collections lack a lease manager,
we use ByIDWithoutLeased via the new NewDistSQLFunctionResolverFromGetter
constructor.

This also simplifies the makeInputConverter signature back to accepting
*kv.DB instead of descs.DB + *descs.Collection.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: mw5h

pkg/sql/catalog/descs/dist_sql_function_resolver.go

@@ -34,6 +34,14 @@ func NewDistSQLFunctionResolver(descs *Collection, txn *kv.Txn) *DistSQLFunction
 	}
 }
 
+// NewDistSQLFunctionResolverFromGetter returns a new DistSQLFunctionResolver
+// using the provided ByIDGetter directly. This is useful when the caller
+// needs to control how descriptors are looked up, for example using
+// ByIDWithoutLeased on a bare-bones collection that lacks a lease manager.
+func NewDistSQLFunctionResolverFromGetter(g ByIDGetter) *DistSQLFunctionResolver {
+	return &DistSQLFunctionResolver{g: g}
+}
+
 // ResolveFunction implements tree.FunctionReferenceResolver interface.
 // It only resolves builtin functions.
 // TODO(chengxiong): extract resolution logics in schema_resolver.go into

pkg/sql/importer/import_processor.go

@@ -13,14 +13,14 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/bulksst"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
-	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
@@ -259,8 +259,7 @@ func makeInputConverter(
 	evalCtx *eval.Context,
 	kvCh chan row.KVBatch,
 	seqChunkProvider *row.SeqChunkProvider,
-	descsDB descs.DB,
-	descriptors *descs.Collection,
+	db *kv.DB,
 ) (inputConverter, error) {
 	if len(spec.Tables) > 1 {
 		return nil, errors.AssertionFailedf("%s only supports reading a single, pre-specified table", spec.Format.Format.String())
@@ -290,32 +289,6 @@ func makeInputConverter(
 		}
 	}
 
-	// If the table has computed columns, set up a FunctionResolver so that
-	// any user-defined functions referenced in computed expressions can be
-	// resolved during type-checking. This mirrors the pattern used by
-	// ColumnBackfiller.InitForDistributedUse.
-	hasComputedCols := false
-	for _, col := range desc.WritableColumns() {
-		if col.IsComputed() {
-			hasComputedCols = true
-			break
-		}
-	}
-	if hasComputedCols {
-		if err := descsDB.DescsTxn(ctx, func(ctx context.Context, txn descs.Txn) error {
-			semaCtx.FunctionResolver = descs.NewDistSQLFunctionResolver(txn.Descriptors(), txn.KV())
-			return nil
-		}); err != nil {
-			return nil, err
-		}
-		// Release leases on any accessed descriptors now that the
-		// resolver is configured. We do this so that leases are not
-		// held for the entire import process.
-		if descriptors != nil {
-			descriptors.ReleaseAll(ctx)
-		}
-	}
-
 	readerParallelism := int(spec.ReaderParallelism)
 	if readerParallelism <= 0 {
 		readerParallelism = int(readerParallelismSetting.Get(&evalCtx.Settings.SV))
@@ -324,7 +297,6 @@ func makeInputConverter(
 		readerParallelism = runtime.GOMAXPROCS(0)
 	}
 
-	db := descsDB.KV()
 	switch spec.Format.Format {
 	case roachpb.IOFileFormat_CSV:
 		isWorkload := true

pkg/sql/importer/import_processor_test.go

@@ -95,7 +95,7 @@ func TestConverterFlushesBatches(t *testing.T) {
 	params := base.TestServerArgs{}
 	s := serverutils.StartServerOnly(t, params)
 	defer s.Stopper().Stop(ctx)
-	descsDB := s.ApplicationLayer().InternalDB().(descs.DB)
+	kvDB := s.ApplicationLayer().DB()
 
 	tests := []testSpec{
 		newTestSpec(ctx, t, csvFormat(), "testdata/csv/data-0"),
@@ -122,7 +122,7 @@ func TestConverterFlushesBatches(t *testing.T) {
 				kvCh := make(chan row.KVBatch, batchSize)
 				semaCtx := tree.MakeSemaContext(nil /* resolver */)
 				conv, err := makeInputConverter(ctx, &semaCtx, converterSpec, &evalCtx, kvCh,
-					nil /* seqChunkProvider */, descsDB, nil /* descriptors */)
+					nil /* seqChunkProvider */, kvDB)
 				if err != nil {
 					t.Fatalf("makeInputConverter() error = %v", err)
 				}

pkg/sql/importer/read_import_base.go

@@ -91,7 +91,7 @@ func runImport(
 	semaCtx := tree.MakeSemaContext(importResolver)
 	conv, convErr := makeInputConverter(
 		ctx, &semaCtx, spec, evalCtx, kvCh, seqChunkProvider,
-		flowCtx.Cfg.DB, flowCtx.Descriptors,
+		flowCtx.Cfg.DB.KV(),
 	)
 	if convErr != nil {
 		return nil, convErr

pkg/sql/row/row_converter.go

@@ -484,20 +484,65 @@ func NewDatumRowConverter(
 		// MakeComputedExprs to map that of Datums.
 		colsOrdered[ri.InsertColIDtoRowIndex.GetDefault(col.GetID())] = col
 	}
-	// Here, computeExprs will be nil if there's no computed column, or
-	// the list of computed expressions (including nil, for those columns
-	// that are not computed) otherwise, according to colsOrdered.
-	c.computedExprs, _, err = schemaexpr.MakeComputedExprs(
-		ctx,
-		colsOrdered,
-		c.tableDesc.PublicColumns(),
-		c.tableDesc,
-		tree.NewUnqualifiedTableName(tree.Name(c.tableDesc.GetName())),
-		c.EvalCtx,
-		c.SemaCtx,
-	)
-	if err != nil {
-		return nil, errors.Wrapf(err, "error type checking and building computed expression for IMPORT INTO")
+
+	// If any computed columns reference UDFs, we need a FunctionResolver
+	// to resolve them by OID during type-checking. Set up a per-instance
+	// resolver using a bare-bones descs.Collection inside a short-lived
+	// txn, and call MakeComputedExprs within the txn so the resolver
+	// remains valid. Each DatumRowConverter instance gets its own
+	// resolver to avoid races when multiple import workers run in
+	// parallel. This mirrors the sequence resolution pattern above.
+	hasComputedCols := false
+	for _, col := range colsOrdered {
+		if col != nil && col.IsComputed() {
+			hasComputedCols = true
+			break
+		}
+	}
+	if hasComputedCols && c.SemaCtx.FunctionResolver == nil && c.db != nil {
+		cf := descs.NewBareBonesCollectionFactory(evalCtx.Settings, evalCtx.Codec)
+		dsdp := catsessiondata.NewDescriptorSessionDataStackProvider(evalCtx.SessionDataStack)
+		descsCol := cf.NewCollection(ctx, descs.WithDescriptorSessionDataProvider(dsdp))
+		defer descsCol.ReleaseAll(ctx)
+		err = c.db.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+			if err := txn.SetFixedTimestamp(ctx, hlc.Timestamp{WallTime: evalCtx.TxnTimestamp.UnixNano()}); err != nil {
+				return err
+			}
+			// Use ByIDWithoutLeased because the bare-bones collection
+			// does not have a lease manager.
+			c.SemaCtx.FunctionResolver = descs.NewDistSQLFunctionResolverFromGetter(
+				descsCol.ByIDWithoutLeased(txn).Get(),
+			)
+			c.computedExprs, _, err = schemaexpr.MakeComputedExprs(
+				ctx,
+				colsOrdered,
+				c.tableDesc.PublicColumns(),
+				c.tableDesc,
+				tree.NewUnqualifiedTableName(tree.Name(c.tableDesc.GetName())),
+				c.EvalCtx,
+				c.SemaCtx,
+			)
+			return err
+		})
+		if err != nil {
+			return nil, errors.Wrapf(err, "error type checking and building computed expression for IMPORT INTO")
+		}
+	} else {
+		// Here, computeExprs will be nil if there's no computed column, or
+		// the list of computed expressions (including nil, for those columns
+		// that are not computed) otherwise, according to colsOrdered.
+		c.computedExprs, _, err = schemaexpr.MakeComputedExprs(
+			ctx,
+			colsOrdered,
+			c.tableDesc.PublicColumns(),
+			c.tableDesc,
+			tree.NewUnqualifiedTableName(tree.Name(c.tableDesc.GetName())),
+			c.EvalCtx,
+			c.SemaCtx,
+		)
+		if err != nil {
+			return nil, errors.Wrapf(err, "error type checking and building computed expression for IMPORT INTO")
+		}
 	}
 
 	// Here, partialIndexExprs will be nil if there are no partial indexes, or a