handle empty strings as nulls for CSV (except for string types) (#2246)

sql3/planner/opbulkinsert.go

@@ -283,6 +283,12 @@ func (i *bulkInsertSourceCSVRowIter) Next(ctx context.Context) (types.Row, error
 		evalValue := rec[mapExpressionResult]
 
 		mapColumn := i.options.mapExpressions[idx]
+
+		// if an empty string and the map type is not a string, treat it as a null
+		if len(evalValue) == 0 && !typeIsString(mapColumn.colType) {
+			result[idx] = nil
+			continue
+		}
 		switch mapColumn.colType.(type) {
 		case *parser.DataTypeID, *parser.DataTypeInt:
 			intVal, err := strconv.ParseInt(evalValue, 10, 64)

sql3/sql_complex_test.go

@@ -1971,6 +1971,64 @@ func TestPlanner_BulkInsert(t *testing.T) {
 		}
 	})
 
+	t.Run("BulkInsertCSVStringNulls", func(t *testing.T) {
+		_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `create table greg-test-n (
+			_id ID,
+			id_col ID,
+			string_col STRING cachetype ranked size 1000,
+			int_col int,
+			decimal_col DECIMAL(2),
+			bool_col BOOL
+			time_col TIMESTAMP,
+			stringset_col STRINGSET,
+			ideset_col IDSET
+		);`)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `BULK INSERT INTO greg-test-n (
+			_id,
+			id_col,
+			string_col,
+			int_col,
+			decimal_col,
+			bool_col,
+			time_col,
+			stringset_col,
+			ideset_col)
+			map (
+			0 ID,
+			1 STRING,
+			2 INT,
+			3 DECIMAL(2),
+			4 BOOL,
+			5 TIMESTAMP,
+			6 STRINGSET,
+			7 IDSET)
+			transform(
+			@0,
+			@0,
+			@1,
+			@2,
+			@3,
+			@4,
+			@5,
+			@6,
+			@7)
+			FROM
+			x'1,,,,,,,
+			2,,,,,,,
+			3,,,,,,,'
+			with
+				BATCHSIZE 10000
+				format 'CSV'
+				input 'STREAM';`)
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+
 	t.Run("BulkInsertAllowMissingValues", func(t *testing.T) {
 		_, _, err = sql_test.MustQueryRows(t, c.GetNode(0).Server, `create table greg-test-amv (
 			_id STRING,