Previously, inserting a value of the wrong type into a column depended
on by a computed column would panic the server. That's because we used
to defer enforcing type checking of insert values until marshalling
time.

This commit fixes the bug by correcting that long-standing problem: we
now enforce insert-value type checks at plan time instead of run time.

To do so, while preserving the nice error messages of the previous
approach that include the names of the columns that had mismatched
types, this commit introduces a new fake AST node called
ValuesClauseWithNames that is a values clause that includes column name
information. This is populated by insert nodes, so that the values
clause can type check itself at plan time and produce error messages
with column names.

Release note (bug fix): fix panic caused by inserting values of the
wrong type into columns depended on by computed columns.

pkg/sql/insert.go

@@ -174,10 +174,14 @@ func (p *planner) Insert(
	// Extract the AST for the data source.
	var insertRows tree.SelectStatement
	arityChecked := false
	colNames := make(tree.NameList, len(insertCols))
	for i := range insertCols {
		colNames[i] = tree.Name(insertCols[i].Name)
	}
	if n.DefaultValues() {
		insertRows = newDefaultValuesClause(defaultExprs, insertCols)
		insertRows = newDefaultValuesClause(defaultExprs, colNames)
	} else {
		src, values, err := extractInsertSource(n.Rows)
		src, values, err := extractInsertSource(colNames, n.Rows)
		if err != nil {
			return nil, err
		}
@@ -671,7 +675,9 @@ func (p *planner) processColumns(
// extractInsertSource removes the parentheses around the data source of an INSERT statement.
// If the data source is a VALUES clause not further qualified with LIMIT/OFFSET and ORDER BY,
// the 3rd return value is a pre-casted pointer to the VALUES clause.
func extractInsertSource(s *tree.Select) (tree.SelectStatement, *tree.ValuesClause, error) {
func extractInsertSource(
	colNames tree.NameList, s *tree.Select,
) (tree.SelectStatement, *tree.ValuesClauseWithNames, error) {
	wrapped := s.Select
	limit := s.Limit
	orderBy := s.OrderBy
@@ -694,25 +700,31 @@ func extractInsertSource(s *tree.Select) (tree.SelectStatement, *tree.ValuesClau

	if orderBy == nil && limit == nil {
		values, _ := wrapped.(*tree.ValuesClause)
		return wrapped, values, nil
		if values != nil {
			return wrapped, &tree.ValuesClauseWithNames{ValuesClause: *values, Names: colNames}, nil
		}
		return wrapped, nil, nil
	}
	return &tree.ParenSelect{
		Select: &tree.Select{Select: wrapped, OrderBy: orderBy, Limit: limit},
	}, nil, nil
}

func newDefaultValuesClause(
	defaultExprs []tree.TypedExpr, cols []sqlbase.ColumnDescriptor,
	defaultExprs []tree.TypedExpr, colNames tree.NameList,
) tree.SelectStatement {
	row := make(tree.Exprs, 0, len(cols))
	for i := range cols {
	row := make(tree.Exprs, 0, len(colNames))
	for i := range colNames {
		if defaultExprs == nil {
			row = append(row, tree.DNull)
			continue
		}
		row = append(row, defaultExprs[i])
	}
	return &tree.ValuesClause{Tuples: []*tree.Tuple{{Exprs: row}}}
	return &tree.ValuesClauseWithNames{
		ValuesClause: tree.ValuesClause{Tuples: []*tree.Tuple{{Exprs: row}}},
		Names:        colNames,
	}
}

// fillDefaults populates default expressions in the provided ValuesClause,
@@ -727,12 +739,14 @@ func newDefaultValuesClause(
//
// The function returns a ValuesClause with defaults filled or an error.
func fillDefaults(
	defaultExprs []tree.TypedExpr, cols []sqlbase.ColumnDescriptor, values *tree.ValuesClause,
) (*tree.ValuesClause, error) {
	defaultExprs []tree.TypedExpr,
	cols []sqlbase.ColumnDescriptor,
	values *tree.ValuesClauseWithNames,
) (*tree.ValuesClauseWithNames, error) {
	ret := values
	copyValues := func() {
		if ret == values {
			ret = &tree.ValuesClause{Tuples: append([]*tree.Tuple(nil), values.Tuples...)}
			ret = &tree.ValuesClauseWithNames{ValuesClause: tree.ValuesClause{Tuples: append([]*tree.Tuple(nil), values.Tuples...)}, Names: values.Names}
		}
	}

@@ -777,6 +791,9 @@ func fillDefaults(
			tuple.Exprs = append(tuple.Exprs, defaultExpr(len(tuple.Exprs)))
		}
	}
	for i := numColsOrig; i < len(cols); i++ {
		ret.Names = append(ret.Names, tree.Name(cols[i].Name))
	}
	return ret, nil
}

pkg/sql/logictest/testdata/logic_test/computed

@@ -107,6 +107,28 @@ DELETE FROM x
statement ok
DROP TABLE x

statement ok
CREATE TABLE x (
  a INT NOT NULL,
  b INT,
  c INT AS (a) STORED,
  d INT AS (a + b) STORED
)

statement ok
INSERT INTO x (a) VALUES (1)

statement error null value in column "a" violates not-null constraint
INSERT INTO x (b) VALUES (1)

query II
SELECT c, d FROM x
----
1  NULL

statement ok
DROP TABLE x

# Check with upserts
statement ok
CREATE TABLE x (
@@ -694,3 +716,13 @@ INSERT INTO x (a) SELECT 1

statement ok
DROP TABLE x

statement ok
CREATE TABLE x (
  a INT PRIMARY KEY,
  b INT AS (a+1) STORED
)

query error value type decimal doesn't match type INT of column "b"
INSERT INTO x VALUES(1.4)

pkg/sql/plan.go

@@ -786,6 +786,8 @@ func (p *planner) newPlan(
		return p.Update(ctx, n, desiredTypes)
	case *tree.ValuesClause:
		return p.Values(ctx, n, desiredTypes)
	case *tree.ValuesClauseWithNames:
		return p.Values(ctx, n, desiredTypes)
	default:
		return nil, errors.Errorf("unknown statement type: %T", stmt)
	}

pkg/sql/sem/tree/values.go

@@ -38,3 +38,15 @@ func (node *ValuesClause) Format(ctx *FmtCtx) {
		ctx.FormatNode(n)
	}
}

// ValuesClauseWithNames is a VALUES clause that has been annotated with column
// names. This is only produced at plan time, never by the parser. It's used to
// pass column names to the VALUES planNode, so it can produce intelligible
// error messages during value type checking.
type ValuesClauseWithNames struct {
	ValuesClause

	// Names is a list of the column names that each tuple in the ValuesClause
	// corresponds to.
	Names NameList
}

pkg/sql/sqlbase/computed_exprs.go

@@ -41,7 +41,11 @@ var _ tree.IndexedVarContainer = &RowIndexedVarContainer{}
func (r *RowIndexedVarContainer) IndexedVarEval(
	idx int, ctx *tree.EvalContext,
) (tree.Datum, error) {
	return r.CurSourceRow[r.Mapping[r.Cols[idx].ID]], nil
	rowIdx, ok := r.Mapping[r.Cols[idx].ID]
	if !ok {
		return tree.DNull, nil
	}
	return r.CurSourceRow[rowIdx], nil
}

// IndexedVarResolvedType implements tree.IndexedVarContainer.

pkg/sql/sqlbase/table.go

@@ -23,6 +23,8 @@ import (

	"github.com/pkg/errors"

	"runtime/debug"

	"github.com/cockroachdb/apd"
	"github.com/cockroachdb/cockroach/pkg/internal/client"
	"github.com/cockroachdb/cockroach/pkg/keys"
@@ -1858,6 +1860,54 @@ func checkElementType(paramType types.T, columnType ColumnType) error {
	return nil
}

// NewMismatchedTypeError creates an error indicating a mismatched value and
// column type.
func NewMismatchedTypeError(
	valType types.T, colType ColumnType_SemanticType, colName string,
) error {
	debug.PrintStack()
	return pgerror.NewErrorWithDepthf(1,
		pgerror.CodeDatatypeMismatchError,
		"value type %s doesn't match type %s of column %q",
		valType, colType, colName)
}

// NewMismatchedLocaleError creates an error indicating a mismatched collated
// string locale in a value and column.
func NewMismatchedLocaleError(valLocale, colLocale string, colName string) error {
	return pgerror.NewErrorWithDepthf(1,
		pgerror.CodeDatatypeMismatchError,
		"locale %q doesn't match locale %q of column %q",
		valLocale, colLocale, colName)
}

// CheckColumnValueType checks to see if valType and colType are compatible,
// returning an error similar to that of MarshalColumnValue if they're not.
func CheckColumnValueType(valType types.T, colType ColumnType, colName string) error {
	switch colType.SemanticType {
	case ColumnType_ARRAY:
		if t, ok := valType.(types.TArray); ok {
			if err := checkElementType(t.Typ, colType); err != nil {
				return err
			}
		}
	case ColumnType_COLLATEDSTRING:
		if t, ok := valType.(types.TCollatedString); ok {
			if t.Locale != *colType.Locale {
				return NewMismatchedLocaleError(t.Locale, *colType.Locale, colName)
			}
		}
	}
	valColType, err := DatumTypeToColumnType(valType)
	if err != nil {
		return err
	}
	if !valColType.Equal(colType) {
		return NewMismatchedTypeError(valType, colType.SemanticType, colName)
	}
	return nil
}

// MarshalColumnValue returns a Go primitive value equivalent of val, of the
// type expected by col. If val's type is incompatible with col, or if
// col's type is not yet implemented, an error is returned.
@@ -1970,8 +2020,7 @@ func MarshalColumnValue(col ColumnDescriptor, val tree.Datum) (roachpb.Value, er
				r.SetString(v.Contents)
				return r, nil
			}
			return r, fmt.Errorf("locale %q doesn't match locale %q of column %q",
				v.Locale, *col.Type.Locale, col.Name)
			return r, NewMismatchedLocaleError(v.Locale, *col.Type.Locale, col.Name)
		}
	case ColumnType_OID:
		if v, ok := val.(*tree.DOid); ok {
@@ -1981,8 +2030,7 @@ func MarshalColumnValue(col ColumnDescriptor, val tree.Datum) (roachpb.Value, er
	default:
		return r, errors.Errorf("unsupported column type: %s", col.Type.SemanticType)
	}
	return r, fmt.Errorf("value type %s doesn't match type %s of column %q",
		val.ResolvedType(), col.Type.SemanticType, col.Name)
	return r, NewMismatchedTypeError(val.ResolvedType(), col.Type.SemanticType, col.Name)
}

const hasNullFlag = 1 << 4

pkg/sql/values.go

@@ -47,12 +47,26 @@ type valuesNode struct {

// Values implements the VALUES clause.
func (p *planner) Values(
	ctx context.Context, n *tree.ValuesClause, desiredTypes []types.T,
	ctx context.Context, origN tree.Statement, desiredTypes []types.T,
) (planNode, error) {
	v := &valuesNode{
		specifiedInQuery: true,
		isConst:          true,
	}

	// If we have names, extract them.
	var n *tree.ValuesClause
	var names tree.NameList
	switch t := origN.(type) {
	case *tree.ValuesClauseWithNames:
		n = &t.ValuesClause
		names = t.Names
	case *tree.ValuesClause:
		n = t
	default:
		log.Fatalf(ctx, "programming error. unhandled case in values: %T %v", origN, origN)
	}

	if len(n.Tuples) == 0 {
		return v, nil
	}
@@ -92,6 +106,25 @@ func (p *planner) Values(
			}

			typ := typedExpr.ResolvedType()
			if names != nil && (!(typ.Equivalent(desired) || typ == types.Unknown)) {
				var colName tree.Name
				if len(names) > i {
					colName = names[i]
				} else {
					colName = "unknown"
				}
				desiredColTyp, err := sqlbase.DatumTypeToColumnType(desired)
				if err != nil {
					return nil, err
				}
				err = sqlbase.CheckColumnValueType(typ, desiredColTyp, string(colName))
				if err != nil {
					return nil, err
				}
				// For some reason we didn't detect a new error. Return a fresh one.
				return nil, sqlbase.NewMismatchedTypeError(typ, desiredColTyp.SemanticType, string(colName))
			}

			if num == 0 {
				v.columns = append(v.columns, sqlbase.ResultColumn{Name: "column" + strconv.Itoa(i+1), Typ: typ})
			} else if v.columns[i].Typ == types.Unknown {