From e6448f435b0f073fc5e91f18081e82fd920b739d Mon Sep 17 00:00:00 2001
From: Marius Posta <marius@cockroachlabs.com>
Date: Thu, 7 Jan 2021 11:00:22 -0500
Subject: [PATCH] sql: turn some catalog.TableDescriptor methods into functions

The catalog.TableDescriptor has ForEach*Index and Find*Index methods
which can just as well be expressed as functions, thereby lightening
that interface definition somewhat.

This patch also adds iterutils.StopIteration() support for the above
ForEach*Index functions, and otherwise also adds comments to many
catalog.Index methods.

Release note: None
---
 pkg/ccl/backupccl/backup_planning.go     |   2 +-
 pkg/ccl/backupccl/restore_planning.go    |   2 +-
 pkg/ccl/backupccl/targets.go             |   2 +-
 pkg/ccl/importccl/import_processor.go    |   2 +-
 pkg/ccl/partitionccl/partition.go        |   2 +-
 pkg/sql/alter_table.go                   |   2 +-
 pkg/sql/catalog/BUILD.bazel              |   1 +
 pkg/sql/catalog/descriptor.go            | 283 +++++++++++++++++++++--
 pkg/sql/catalog/tabledesc/index.go       |   8 +-
 pkg/sql/catalog/tabledesc/safe_format.go |   2 +-
 pkg/sql/catalog/tabledesc/structured.go  |  16 +-
 pkg/sql/catalog/tabledesc/table_desc.go  | 209 ++++-------------
 pkg/sql/crdb_internal.go                 |  14 +-
 pkg/sql/create_table.go                  |   2 +-
 pkg/sql/drop_index.go                    |   4 +-
 pkg/sql/information_schema.go            |   2 +-
 pkg/sql/partition_utils.go               |   2 +-
 pkg/sql/pg_catalog.go                    |   8 +-
 pkg/sql/rowexec/joinreader_test.go       |   2 +-
 pkg/sql/truncate.go                      |   2 +-
 20 files changed, 346 insertions(+), 221 deletions(-)

diff --git a/pkg/ccl/backupccl/backup_planning.go b/pkg/ccl/backupccl/backup_planning.go
index 0e3fab9a86cf..4276f65345d5 100644
--- a/pkg/ccl/backupccl/backup_planning.go
+++ b/pkg/ccl/backupccl/backup_planning.go
@@ -182,7 +182,7 @@ func getLogicallyMergedTableSpans(
 	checkForKVInBounds func(start, end roachpb.Key, endTime hlc.Timestamp) (bool, error),
 ) ([]roachpb.Span, error) {
 	var nonDropIndexIDs []descpb.IndexID
-	if err := table.ForEachNonDropIndex(func(idx catalog.Index) error {
+	if err := catalog.ForEachNonDropIndex(table, func(idx catalog.Index) error {
 		key := tableAndIndex{tableID: table.GetID(), indexID: idx.GetID()}
 		if added[key] {
 			return nil
diff --git a/pkg/ccl/backupccl/restore_planning.go b/pkg/ccl/backupccl/restore_planning.go
index 3e13127e297c..faf3da356b85 100644
--- a/pkg/ccl/backupccl/restore_planning.go
+++ b/pkg/ccl/backupccl/restore_planning.go
@@ -1021,7 +1021,7 @@ func RewriteTableDescs(
 			return err
 		}
 
-		if err := table.ForEachNonDropIndex(func(indexI catalog.Index) error {
+		if err := catalog.ForEachNonDropIndex(table, func(indexI catalog.Index) error {
 			index := indexI.IndexDesc()
 			// Verify that for any interleaved index being restored, the interleave
 			// parent is also being restored. Otherwise, the interleave entries in the
diff --git a/pkg/ccl/backupccl/targets.go b/pkg/ccl/backupccl/targets.go
index 93b8ec72761b..15cd4b683fec 100644
--- a/pkg/ccl/backupccl/targets.go
+++ b/pkg/ccl/backupccl/targets.go
@@ -668,7 +668,7 @@ func ensureInterleavesIncluded(tables []catalog.TableDescriptor) error {
 	}
 
 	for _, table := range tables {
-		if err := table.ForEachIndex(catalog.IndexOpts{
+		if err := catalog.ForEachIndex(table, catalog.IndexOpts{
 			AddMutations: true,
 		}, func(index catalog.Index) error {
 			for i := 0; i < index.NumInterleaveAncestors(); i++ {
diff --git a/pkg/ccl/importccl/import_processor.go b/pkg/ccl/importccl/import_processor.go
index 5bb40aa3fe75..9da93ace868f 100644
--- a/pkg/ccl/importccl/import_processor.go
+++ b/pkg/ccl/importccl/import_processor.go
@@ -191,7 +191,7 @@ func makeInputConverter(
 	}
 
 	if singleTable != nil {
-		if idx := singleTable.FindDeletableNonPrimaryIndex(func(idx catalog.Index) bool {
+		if idx := catalog.FindDeletableNonPrimaryIndex(singleTable, func(idx catalog.Index) bool {
 			return idx.IsPartial()
 		}); idx != nil {
 			return nil, unimplemented.NewWithIssue(50225, "cannot import into table with partial indexes")
diff --git a/pkg/ccl/partitionccl/partition.go b/pkg/ccl/partitionccl/partition.go
index e61cd183e1d5..5b3941281c89 100644
--- a/pkg/ccl/partitionccl/partition.go
+++ b/pkg/ccl/partitionccl/partition.go
@@ -275,7 +275,7 @@ func selectPartitionExprs(
 
 	a := &rowenc.DatumAlloc{}
 	var prefixDatums []tree.Datum
-	if err := tableDesc.ForEachIndex(catalog.IndexOpts{
+	if err := catalog.ForEachIndex(tableDesc, catalog.IndexOpts{
 		AddMutations: true,
 	}, func(idx catalog.Index) error {
 		return selectPartitionExprsByName(
diff --git a/pkg/sql/alter_table.go b/pkg/sql/alter_table.go
index 46fed26abea7..6d4e170bdb40 100644
--- a/pkg/sql/alter_table.go
+++ b/pkg/sql/alter_table.go
@@ -816,7 +816,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 			// new name exists. This is what postgres does.
 			switch details.Kind {
 			case descpb.ConstraintTypeUnique, descpb.ConstraintTypePK:
-				if n.tableDesc.FindNonDropIndex(func(idx catalog.Index) bool {
+				if catalog.FindNonDropIndex(n.tableDesc, func(idx catalog.Index) bool {
 					return idx.GetName() == string(t.NewName)
 				}) != nil {
 					return pgerror.Newf(pgcode.DuplicateRelation,
diff --git a/pkg/sql/catalog/BUILD.bazel b/pkg/sql/catalog/BUILD.bazel
index 2412f8841bab..36575a9cefe6 100644
--- a/pkg/sql/catalog/BUILD.bazel
+++ b/pkg/sql/catalog/BUILD.bazel
@@ -26,6 +26,7 @@ go_library(
         "//pkg/sql/types",
         "//pkg/util",
         "//pkg/util/hlc",
+        "//pkg/util/iterutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
     ],
diff --git a/pkg/sql/catalog/descriptor.go b/pkg/sql/catalog/descriptor.go
index e83724143271..cc5001dfe33c 100644
--- a/pkg/sql/catalog/descriptor.go
+++ b/pkg/sql/catalog/descriptor.go
@@ -22,11 +22,13 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
 
-// IndexOpts configures the behavior of TableDescriptor.ForEachIndex.
+// IndexOpts configures the behavior of catalog.ForEachIndex and
+// catalog.FindIndex.
 type IndexOpts struct {
 	// NonPhysicalPrimaryIndex should be included.
 	NonPhysicalPrimaryIndex bool
@@ -112,40 +114,78 @@ type TableDescriptor interface {
 	GetFormatVersion() descpb.FormatVersion
 
 	GetPrimaryIndexID() descpb.IndexID
+	GetPrimaryIndex() Index
 	PrimaryIndexSpan(codec keys.SQLCodec) roachpb.Span
 	IndexSpan(codec keys.SQLCodec, id descpb.IndexID) roachpb.Span
 	GetIndexMutationCapabilities(id descpb.IndexID) (isMutation, isWriteOnly bool)
 	KeysPerRow(id descpb.IndexID) (int, error)
 
-	GetPrimaryIndex() Index
+	// AllIndexes returns a slice with all indexes, public and non-public,
+	// in the underlying proto, in their canonical order:
+	// - the primary index,
+	// - the public non-primary indexes in the Indexes array, in order,
+	// - the non-public indexes present in the Mutations array, in order.
+	//
+	// See also Index.Ordinal().
 	AllIndexes() []Index
+
+	// ActiveIndexes returns a slice with all public indexes in the underlying
+	// proto, in their canonical order:
+	// - the primary index,
+	// - the public non-primary indexes in the Indexes array, in order.
+	//
+	// See also Index.Ordinal().
 	ActiveIndexes() []Index
+
+	// NonDropIndexes returns a slice of all non-drop indexes in the underlying
+	// proto, in their canonical order. This means:
+	// - the primary index, if the table is a physical table,
+	// - the public non-primary indexes in the Indexes array, in order,
+	// - the non-public indexes present in the Mutations array, in order,
+	//   if the mutation is not a drop.
+	//
+	// See also Index.Ordinal().
 	NonDropIndexes() []Index
+
+	// NonDropIndexes returns a slice of all partial indexes in the underlying
+	// proto, in their canonical order. This is equivalent to taking the slice
+	// produced by AllIndexes and removing indexes with empty expressions.
 	PartialIndexes() []Index
+
+	// PublicNonPrimaryIndexes returns a slice of all active secondary indexes,
+	// in their canonical order. This is equivalent to the Indexes array in the
+	// proto.
 	PublicNonPrimaryIndexes() []Index
+
+	// WritableNonPrimaryIndexes returns a slice of all non-primary indexes which
+	// allow being written to: public + delete-and-write-only, in their canonical
+	// order. This is equivalent to taking the slice produced by
+	// DeletableNonPrimaryIndexes and removing the indexes which are in mutations
+	// in the delete-only state.
 	WritableNonPrimaryIndexes() []Index
+
+	// DeletableNonPrimaryIndexes returns a slice of all non-primary indexes
+	// which allow being deleted from: public + delete-and-write-only +
+	// delete-only, in  their canonical order. This is equivalent to taking
+	// the slice produced by AllIndexes and removing the primary index.
 	DeletableNonPrimaryIndexes() []Index
-	DeleteOnlyNonPrimaryIndexes() []Index
 
-	ForEachIndex(opts IndexOpts, f func(idx Index) error) error
-	ForEachActiveIndex(f func(idx Index) error) error
-	ForEachNonDropIndex(f func(idx Index) error) error
-	ForEachPartialIndex(f func(idx Index) error) error
-	ForEachPublicNonPrimaryIndex(f func(idx Index) error) error
-	ForEachWritableNonPrimaryIndex(f func(idx Index) error) error
-	ForEachDeletableNonPrimaryIndex(f func(idx Index) error) error
-	ForEachDeleteOnlyNonPrimaryIndex(f func(idx Index) error) error
-
-	FindIndex(opts IndexOpts, test func(idx Index) bool) Index
-	FindActiveIndex(test func(idx Index) bool) Index
-	FindNonDropIndex(test func(idx Index) bool) Index
-	FindPartialIndex(test func(idx Index) bool) Index
-	FindPublicNonPrimaryIndex(test func(idx Index) bool) Index
-	FindWritableNonPrimaryIndex(test func(idx Index) bool) Index
-	FindDeletableNonPrimaryIndex(test func(idx Index) bool) Index
-	FindDeleteOnlyNonPrimaryIndex(test func(idx Index) bool) Index
+	// DeleteOnlyNonPrimaryIndexes returns a slice of all non-primary indexes
+	// which allow only being deleted from, in their canonical order. This is
+	// equivalent to taking the slice produced by DeletableNonPrimaryIndexes and
+	// removing the indexes which are not in mutations or not in the delete-only
+	// state.
+	DeleteOnlyNonPrimaryIndexes() []Index
 
+	// FindIndexWithID returns the first catalog.Index that matches the id
+	// in the set of all indexes, or an error if none was found. The order of
+	// traversal is the canonical order, see Index.Ordinal().
 	FindIndexWithID(id descpb.IndexID) (Index, error)
+
+	// FindIndexWithName returns the first catalog.Index that matches the name in
+	// the set of all indexes, excluding the primary index of non-physical
+	// tables, or an error if none was found. The order of traversal is the
+	// canonical order, see Index.Ordinal().
 	FindIndexWithName(name string) (Index, error)
 
 	HasPrimaryKey() bool
@@ -208,17 +248,52 @@ type TableDescriptor interface {
 
 // Index is an interface around the index descriptor types.
 type Index interface {
+
+	// IndexDesc returns the underlying protobuf descriptor.
+	// Ideally, this method should be called as rarely as possible.
 	IndexDesc() *descpb.IndexDescriptor
+
+	// IndexDescDeepCopy returns a deep copy of the underlying proto.
 	IndexDescDeepCopy() descpb.IndexDescriptor
 
+	// Ordinal returns the ordinal of the index in its parent table descriptor.
+	//
+	// The ordinal of an index in a `tableDesc descpb.TableDescriptor` is
+	// defined as follows:
+	// - 0 is the ordinal of the primary index,
+	// - [1:1+len(tableDesc.Indexes)] is the range of public non-primary indexes,
+	// - [1+len(tableDesc.Indexes):] is the range of non-public indexes.
+	//
+	// In terms of a `table catalog.TableDescriptor` interface, it is defined
+	// as the catalog.Index object's position in the table.AllIndexes() slice.
 	Ordinal() int
+
+	// Primary returns true iff the index is the primary index for the table
+	// descriptor.
 	Primary() bool
+
+	// Public returns true iff the index is active, i.e. readable, in the table
+	// descriptor.
 	Public() bool
+
+	// WriteAndDeleteOnly returns true iff the index is a mutation in the
+	// delete-and-write-only state in the table descriptor.
 	WriteAndDeleteOnly() bool
+
+	// DeleteOnly returns true iff the index is a mutation in the delete-only
+	// state in the table descriptor.
 	DeleteOnly() bool
+
+	// Adding returns true iff the index is an add mutation in the table
+	// descriptor.
 	Adding() bool
+
+	// Dropped returns true iff the index is a drop mutation in the table
+	// descriptor.
 	Dropped() bool
 
+	// The remaining methods operate on the underlying descpb.IndexDescriptor object.
+
 	GetID() descpb.IndexID
 	GetName() string
 	IsInterleaved() bool
@@ -229,34 +304,44 @@ type Index interface {
 	IsCreatedExplicitly() bool
 	GetPredicate() string
 	GetType() descpb.IndexDescriptor_Type
-	IsValidOriginIndex(originColIDs descpb.ColumnIDs) bool
-	IsValidReferencedIndex(referencedColIDs descpb.ColumnIDs) bool
 	GetGeoConfig() geoindex.Config
-	GetSharded() descpb.ShardedDescriptor
-	GetShardColumnName() string
 	GetVersion() descpb.IndexDescriptorVersion
 	GetEncodingType() descpb.IndexDescriptorEncodingType
+
+	GetSharded() descpb.ShardedDescriptor
+	GetShardColumnName() string
+
+	IsValidOriginIndex(originColIDs descpb.ColumnIDs) bool
+	IsValidReferencedIndex(referencedColIDs descpb.ColumnIDs) bool
+
 	GetPartitioning() descpb.PartitioningDescriptor
 	FindPartitionByName(name string) descpb.PartitioningDescriptor
 	PartitionNames() []string
+
 	NumInterleaveAncestors() int
 	GetInterleaveAncestor(ancestorOrdinal int) descpb.InterleaveDescriptor_Ancestor
+
 	NumInterleavedBy() int
 	GetInterleavedBy(interleavedByOrdinal int) descpb.ForeignKeyReference
+
 	NumColumns() int
 	GetColumnID(columnOrdinal int) descpb.ColumnID
 	GetColumnName(columnOrdinal int) string
 	GetColumnDirection(columnOrdinal int) descpb.IndexDescriptor_Direction
+
+	ForEachColumnID(func(id descpb.ColumnID) error) error
 	ContainsColumnID(colID descpb.ColumnID) bool
 	InvertedColumnID() descpb.ColumnID
 	InvertedColumnName() string
-	ForEachColumnID(func(id descpb.ColumnID) error) error
+
 	NumStoredColumns() int
 	GetStoredColumnID(storedColumnOrdinal int) descpb.ColumnID
 	GetStoredColumnName(storedColumnOrdinal int) string
 	HasOldStoredColumns() bool
+
 	NumExtraColumns() int
 	GetExtraColumnID(extraColumnOrdinal int) descpb.ColumnID
+
 	NumCompositeColumns() int
 	GetCompositeColumnID(compositeColumnOrdinal int) descpb.ColumnID
 }
@@ -356,3 +441,151 @@ func FormatSafeDescriptorProperties(w *redact.StringBuilder, desc Descriptor) {
 		w.Printf(", NumDrainingNames: %d", len(drainingNames))
 	}
 }
+
+func isIndexInSearchSet(desc TableDescriptor, opts IndexOpts, idx Index) bool {
+	if !opts.NonPhysicalPrimaryIndex && idx.Primary() && !desc.IsPhysicalTable() {
+		return false
+	}
+	if !opts.AddMutations && idx.Adding() {
+		return false
+	}
+	if !opts.DropMutations && idx.Dropped() {
+		return false
+	}
+	return true
+}
+
+// ForEachIndex runs f over each index in the table descriptor according to
+// filter parameters in opts. Indexes are visited in their canonical order,
+// see Index.Ordinal(). ForEachIndex supports iterutil.StopIteration().
+func ForEachIndex(desc TableDescriptor, opts IndexOpts, f func(idx Index) error) error {
+	for _, idx := range desc.AllIndexes() {
+		if !isIndexInSearchSet(desc, opts, idx) {
+			continue
+		}
+		if err := f(idx); err != nil {
+			if iterutil.Done(err) {
+				return nil
+			}
+			return err
+		}
+	}
+	return nil
+}
+
+func forEachIndex(slice []Index, f func(idx Index) error) error {
+	for _, idx := range slice {
+		if err := f(idx); err != nil {
+			if iterutil.Done(err) {
+				return nil
+			}
+			return err
+		}
+	}
+	return nil
+}
+
+// ForEachActiveIndex is like ForEachIndex over ActiveIndexes().
+func ForEachActiveIndex(desc TableDescriptor, f func(idx Index) error) error {
+	return forEachIndex(desc.ActiveIndexes(), f)
+}
+
+// ForEachNonDropIndex is like ForEachIndex over NonDropIndexes().
+func ForEachNonDropIndex(desc TableDescriptor, f func(idx Index) error) error {
+	return forEachIndex(desc.NonDropIndexes(), f)
+}
+
+// ForEachPartialIndex is like ForEachIndex over PartialIndexes().
+func ForEachPartialIndex(desc TableDescriptor, f func(idx Index) error) error {
+	return forEachIndex(desc.PartialIndexes(), f)
+}
+
+// ForEachPublicNonPrimaryIndex is like ForEachIndex over
+// PublicNonPrimaryIndexes().
+func ForEachPublicNonPrimaryIndex(desc TableDescriptor, f func(idx Index) error) error {
+	return forEachIndex(desc.PublicNonPrimaryIndexes(), f)
+}
+
+// ForEachWritableNonPrimaryIndex is like ForEachIndex over
+// WritableNonPrimaryIndexes().
+func ForEachWritableNonPrimaryIndex(desc TableDescriptor, f func(idx Index) error) error {
+	return forEachIndex(desc.WritableNonPrimaryIndexes(), f)
+}
+
+// ForEachDeletableNonPrimaryIndex is like ForEachIndex over
+// DeletableNonPrimaryIndexes().
+func ForEachDeletableNonPrimaryIndex(desc TableDescriptor, f func(idx Index) error) error {
+	return forEachIndex(desc.DeletableNonPrimaryIndexes(), f)
+}
+
+// ForEachDeleteOnlyNonPrimaryIndex is like ForEachIndex over
+// DeleteOnlyNonPrimaryIndexes().
+func ForEachDeleteOnlyNonPrimaryIndex(desc TableDescriptor, f func(idx Index) error) error {
+	return forEachIndex(desc.DeleteOnlyNonPrimaryIndexes(), f)
+}
+
+// FindIndex returns the first index for which test returns true, nil otherwise,
+// according to the parameters in opts just like ForEachIndex.
+// Indexes are visited in their canonical order, see Index.Ordinal().
+func FindIndex(desc TableDescriptor, opts IndexOpts, test func(idx Index) bool) Index {
+	for _, idx := range desc.AllIndexes() {
+		if !isIndexInSearchSet(desc, opts, idx) {
+			continue
+		}
+		if test(idx) {
+			return idx
+		}
+	}
+	return nil
+}
+
+func findIndex(slice []Index, test func(idx Index) bool) Index {
+	for _, idx := range slice {
+		if test(idx) {
+			return idx
+		}
+	}
+	return nil
+}
+
+// FindActiveIndex returns the first index in ActiveIndex() for which test
+// returns true.
+func FindActiveIndex(desc TableDescriptor, test func(idx Index) bool) Index {
+	return findIndex(desc.ActiveIndexes(), test)
+}
+
+// FindNonDropIndex returns the first index in NonDropIndex() for which test
+// returns true.
+func FindNonDropIndex(desc TableDescriptor, test func(idx Index) bool) Index {
+	return findIndex(desc.NonDropIndexes(), test)
+}
+
+// FindPartialIndex returns the first index in PartialIndex() for which test
+// returns true.
+func FindPartialIndex(desc TableDescriptor, test func(idx Index) bool) Index {
+	return findIndex(desc.PartialIndexes(), test)
+}
+
+// FindPublicNonPrimaryIndex returns the first index in PublicNonPrimaryIndex()
+// for which test returns true.
+func FindPublicNonPrimaryIndex(desc TableDescriptor, test func(idx Index) bool) Index {
+	return findIndex(desc.PublicNonPrimaryIndexes(), test)
+}
+
+// FindWritableNonPrimaryIndex returns the first index in
+// WritableNonPrimaryIndex() for which test returns true.
+func FindWritableNonPrimaryIndex(desc TableDescriptor, test func(idx Index) bool) Index {
+	return findIndex(desc.WritableNonPrimaryIndexes(), test)
+}
+
+// FindDeletableNonPrimaryIndex returns the first index in
+// DeletableNonPrimaryIndex() for which test returns true.
+func FindDeletableNonPrimaryIndex(desc TableDescriptor, test func(idx Index) bool) Index {
+	return findIndex(desc.DeletableNonPrimaryIndexes(), test)
+}
+
+// FindDeleteOnlyNonPrimaryIndex returns the first index in
+// DeleteOnlyNonPrimaryIndex() for which test returns true.
+func FindDeleteOnlyNonPrimaryIndex(desc TableDescriptor, test func(idx Index) bool) Index {
+	return findIndex(desc.DeleteOnlyNonPrimaryIndexes(), test)
+}
diff --git a/pkg/sql/catalog/tabledesc/index.go b/pkg/sql/catalog/tabledesc/index.go
index 9a4f76792ab5..e13cc7fcfe5a 100644
--- a/pkg/sql/catalog/tabledesc/index.go
+++ b/pkg/sql/catalog/tabledesc/index.go
@@ -40,7 +40,11 @@ func (w index) IndexDescDeepCopy() descpb.IndexDescriptor {
 	return *protoutil.Clone(w.desc).(*descpb.IndexDescriptor)
 }
 
-// Ordinal returns the ordinal of the index within the table descriptor.
+// Ordinal returns the ordinal of the index in its parent TableDescriptor.
+// The ordinal is defined as follows:
+// - 0 is the ordinal of the primary index,
+// - [1:1+len(desc.Indexes)] is the range of public non-primary indexes,
+// - [1+len(desc.Indexes):] is the range of non-public indexes.
 func (w index) Ordinal() int {
 	return w.ordinal
 }
@@ -61,7 +65,7 @@ func (w index) Adding() bool {
 	return w.mutationDirection == descpb.DescriptorMutation_ADD
 }
 
-// Adding returns true iff the index is a drop mutation in the table descriptor.
+// Dropped returns true iff the index is a drop mutation in the table descriptor.
 func (w index) Dropped() bool {
 	return w.mutationDirection == descpb.DescriptorMutation_DROP
 }
diff --git a/pkg/sql/catalog/tabledesc/safe_format.go b/pkg/sql/catalog/tabledesc/safe_format.go
index 499205430a54..db975755e374 100644
--- a/pkg/sql/catalog/tabledesc/safe_format.go
+++ b/pkg/sql/catalog/tabledesc/safe_format.go
@@ -110,7 +110,7 @@ func formatSafeTableIndexes(w *redact.StringBuilder, desc catalog.TableDescripto
 	w.Printf(", PrimaryIndex: %d", desc.GetPrimaryIndexID())
 	w.Printf(", NextIndexID: %d", desc.TableDesc().NextIndexID)
 	w.Printf(", Indexes: [")
-	_ = desc.ForEachActiveIndex(func(idx catalog.Index) error {
+	_ = catalog.ForEachActiveIndex(desc, func(idx catalog.Index) error {
 		if !idx.Primary() {
 			w.Printf(", ")
 		}
diff --git a/pkg/sql/catalog/tabledesc/structured.go b/pkg/sql/catalog/tabledesc/structured.go
index 2b9019ddbea6..1ad6d78a04be 100644
--- a/pkg/sql/catalog/tabledesc/structured.go
+++ b/pkg/sql/catalog/tabledesc/structured.go
@@ -847,7 +847,7 @@ func ForEachExprStringInTableDesc(descI catalog.TableDescriptor, f func(expr *st
 	}
 
 	// Process all indexes.
-	if err := descI.ForEachIndex(catalog.IndexOpts{
+	if err := catalog.ForEachIndex(descI, catalog.IndexOpts{
 		NonPhysicalPrimaryIndex: true,
 		DropMutations:           true,
 		AddMutations:            true,
@@ -1048,7 +1048,7 @@ func (desc *Mutable) allocateIndexIDs(columnNames map[string]descpb.ColumnID) er
 	}
 
 	// Assign names to unnamed indexes.
-	_ = desc.ForEachDeletableNonPrimaryIndex(func(idx catalog.Index) error {
+	_ = catalog.ForEachDeletableNonPrimaryIndex(desc, func(idx catalog.Index) error {
 		if len(idx.GetName()) == 0 {
 			idx.IndexDesc().Name = buildIndexName(desc, idx)
 		}
@@ -1437,7 +1437,7 @@ func (desc *wrapper) validateCrossReferences(ctx context.Context, dg catalog.Des
 		// un-upgraded foreign key references on the other table. This logic
 		// somewhat parallels the logic in maybeUpgradeForeignKeyRepOnIndex.
 		unupgradedFKsPresent := false
-		if err := referencedTable.ForEachIndex(catalog.IndexOpts{}, func(referencedIdx catalog.Index) error {
+		if err := catalog.ForEachIndex(referencedTable, catalog.IndexOpts{}, func(referencedIdx catalog.Index) error {
 			if found {
 				// TODO (lucy): If we ever revisit the tabledesc.Immutable methods, add
 				// a way to break out of the index loop.
@@ -1516,7 +1516,7 @@ func (desc *wrapper) validateCrossReferences(ctx context.Context, dg catalog.Des
 		// un-upgraded foreign key references on the other table. This logic
 		// somewhat parallels the logic in maybeUpgradeForeignKeyRepOnIndex.
 		unupgradedFKsPresent := false
-		if err := originTable.ForEachIndex(catalog.IndexOpts{}, func(originIdx catalog.Index) error {
+		if err := catalog.ForEachIndex(originTable, catalog.IndexOpts{}, func(originIdx catalog.Index) error {
 			if found {
 				// TODO (lucy): If we ever revisit the tabledesc.Immutable methods, add
 				// a way to break out of the index loop.
@@ -1740,7 +1740,7 @@ func ValidateTableLocalityConfig(
 
 // ValidateIndexNameIsUnique validates that the index name does not exist.
 func (desc *wrapper) ValidateIndexNameIsUnique(indexName string) error {
-	if desc.FindNonDropIndex(func(idx catalog.Index) bool {
+	if catalog.FindNonDropIndex(desc, func(idx catalog.Index) bool {
 		return idx.GetName() == indexName
 	}) != nil {
 		return sqlerrors.NewRelationAlreadyExistsError(indexName)
@@ -2494,7 +2494,7 @@ func (desc *wrapper) validatePartitioning() error {
 	partitionNames := make(map[string]string)
 
 	a := &rowenc.DatumAlloc{}
-	return desc.ForEachNonDropIndex(func(idx catalog.Index) error {
+	return catalog.ForEachNonDropIndex(desc, func(idx catalog.Index) error {
 		idxDesc := idx.IndexDesc()
 		return desc.validatePartitioningDescriptor(
 			a, idxDesc, &idxDesc.Partitioning, 0 /* colOffset */, partitionNames,
@@ -3228,7 +3228,7 @@ func (desc *wrapper) FindFKByName(name string) (*descpb.ForeignKeyConstraint, er
 // IsInterleaved returns true if any part of this this table is interleaved with
 // another table's data.
 func (desc *wrapper) IsInterleaved() bool {
-	return nil != desc.FindNonDropIndex(func(idx catalog.Index) bool {
+	return nil != catalog.FindNonDropIndex(desc, func(idx catalog.Index) bool {
 		return idx.IsInterleaved()
 	})
 }
@@ -4071,7 +4071,7 @@ func (desc *Immutable) MutationColumns() []descpb.ColumnDescriptor {
 // IsShardColumn returns true if col corresponds to a non-dropped hash sharded
 // index. This method assumes that col is currently a member of desc.
 func (desc *Mutable) IsShardColumn(col *descpb.ColumnDescriptor) bool {
-	return nil != desc.FindNonDropIndex(func(idx catalog.Index) bool {
+	return nil != catalog.FindNonDropIndex(desc, func(idx catalog.Index) bool {
 		return idx.IsSharded() && idx.GetShardColumnName() == col.Name
 	})
 }
diff --git a/pkg/sql/catalog/tabledesc/table_desc.go b/pkg/sql/catalog/tabledesc/table_desc.go
index 1290f2147a75..fa83f8907817 100644
--- a/pkg/sql/catalog/tabledesc/table_desc.go
+++ b/pkg/sql/catalog/tabledesc/table_desc.go
@@ -267,199 +267,84 @@ func (desc *wrapper) getExistingOrNewIndexCache() *indexCache {
 	return desc.indexCache
 }
 
-// AllIndexes returns a slice containing all indexes represented in the table
-// descriptor, including mutations.
+// AllIndexes returns a slice with all indexes, public and non-public,
+// in the underlying proto, in their canonical order:
+// - the primary index,
+// - the public non-primary indexes in the Indexes array, in order,
+// - the non-public indexes present in the Mutations array, in order.
+//
+// See also catalog.Index.Ordinal().
 func (desc *wrapper) AllIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().allIndexes(desc)
 }
 
-// ActiveIndexes returns a slice of all active (aka public) indexes.
+// ActiveIndexes returns a slice with all public indexes in the underlying
+// proto, in their canonical order:
+// - the primary index,
+// - the public non-primary indexes in the Indexes array, in order.
+//
+// See also catalog.Index.Ordinal().
 func (desc *wrapper) ActiveIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().activeIndexes(desc)
 }
 
-// NonDropIndexes returns a slice of all indexes (including mutations) which are
-// not being dropped.
+// NonDropIndexes returns a slice of all non-drop indexes in the underlying
+// proto, in their canonical order. This means:
+// - the primary index, if the table is a physical table,
+// - the public non-primary indexes in the Indexes array, in order,
+// - the non-public indexes present in the Mutations array, in order,
+//   if the mutation is not a drop.
+//
+// See also catalog.Index.Ordinal().
 func (desc *wrapper) NonDropIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().nonDropIndexes(desc)
 }
 
-// PartialIndexes returns a slice of all partial indexes in the table
-// descriptor, including mutations.
+// NonDropIndexes returns a slice of all partial indexes in the underlying
+// proto, in their canonical order. This is equivalent to taking the slice
+// produced by AllIndexes and filtering indexes with non-empty expressions.
 func (desc *wrapper) PartialIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().partialIndexes(desc)
 }
 
-// PublicNonPrimaryIndexes returns a slice of all active secondary indexes.
+// PublicNonPrimaryIndexes returns a slice of all active secondary indexes,
+// in their canonical order. This is equivalent to the Indexes array in the
+// proto.
 func (desc *wrapper) PublicNonPrimaryIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().publicNonPrimaryIndexes(desc)
 }
 
-// WritableNonPrimaryIndexes returns a slice of all secondary indexes which
-// allow being written to: active + delete-and-write-only.
+// WritableNonPrimaryIndexes returns a slice of all non-primary indexes which
+// allow being written to: public + delete-and-write-only, in their canonical
+// order. This is equivalent to taking the slice produced by
+// DeletableNonPrimaryIndexes and removing the indexes which are in mutations
+// in the delete-only state.
 func (desc *wrapper) WritableNonPrimaryIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().writableNonPrimaryIndexes(desc)
 }
 
-// DeletableNonPrimaryIndexes returns a slice of all secondary indexes which
-// allow being deleted from: active + delete-and-write-only + delete-only.
+// DeletableNonPrimaryIndexes returns a slice of all non-primary indexes
+// which allow being deleted from: public + delete-and-write-only +
+// delete-only, in  their canonical order. This is equivalent to taking
+// the slice produced by AllIndexes and removing the primary index.
 func (desc *wrapper) DeletableNonPrimaryIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().deletableNonPrimaryIndexes(desc)
 }
 
-// DeletableNonPrimaryIndexes returns a slice of all secondary indexes which
-// only allow being deleted from.
+// DeleteOnlyNonPrimaryIndexes returns a slice of all non-primary indexes
+// which allow only being deleted from, in their canonical order. This is
+// equivalent to taking the slice produced by DeletableNonPrimaryIndexes and
+// removing the indexes which are not in mutations or not in the delete-only
+// state.
 func (desc *wrapper) DeleteOnlyNonPrimaryIndexes() []catalog.Index {
 	return desc.getExistingOrNewIndexCache().deleteOnlyNonPrimaryIndexes(desc)
 }
 
-// ForEachIndex runs f over each index in the table descriptor according to
-// filter parameters in opts.
-func (desc *wrapper) ForEachIndex(opts catalog.IndexOpts, f func(idx catalog.Index) error) error {
-	for _, idx := range desc.AllIndexes() {
-		if !opts.NonPhysicalPrimaryIndex && idx.Primary() && !desc.IsPhysicalTable() {
-			continue
-		}
-		if !opts.AddMutations && idx.Adding() {
-			continue
-		}
-		if !opts.DropMutations && idx.Dropped() {
-			continue
-		}
-		if err := f(idx); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func forEachIndex(slice []catalog.Index, f func(idx catalog.Index) error) error {
-	for _, idx := range slice {
-		if err := f(idx); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-// ForEachActiveIndex is like ForEachIndex over ActiveIndexes().
-func (desc *wrapper) ForEachActiveIndex(f func(idx catalog.Index) error) error {
-	return forEachIndex(desc.ActiveIndexes(), f)
-}
-
-// ForEachNonDropIndex is like ForEachIndex over NonDropIndexes().
-func (desc *wrapper) ForEachNonDropIndex(f func(idx catalog.Index) error) error {
-	return forEachIndex(desc.NonDropIndexes(), f)
-}
-
-// ForEachPartialIndex is like ForEachIndex over PartialIndexes().
-func (desc *wrapper) ForEachPartialIndex(f func(idx catalog.Index) error) error {
-	return forEachIndex(desc.PartialIndexes(), f)
-}
-
-// ForEachPublicNonPrimaryIndex is like ForEachIndex over
-// PublicNonPrimaryIndexes().
-func (desc *wrapper) ForEachPublicNonPrimaryIndex(f func(idx catalog.Index) error) error {
-	return forEachIndex(desc.PublicNonPrimaryIndexes(), f)
-}
-
-// ForEachWritableNonPrimaryIndex is like ForEachIndex over
-// WritableNonPrimaryIndexes().
-func (desc *wrapper) ForEachWritableNonPrimaryIndex(f func(idx catalog.Index) error) error {
-	return forEachIndex(desc.WritableNonPrimaryIndexes(), f)
-}
-
-// ForEachDeletableNonPrimaryIndex is like ForEachIndex over
-// DeletableNonPrimaryIndexes().
-func (desc *wrapper) ForEachDeletableNonPrimaryIndex(f func(idx catalog.Index) error) error {
-	return forEachIndex(desc.DeletableNonPrimaryIndexes(), f)
-}
-
-// ForEachDeleteOnlyNonPrimaryIndex is like ForEachIndex over
-// DeleteOnlyNonPrimaryIndexes().
-func (desc *wrapper) ForEachDeleteOnlyNonPrimaryIndex(f func(idx catalog.Index) error) error {
-	return forEachIndex(desc.DeleteOnlyNonPrimaryIndexes(), f)
-}
-
-// FindIndex returns the first index for which test returns true, nil otherwise,
-// according to the parameters in opts just like ForEachIndex.
-func (desc *wrapper) FindIndex(
-	opts catalog.IndexOpts, test func(idx catalog.Index) bool,
-) catalog.Index {
-	for _, idx := range desc.AllIndexes() {
-		if !opts.NonPhysicalPrimaryIndex && idx.Primary() && !desc.IsPhysicalTable() {
-			continue
-		}
-		if !opts.AddMutations && idx.Adding() {
-			continue
-		}
-		if !opts.DropMutations && idx.Dropped() {
-			continue
-		}
-		if test(idx) {
-			return idx
-		}
-	}
-	return nil
-}
-
-func findIndex(slice []catalog.Index, test func(idx catalog.Index) bool) catalog.Index {
-	for _, idx := range slice {
-		if test(idx) {
-			return idx
-		}
-	}
-	return nil
-}
-
-// FindActiveIndex returns the first index in ActiveIndex() for which test
-// returns true.
-func (desc *wrapper) FindActiveIndex(test func(idx catalog.Index) bool) catalog.Index {
-	return findIndex(desc.ActiveIndexes(), test)
-}
-
-// FindNonDropIndex returns the first index in NonDropIndex() for which test
-// returns true.
-func (desc *wrapper) FindNonDropIndex(test func(idx catalog.Index) bool) catalog.Index {
-	return findIndex(desc.NonDropIndexes(), test)
-}
-
-// FindPartialIndex returns the first index in PartialIndex() for which test
-// returns true.
-func (desc *wrapper) FindPartialIndex(test func(idx catalog.Index) bool) catalog.Index {
-	return findIndex(desc.PartialIndexes(), test)
-}
-
-// FindPublicNonPrimaryIndex returns the first index in PublicNonPrimaryIndex()
-// for which test returns true.
-func (desc *wrapper) FindPublicNonPrimaryIndex(test func(idx catalog.Index) bool) catalog.Index {
-	return findIndex(desc.PublicNonPrimaryIndexes(), test)
-}
-
-// FindWritableNonPrimaryIndex returns the first index in
-// WritableNonPrimaryIndex() for which test returns true.
-func (desc *wrapper) FindWritableNonPrimaryIndex(test func(idx catalog.Index) bool) catalog.Index {
-	return findIndex(desc.WritableNonPrimaryIndexes(), test)
-}
-
-// FindDeletableNonPrimaryIndex returns the first index in
-// DeletableNonPrimaryIndex() for which test returns true.
-func (desc *wrapper) FindDeletableNonPrimaryIndex(test func(idx catalog.Index) bool) catalog.Index {
-	return findIndex(desc.DeletableNonPrimaryIndexes(), test)
-}
-
-// FindDeleteOnlyNonPrimaryIndex returns the first index in
-// DeleteOnlyNonPrimaryIndex() for which test returns true.
-func (desc *wrapper) FindDeleteOnlyNonPrimaryIndex(
-	test func(idx catalog.Index) bool,
-) catalog.Index {
-	return findIndex(desc.DeleteOnlyNonPrimaryIndexes(), test)
-}
-
 // FindIndexWithID returns the first catalog.Index that matches the id
-// in the set of all indexes.
+// in the set of all indexes, or an error if none was found. The order of
+// traversal is the canonical order, see catalog.Index.Ordinal().
 func (desc *wrapper) FindIndexWithID(id descpb.IndexID) (catalog.Index, error) {
-	if idx := desc.FindIndex(catalog.IndexOpts{
+	if idx := catalog.FindIndex(desc, catalog.IndexOpts{
 		NonPhysicalPrimaryIndex: true,
 		DropMutations:           true,
 		AddMutations:            true,
@@ -477,9 +362,11 @@ func (desc *wrapper) FindIndexWithID(id descpb.IndexID) (catalog.Index, error) {
 }
 
 // FindIndexWithName returns the first catalog.Index that matches the name in
-// the set of all indexes, excluding the primary index of non-physical tables.
+// the set of all indexes, excluding the primary index of non-physical
+// tables, or an error if none was found. The order of traversal is the
+// canonical order, see catalog.Index.Ordinal().
 func (desc *wrapper) FindIndexWithName(name string) (catalog.Index, error) {
-	if idx := desc.FindIndex(catalog.IndexOpts{
+	if idx := catalog.FindIndex(desc, catalog.IndexOpts{
 		NonPhysicalPrimaryIndex: false,
 		DropMutations:           true,
 		AddMutations:            true,
diff --git a/pkg/sql/crdb_internal.go b/pkg/sql/crdb_internal.go
index ebb6313a1961..b5f1e76064e7 100644
--- a/pkg/sql/crdb_internal.go
+++ b/pkg/sql/crdb_internal.go
@@ -1823,7 +1823,7 @@ CREATE TABLE crdb_internal.create_statements (
 		if createNofk == "" {
 			createNofk = stmt
 		}
-		hasPartitions := nil != table.FindIndex(catalog.IndexOpts{}, func(idx catalog.Index) bool {
+		hasPartitions := nil != catalog.FindIndex(table, catalog.IndexOpts{}, func(idx catalog.Index) bool {
 			return idx.GetPartitioning().NumColumns != 0
 		})
 		return addRow(
@@ -2050,7 +2050,7 @@ CREATE TABLE crdb_internal.table_indexes (
 					tableName := tree.NewDString(table.GetName())
 					// We report the primary index of non-physical tables here. These
 					// indexes are not reported as a part of ForeachIndex.
-					return table.ForEachIndex(catalog.IndexOpts{
+					return catalog.ForEachIndex(table, catalog.IndexOpts{
 						NonPhysicalPrimaryIndex: true,
 					}, func(idx catalog.Index) error {
 						row = row[:0]
@@ -2179,7 +2179,7 @@ CREATE TABLE crdb_internal.index_columns (
 					return nil
 				}
 
-				return table.ForEachIndex(catalog.IndexOpts{NonPhysicalPrimaryIndex: true}, reportIndex)
+				return catalog.ForEachIndex(table, catalog.IndexOpts{NonPhysicalPrimaryIndex: true}, reportIndex)
 			})
 	},
 }
@@ -2261,7 +2261,7 @@ CREATE TABLE crdb_internal.backward_dependencies (
 			}
 
 			// Record the backward references of the primary index.
-			if err := table.ForEachIndex(catalog.IndexOpts{}, reportIdxDeps); err != nil {
+			if err := catalog.ForEachIndex(table, catalog.IndexOpts{}, reportIdxDeps); err != nil {
 				return err
 			}
 
@@ -2390,7 +2390,7 @@ CREATE TABLE crdb_internal.forward_dependencies (
 				}
 
 				// Record the backward references of the primary index.
-				if err := table.ForEachIndex(catalog.IndexOpts{}, reportIdxDeps); err != nil {
+				if err := catalog.ForEachIndex(table, catalog.IndexOpts{}, reportIdxDeps); err != nil {
 					return err
 				}
 				reportDependedOnBy := func(
@@ -2820,7 +2820,7 @@ CREATE TABLE crdb_internal.zones (
 				}
 
 				for i, s := range subzones {
-					index := table.FindActiveIndex(func(idx catalog.Index) bool {
+					index := catalog.FindActiveIndex(table, func(idx catalog.Index) bool {
 						return idx.GetID() == descpb.IndexID(s.IndexID)
 					})
 					if index == nil {
@@ -3404,7 +3404,7 @@ CREATE TABLE crdb_internal.partitions (
 		worker := func(pusher rowPusher) error {
 			return forEachTableDescAll(ctx, p, dbContext, hideVirtual, /* virtual tables have no partitions*/
 				func(db *dbdesc.Immutable, _ string, table catalog.TableDescriptor) error {
-					return table.ForEachIndex(catalog.IndexOpts{
+					return catalog.ForEachIndex(table, catalog.IndexOpts{
 						AddMutations: true,
 					}, func(index catalog.Index) error {
 						return addPartitioningRows(ctx, p, dbName, table, index.IndexDesc(), &index.IndexDesc().Partitioning,
diff --git a/pkg/sql/create_table.go b/pkg/sql/create_table.go
index 75e158efce9a..57df9d1981d7 100644
--- a/pkg/sql/create_table.go
+++ b/pkg/sql/create_table.go
@@ -1886,7 +1886,7 @@ func NewTableDesc(
 	}
 
 	// Record the types of indexes that the table has.
-	if err := desc.ForEachNonDropIndex(func(idx catalog.Index) error {
+	if err := catalog.ForEachNonDropIndex(&desc, func(idx catalog.Index) error {
 		if idx.IsSharded() {
 			telemetry.Inc(sqltelemetry.HashShardedIndexCounter)
 		}
diff --git a/pkg/sql/drop_index.go b/pkg/sql/drop_index.go
index 5627069b7dbf..2beb76a77bb6 100644
--- a/pkg/sql/drop_index.go
+++ b/pkg/sql/drop_index.go
@@ -202,7 +202,7 @@ func (n *dropIndexNode) maybeDropShardColumn(
 	if dropped {
 		return nil
 	}
-	if tableDesc.FindNonDropIndex(func(otherIdx catalog.Index) bool {
+	if catalog.FindNonDropIndex(tableDesc, func(otherIdx catalog.Index) bool {
 		return otherIdx.ContainsColumnID(shardColDesc.ID)
 	}) != nil {
 		return nil
@@ -456,7 +456,7 @@ func (p *planner) dropIndexByName(
 		)
 	}
 
-	foundIndex := tableDesc.FindPublicNonPrimaryIndex(func(idxEntry catalog.Index) bool {
+	foundIndex := catalog.FindPublicNonPrimaryIndex(tableDesc, func(idxEntry catalog.Index) bool {
 		return idxEntry.GetID() == idx.ID
 	})
 
diff --git a/pkg/sql/information_schema.go b/pkg/sql/information_schema.go
index ccaaa4b6f489..d4dfae34c806 100644
--- a/pkg/sql/information_schema.go
+++ b/pkg/sql/information_schema.go
@@ -1297,7 +1297,7 @@ CREATE TABLE information_schema.statistics (
 					)
 				}
 
-				return table.ForEachIndex(catalog.IndexOpts{}, func(index catalog.Index) error {
+				return catalog.ForEachIndex(table, catalog.IndexOpts{}, func(index catalog.Index) error {
 					// Columns in the primary key that aren't in index.ColumnNames or
 					// index.StoreColumnNames are implicit columns in the index.
 					var implicitCols map[string]struct{}
diff --git a/pkg/sql/partition_utils.go b/pkg/sql/partition_utils.go
index 3e74449a6ac1..f354a28c7e2b 100644
--- a/pkg/sql/partition_utils.go
+++ b/pkg/sql/partition_utils.go
@@ -100,7 +100,7 @@ func GenerateSubzoneSpans(
 
 	var indexCovering covering.Covering
 	var partitionCoverings []covering.Covering
-	if err := tableDesc.ForEachIndex(catalog.IndexOpts{
+	if err := catalog.ForEachIndex(tableDesc, catalog.IndexOpts{
 		AddMutations: true,
 	}, func(idx catalog.Index) error {
 		_, indexSubzoneExists := subzoneIndexByIndexID[idx.GetID()]
diff --git a/pkg/sql/pg_catalog.go b/pkg/sql/pg_catalog.go
index 741106caeb0a..0e0e18dce763 100644
--- a/pkg/sql/pg_catalog.go
+++ b/pkg/sql/pg_catalog.go
@@ -445,7 +445,7 @@ https://www.postgresql.org/docs/12/catalog-pg-attribute.html`,
 
 		// Columns for each index.
 		columnIdxMap := table.ColumnIdxMap()
-		return table.ForEachIndex(catalog.IndexOpts{}, func(index catalog.Index) error {
+		return catalog.ForEachIndex(table, catalog.IndexOpts{}, func(index catalog.Index) error {
 			for i := 0; i < index.NumColumns(); i++ {
 				colID := index.GetColumnID(i)
 				idxID := h.IndexOid(table.GetID(), index.GetID())
@@ -625,7 +625,7 @@ https://www.postgresql.org/docs/9.5/catalog-pg-class.html`,
 		}
 
 		// Indexes.
-		return table.ForEachIndex(catalog.IndexOpts{}, func(index catalog.Index) error {
+		return catalog.ForEachIndex(table, catalog.IndexOpts{}, func(index catalog.Index) error {
 			indexType := forwardIndexOid
 			if index.GetType() == descpb.IndexDescriptor_INVERTED {
 				indexType = invertedIndexOid
@@ -1442,7 +1442,7 @@ https://www.postgresql.org/docs/9.5/catalog-pg-index.html`,
 		return forEachTableDesc(ctx, p, dbContext, hideVirtual, /* virtual tables do not have indexes */
 			func(db *dbdesc.Immutable, scName string, table catalog.TableDescriptor) error {
 				tableOid := tableOid(table.GetID())
-				return table.ForEachIndex(catalog.IndexOpts{}, func(index catalog.Index) error {
+				return catalog.ForEachIndex(table, catalog.IndexOpts{}, func(index catalog.Index) error {
 					isMutation, isWriteOnly :=
 						table.GetIndexMutationCapabilities(index.GetID())
 					isReady := isMutation && isWriteOnly
@@ -1524,7 +1524,7 @@ https://www.postgresql.org/docs/9.5/view-pg-indexes.html`,
 			func(db *dbdesc.Immutable, scName string, table catalog.TableDescriptor, tableLookup tableLookupFn) error {
 				scNameName := tree.NewDName(scName)
 				tblName := tree.NewDName(table.GetName())
-				return table.ForEachIndex(catalog.IndexOpts{}, func(index catalog.Index) error {
+				return catalog.ForEachIndex(table, catalog.IndexOpts{}, func(index catalog.Index) error {
 					def, err := indexDefFromDescriptor(ctx, p, db, table, index.IndexDesc(), tableLookup)
 					if err != nil {
 						return err
diff --git a/pkg/sql/rowexec/joinreader_test.go b/pkg/sql/rowexec/joinreader_test.go
index 8ddbfde95fe8..83f738d3730d 100644
--- a/pkg/sql/rowexec/joinreader_test.go
+++ b/pkg/sql/rowexec/joinreader_test.go
@@ -1274,7 +1274,7 @@ func BenchmarkJoinReader(b *testing.B) {
 							// Get the table descriptor and find the index that will provide us with
 							// the expected match ratio.
 							tableDesc := catalogkv.TestingGetTableDescriptor(kvDB, keys.SystemSQLCodec, "test", tableName)
-							foundIndex := tableDesc.FindPublicNonPrimaryIndex(func(idx catalog.Index) bool {
+							foundIndex := catalog.FindPublicNonPrimaryIndex(tableDesc, func(idx catalog.Index) bool {
 								require.Equal(b, 1, idx.NumColumns(), "all indexes created in this benchmark should only contain one column")
 								return idx.GetColumnName(0) == columnDef.name
 							})
diff --git a/pkg/sql/truncate.go b/pkg/sql/truncate.go
index 4de6cf76eab8..d1b54b0b40f6 100644
--- a/pkg/sql/truncate.go
+++ b/pkg/sql/truncate.go
@@ -376,7 +376,7 @@ func (p *planner) reassignInterleaveIndexReferences(
 ) error {
 	for _, table := range tables {
 		changed := false
-		if err := table.ForEachNonDropIndex(func(indexI catalog.Index) error {
+		if err := catalog.ForEachNonDropIndex(table, func(indexI catalog.Index) error {
 			index := indexI.IndexDesc()
 			for j, a := range index.Interleave.Ancestors {
 				if a.TableID == truncatedID {