Merge #45865 #46857 #47582 #47583 #47629 #47756

45865: backupccl: remove unused username param in WriteTableDescs r=dt a=pbardea

Release note: None

46857: ui: Sortable loading state r=dhartunian a=elkmaster

Added global loading state to sort tables
Added loading state to database tables

![loading-database](https://user-images.githubusercontent.com/12850886/78561401-4e875500-7820-11ea-8462-53bd7310adf0.png)

Resolves: #46568

Release justification: low risk, high benefit changes to existing functionality

Release note (ui): database loading state design updates

47582:  backupccl: fix flake in TestProtectedTimestampDuringBackup  r=pbardea a=pbardea

TestProtectedTimestampDuringBackup would sometimes flake as the GC queue
would assign it a low priority that would be below the threshold to turn
shouldQueue to true. However, the priority is non-zero indicating that
the timestamp was indeed not protected. This change aims to remove the
flake by checking for a non-zero priority rather than if shouldQueue is
true.

This PR also makes the same change to the ImportInto variant of the test since
they share the same test structure.

Fixes #47522.

Release note: None

47583: col*: use logical types throughout the vectorized engine r=yuzefovich a=yuzefovich

**col...: create new package and move some code**

This commit introduces `colbase` package which currently contains the
following (that are extracted from `colexec` package):
- `allocator.go`
- `random_testutils.go`
- `Operator` interface
- `CopyBatch` method.

The reason for extracting these things is that they are used by multiple
col* packages, so I think it's a good hygiene to separate them out.

This commit also renames `execerror` package to `vecerror` and then
moves it inside of `colbase` directory. It also changes the panic
matching so that now we catch all panics coming from a package that has
`pkg/sql/col` in its path (this will make sure that we don't forget to
register newly added packages that use panic-catching mechanism of the
vectorized engine with the panic-catcher). Additionally, this commit
renames the methods of `execerror` package.

Next, it moves `colexec/typeconv` package into `colbase` as well as
`colexec/testutils.go` file.

Finally, it removes `vecerror.NonVectorizedPanic` in favor of
`vecerror.ExpectedError`. The guidance on whether `InternalError` or
`ExpectedError` method should be used has been updated: the distinction
is whether the vectorized engine ends up in an unexpected - invalid -
state (for example, we expect that the vectorized engine might be
performing division by zero when evaluating a binary expression).

Release note: None

**col...: use logical types throughout the vectorized engine**

This commit transitions the vectorized engine to use SQL logical types
wherever possible, only converting to its physical type equivalent when
necessary (for example, when choosing which instance of projection
operator to use). This will allow us to have access to the actual type
whever we need and will allow us implement `coltypes.Datum` for all
unoptimized types.

This commit also moves some a few things around to clean up the
dependency graph (namely, `BytesEncodeFormat` is moved from
`sql/sessiondata` to `sql/lex`). Additionally, it replaces a single call
to `util/log.Infof` with a print statement in `util/protoutil` package
to remove the dependency on `util/log` (which depends on a bunch of
other things) so that `pkg/workload` would need to import less things.

Another thing worth calling out is the creation of copies of types in
`execplan` file to make sure that we don't override the input sync
specs.

Addresses: #43559.

Release note: None

**col...: introduce new package and more code movement**

Move `coldata/random_testutils.go` into newly created package
`coldatatestutils`.

Move contents of `colbase/random_testutils.go` into `coldatatestutils`
package.

Rename `colbase` to `colexecbase`. Also remove templated comments from
import sections of `_tmpl` files in favor of adding vars that remove
unused warnings (those templated comments work poorly when
moving/renaming the dependencies).

Rename `vecerror` to `colexecerror`.

Move `CopyBatch` from `colexecbase` into `coldatatestutils`. Also remove
memory accounting from `CopyBatch`.

Move `typeconv` from `colexecbase` into `coltypes` folder.

Move `coldata/vec_test.go` into `coldata_test` package to prevent an
import cycle. Also move one unit test from `coldata/vec_test.go` into
`coldata/bytes_test.go`.

Move `colexecbase/allocator.go` into newly created `colmem` package.

Release note: None

47629: sql: use Clock.PhysicalTime in beginTransactionTimestampsAndReadMode r=nvanbenschoten a=nvanbenschoten

Synchronizing with the HLC clock doesn't look to be necessary. I'm confused
about this though. The comment on beginTransactionTimestampsAndReadMode says
that "txnSQLTimestamp propagates to become the TxnTimestamp". Is this trying to
say that the timestamp makes it way into the kv.Txn? Because that's not true.

Regardless, the one reason not to make this change is that PhysicalTime is not
guaranteed to be monotonic on some systems and can generally diverge from the
HLC's clock. If we're worried about that though, we should use the HLC here and
feed that directly into the kv.Txn. We shouldn't need to grab two timestamps
from the HLC per txn.

47756: cli,base: surface stored critical errors at the right moment r=tbg a=knz

Fixes #44041

If/when storage detects an important error, the text for this error is
stored in a file named `_CRITICAL_ERROR.txt` in the auxiliary
directory. The intention is to block subsequent server restarts until
the error is investigated and the file (manually) removed.

Prior to this patch, this check was done in the startup sequence 1)
before logging was fully initialized 2) using a `log.Fatal` to
announce the critical error.

