sql: clean up mutable not-null columns hack

Mutation columns in some cases need to be scanned even if they haven't
been backfilled yet, which means that we may retrieve NULL values even
if they are market as not-nullable.

We currently have a hack in the table descriptor which changes the
nullable flags in the column descriptors when `ReadableColumns()` is
used. It is very surprising that we can get different descriptors for
a given ColumnID depending if we look for it in `ReadableColumns()` or
in `AllColumns()` (e.g. via FindColumnWithID.

This commit cleans this up, changing the scanning code to check for
`Public()` instead.

Release note: None

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,15 @@ 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)
+		//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{