s1ny1998 commented on code in PR #43000:
URL: https://github.com/apache/superset/pull/43000#discussion_r3748863641
##########
superset/db_engine_specs/clickhouse.py:
##########
@@ -419,6 +424,125 @@ def get_datatype(cls, type_code: str) -> str:
# keep it lowercase, as ClickHouse types aren't typical SHOUTCASE ANSI
SQL
return type_code
+ @classmethod
+ def df_to_sql(
+ cls,
+ database: Database,
+ table: Table,
+ df: Any,
+ to_sql_kwargs: dict[str, Any],
+ ) -> None:
+ """Upload a DataFrame to ClickHouse.
+
+ ClickHouse requires every table to declare a table engine, which the
+ `CREATE TABLE` that pandas' ``to_sql`` emits does not — so instead of
+ letting pandas create the table we create it ourselves with a MergeTree
+ engine and then append the rows.
+
+ The engine can be tuned per database through its ``extra`` JSON::
+
+ {"clickhouse_file_upload": {
+ "order_by": ["col1", "col2"],
+ "partition_by": "toYYYYMM(col1)",
+ "primary_key": "col1",
+ "settings": {"index_granularity": 8192}
+ }}
+
+ ``order_by`` defaults to ``tuple()`` (no sort key) so any upload works
+ without configuration.
+ """
+ # pylint: disable=import-outside-toplevel, import-error
+ import pandas as pd
+ from clickhouse_connect.cc_sqlalchemy.ddl.tableengine import MergeTree
+ from sqlalchemy import (
+ Column,
+ inspect,
+ MetaData,
+ Table as SqlaTable,
+ text,
+ types as sqltypes,
+ )
+
+ if_exists = to_sql_kwargs.get("if_exists", "fail")
+
+ if to_sql_kwargs.get("index"):
+ # Fold the index into columns so the table we create matches what
+ # gets inserted; we always append with index=False below.
+ df = df.reset_index()
+
+ config = (database.get_extra() or {}).get("clickhouse_file_upload", {})
+ order_by = config.get("order_by")
+ if order_by:
+ key_columns = set(
+ order_by if isinstance(order_by, (list, tuple)) else [order_by]
+ )
+ engine_kwargs: dict[str, Any] = {"order_by": order_by}
+ else:
+ # MergeTree still requires an ORDER BY; an empty tuple means
"none".
+ key_columns = set()
+ engine_kwargs = {"order_by": text("tuple()")}
+ for key in ("partition_by", "primary_key", "settings"):
+ if config.get(key):
+ engine_kwargs[key] = config[key]
+
+ def _column_type(dtype: Any) -> sqltypes.TypeEngine:
+ if pd.api.types.is_bool_dtype(dtype):
+ return sqltypes.Boolean()
+ if pd.api.types.is_integer_dtype(dtype):
+ return sqltypes.BigInteger()
+ if pd.api.types.is_float_dtype(dtype):
+ return sqltypes.Float()
+ if pd.api.types.is_datetime64_any_dtype(dtype):
+ return sqltypes.DateTime()
+ return sqltypes.String()
+
+ with cls.get_engine(
+ database, catalog=table.catalog, schema=table.schema
+ ) as engine:
+ has_table = inspect(engine).has_table(table.table,
schema=table.schema)
+ if has_table and if_exists == "fail":
+ # Raise ValueError so the uploader surfaces its friendly
+ # "table already exists" message (see UploadCommand).
+ raise ValueError(f"Table {table.table} already exists.")
+ if has_table and if_exists == "replace":
+ SqlaTable(table.table, MetaData(), schema=table.schema).drop(
+ engine, checkfirst=True
+ )
+ has_table = False
Review Comment:
pandas to_sql(if_exists="replace") is used by the base uploader for every
engine, it has identical drop-then-create behavior, and file upload is a single
admin action, not a concurrent path. A staged swap (temp table + EXCHANGE
TABLES) would be ClickHouse-specific complexity inconsistent with the rest of
the codebase.
I think that's the right call for this PR, but it's a judgment call. Idk, I
may be wrong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