The first aspect is problematic because it logs before logging flags
are applied. The second is problematic because it makes the failure
super-verbose and buries the lede.

This patch simplifies the code and makes the error reported at the
right place.

Example, before:
```
kena@kenax ....com/cockroachdb/cockroach % ./cockroach start-single-node
F200421 14:24:02.303675 1 cli/start.go:478  From .../auxiliary/_CRITICAL_ALERT.txt:

boom

goroutine 1 [running]:
github.com/cockroachdb/cockroach/pkg/util/log.getStacks(0x6a7ca00, 0xed630f902, 0x0, 0x1000)
        ...//src/github.com/cockroachdb/cockroach/pkg/util/log/get_stacks.go:25 +0xb8
github.com/cockroachdb/cockroach/pkg/util/log.(*loggerT).outputLogEntry(0x6a79800, 0xc000000004, 0x33eb7b9, 0xc, 0x1de, 0xc000ba8180, 0x76)
        ...//src/github.com/cockroachdb/cockroach/pkg/util/log/clog.go:210 +0xa92
github.com/cockroachdb/cockroach/pkg/util/log.addStructured(0x15e3420, 0xc000078168, 0x4, 0x2, 0x0, 0x0, 0xc00063f7e8, 0x1, 0x1)
        ...//src/github.com/cockroachdb/cockroach/pkg/util/log/structured.go:66 +0x2c9
...
[147/245]
****************************************************************************

This node experienced a fatal error (printed above), and as a result the
process is terminating.

Fatal errors can occur due to faulty hardware (disks, memory, clocks) or a
...
    support@cockroachlabs.com

The Cockroach Labs team appreciates your feedback.
```

Example, after:

```
kena@kenax ....com/cockroachdb/cockroach % ./cockroach start-single-node
*
* ERROR: startup forbidden by prior critical alert
* DETAIL: From /data/home/kena/src/go/src/github.com/cockroachdb/cockroach/cockroach-data/auxiliary/_CRITICAL_ALERT.txt:
* boom
*
```

Release note (cli change): The error message displayed upon `cockroach
start` / `cockroach start-single-node` when manual intervention is
needed in the store directory is now clearer.

Co-authored-by: Paul Bardea <pbardea@gmail.com>
Co-authored-by: Vlad Los <carrott9@gmail.com>
Co-authored-by: Paul Bardea <paul@pbardea.com>
Co-authored-by: Yahor Yuzefovich <yahor@cockroachlabs.com>
Co-authored-by: Nathan VanBenschoten <nvanbenschoten@gmail.com>
Co-authored-by: Raphael 'kena' Poss <knz@thaumogen.net>

pkg/base/store_spec.go

@@ -26,8 +26,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/netutil"
+	"github.com/cockroachdb/errors"
 	humanize "github.com/dustin/go-humanize"
-	"github.com/pkg/errors"
 	"github.com/spf13/pflag"
 )
 
@@ -369,12 +369,23 @@ func PreventedStartupFile(dir string) string {
 	return filepath.Join(dir, "_CRITICAL_ALERT.txt")
 }
 
