importccl,workloadccl: plumb semaCtx with type resolver to workload

This change plumbs the semaCtx that has a type resolver through
to the import workload workers. This allows for resolution of
UDTs in tables being imported into via workload fixtures.

Release note: None

pkg/ccl/importccl/bench_test.go

@@ -179,7 +179,8 @@ func benchmarkConvertToKVs(b *testing.B, g workload.Generator) {
 				SessionDataStack: sessiondata.NewStack(&sessiondata.SessionData{}),
 				Codec:            keys.SystemSQLCodec,
 			}
-			return wc.Worker(ctx, evalCtx)
+			semaCtx := tree.MakeSemaContext()
+			return wc.Worker(ctx, evalCtx, &semaCtx)
 		})
 		for kvBatch := range kvCh {
 			for i := range kvBatch.KVs {

pkg/ccl/importccl/import_processor.go

@@ -273,7 +273,7 @@ func makeInputConverter(
 			}
 		}
 		if isWorkload {
-			return newWorkloadReader(kvCh, singleTable, evalCtx), nil
+			return newWorkloadReader(semaCtx, evalCtx, singleTable, kvCh), nil
 		}
 		return newCSVInputReader(
 			semaCtx, kvCh, spec.Format.Csv, spec.WalltimeNanos, int(spec.ReaderParallelism),

pkg/ccl/importccl/read_import_workload.go

@@ -35,6 +35,7 @@ import (
 )
 
 type workloadReader struct {
+	semaCtx *tree.SemaContext
 	evalCtx *tree.EvalContext
 	table   catalog.TableDescriptor
 	kvCh    chan row.KVBatch
@@ -43,9 +44,12 @@ type workloadReader struct {
 var _ inputConverter = &workloadReader{}
 
 func newWorkloadReader(
-	kvCh chan row.KVBatch, table catalog.TableDescriptor, evalCtx *tree.EvalContext,
+	semaCtx *tree.SemaContext,
+	evalCtx *tree.EvalContext,
+	table catalog.TableDescriptor,
+	kvCh chan row.KVBatch,
 ) *workloadReader {
-	return &workloadReader{evalCtx: evalCtx, table: table, kvCh: kvCh}
+	return &workloadReader{semaCtx: semaCtx, evalCtx: evalCtx, table: table, kvCh: kvCh}
 }
 
 func (w *workloadReader) start(ctx ctxgroup.Group) {
@@ -164,7 +168,7 @@ func (w *workloadReader) readFiles(
 	for _, wc := range wcs {
 		if err := ctxgroup.GroupWorkers(ctx, runtime.GOMAXPROCS(0), func(ctx context.Context, _ int) error {
 			evalCtx := w.evalCtx.Copy()
-			return wc.Worker(ctx, evalCtx)
+			return wc.Worker(ctx, evalCtx, w.semaCtx)
 		}); err != nil {
 			return err
 		}
@@ -216,9 +220,10 @@ func NewWorkloadKVConverter(
 // minimzing the amount of overlapping SSTs ingested.
 //
 // This worker needs its own EvalContext and DatumAlloc.
-func (w *WorkloadKVConverter) Worker(ctx context.Context, evalCtx *tree.EvalContext) error {
-	semaCtx := tree.MakeSemaContext()
-	conv, err := row.NewDatumRowConverter(ctx, &semaCtx, w.tableDesc, nil, /* targetColNames */
+func (w *WorkloadKVConverter) Worker(
+	ctx context.Context, evalCtx *tree.EvalContext, semaCtx *tree.SemaContext,
+) error {
+	conv, err := row.NewDatumRowConverter(ctx, semaCtx, w.tableDesc, nil, /* targetColNames */
 		evalCtx, w.kvCh, nil /* seqChunkProvider */, nil /* metrics */)
 	if err != nil {
 		return err

pkg/ccl/workloadccl/format/sstable.go

@@ -85,7 +85,8 @@ func ToSSTable(t workload.Table, tableID descpb.ID, ts time.Time) ([]byte, error
 			SessionDataStack: sessiondata.NewStack(&sessiondata.SessionData{}),
 			Codec:            keys.SystemSQLCodec,
 		}
-		return wc.Worker(ctx, evalCtx)
+		semaCtx := tree.MakeSemaContext()
+		return wc.Worker(ctx, evalCtx, &semaCtx)
 	})
 	var sst []byte
 	var kvs sortableKVs

pkg/sql/sem/tree/type_name.go

@@ -164,7 +164,7 @@ func ResolveType(
 		return resolver.ResolveType(ctx, t)
 	case *OIDTypeReference:
 		if resolver == nil {
-			return nil, pgerror.Newf(pgcode.UndefinedObject, "type OID %d does not exist", t.OID)
+			return nil, pgerror.Newf(pgcode.UndefinedObject, "type resolver unavailable to resolve type OID %d", t.OID)
 		}
 		return resolver.ResolveTypeByOID(ctx, t.OID)
 	default: