Merge #51462 #51489

51462: storage: optimize MVCCGarbageCollect for large number of undeleted version r=itsbilal a=ajwerner

This is a better-tested take-2 of #51184.

* The first commit introduces a `SupportsPrev` method to the `Iterator` interface.
* The second commit adjusts the benchmark to use a regular batch.
* The third commit adjusts the logic in #51184 to fix the bugs there and adds testing.

The win is now only pronounced on pebble, as might be expected. 

51489: opt, cat: rename Table.AllColumnCount to Table.ColumnCount r=RaduBerinde a=RaduBerinde

This change renames AllColumnCount to ColumnCount. In #51400 I meant
to rename before merging but I forgot.

Release note: None

Co-authored-by: Andrew Werner <[email protected]>
Co-authored-by: Radu Berinde <[email protected]>

pkg/kv/kvserver/spanset/batch.go

@@ -206,6 +206,11 @@ func (i *Iterator) Stats() storage.IteratorStats {
 	return i.i.Stats()
 }
 
+// SupportsPrev is part of the engine.Iterator interface.
+func (i *Iterator) SupportsPrev() bool {
+	return i.i.SupportsPrev()
+}
+
 // MVCCOpsSpecialized is part of the engine.MVCCIterator interface.
 func (i *Iterator) MVCCOpsSpecialized() bool {
 	if mvccIt, ok := i.i.(storage.MVCCIterator); ok {

pkg/sql/opt/cat/table.go

@@ -41,9 +41,9 @@ type Table interface {
 	// information_schema tables.
 	IsVirtualTable() bool
 
-	// AllColumnCount returns the number of columns in the table. This includes
+	// ColumnCount returns the number of columns in the table. This includes
 	// public columns, write-only columns, etc.
-	AllColumnCount() int
+	ColumnCount() int
 
 	// Column returns a Column interface to the column at the ith ordinal
 	// position within the table, where i < ColumnCount. Note that the Columns

pkg/sql/opt/cat/utils.go

@@ -130,7 +130,7 @@ func FormatTable(cat Catalog, tab Table, tp treeprinter.Node) {
 	}
 
 	var buf bytes.Buffer
-	for i := 0; i < tab.AllColumnCount(); i++ {
+	for i := 0; i < tab.ColumnCount(); i++ {
 		buf.Reset()
 		formatColumn(tab.Column(i), IsMutationColumn(tab, i), &buf)
 		child.Child(buf.String())

pkg/sql/opt/exec/execbuilder/mutation.go

@@ -702,7 +702,7 @@ func mutationOutputColMap(mutation memo.RelExpr) opt.ColMap {
 
 	var colMap opt.ColMap
 	ord := 0
-	for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, tab.ColumnCount(); i < n; i++ {
 		colID := private.Table.ColumnID(i)
 		if outCols.Contains(colID) {
 			colMap.Set(int(colID), ord)

pkg/sql/opt/exec/execbuilder/relational.go

@@ -411,7 +411,7 @@ func (b *Builder) getColumns(
 	var needed exec.TableColumnOrdinalSet
 	var output opt.ColMap
 
-	columnCount := b.mem.Metadata().Table(tableID).AllColumnCount()
+	columnCount := b.mem.Metadata().Table(tableID).ColumnCount()
 	n := 0
 	for i := 0; i < columnCount; i++ {
 		colID := tableID.ColumnID(i)

pkg/sql/opt/memo/check_expr.go

@@ -173,13 +173,13 @@ func (m *Memo) CheckExpr(e opt.Expr) {
 
 	case *InsertExpr:
 		tab := m.Metadata().Table(t.Table)
-		m.checkColListLen(t.InsertCols, tab.AllColumnCount(), "InsertCols")
+		m.checkColListLen(t.InsertCols, tab.ColumnCount(), "InsertCols")
 		m.checkColListLen(t.FetchCols, 0, "FetchCols")
 		m.checkColListLen(t.UpdateCols, 0, "UpdateCols")
 
 		// Ensure that insert columns include all columns except for delete-only
 		// mutation columns (which do not need to be part of INSERT).
-		for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+		for i, n := 0, tab.ColumnCount(); i < n; i++ {
 			if tab.ColumnKind(i) != cat.DeleteOnly && t.InsertCols[i] == 0 {
 				panic(errors.AssertionFailedf("insert values not provided for all table columns"))
 			}
@@ -190,8 +190,8 @@ func (m *Memo) CheckExpr(e opt.Expr) {
 	case *UpdateExpr:
 		tab := m.Metadata().Table(t.Table)
 		m.checkColListLen(t.InsertCols, 0, "InsertCols")
-		m.checkColListLen(t.FetchCols, tab.AllColumnCount(), "FetchCols")
-		m.checkColListLen(t.UpdateCols, tab.AllColumnCount(), "UpdateCols")
+		m.checkColListLen(t.FetchCols, tab.ColumnCount(), "FetchCols")
+		m.checkColListLen(t.UpdateCols, tab.ColumnCount(), "UpdateCols")
 		m.checkMutationExpr(t, &t.MutationPrivate)
 
 	case *ZigzagJoinExpr:
@@ -250,7 +250,7 @@ func (m *Memo) checkMutationExpr(rel RelExpr, private *MutationPrivate) {
 	// Output columns should never include mutation columns.
 	tab := m.Metadata().Table(private.Table)
 	var mutCols opt.ColSet
-	for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, tab.ColumnCount(); i < n; i++ {
 		if cat.IsMutationColumn(tab, i) {
 			mutCols.Add(private.Table.ColumnID(i))
 		}

pkg/sql/opt/memo/expr.go

@@ -640,7 +640,7 @@ func (m *MutationPrivate) MapToInputCols(tabCols opt.ColSet) opt.ColSet {
 // AddEquivTableCols adds an FD to the given set that declares an equivalence
 // between each table column and its corresponding input column.
 func (m *MutationPrivate) AddEquivTableCols(md *opt.Metadata, fdset *props.FuncDepSet) {
-	for i, n := 0, md.Table(m.Table).AllColumnCount(); i < n; i++ {
+	for i, n := 0, md.Table(m.Table).ColumnCount(); i < n; i++ {
 		t := m.Table.ColumnID(i)
 		id := m.MapToInputID(t)
 		if id != 0 {

pkg/sql/opt/memo/logical_props_builder.go

@@ -1268,7 +1268,7 @@ func (b *logicalPropsBuilder) buildMutationProps(mutation RelExpr, rel *props.Re
 	// Output Columns
 	// --------------
 	// Only non-mutation columns are output columns.
-	for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, tab.ColumnCount(); i < n; i++ {
 		if private.IsColumnOutput(i) {
 			colID := private.Table.ColumnID(i)
 			rel.OutputCols.Add(colID)
@@ -1289,7 +1289,7 @@ func (b *logicalPropsBuilder) buildMutationProps(mutation RelExpr, rel *props.Re
 	// null or the corresponding insert and fetch/update columns are not null. In
 	// other words, if either the source or destination column is not null, then
 	// the column must be not null.
-	for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, tab.ColumnCount(); i < n; i++ {
 		tabColID := private.Table.ColumnID(i)
 		if !rel.OutputCols.Contains(tabColID) {
 			continue
@@ -1562,7 +1562,7 @@ func MakeTableFuncDep(md *opt.Metadata, tabID opt.TableID) *props.FuncDepSet {
 	// Make now and annotate the metadata table with it for next time.
 	var allCols opt.ColSet
 	tab := md.Table(tabID)
-	for i := 0; i < tab.AllColumnCount(); i++ {
+	for i := 0; i < tab.ColumnCount(); i++ {
 		allCols.Add(tabID.ColumnID(i))
 	}
 
@@ -1819,7 +1819,7 @@ func tableNotNullCols(md *opt.Metadata, tabID opt.TableID) opt.ColSet {
 
 	// Only iterate over non-mutation columns, since even non-null mutation
 	// columns can be null during backfill.
-	for i := 0; i < tab.AllColumnCount(); i++ {
+	for i := 0; i < tab.ColumnCount(); i++ {
 		// Non-null mutation columns can be null during backfill.
 		if !cat.IsMutationColumn(tab, i) && !tab.Column(i).IsNullable() {
 			cs.Add(tabID.ColumnID(i))

pkg/sql/opt/memo/multiplicity_builder.go

@@ -107,7 +107,7 @@ func deriveUnfilteredCols(in RelExpr) opt.ColSet {
 		md := t.Memo().Metadata()
 		baseTable := md.Table(t.Table)
 		if t.IsUnfiltered(md) {
-			for i, cnt := 0, baseTable.AllColumnCount(); i < cnt; i++ {
+			for i, cnt := 0, baseTable.ColumnCount(); i < cnt; i++ {
 				unfilteredCols.Add(t.Table.ColumnID(i))
 			}
 		}

pkg/sql/opt/memo/multiplicity_builder_test.go

@@ -329,7 +329,7 @@ func (ob *testOpBuilder) makeScan(tableName tree.Name) (scan RelExpr, vars []*Va
 	tab := ob.cat.Table(tn)
 	tabID := ob.mem.Metadata().AddTable(tab, tn)
 	var cols opt.ColSet
-	for i := 0; i < tab.AllColumnCount(); i++ {
+	for i := 0; i < tab.ColumnCount(); i++ {
 		col := tabID.ColumnID(i)
 		cols.Add(col)
 		newVar := ob.mem.MemoizeVariable(col)

pkg/sql/opt/memo/statistics_builder_test.go

@@ -99,7 +99,7 @@ func TestGetStatsFromConstraint(t *testing.T) {
 		t.Helper()
 
 		var cols opt.ColSet
-		for i := 0; i < tab.AllColumnCount(); i++ {
+		for i := 0; i < tab.ColumnCount(); i++ {
 			cols.Add(tabID.ColumnID(i))
 		}
 

pkg/sql/opt/metadata.go

@@ -336,7 +336,7 @@ func (md *Metadata) AddTable(tab cat.Table, alias *tree.TableName) TableID {
 	}
 	md.tables = append(md.tables, TableMeta{MetaID: tabID, Table: tab, Alias: *alias})
 
-	colCount := tab.AllColumnCount()
+	colCount := tab.ColumnCount()
 	if md.cols == nil {
 		md.cols = make([]ColumnMeta, 0, colCount)
 	}

pkg/sql/opt/optbuilder/insert.go

@@ -366,7 +366,7 @@ func (mb *mutationBuilder) needExistingRows() bool {
 	// TODO(andyk): This is not true in the case of composite key encodings. See
 	// issue #34518.
 	keyOrds := getIndexLaxKeyOrdinals(mb.tab.Index(cat.PrimaryIndex))
-	for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 		if keyOrds.Contains(i) {
 			// #1: Don't consider key columns.
 			continue
@@ -512,7 +512,7 @@ func (mb *mutationBuilder) addTargetTableColsForInsert(maxCols int) {
 	// Only consider non-mutation columns, since mutation columns are hidden from
 	// the SQL user.
 	numCols := 0
-	for i, n := 0, mb.tab.AllColumnCount(); i < n && numCols < maxCols; i++ {
+	for i, n := 0, mb.tab.ColumnCount(); i < n && numCols < maxCols; i++ {
 		// Skip mutation or hidden columns.
 		if cat.IsMutationColumn(mb.tab, i) || mb.tab.Column(i).IsHidden() {
 			continue
@@ -559,8 +559,8 @@ func (mb *mutationBuilder) buildInputForInsert(inScope *scope, inputRows *tree.S
 			desiredTypes[i] = mb.md.ColumnMeta(colID).Type
 		}
 	} else {
-		desiredTypes = make([]*types.T, 0, mb.tab.AllColumnCount())
-		for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+		desiredTypes = make([]*types.T, 0, mb.tab.ColumnCount())
+		for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 			if !cat.IsMutationColumn(mb.tab, i) {
 				tabCol := mb.tab.Column(i)
 				if !tabCol.IsHidden() {
@@ -759,7 +759,7 @@ func (mb *mutationBuilder) buildInputForDoNothing(inScope *scope, conflictOrds u
 			conflictCols, mb.outScope, true /* nullsAreDistinct */, "" /* errorOnDup */)
 	}
 
-	mb.targetColList = make(opt.ColList, 0, mb.tab.AllColumnCount())
+	mb.targetColList = make(opt.ColList, 0, mb.tab.ColumnCount())
 	mb.targetColSet = opt.ColSet{}
 }
 
@@ -873,7 +873,7 @@ func (mb *mutationBuilder) buildInputForUpsert(
 		mb.b.buildWhere(where, mb.outScope)
 	}
 
-	mb.targetColList = make(opt.ColList, 0, mb.tab.AllColumnCount())
+	mb.targetColList = make(opt.ColList, 0, mb.tab.ColumnCount())
 	mb.targetColSet = opt.ColSet{}
 }
 
@@ -912,7 +912,7 @@ func (mb *mutationBuilder) setUpsertCols(insertCols tree.NameList) {
 	}
 
 	// Never update mutation columns.
-	for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 		if cat.IsMutationColumn(mb.tab, i) {
 			mb.updateOrds[i] = -1
 		}
@@ -971,7 +971,7 @@ func (mb *mutationBuilder) projectUpsertColumns() {
 
 	// Add a new column for each target table column that needs to be upserted.
 	// This can include mutation columns.
-	for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 		insertScopeOrd := mb.insertOrds[i]
 		updateScopeOrd := mb.updateOrds[i]
 		if updateScopeOrd == -1 {
@@ -1058,7 +1058,7 @@ func (mb *mutationBuilder) mapPublicColumnNamesToOrdinals(names tree.NameList) u
 	var ords util.FastIntSet
 	for _, name := range names {
 		found := false
-		for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+		for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 			tabCol := mb.tab.Column(i)
 			if tabCol.ColName() == name && !cat.IsMutationColumn(mb.tab, i) {
 				ords.Add(i)

pkg/sql/opt/optbuilder/mutation_builder.go

@@ -170,7 +170,7 @@ func (mb *mutationBuilder) init(b *Builder, opName string, tab cat.Table, alias
 	mb.tab = tab
 	mb.alias = alias
 
-	n := tab.AllColumnCount()
+	n := tab.ColumnCount()
 	mb.targetColList = make(opt.ColList, 0, n)
 
 	// Allocate segmented array of scope column ordinals.
@@ -546,7 +546,7 @@ func (mb *mutationBuilder) addSynthesizedCols(
 ) {
 	var projectionsScope *scope
 
-	for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 		if mb.tab.ColumnKind(i) == cat.DeleteOnly {
 			// Skip delete-only mutation columns, since they are ignored by all
 			// mutation operators that synthesize columns.
@@ -779,7 +779,7 @@ func (mb *mutationBuilder) addPartialIndexCols(aliasPrefix string, ords []scopeO
 func (mb *mutationBuilder) disambiguateColumns() {
 	// Determine the set of scope columns that will have their names preserved.
 	var preserve util.FastIntSet
-	for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 		scopeOrd := mb.mapToReturnScopeOrd(i)
 		if scopeOrd != -1 {
 			preserve.Add(int(scopeOrd))
@@ -829,8 +829,8 @@ func (mb *mutationBuilder) makeMutationPrivate(needResults bool) *memo.MutationP
 	}
 
 	if needResults {
-		private.ReturnCols = make(opt.ColList, mb.tab.AllColumnCount())
-		for i, n := 0, mb.tab.AllColumnCount(); i < n; i++ {
+		private.ReturnCols = make(opt.ColList, mb.tab.ColumnCount())
+		for i, n := 0, mb.tab.ColumnCount(); i < n; i++ {
 			if cat.IsMutationColumn(mb.tab, i) {
 				// Only non-mutation columns are output columns.
 				continue
@@ -941,7 +941,7 @@ func (mb *mutationBuilder) checkNumCols(expected, actual int) {
 // reuse.
 func (mb *mutationBuilder) parseDefaultOrComputedExpr(colID opt.ColumnID) tree.Expr {
 	if mb.parsedExprs == nil {
-		mb.parsedExprs = make([]tree.Expr, mb.tab.AllColumnCount())
+		mb.parsedExprs = make([]tree.Expr, mb.tab.ColumnCount())
 	}
 
 	// Return expression from cache, if it was already parsed previously.

pkg/sql/opt/optbuilder/scope.go

@@ -227,9 +227,9 @@ func (s *scope) appendColumnsFromScope(src *scope) {
 func (s *scope) appendColumnsFromTable(tabMeta *opt.TableMeta, alias *tree.TableName) {
 	tab := tabMeta.Table
 	if s.cols == nil {
-		s.cols = make([]scopeColumn, 0, tab.AllColumnCount())
+		s.cols = make([]scopeColumn, 0, tab.ColumnCount())
 	}
-	for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, tab.ColumnCount(); i < n; i++ {
 		if cat.IsMutationColumn(tab, i) {
 			continue
 		}

pkg/sql/opt/optbuilder/select.go

@@ -463,8 +463,8 @@ func (b *Builder) buildScan(
 	if ordinals == nil {
 		// If no ordinals are requested, then add in all of the table columns
 		// (including mutation columns if scanMutationCols is true).
-		outScope.cols = make([]scopeColumn, 0, tab.AllColumnCount())
-		for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+		outScope.cols = make([]scopeColumn, 0, tab.ColumnCount())
+		for i, n := 0, tab.ColumnCount(); i < n; i++ {
 			if scanMutationCols || !cat.IsMutationColumn(tab, i) {
 				addCol(i)
 			}
@@ -579,7 +579,7 @@ func (b *Builder) addCheckConstraintsForTable(tabMeta *opt.TableMeta) {
 	// Find the non-nullable table columns. Mutation columns can be NULL during
 	// backfill, so they should be excluded.
 	var notNullCols opt.ColSet
-	for i := 0; i < tab.AllColumnCount(); i++ {
+	for i := 0; i < tab.ColumnCount(); i++ {
 		if !tab.Column(i).IsNullable() && !cat.IsMutationColumn(tab, i) {
 			notNullCols.Add(tabMeta.MetaID.ColumnID(i))
 		}
@@ -628,7 +628,7 @@ func (b *Builder) addCheckConstraintsForTable(tabMeta *opt.TableMeta) {
 func (b *Builder) addComputedColsForTable(tabMeta *opt.TableMeta) {
 	var tableScope *scope
 	tab := tabMeta.Table
-	for i, n := 0, tab.AllColumnCount(); i < n; i++ {
+	for i, n := 0, tab.ColumnCount(); i < n; i++ {
 		tabCol := tab.Column(i)
 		if !tabCol.IsComputed() {
 			continue

pkg/sql/opt/optbuilder/util.go

@@ -644,7 +644,7 @@ func resolveNumericColumnRefs(tab cat.Table, columns []tree.ColumnID) (ordinals
 	ordinals = make([]int, len(columns))
 	for i, c := range columns {
 		ord := 0
-		cnt := tab.AllColumnCount()
+		cnt := tab.ColumnCount()
 		for ord < cnt {
 			if tab.Column(ord).ColID() == cat.StableID(c) && !cat.IsMutationColumn(tab, ord) {
 				break
@@ -663,7 +663,7 @@ func resolveNumericColumnRefs(tab cat.Table, columns []tree.ColumnID) (ordinals
 // having the given name, if one exists in the given table. Otherwise, it
 // returns -1.
 func findPublicTableColumnByName(tab cat.Table, name tree.Name) int {
-	for ord, n := 0, tab.AllColumnCount(); ord < n; ord++ {
+	for ord, n := 0, tab.ColumnCount(); ord < n; ord++ {
 		if tab.Column(ord).ColName() == name && !cat.IsMutationColumn(tab, ord) {
 			return ord
 		}

pkg/sql/opt/testutils/testcat/create_table.go

@@ -94,7 +94,7 @@ func (tc *Catalog) CreateTable(stmt *tree.CreateTable) *Table {
 
 	if !hasPrimaryIndex {
 		rowid := &Column{
-			Ordinal:     tab.AllColumnCount(),
+			Ordinal:     tab.ColumnCount(),
 			Name:        "rowid",
 			Type:        types.Int,
 			Hidden:      true,
@@ -254,7 +254,7 @@ func (tc *Catalog) CreateTableAs(name tree.TableName, columns []*Column) *Table
 	tab := &Table{TabID: tc.nextStableID(), TabName: name, Catalog: tc, Columns: columns}
 
 	rowid := &Column{
-		Ordinal:     tab.AllColumnCount(),
+		Ordinal:     tab.ColumnCount(),
 		Name:        "rowid",
 		Type:        types.Int,
 		Hidden:      true,
@@ -380,7 +380,7 @@ func (tc *Catalog) resolveFK(tab *Table, d *tree.ForeignKeyConstraintTableDef) {
 func (tt *Table) addColumn(def *tree.ColumnTableDef) {
 	nullable := !def.PrimaryKey.IsPrimaryKey && def.Nullable.Nullability != tree.NotNull
 	col := &Column{
-		Ordinal:  tt.AllColumnCount(),
+		Ordinal:  tt.ColumnCount(),
 		Name:     string(def.Name),
 		Type:     tree.MustBeStaticallyKnownType(def.Type),
 		Nullable: nullable,
@@ -480,7 +480,7 @@ func (tt *Table) addIndex(def *tree.IndexTableDef, typ indexType) *Index {
 			pkOrdinals.Add(c.Ordinal)
 		}
 		// Add the rest of the columns in the table.
-		for i, n := 0, tt.AllColumnCount(); i < n; i++ {
+		for i, n := 0, tt.ColumnCount(); i < n; i++ {
 			if !pkOrdinals.Contains(i) {
 				idx.addColumnByOrdinal(tt, i, tree.Ascending, nonKeyCol)
 			}

pkg/sql/opt/testutils/testcat/test_catalog.go

@@ -596,8 +596,8 @@ func (tt *Table) IsVirtualTable() bool {
 	return tt.IsVirtual
 }
 
-// AllColumnCount is part of the cat.Table interface.
-func (tt *Table) AllColumnCount() int {
+// ColumnCount is part of the cat.Table interface.
+func (tt *Table) ColumnCount() int {
 	return len(tt.Columns)
 }