fix: Upload CSV as Dataset

superset/commands/database/uploaders/base.py

@@ -34,6 +34,7 @@
 )
 from superset.connectors.sqla.models import SqlaTable
 from superset.daos.database import DatabaseDAO
+from superset.exceptions import SupersetException
 from superset.models.core import Database
 from superset.sql.parse import Table
 from superset.utils.backports import StrEnum
@@ -124,6 +125,8 @@ def _dataframe_to_database(
                 df,
                 to_sql_kwargs=to_sql_kwargs,
             )
+        except SupersetException:
+            raise
         except ValueError as ex:
             raise DatabaseUploadFailed(
                 message=_(

superset/commands/database/uploaders/csv_reader.py

@@ -154,6 +154,7 @@ def _read_csv(  # noqa: C901
         kwargs["low_memory"] = False
 
         try:
+            types = kwargs.pop("dtype", None)
             if "chunksize" in kwargs:
                 chunks = []
                 total_rows = 0
@@ -188,13 +189,32 @@ def _read_csv(  # noqa: C901
                         index_col = kwargs.get("index_col")
                         if isinstance(index_col, str):
                             result.index.name = index_col
-                    return result
-                return pd.DataFrame()
+                    df = result
+            else:
+                df = pd.read_csv(
+                    filepath_or_buffer=file.stream,
+                    **kwargs,
+                )
 
-            return pd.read_csv(
-                filepath_or_buffer=file.stream,
-                **kwargs,
-            )
+            if types:
+                for column, dtype in types.items():
+                    try:
+                        df[column] = df[column].astype(dtype)
+                    except ValueError as ex:
+                        error_msg = f"Non {dtype} value found in column '{column}'."
+                        ex_msg = str(ex)
+                        invalid_value = ex_msg.split(":")[-1].strip().strip("'")
+                        error_msg += (
+                            f" Value: '{invalid_value}'" if invalid_value else ""
+                        )
+                        mask = df[column] == invalid_value
+                        if mask.any() and invalid_value:
+                            line_number = mask.idxmax() + kwargs.get("header", 0) + 1
+                            error_msg += f", line: {line_number}."
+                        raise DatabaseUploadFailed(message=error_msg) from ex
+            return df
+        except DatabaseUploadFailed:
+            raise
         except UnicodeDecodeError as ex:
             if encoding != DEFAULT_ENCODING:
                 raise DatabaseUploadFailed(

tests/unit_tests/commands/databases/csv_reader_test.py

@@ -421,6 +421,175 @@ def test_csv_reader_file_metadata_invalid_file():
     )
 
 
+def test_csv_reader_integer_in_float_column():
+    csv_data = [
+        ["Name", "Score", "City"],
+        ["name1", 25.5, "city1"],
+        ["name2", 25, "city2"],
+    ]
+
+    csv_reader = CSVReader(
+        options=CSVReaderOptions(column_data_types={"Score": "float"})
+    )
+
+    df = csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert df.shape == (2, 3)
+    assert df["Score"].dtype == "float64"
+
+
+def test_csv_reader_object_type_auto_inferring():
+    # this case below won't raise a error
+    csv_data = [
+        ["Name", "id", "City"],
+        ["name1", 25.5, "city1"],
+        ["name2", 15, "city2"],
+        ["name3", 123456789086, "city3"],
+        ["name4", "abc", "city4"],
+        ["name5", 4.75, "city5"],
+    ]
+
+    csv_reader = CSVReader()
+
+    df = csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert df.shape == (5, 3)
+    # pandas automatically infers the type if column_data_types is not informed
+    # if there's only one string in the column it converts the whole column to object
+    assert df["id"].dtype == "object"
+
+
+def test_csv_reader_float_type_auto_inferring():
+    csv_data = [
+        ["Name", "id", "City"],
+        ["name1", "25", "city1"],
+        ["name2", "15", "city2"],
+        ["name3", "123456789086", "city3"],
+        ["name5", "4.75", "city5"],
+    ]
+
+    csv_reader = CSVReader()
+
+    df = csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert df.shape == (4, 3)
+    # The type here is automatically inferred to float due to 4.75 value
+    assert df["id"].dtype == "float64"
+
+
+def test_csv_reader_int_type_auto_inferring():
+    csv_data = [
+        ["Name", "id", "City"],
+        ["name1", "0", "city1"],
+        ["name2", "15", "city2"],
+        ["name3", "123456789086", "city3"],
+        ["name5", "45", "city5"],
+    ]
+
+    csv_reader = CSVReader()
+
+    df = csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert df.shape == (4, 3)
+    assert df["id"].dtype == "int64"
+
+
+def test_csv_reader_bigint_type_auto_inferring():
+    csv_data = [
+        ["Name", "id", "City"],
+        ["name1", "9223372036854775807", "city1"],
+        ["name2", "9223372036854775806", "city2"],
+        ["name3", "1234567890123456789", "city3"],
+        ["name4", "0", "city4"],
+        ["name5", "-9223372036854775808", "city5"],
+    ]
+
+    csv_reader = CSVReader()
+
+    df = csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert df.shape == (5, 3)
+    assert df["id"].dtype == "int64"
+    assert df.iloc[0]["id"] == 9223372036854775807
+    assert df.iloc[4]["id"] == -9223372036854775808
+
+
+def test_csv_reader_int_typing():
+    csv_data = [
+        ["Name", "id", "City"],
+        ["name1", "0", "city1"],
+        ["name2", "15", "city2"],
+        ["name3", "123456789086", "city3"],
+        ["name5", "45", "city5"],
+    ]
+
+    csv_reader = CSVReader(options=CSVReaderOptions(column_data_types={"id": "int"}))
+
+    df = csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert df.shape == (4, 3)
+    assert df["id"].dtype == "int64"
+
+
+def test_csv_reader_float_typing():
+    csv_data = [
+        ["Name", "score", "City"],
+        ["name1", "0", "city1"],
+        ["name2", "15.3", "city2"],
+        ["name3", "45", "city3"],
+        ["name5", "23.1342", "city5"],
+    ]
+
+    csv_reader = CSVReader(
+        options=CSVReaderOptions(column_data_types={"score": "float"})
+    )
+
+    df = csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert df.shape == (4, 3)
+    assert df["score"].dtype == "float64"
+
+
+def test_csv_reader_non_numeric_in_integer_column():
+    csv_data = [
+        ["Name", "Age", "City"],
+        ["name1", "abc", "city1"],
+        ["name2", "25", "city2"],
+    ]
+
+    csv_reader = CSVReader(options=CSVReaderOptions(column_data_types={"Age": "int"}))
+
+    with pytest.raises(DatabaseUploadFailed) as ex:
+        csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert (
+        str(ex.value) == "Non int value found in column 'Age'. Value: 'abc', line: 1."
+    )
+
+
+def test_csv_reader_non_numeric_in_float_column():
+    csv_data = [
+        ["Name", "Score", "City"],
+        ["name1", "5.3", "city1"],
+        ["name2", "25.5", "city2"],
+        ["name3", "24.5", "city3"],
+        ["name4", "1.0", "city4"],
+        ["name5", "one point five", "city5"],
+    ]
+
+    csv_reader = CSVReader(
+        options=CSVReaderOptions(column_data_types={"Score": "float"})
+    )
+
+    with pytest.raises(DatabaseUploadFailed) as ex:
+        csv_reader.file_to_dataframe(create_csv_file(csv_data))
+
+    assert (
+        str(ex.value)
+        == "Non float value found in column 'Score'. Value: 'one point five', line: 5."
+    )
+
+
 def test_csv_reader_chunking_large_file():
     """Test that chunking is used for large files."""
     # Create a large CSV with more than 100k rows