importccl: fix rejected file overwriting the original file

There was severe error introduced by cockroachdb#41430 that causes the original
IMPORT file to be overwritten by the rejected file.

Release note (bug fix): fix bug when using ‘experimental_save_rejected’
that would cause the rejected row file to overwrite the original input
file.

pkg/ccl/importccl/read_import_base.go

@@ -205,13 +205,11 @@ func readInputFiles(
 						// no rejected rows
 						return nil
 					}
-					rejectedFile := dataFile
-					parsedURI, err := url.Parse(rejectedFile)
+					rejFn, err := rejectedFilename(dataFile)
 					if err != nil {
 						return err
 					}
-					parsedURI.Path = parsedURI.Path + ".rejected"
-					conf, err := cloud.ExternalStorageConfFromURI(rejectedFile)
+					conf, err := cloud.ExternalStorageConfFromURI(rejFn)
 					if err != nil {
 						return err
 					}
@@ -250,6 +248,15 @@ func readInputFiles(
 	return nil
 }
 
+func rejectedFilename(datafile string) (string, error) {
+	parsedURI, err := url.Parse(datafile)
+	if err != nil {
+		return "", err
+	}
+	parsedURI.Path = parsedURI.Path + ".rejected"
+	return parsedURI.String(), nil
+}
+
 func decompressingReader(
 	in io.Reader, name string, hint roachpb.IOFileFormat_Compression,
 ) (io.ReadCloser, error) {

pkg/ccl/importccl/read_import_base_test.go

@@ -0,0 +1,44 @@
+// Copyright 2019 The Cockroach Authors.
+//
+// Licensed as a CockroachDB Enterprise file under the Cockroach Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//     https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt
+
+package importccl
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+)
+
+func TestRejectedFilename(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	tests := []struct {
+		name     string
+		fname    string
+		rejected string
+	}{
+		{
+			name:     "http",
+			fname:    "http://127.0.0.1",
+			rejected: "http://127.0.0.1/.rejected",
+		},
+		{
+			name:     "nodelocal",
+			fname:    "nodelocal:///file.csv",
+			rejected: "nodelocal:///file.csv.rejected",
+		},
+	}
+	for _, tc := range tests {
+		rej, err := rejectedFilename(tc.fname)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if tc.rejected != rej {
+			t.Errorf("expected:\n%v\ngot:\n%v\n", tc.rejected, rej)
+		}
+	}
+}