You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by be...@apache.org on 2025/08/07 17:52:40 UTC
(superset) branch improve-csv-upload-perf created (now 9c4f34156c)
This is an automated email from the ASF dual-hosted git repository.
beto pushed a change to branch improve-csv-upload-perf
in repository https://gitbox.apache.org/repos/asf/superset.git
at 9c4f34156c feat: improve perf of CSV uploads
This branch includes the following new commits:
new 9c4f34156c feat: improve perf of CSV uploads
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(superset) 01/01: feat: improve perf of CSV uploads
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
beto pushed a commit to branch improve-csv-upload-perf
in repository https://gitbox.apache.org/repos/asf/superset.git
commit 9c4f34156c7864a6bc07914f3749b5d91bde48d6
Author: Beto Dealmeida <ro...@dealmeida.net>
AuthorDate: Thu Aug 7 13:51:40 2025 -0400
feat: improve perf of CSV uploads
---
superset/commands/database/uploaders/csv_reader.py | 106 +++++++++++++++++----
superset/db_engine_specs/clickhouse.py | 1 +
superset/db_engine_specs/db2.py | 1 +
superset/db_engine_specs/duckdb.py | 1 +
superset/db_engine_specs/mysql.py | 1 +
superset/db_engine_specs/oracle.py | 1 +
superset/db_engine_specs/postgres.py | 1 +
superset/db_engine_specs/sqlite.py | 1 +
8 files changed, 92 insertions(+), 21 deletions(-)
diff --git a/superset/commands/database/uploaders/csv_reader.py b/superset/commands/database/uploaders/csv_reader.py
index 0072a7d5cf..41ba8422f6 100644
--- a/superset/commands/database/uploaders/csv_reader.py
+++ b/superset/commands/database/uploaders/csv_reader.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
import logging
+from importlib import util
from typing import Any, Optional
import pandas as pd
@@ -30,8 +31,10 @@ from superset.commands.database.uploaders.base import (
logger = logging.getLogger(__name__)
-READ_CSV_CHUNK_SIZE = 1000
-ROWS_TO_READ_METADATA = 2
+READ_CSV_CHUNK_SIZE = 50000
+ROWS_TO_READ_METADATA = 100
+DEFAULT_ENCODING = "utf-8"
+ENCODING_FALLBACKS = ["utf-8", "latin-1", "cp1252", "iso-8859-1"]
class CSVReaderOptions(ReaderOptions, total=False):
@@ -60,24 +63,69 @@ class CSVReader(BaseDataReader):
options=dict(options),
)
+ @staticmethod
+ def _detect_encoding(file: FileStorage) -> str:
+ """Detect file encoding with fallback options"""
+ sample = file.read(10000)
+ file.seek(0)
+
+ for encoding in ENCODING_FALLBACKS:
+ try:
+ sample.decode(encoding)
+ return encoding
+ except UnicodeDecodeError:
+ continue
+ return DEFAULT_ENCODING
+
@staticmethod
def _read_csv(file: FileStorage, kwargs: dict[str, Any]) -> pd.DataFrame:
+ encoding = kwargs.get("encoding", DEFAULT_ENCODING)
+
+ if "pyarrow" not in str(pd.__version__):
+ kwargs["engine"] = "pyarrow" if util.find_spec("pyarrow") else "c"
+ else:
+ kwargs["engine"] = "c"
+
+ kwargs["low_memory"] = False
+
try:
if "chunksize" in kwargs:
- return pd.concat(
- pd.read_csv(
- filepath_or_buffer=file.stream,
- **kwargs,
- )
+ chunks = []
+ chunk_iterator = pd.read_csv(
+ filepath_or_buffer=file.stream,
+ **kwargs,
+ )
+
+ for chunk in chunk_iterator:
+ chunks.append(chunk)
+ if "nrows" in kwargs and len(pd.concat(chunks)) >= kwargs["nrows"]:
+ break
+
+ return (
+ pd.concat(chunks, ignore_index=True) if chunks else pd.DataFrame()
)
+
return pd.read_csv(
filepath_or_buffer=file.stream,
**kwargs,
)
+ except UnicodeDecodeError as ex:
+ if encoding != DEFAULT_ENCODING:
+ raise DatabaseUploadFailed(
+ message=_("Parsing error: %(error)s", error=str(ex))
+ ) from ex
+
+ file.seek(0)
+ detected_encoding = CSVReader._detect_encoding(file)
+ if detected_encoding != encoding:
+ kwargs["encoding"] = detected_encoding
+ return CSVReader._read_csv(file, kwargs)
+ raise DatabaseUploadFailed(
+ message=_("Parsing error: %(error)s", error=str(ex))
+ ) from ex
except (
pd.errors.ParserError,
pd.errors.EmptyDataError,
- UnicodeDecodeError,
ValueError,
) as ex:
raise DatabaseUploadFailed(
@@ -93,31 +141,45 @@ class CSVReader(BaseDataReader):
:return: pandas DataFrame
:throws DatabaseUploadFailed: if there is an error reading the file
"""
+ rows_to_read = self._options.get("rows_to_read")
+
+ use_chunking = rows_to_read is None or rows_to_read > READ_CSV_CHUNK_SIZE * 2
+
kwargs = {
- "chunksize": READ_CSV_CHUNK_SIZE,
- "encoding": "utf-8",
+ "encoding": self._options.get("encoding", DEFAULT_ENCODING),
"header": self._options.get("header_row", 0),
"decimal": self._options.get("decimal_character", "."),
"index_col": self._options.get("index_column"),
"dayfirst": self._options.get("day_first", False),
- "iterator": True,
"keep_default_na": not self._options.get("null_values"),
- "usecols": self._options.get("columns_read")
- if self._options.get("columns_read") # None if an empty list
- else None,
- "na_values": self._options.get("null_values")
- if self._options.get("null_values") # None if an empty list
- else None,
- "nrows": self._options.get("rows_to_read"),
+ "usecols": (
+ self._options.get("columns_read")
+ if self._options.get("columns_read") # None if an empty list
+ else None
+ ),
+ "na_values": (
+ self._options.get("null_values")
+ if self._options.get("null_values") # None if an empty list
+ else None
+ ),
+ "nrows": rows_to_read,
"parse_dates": self._options.get("column_dates"),
"sep": self._options.get("delimiter", ","),
"skip_blank_lines": self._options.get("skip_blank_lines", False),
"skipinitialspace": self._options.get("skip_initial_space", False),
"skiprows": self._options.get("skip_rows", 0),
- "dtype": self._options.get("column_data_types")
- if self._options.get("column_data_types")
- else None,
+ "dtype": (
+ self._options.get("column_data_types")
+ if self._options.get("column_data_types")
+ else None
+ ),
+ "cache_dates": True,
}
+
+ if use_chunking:
+ kwargs["chunksize"] = READ_CSV_CHUNK_SIZE
+ kwargs["iterator"] = True
+
return self._read_csv(file, kwargs)
def file_metadata(self, file: FileStorage) -> FileMetadata:
@@ -131,6 +193,8 @@ class CSVReader(BaseDataReader):
"nrows": ROWS_TO_READ_METADATA,
"header": self._options.get("header_row", 0),
"sep": self._options.get("delimiter", ","),
+ "encoding": self._options.get("encoding", DEFAULT_ENCODING),
+ "low_memory": False,
}
df = self._read_csv(file, kwargs)
return {
diff --git a/superset/db_engine_specs/clickhouse.py b/superset/db_engine_specs/clickhouse.py
index ada2f9f5e4..d743e09be3 100644
--- a/superset/db_engine_specs/clickhouse.py
+++ b/superset/db_engine_specs/clickhouse.py
@@ -54,6 +54,7 @@ class ClickHouseBaseEngineSpec(BaseEngineSpec):
"""Shared engine spec for ClickHouse."""
time_groupby_inline = True
+ supports_multivalues_insert = True
_time_grain_expressions = {
None: "{col}",
diff --git a/superset/db_engine_specs/db2.py b/superset/db_engine_specs/db2.py
index 8dd7b00a6b..8a09f5af3c 100644
--- a/superset/db_engine_specs/db2.py
+++ b/superset/db_engine_specs/db2.py
@@ -36,6 +36,7 @@ class Db2EngineSpec(BaseEngineSpec):
max_column_name_length = 30
supports_dynamic_schema = True
+ supports_multivalues_insert = True
_time_grain_expressions = {
None: "{col}",
diff --git a/superset/db_engine_specs/duckdb.py b/superset/db_engine_specs/duckdb.py
index 5cf11febca..3e7310f05f 100644
--- a/superset/db_engine_specs/duckdb.py
+++ b/superset/db_engine_specs/duckdb.py
@@ -196,6 +196,7 @@ class DuckDBEngineSpec(DuckDBParametersMixin, BaseEngineSpec):
default_driver = "duckdb_engine"
sqlalchemy_uri_placeholder = "duckdb:////path/to/duck.db"
+ supports_multivalues_insert = True
_time_grain_expressions = {
None: "{col}",
diff --git a/superset/db_engine_specs/mysql.py b/superset/db_engine_specs/mysql.py
index fc0606697b..8c071f8dab 100644
--- a/superset/db_engine_specs/mysql.py
+++ b/superset/db_engine_specs/mysql.py
@@ -74,6 +74,7 @@ class MySQLEngineSpec(BasicParametersMixin, BaseEngineSpec):
encryption_parameters = {"ssl": "1"}
supports_dynamic_schema = True
+ supports_multivalues_insert = True
column_type_mappings = (
(
diff --git a/superset/db_engine_specs/oracle.py b/superset/db_engine_specs/oracle.py
index 37c779b8aa..e40223730b 100644
--- a/superset/db_engine_specs/oracle.py
+++ b/superset/db_engine_specs/oracle.py
@@ -28,6 +28,7 @@ class OracleEngineSpec(BaseEngineSpec):
engine_name = "Oracle"
force_column_alias_quotes = True
max_column_name_length = 128
+ supports_multivalues_insert = True
_time_grain_expressions = {
None: "{col}",
diff --git a/superset/db_engine_specs/postgres.py b/superset/db_engine_specs/postgres.py
index b86172cb3a..3e95ecc790 100644
--- a/superset/db_engine_specs/postgres.py
+++ b/superset/db_engine_specs/postgres.py
@@ -99,6 +99,7 @@ class PostgresBaseEngineSpec(BaseEngineSpec):
engine = ""
engine_name = "PostgreSQL"
+ supports_multivalues_insert = True
_time_grain_expressions = {
None: "{col}",
diff --git a/superset/db_engine_specs/sqlite.py b/superset/db_engine_specs/sqlite.py
index a7e9a96e33..f8604bcfea 100644
--- a/superset/db_engine_specs/sqlite.py
+++ b/superset/db_engine_specs/sqlite.py
@@ -42,6 +42,7 @@ class SqliteEngineSpec(BaseEngineSpec):
engine_name = "SQLite"
disable_ssh_tunneling = True
+ supports_multivalues_insert = True
_time_grain_expressions = {
None: "{col}",