-// GetPreventedStartupMessage attempts to read the PreventedStartupFile for each
-// store directory and returns their concatenated contents. These files
-// typically request operator intervention after a corruption event by
-// preventing the affected node(s) from starting back up.
-func (ssl StoreSpecList) GetPreventedStartupMessage() (string, error) {
-	var buf strings.Builder
+// PriorCriticalAlertError attempts to read the
+// PreventedStartupFile for each store directory and returns their
+// contents as a structured error.
+//
+// These files typically request operator intervention after a
+// corruption event by preventing the affected node(s) from starting
+// back up.
+func (ssl StoreSpecList) PriorCriticalAlertError() (err error) {
+	addError := func(newErr error) {
+		if err == nil {
+			err = errors.New("startup forbidden by prior critical alert")
+		}
+		// We use WithDetailf here instead of errors.CombineErrors
+		// because we want the details to be printed to the screen
+		// (combined errors only show up via %+v).
+		err = errors.WithDetailf(err, "%v", newErr)
+	}
 	for _, ss := range ssl.Specs {
 		path := ss.PreventedStartupFile()
 		if path == "" {
@@ -383,15 +394,13 @@ func (ssl StoreSpecList) GetPreventedStartupMessage() (string, error) {
 		b, err := ioutil.ReadFile(path)
 		if err != nil {
 			if !os.IsNotExist(err) {
-				return "", err
+				addError(errors.Wrapf(err, "%s", path))
 			}
 			continue
 		}
-		fmt.Fprintf(&buf, "From %s:\n\n", path)
-		_, _ = buf.Write(b)
-		fmt.Fprintln(&buf)
+		addError(errors.Newf("From %s:\n\n%s\n", path, b))
 	}
-	return buf.String(), nil
+	return err
 }
 
 // PreventedStartupFile returns the path to a file which, if it exists, should

pkg/base/store_spec_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -245,13 +246,13 @@ func TestStoreSpecListPreventedStartupMessage(t *testing.T) {
 		},
 	}
 
-	msg, err := ssl.GetPreventedStartupMessage()
+	err := ssl.PriorCriticalAlertError()
 	require.NoError(t, err)
-	require.Empty(t, msg)
 
 	require.NoError(t, ioutil.WriteFile(ssl.Specs[2].PreventedStartupFile(), []byte("boom"), 0644))
 
-	msg, err = ssl.GetPreventedStartupMessage()
-	require.NoError(t, err)
-	require.Contains(t, msg, "boom")
+	err = ssl.PriorCriticalAlertError()
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "startup forbidden by prior critical alert")
+	require.Contains(t, errors.FlattenDetails(err), "boom")
 }

pkg/ccl/backupccl/backup_job.go

@@ -541,6 +541,14 @@ func (b *backupResumer) Resume(
 	}
 	b.deleteCheckpoint(ctx, p.ExecCfg())
 
