Merge #39032 #39130

39032: pkg: Allows no target cols to be specified in IMPORT INTO query. r=adityamaru27 a=adityamaru27

It adds the grammar required to support queries such as:
`IMPORT INTO test CSV DATA ('...');`

No target columns implies that all columns are imported into from the data sources.

39130: batcheval: check GCThreshold in RevertRange command r=dt a=dt

I think somehow I got mixed up about when a key is GC'ed in that original comment.
So that we can time-travel to any time within GC window, we already only GC a key
when there is a newer _expired_ key over it, and the requirements for being able
to revert to a time are no different than being able to read at that time, so the
same GCThreshold check we do elsewhere should also be good enough here.

Release note: none.

Co-authored-by: Aditya Maru <adityamaru@cockroachlabs.com>
Co-authored-by: David Taylor <tinystatemachine@gmail.com>

docs/generated/sql/bnf/import_csv.bnf

@@ -5,3 +5,5 @@ import_stmt ::=
 	| 'IMPORT' 'TABLE' table_name '(' table_elem_list ')' 'CSV' 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 
 	| 'IMPORT' 'INTO' table_name '(' insert_column_list ')' 'CSV' 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 'WITH' kv_option_list
 	| 'IMPORT' 'INTO' table_name '(' insert_column_list ')' 'CSV' 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 
+	| 'IMPORT' 'INTO' table_name 'CSV' 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 'WITH' kv_option_list
+	| 'IMPORT' 'INTO' table_name 'CSV' 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 

docs/generated/sql/bnf/import_dump.bnf

@@ -5,3 +5,5 @@ import_stmt ::=
 	| 'IMPORT' 'TABLE' table_name 'FROM' import_format file_location 
 	| 'IMPORT' 'INTO' table_name '(' insert_column_list ')' import_format 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 'WITH' kv_option_list
 	| 'IMPORT' 'INTO' table_name '(' insert_column_list ')' import_format 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 
+	| 'IMPORT' 'INTO' table_name import_format 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 'WITH' kv_option_list
+	| 'IMPORT' 'INTO' table_name import_format 'DATA' '(' file_location ( ( ',' file_location ) )* ')' 

docs/generated/sql/bnf/stmt_block.bnf

@@ -130,6 +130,7 @@ import_stmt ::=
 	| 'IMPORT' 'TABLE' table_name 'CREATE' 'USING' string_or_placeholder import_format 'DATA' '(' string_or_placeholder_list ')' opt_with_options
 	| 'IMPORT' 'TABLE' table_name '(' table_elem_list ')' import_format 'DATA' '(' string_or_placeholder_list ')' opt_with_options
 	| 'IMPORT' 'INTO' table_name '(' insert_column_list ')' import_format 'DATA' '(' string_or_placeholder_list ')' opt_with_options
+	| 'IMPORT' 'INTO' table_name import_format 'DATA' '(' string_or_placeholder_list ')' opt_with_options
 
 insert_stmt ::=
 	opt_with_clause 'INSERT' 'INTO' insert_target insert_rest returning_clause

pkg/ccl/importccl/import_stmt_test.go

@@ -1743,6 +1743,45 @@ func TestImportIntoCSV(t *testing.T) {
 		sqlDB.Exec(t, "DROP DATABASE targetcols")
 	})
 
+	// Tests IMPORT INTO without any target columns specified. This implies an
+	// import of all columns in the exisiting table.
+	t.Run("no-target-cols-specified", func(t *testing.T) {
+		sqlDB.Exec(t, "CREATE DATABASE targetcols; USE targetcols")
+		sqlDB.Exec(t, `CREATE TABLE t (a INT, b STRING)`)
+
+		// Insert the test data
+		insert := []string{"''", "'text'", "'a'", "'e'", "'l'", "'t'", "'z'"}
+
+		if tx, err := db.Begin(); err != nil {
+			t.Fatal(err)
+		} else {
+			for i, v := range insert {
+				sqlDB.Exec(t, fmt.Sprintf("INSERT INTO t (a, b) VALUES (%d, %s)", i, v))
+			}
+
+			if err := tx.Commit(); err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		sqlDB.Exec(t, fmt.Sprintf("IMPORT INTO t CSV DATA (%s)", testFiles.files[0]))
+
+		var result int
+		numExistingRows := len(insert)
+		// Verify that all columns have been populated with imported data.
+		sqlDB.QueryRow(t, `SELECT count(*) FROM t WHERE a IS NOT NULL`).Scan(&result)
+		if expect := numExistingRows + rowsPerFile; result != expect {
+			t.Fatalf("expected %d rows, got %d", expect, result)
+		}
+
+		sqlDB.QueryRow(t, `SELECT count(*) FROM t WHERE b IS NOT NULL`).Scan(&result)
+		if expect := numExistingRows + rowsPerFile; result != expect {
+			t.Fatalf("expected %d rows, got %d", expect, result)
+		}
+
+		sqlDB.Exec(t, "DROP DATABASE targetcols")
+	})
+
 	// IMPORT INTO does not support DEFAULT expressions for either target or
 	// non-target columns.
 	t.Run("import-into-check-no-default-cols", func(t *testing.T) {
@@ -1755,8 +1794,8 @@ func TestImportIntoCSV(t *testing.T) {
 		if tx, err := db.Begin(); err != nil {
 			t.Fatal(err)
 		} else {
-			for i := range insert {
-				sqlDB.Exec(t, fmt.Sprintf("INSERT INTO t (a, b) VALUES (%d, %s)", i, insert[i]))
+			for i, v := range insert {
+				sqlDB.Exec(t, fmt.Sprintf("INSERT INTO t (a, b) VALUES (%d, %s)", i, v))
 			}
 
 			if err := tx.Commit(); err != nil {
@@ -1771,8 +1810,6 @@ func TestImportIntoCSV(t *testing.T) {
 
 		sqlDB.Exec(t, "DROP DATABASE targetcols")
 	})
-	// TODO(adityamaru): Add test for IMPORT INTO without target columns specified
-	// once grammar has been added.
 }
 
 func BenchmarkImport(b *testing.B) {