importccl: lift computed column check to input converter

The current implementation of checking for validating the number of rows
for computed columns for in non-IMPORT INTO backups is inefficient. This
commit moves the check from being performed on every row to only being
performed once per import.

Release note: None

pkg/ccl/importccl/import_processor.go

@@ -145,6 +145,24 @@ func makeInputConverter(
 				return nil, unimplemented.NewWithIssue(50225, "cannot import into table with partial indexes")
 			}
 		}
+
+		// If we're using a format like CSV where data columns are not "named", and
+		// therefore cannot be mapped to schema columns, then require the user to
+		// use IMPORT INTO.
+		//
+		// We could potentially do something smarter here and check that only a
+		// suffix of the columns are computed, and then expect the data file to have
+		// #(visible columns) - #(computed columns).
+		if len(singleTableTargetCols) == 0 && !formatHasNamedColumns(spec.Format.Format) {
+			for _, col := range singleTable.VisibleColumns() {
+				if col.IsComputed() {
+					// TODO(before merge): Replace this issue number with one that
+					// summarizes the above comment in a ticket.
+					return nil, unimplemented.NewWithIssueDetail(42846, "import.computed",
+						"to use computed columns, use IMPORT INTO")
+				}
+			}
+		}
 	}
 
 	switch spec.Format.Format {

pkg/ccl/importccl/import_stmt_test.go

@@ -3747,8 +3747,8 @@ func TestImportComputed(t *testing.T) {
 	}
 	avroData := createAvroData(t, "t", avroField, avroRows)
 	pgdumpData := `
-CREATE TABLE users (a INT, b INT, c INT AS (a + b) STORED);
-INSERT INTO users (a, b) VALUES (1, 2), (3, 4);
+CREATE TABLE users (a INT, b INT, c INT AS (a + b) STORED);		
+INSERT INTO users (a, b) VALUES (1, 2), (3, 4);		
 `
 	defer srv.Close()
 	tests := []struct {
@@ -3812,18 +3812,16 @@ INSERT INTO users (a, b) VALUES (1, 2), (3, 4);
 			name:          "import-table-csv",
 			data:          "35,23\n67,10",
 			create:        "a INT, c INT AS (a + b) STORED, b INT",
-			targetCols:    "a, b",
 			format:        "CSV",
-			expectedError: "requires targeted column specification",
+			expectedError: "to use computed columns, use IMPORT INTO",
 		},
 		{
 			into:            false,
 			name:            "import-table-avro",
 			data:            avroData,
-			create:          "a INT, b INT, c INT AS (a + b) STORED",
-			targetCols:      "a, b",
+			create:          "a INT, c INT AS (a + b) STORED, b INT",
 			format:          "AVRO",
-			expectedResults: [][]string{{"1", "2", "3"}, {"3", "4", "7"}},
+			expectedResults: [][]string{{"1", "3", "2"}, {"3", "7", "4"}},
 		},
 		{
 			into:            false,

pkg/ccl/importccl/read_import_base.go

@@ -345,6 +345,18 @@ type inputConverter interface {
 		format roachpb.IOFileFormat, makeExternalStorage cloud.ExternalStorageFactory, user string) error
 }
 
+// formatHasNamedColumns returns true if the data in the input files can be
+// mapped specifically to a particular data column.
+func formatHasNamedColumns(format roachpb.IOFileFormat_FileFormat) bool {
+	switch format {
+	case roachpb.IOFileFormat_Avro,
+		roachpb.IOFileFormat_Mysqldump,
+		roachpb.IOFileFormat_PgDump:
+		return true
+	}
+	return false
+}
+
 func isMultiTableFormat(format roachpb.IOFileFormat_FileFormat) bool {
 	switch format {
 	case roachpb.IOFileFormat_Mysqldump,

pkg/ccl/importccl/read_import_csv.go

@@ -134,15 +134,6 @@ func (p *csvRowProducer) Row() (interface{}, error) {
 	p.rowNum++
 	expectedColsLen := len(p.expectedColumns)
 	if expectedColsLen == 0 {
-		// TODO(anzoteh96): this should really be only checked once per import instead of every row.
-		for _, col := range p.importCtx.tableDesc.VisibleColumns() {
-			if col.IsComputed() {
-				return nil,
-					errors.Newf(
-						"IMPORT CSV with computed column %q requires targeted column specification",
-						col.Name)
-			}
-		}
 		expectedColsLen = len(p.importCtx.tableDesc.VisibleColumns())
 	}