Merge #38656

38656: pkg: Migrate RowConverter to sql/row r=adityamaru27 a=adityamaru27

There is no modification of logic, this is only a change of package
to allow access of the `RowConverter` from `distsqlrun`.
Also involves renaming RowConverter due to stutter lint.

Co-authored-by: Aditya Maru <adityamaru@cockroachlabs.com>

pkg/ccl/importccl/load.go

@@ -288,7 +288,7 @@ func insertStmtToKVs(
 		return errors.Errorf("load insert: expected VALUES clause: %q", stmt)
 	}
 
-	b := sql.Inserter(f)
+	b := row.KVInserter(f)
 	computedIVarContainer := sqlbase.RowIndexedVarContainer{
 		Mapping: ri.InsertColIDtoRowIndex,
 		Cols:    tableDesc.Columns,
@@ -315,7 +315,7 @@ func insertStmtToKVs(
 		var computeExprs []tree.TypedExpr
 		var computedCols []sqlbase.ColumnDescriptor
 
-		insertRow, err := sql.GenerateInsertRow(
+		insertRow, err := row.GenerateInsertRow(
 			defaultExprs, computeExprs, cols, computedCols, evalCtx, tableDesc, insertRow, &computedIVarContainer,
 		)
 		if err != nil {

pkg/ccl/importccl/read_import_csv.go

@@ -15,8 +15,8 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
-	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
@@ -162,7 +162,7 @@ func (c *csvInputReader) convertRecordWorker(ctx context.Context) error {
 	// Create a new evalCtx per converter so each go routine gets its own
 	// collationenv, which can't be accessed in parallel.
 	evalCtx := c.evalCtx.Copy()
-	conv, err := sql.NewRowConverter(c.tableDesc, evalCtx, c.kvCh)
+	conv, err := row.NewDatumRowConverter(c.tableDesc, evalCtx, c.kvCh)
 	if err != nil {
 		return err
 	}

pkg/ccl/importccl/read_import_mysql.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/lex"
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
+	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -42,7 +43,7 @@ import (
 // KVs using the mapped converter and sent to kvCh.
 type mysqldumpReader struct {
 	evalCtx  *tree.EvalContext
-	tables   map[string]*sql.RowConverter
+	tables   map[string]*row.DatumRowConverter
 	kvCh     chan []roachpb.KeyValue
 	debugRow func(tree.Datums)
 }
@@ -56,13 +57,13 @@ func newMysqldumpReader(
 ) (*mysqldumpReader, error) {
 	res := &mysqldumpReader{evalCtx: evalCtx, kvCh: kvCh}
 
-	converters := make(map[string]*sql.RowConverter, len(tables))
+	converters := make(map[string]*row.DatumRowConverter, len(tables))
 	for name, table := range tables {
 		if table == nil {
 			converters[name] = nil
 			continue
 		}
-		conv, err := sql.NewRowConverter(table, evalCtx, kvCh)
+		conv, err := row.NewDatumRowConverter(table, evalCtx, kvCh)
 		if err != nil {
 			return nil, err
 		}

pkg/ccl/importccl/read_import_mysqlout.go

@@ -16,15 +16,15 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
-	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 )
 
 type mysqloutfileReader struct {
-	conv sql.RowConverter
+	conv row.DatumRowConverter
 	opts roachpb.MySQLOutfileOptions
 }
 
@@ -36,7 +36,7 @@ func newMysqloutfileReader(
 	tableDesc *sqlbase.TableDescriptor,
 	evalCtx *tree.EvalContext,
 ) (*mysqloutfileReader, error) {
-	conv, err := sql.NewRowConverter(tableDesc, evalCtx, kvCh)
+	conv, err := row.NewDatumRowConverter(tableDesc, evalCtx, kvCh)
 	if err != nil {
 		return nil, err
 	}

pkg/ccl/importccl/read_import_pgcopy.go

@@ -19,8 +19,8 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
-	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
@@ -32,7 +32,7 @@ import (
 const defaultScanBuffer = 1024 * 1024 * 4
 
 type pgCopyReader struct {
-	conv sql.RowConverter
+	conv row.DatumRowConverter
 	opts roachpb.PgCopyOptions
 }
 
@@ -44,7 +44,7 @@ func newPgCopyReader(
 	tableDesc *sqlbase.TableDescriptor,
 	evalCtx *tree.EvalContext,
 ) (*pgCopyReader, error) {
-	conv, err := sql.NewRowConverter(tableDesc, evalCtx, kvCh)
+	conv, err := row.NewDatumRowConverter(tableDesc, evalCtx, kvCh)
 	if err != nil {
 		return nil, err
 	}

pkg/ccl/importccl/read_import_pgdump.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
@@ -391,7 +392,7 @@ func getTableName2(u *tree.UnresolvedObjectName) (string, error) {
 }
 
 type pgDumpReader struct {
-	tables map[string]*sql.RowConverter
+	tables map[string]*row.DatumRowConverter
 	descs  map[string]*sqlbase.TableDescriptor
 	kvCh   chan []roachpb.KeyValue
 	opts   roachpb.PgDumpOptions
@@ -406,10 +407,10 @@ func newPgDumpReader(
 	descs map[string]*sqlbase.TableDescriptor,
 	evalCtx *tree.EvalContext,
 ) (*pgDumpReader, error) {
-	converters := make(map[string]*sql.RowConverter, len(descs))
+	converters := make(map[string]*row.DatumRowConverter, len(descs))
 	for name, desc := range descs {
 		if desc.IsTable() {
-			conv, err := sql.NewRowConverter(desc, evalCtx, kvCh)
+			conv, err := row.NewDatumRowConverter(desc, evalCtx, kvCh)
 			if err != nil {
 				return nil, err
 			}

pkg/ccl/importccl/read_import_workload.go

@@ -21,9 +21,9 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
-	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/exec/coldata"
 	"github.com/cockroachdb/cockroach/pkg/sql/exec/types"
+	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	sqltypes "github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -231,7 +231,7 @@ func NewWorkloadKVConverter(
 func (w *WorkloadKVConverter) Worker(
 	ctx context.Context, evalCtx *tree.EvalContext, finishedBatchFn func(),
 ) error {
-	conv, err := sql.NewRowConverter(w.tableDesc, evalCtx, w.kvCh)
+	conv, err := row.NewDatumRowConverter(w.tableDesc, evalCtx, w.kvCh)
 	if err != nil {
 		return err
 	}

pkg/sql/insert.go

@@ -25,7 +25,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/errorutil/unimplemented"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
-	"github.com/cockroachdb/errors"
 )
 
 var insertNodePool = sync.Pool{
@@ -522,7 +521,7 @@ func (n *insertNode) processSourceRow(params runParams, sourceVals tree.Datums)
 	// Process the incoming row tuple and generate the full inserted
 	// row. This fills in the defaults, computes computed columns, and
 	// check the data width complies with the schema constraints.
-	rowVals, err := GenerateInsertRow(
+	rowVals, err := row.GenerateInsertRow(
 		n.run.defaultExprs,
 		n.run.computeExprs,
 		n.run.insertCols,
@@ -601,110 +600,6 @@ func (n *insertNode) enableAutoCommit() {
 	n.run.ti.enableAutoCommit()
 }
 
-// GenerateInsertRow prepares a row tuple for insertion. It fills in default
-// expressions, verifies non-nullable columns, and checks column widths.
-//
-// The result is a row tuple providing values for every column in insertCols.
-// This results contains:
-//
-// - the values provided by rowVals, the tuple of source values. The
-//   caller ensures this provides values 1-to-1 to the prefix of
-//   insertCols that was specified explicitly in the INSERT statement.
-// - the default values for any additional columns in insertCols that
-//   have default values in defaultExprs.
-// - the computed values for any additional columns in insertCols
-//   that are computed. The mapping in rowContainerForComputedCols
-//   maps the indexes of the comptuedCols/computeExpr slices
-//   back into indexes in the result row tuple.
-//
-func GenerateInsertRow(
-	defaultExprs []tree.TypedExpr,
-	computeExprs []tree.TypedExpr,
-	insertCols []sqlbase.ColumnDescriptor,
-	computedCols []sqlbase.ColumnDescriptor,
-	evalCtx *tree.EvalContext,
-	tableDesc *sqlbase.ImmutableTableDescriptor,
-	rowVals tree.Datums,
-	rowContainerForComputedVals *sqlbase.RowIndexedVarContainer,
-) (tree.Datums, error) {
-	// The values for the row may be shorter than the number of columns being
-	// inserted into. Generate default values for those columns using the
-	// default expressions. This will not happen if the row tuple was produced
-	// by a ValuesClause, because all default expressions will have been populated
-	// already by fillDefaults.
-	if len(rowVals) < len(insertCols) {
-		// It's not cool to append to the slice returned by a node; make a copy.
-		oldVals := rowVals
-		rowVals = make(tree.Datums, len(insertCols))
-		copy(rowVals, oldVals)
-
-		for i := len(oldVals); i < len(insertCols); i++ {
-			if defaultExprs == nil {
-				rowVals[i] = tree.DNull
-				continue
-			}
-			d, err := defaultExprs[i].Eval(evalCtx)
-			if err != nil {
-				return nil, err
-			}
-			rowVals[i] = d
-		}
-	}
-
-	// Generate the computed values, if needed.
-	if len(computeExprs) > 0 {
-		rowContainerForComputedVals.CurSourceRow = rowVals
-		evalCtx.PushIVarContainer(rowContainerForComputedVals)
-		for i := range computedCols {
-			// Note that even though the row is not fully constructed at this point,
-			// since we disallow computed columns from referencing other computed
-			// columns, all the columns which could possibly be referenced *are*
-			// available.
-			d, err := computeExprs[i].Eval(evalCtx)
-			if err != nil {
-				return nil, errors.Wrapf(err, "computed column %s", tree.ErrString((*tree.Name)(&computedCols[i].Name)))
-			}
-			rowVals[rowContainerForComputedVals.Mapping[computedCols[i].ID]] = d
-		}
-		evalCtx.PopIVarContainer()
-	}
-
-	// Verify the column constraints.
-	//
-	// We would really like to use enforceLocalColumnConstraints() here,
-	// but this is not possible because of some brain damage in the
-	// Insert() constructor, which causes insertCols to contain
-	// duplicate columns descriptors: computed columns are listed twice,
-	// one will receive a NULL value and one will receive a comptued
-	// value during execution. It "works out in the end" because the
-	// latter (non-NULL) value overwrites the earlier, but
-	// enforceLocalColumnConstraints() does not know how to reason about
-	// this.
-	//
-	// In the end it does not matter much, this code is going away in
-	// favor of the (simpler, correct) code in the CBO.
-
-	// Check to see if NULL is being inserted into any non-nullable column.
-	for _, col := range tableDesc.WritableColumns() {
-		if !col.Nullable {
-			if i, ok := rowContainerForComputedVals.Mapping[col.ID]; !ok || rowVals[i] == tree.DNull {
-				return nil, sqlbase.NewNonNullViolationError(col.Name)
-			}
-		}
-	}
-
-	// Ensure that the values honor the specified column widths.
-	for i := 0; i < len(insertCols); i++ {
-		outVal, err := sqlbase.LimitValueWidth(&insertCols[i].Type, rowVals[i], &insertCols[i].Name)
-		if err != nil {
-			return nil, err
-		}
-		rowVals[i] = outVal
-	}
-
-	return rowVals, nil
-}
-
 // processColumns returns the column descriptors identified by the
 // given name list. It also checks that a given column name is only
 // listed once. If no column names are given (special case for INSERT)