sql: clean up mutable not-null columns hack

pkg/sql/catalog/descriptor.go

@@ -447,9 +447,8 @@ type TableDescriptor interface {
 	// details.
 	AccessibleColumns() []Column
 	// ReadableColumns is a list of columns (including those undergoing a schema
-	// change) which can be scanned. Columns in the process of a schema change
-	// are all set to nullable while column backfilling is still in
-	// progress, as mutation columns may have NULL values.
+	// change) which can be scanned. Note that mutation columns may produce NULL
+	// values when scanned, even if they are marked as not nullable.
 	ReadableColumns() []Column
 	// UserDefinedTypeColumns returns a slice of Column interfaces
 	// containing the table's columns with user defined types, in the

pkg/sql/catalog/tabledesc/column.go

@@ -315,23 +315,13 @@ func newColumnCache(desc *descpb.TableDescriptor, mutations *mutationCache) *col
 		c.writable = c.public
 		c.nonDrop = c.public
 	} else {
-		readableDescs := make([]descpb.ColumnDescriptor, 0, numMutations)
-		readableBackingStructs := make([]column, 0, numMutations)
 		for _, col := range c.deletable {
 			if !col.DeleteOnly() {
 				lazyAllocAppendColumn(&c.writable, col, numDeletable)
 			}
 			if !col.Dropped() {
 				lazyAllocAppendColumn(&c.nonDrop, col, numDeletable)
 			}
-			if !col.Public() && !col.IsNullable() {
-				j := len(readableDescs)
-				readableDescs = append(readableDescs, *col.ColumnDesc())
-				readableDescs[j].Nullable = true
-				readableBackingStructs = append(readableBackingStructs, *col.(*column))
-				readableBackingStructs[j].desc = &readableDescs[j]
-				col = &readableBackingStructs[j]
-			}
 			lazyAllocAppendColumn(&c.readable, col, numDeletable)
 		}
 	}

pkg/sql/colfetcher/cfetcher.go

@@ -1379,7 +1379,7 @@ func (rf *cFetcher) fillNulls() error {
 		if table.compositeIndexColOrdinals.Contains(i) {
 			continue
 		}
-		if !table.cols[i].IsNullable() {
+		if !table.cols[i].IsNullable() && table.cols[i].Public() {
 			var indexColValues strings.Builder
 			rf.writeDecodedCols(&indexColValues, table.indexColOrdinals, ',')
 			return scrub.WrapError(scrub.UnexpectedNullValueError, errors.Errorf(

pkg/sql/row/fetcher.go

@@ -1215,15 +1215,16 @@ func (rf *Fetcher) finalizeRow() error {
 	}
 
 	// Fill in any missing values with NULLs
-	for i := range table.cols {
+	for i, col := range table.cols {
 		if rf.valueColsFound == table.neededValueCols {
 			// Found all cols - done!
 			return nil
 		}
-		if table.neededCols.Contains(int(table.cols[i].GetID())) && table.row[i].IsUnset() {
+		if table.neededCols.Contains(int(col.GetID())) && table.row[i].IsUnset() {
 			// If the row was deleted, we'll be missing any non-primary key
-			// columns, including nullable ones, but this is expected.
-			if !table.cols[i].IsNullable() && !table.rowIsDeleted && !rf.IgnoreUnexpectedNulls {
+			// columns, including nullable ones, but this is expected. If the column
+			// is not yet active, we can also expect NULLs.
+			if !col.IsNullable() && !table.rowIsDeleted && !rf.IgnoreUnexpectedNulls && col.Public() {
 				var indexColValues []string
 				for _, idx := range table.indexColIdx {
 					if idx != -1 {
@@ -1234,7 +1235,7 @@ func (rf *Fetcher) finalizeRow() error {
 				}
 				return errors.AssertionFailedf(
 					"Non-nullable column \"%s:%s\" with no value! Index scanned was %q with the index key columns (%s) and the values (%s)",
-					table.desc.GetName(), table.cols[i].GetName(), table.index.GetName(),
+					table.desc.GetName(), col.GetName(), table.index.GetName(),
 					strings.Join(table.index.IndexDesc().KeyColumnNames, ","), strings.Join(indexColValues, ","))
 			}
 			table.row[i] = rowenc.EncDatum{