codeant-ai-for-open-source[bot] commented on code in PR #43000:
URL: https://github.com/apache/superset/pull/43000#discussion_r3748866413
##########
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:
✅ **Customized review instruction saved!**
**Instruction:**
> Do not flag concurrency risks for the standard pandas to_sql replace flow
in this ClickHouse uploader; file uploads are treated as single-admin actions,
and staged table swaps are intentionally out of scope.
**Applied to:**
- `superset/db_engine_specs/clickhouse.py`
---
💡 *To manage or update this instruction, visit: [CodeAnt AI
Settings](https://app.codeant.ai/org/settings/learnings)*
--
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]