+	if ptsID != nil && !b.testingKnobs.ignoreProtectedTimestamps {
+		if err := p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+			return b.releaseProtectedTimestamp(ctx, txn, p.ExecCfg().ProtectedTimestampProvider)
+		}); err != nil {
+			log.Errorf(ctx, "failed to release protected timestamp: %v", err)
+		}
+	}
+
 	resultsCh <- tree.Datums{
 		tree.NewDInt(tree.DInt(*b.job.ID())),
 		tree.NewDString(string(jobs.StatusSucceeded)),
@@ -550,14 +558,6 @@ func (b *backupResumer) Resume(
 		tree.NewDInt(tree.DInt(res.DataSize)),
 	}
 
-	if ptsID != nil && !b.testingKnobs.ignoreProtectedTimestamps {
-		if err := p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
-			return b.releaseProtectedTimestamp(ctx, txn, p.ExecCfg().ProtectedTimestampProvider)
-		}); err != nil {
-			log.Errorf(ctx, "failed to release protected timestamp: %v", err)
-		}
-	}
-
 	// Collect telemetry.
 	{
 		telemetry.Count("backup.total.succeeded")

pkg/ccl/backupccl/backup_test.go

@@ -3626,7 +3626,14 @@ func TestProtectedTimestampsDuringBackup(t *testing.T) {
 	rowCount := runner.QueryStr(t, "SELECT * FROM foo")
 
 	go func() {
-		runner.Exec(t, `BACKUP TABLE FOO TO 'nodelocal://1/foo'`)
+		// N.B. We use the conn rather than the runner here since the test may
+		// finish before the job finishes. The test will finish as soon as the
+		// timestamp is no longer protected. If the test starts tearing down the
+		// cluster before the backup job is done, the test may still fail when the
+		// backup fails. This test does not particularly care if the BACKUP
+		// completes with a success or failure, as long as the timestamp is released
+		// shortly after the BACKUP is unblocked.
+		_, _ = conn.Exec(`BACKUP TABLE FOO TO 'nodelocal://1/foo'`) // ignore error.
 	}()
 
 	var jobID string
@@ -3669,11 +3676,15 @@ func TestProtectedTimestampsDuringBackup(t *testing.T) {
 
 	// Wait for the ranges to learn about the removed record and ensure that we
 	// can GC from the range soon.
-	gcRanRE := regexp.MustCompile("(?s)shouldQueue=true.*processing replica.*GC score after GC")
+	// This regex matches when all float priorities other than 0.00000. It does
+	// this by matching either a float >= 1 (e.g. 1230.012) or a float < 1 (e.g.
+	// 0.000123).
+	matchNonZero := "[1-9]\\d*\\.\\d+|0\\.\\d*[1-9]\\d*"
+	nonZeroProgressRE := regexp.MustCompile(fmt.Sprintf("priority=(%s)", matchNonZero))
 	testutils.SucceedsSoon(t, func() error {
 		writeGarbage(3, 10)
-		if trace := gcTable(false /* skipShouldQueue */); !gcRanRE.MatchString(trace) {
-			return fmt.Errorf("expected %v in trace: %v", gcRanRE, trace)
+		if trace := gcTable(false /* skipShouldQueue */); !nonZeroProgressRE.MatchString(trace) {
+			return fmt.Errorf("expected %v in trace: %v", nonZeroProgressRE, trace)
 		}
 		return nil
 	})

pkg/ccl/backupccl/restore_job.go

@@ -432,7 +432,6 @@ func WriteTableDescs(
 	databases []*sqlbase.DatabaseDescriptor,
 	tables []*sqlbase.TableDescriptor,
 	descCoverage tree.DescriptorCoverage,
-	user string,
 	settings *cluster.Settings,
 	extra []roachpb.KeyValue,
 ) error {
@@ -924,7 +923,7 @@ func createImportingTables(
 	if !details.PrepareCompleted {
 		err := p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
 			// Write the new TableDescriptors which are set in the OFFLINE state.
-			if err := WriteTableDescs(ctx, txn, databases, tables, details.DescriptorCoverage, r.job.Payload().Username, r.settings, nil /* extra */); err != nil {
+			if err := WriteTableDescs(ctx, txn, databases, tables, details.DescriptorCoverage, r.settings, nil /* extra */); err != nil {
 				return errors.Wrapf(err, "restoring %d TableDescriptors from %d databases", len(r.tables), len(databases))
 			}
 

pkg/ccl/importccl/import_into_test.go

@@ -147,11 +147,15 @@ func TestProtectedTimestampsDuringImportInto(t *testing.T) {
 
 	// Wait for the ranges to learn about the removed record and ensure that we
 	// can GC from the range soon.
-	gcRanRE := regexp.MustCompile("(?s)shouldQueue=true.*processing replica.*GC score after GC")
+	// This regex matches when all float priorities other than 0.00000. It does
+	// this by matching either a float >= 1 (e.g. 1230.012) or a float < 1 (e.g.
+	// 0.000123).
+	matchNonZero := "[1-9]\\d*\\.\\d+|0\\.\\d*[1-9]\\d*"
+	nonZeroProgressRE := regexp.MustCompile(fmt.Sprintf("priority=(%s)", matchNonZero))
 	testutils.SucceedsSoon(t, func() error {
 		writeGarbage(3, 10)
-		if trace := gcTable(false /* skipShouldQueue */); !gcRanRE.MatchString(trace) {
-			return fmt.Errorf("expected %v in trace: %v", gcRanRE, trace)
+		if trace := gcTable(false /* skipShouldQueue */); !nonZeroProgressRE.MatchString(trace) {
+			return fmt.Errorf("expected %v in trace: %v", nonZeroProgressRE, trace)
 		}
 		return nil
 	})

pkg/ccl/importccl/import_stmt.go

@@ -902,7 +902,7 @@ func prepareNewTableDescsForIngestion(
 	// Write the new TableDescriptors and flip the namespace entries over to
 	// them. After this call, any queries on a table will be served by the newly
 	// imported data.
-	if err := backupccl.WriteTableDescs(ctx, txn, nil /* databases */, tableDescs, tree.RequestedDescriptors, p.User(), p.ExecCfg().Settings, seqValKVs); err != nil {
+	if err := backupccl.WriteTableDescs(ctx, txn, nil /* databases */, tableDescs, tree.RequestedDescriptors, p.ExecCfg().Settings, seqValKVs); err != nil {
 		return nil, errors.Wrapf(err, "creating tables")
 	}
 

pkg/cli/sql_util.go

@@ -30,7 +30,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/sql/lex"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
-	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -921,7 +920,7 @@ func formatVal(val driver.Value, showPrintableUnicode bool, showNewLinesAndTabs
 		// that we can let the user see and control the result using
 		// `bytea_output`.
 		return lex.EncodeByteArrayToRawBytes(string(t),
-			sessiondata.BytesEncodeEscape, false /* skipHexPrefix */)
+			lex.BytesEncodeEscape, false /* skipHexPrefix */)
 
 	case time.Time:
 		return t.Format(tree.TimestampOutputFormat)

pkg/cli/start.go

@@ -470,14 +470,6 @@ func runStart(cmd *cobra.Command, args []string, disableReplication bool) error
 	// not be world-readable.
 	disableOtherPermissionBits()
 
-	// TODO(knz): the following call is not in the right place.
-	// See: https://github.com/cockroachdb/cockroach/issues/44041
-	if s, err := serverCfg.Stores.GetPreventedStartupMessage(); err != nil {
-		return err
-	} else if s != "" {
-		log.Fatal(context.Background(), s)
-	}
-
 	// Set up the signal handlers. This also ensures that any of these
 	// signals received beyond this point do not interrupt the startup
 	// sequence until the point signals are checked below.
@@ -521,6 +513,12 @@ func runStart(cmd *cobra.Command, args []string, disableReplication bool) error
 		return err
 	}
 
+	// If any store has something to say against a server start-up
+	// (e.g. previously detected corruption), listen to them now.
+	if err := serverCfg.Stores.PriorCriticalAlertError(); err != nil {
+		return err
+	}
+
 	// We don't care about GRPCs fairly verbose logs in most client commands,
 	// but when actually starting a server, we enable them.
 	grpcutil.SetSeverity(log.Severity_WARNING)

pkg/col/coldata/batch.go

@@ -17,6 +17,8 @@ import (
 	"sync/atomic"
 
 	"github.com/cockroachdb/cockroach/pkg/col/coltypes"
+	"github.com/cockroachdb/cockroach/pkg/col/coltypes/typeconv"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/errors"
 )
 
@@ -47,14 +49,14 @@ type Batch interface {
 	// provided Vec.
 	ReplaceCol(Vec, int)
 	// Reset modifies the caller in-place to have the given length and columns
-	// with the given coltypes. If it's possible, Reset will reuse the existing
+	// with the given types. If it's possible, Reset will reuse the existing
 	// columns and allocations, invalidating existing references to the Batch or
 	// its Vecs. However, Reset does _not_ zero out the column data.
 	//
 	// NOTE: Reset can allocate a new Batch, so when calling from the vectorized
 	// engine consider either allocating a new Batch explicitly via
 	// colexec.Allocator or calling ResetInternalBatch.
-	Reset(types []coltypes.T, length int)
+	Reset(types []types.T, length int)
 	// ResetInternalBatch resets a batch and its underlying Vecs for reuse. It's
 	// important for callers to call ResetInternalBatch if they own internal
 	// batches that they reuse as not doing this could result in correctness
@@ -98,24 +100,24 @@ func ResetBatchSizeForTests() {
 // NewMemBatch allocates a new in-memory Batch. A coltypes.Unknown type
 // will create a placeholder Vec that may not be accessed.
 // TODO(jordan): pool these allocations.
-func NewMemBatch(types []coltypes.T) Batch {
+func NewMemBatch(types []types.T) Batch {
 	return NewMemBatchWithSize(types, BatchSize())
 }
 
 // NewMemBatchWithSize allocates a new in-memory Batch with the given column
 // size. Use for operators that have a precisely-sized output batch.
-func NewMemBatchWithSize(types []coltypes.T, size int) Batch {
+func NewMemBatchWithSize(types []types.T, size int) Batch {
 	b := NewMemBatchNoCols(types, size).(*MemBatch)
 	for i, t := range types {
-		b.b[i] = NewMemColumn(t, size)
+		b.b[i] = NewMemColumn(&t, size)
 	}
 	return b
 }
 
 // NewMemBatchNoCols creates a "skeleton" of new in-memory Batch. It allocates
 // memory for the selection vector but does *not* allocate any memory for the
 // column vectors - those will have to be added separately.
-func NewMemBatchNoCols(types []coltypes.T, size int) Batch {
+func NewMemBatchNoCols(types []types.T, size int) Batch {
 	if max := math.MaxUint16; size > max {
 		panic(fmt.Sprintf(`batches cannot have length larger than %d; requested %d`, max, size))
 	}
@@ -156,7 +158,7 @@ func (b *zeroBatch) ReplaceCol(Vec, int) {
 	panic("no columns should be replaced in zero batch")
 }
 
-func (b *zeroBatch) Reset([]coltypes.T, int) {
+func (b *zeroBatch) Reset([]types.T, int) {
 	panic("zero batch should not be reset")
 }
 
@@ -226,7 +228,7 @@ func (m *MemBatch) ReplaceCol(col Vec, colIdx int) {
 }
 
 // Reset implements the Batch interface.
-func (m *MemBatch) Reset(types []coltypes.T, length int) {
+func (m *MemBatch) Reset(types []types.T, length int) {
 	ResetNoTruncation(m, types, length)
 	m.b = m.b[:len(types)]
 }
@@ -236,14 +238,15 @@ func (m *MemBatch) Reset(types []coltypes.T, length int) {
 // the prefix of already present columns matches the desired type schema,
 // the batch will be reused (meaning this method does *not* truncate the
 // type schema).
-func ResetNoTruncation(m *MemBatch, types []coltypes.T, length int) {
+func ResetNoTruncation(m *MemBatch, types []types.T, length int) {
 	// The columns are always sized the same as the selection vector, so use it as
 	// a shortcut for the capacity (like a go slice, the batch's `Length` could be
 	// shorter than the capacity). We could be more defensive and type switch
 	// every column to verify its capacity, but that doesn't seem necessary yet.
 	cannotReuse := m == nil || len(m.sel) < length || m.Width() < len(types)
 	for i := 0; i < len(types) && !cannotReuse; i++ {
-		if m.ColVec(i).Type() != types[i] {
+		// TODO(yuzefovich): change this when coltypes.Datum is introduced.
+		if m.ColVec(i).Type() != typeconv.FromColumnType(&types[i]) {
 			cannotReuse = true
 		}
 	}

pkg/col/coldata/batch_test.go

@@ -17,14 +17,16 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/col/coldata"
 	"github.com/cockroachdb/cockroach/pkg/col/coltypes"
+	"github.com/cockroachdb/cockroach/pkg/col/coltypes/typeconv"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestBatchReset(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
-	resetAndCheck := func(b coldata.Batch, typs []coltypes.T, n int, shouldReuse bool) {
+	resetAndCheck := func(b coldata.Batch, typs []types.T, n int, shouldReuse bool) {
 		t.Helper()
 		// Use the data backing the ColVecs slice as a proxy for when things get
 		// reallocated.
@@ -47,7 +49,7 @@ func TestBatchReset(t *testing.T) {
 		for i, vec := range b.ColVecs() {
 			assert.False(t, vec.MaybeHasNulls())
 			assert.False(t, vec.Nulls().NullAt(0))
-			assert.Equal(t, typs[i], vec.Type())
+			assert.Equal(t, typeconv.FromColumnType(&typs[i]), vec.Type())
 			// Sanity check that we can actually use the column. This is mostly for
 			// making sure a flat bytes column gets reset.
 			vec.Nulls().SetNull(0)
@@ -68,9 +70,9 @@ func TestBatchReset(t *testing.T) {
 		}
 	}
 
-	typsInt := []coltypes.T{coltypes.Int64}
-	typsBytes := []coltypes.T{coltypes.Bytes}
-	typsIntBytes := []coltypes.T{coltypes.Int64, coltypes.Bytes}
+	typsInt := []types.T{*types.Int}
+	typsBytes := []types.T{*types.Bytes}
+	typsIntBytes := []types.T{*types.Int, *types.Bytes}
 	var b coldata.Batch
 
 	// Simple case, reuse