Clickhouse

.github/workflows/test_destination_clickhouse.yml

@@ -0,0 +1,78 @@
+
+name: test | clickhouse
+
+on:
+  pull_request:
+    branches:
+      - master
+      - devel
+  workflow_dispatch:
+  schedule:
+    - cron:  '0 2 * * *'
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+env:
+  RUNTIME__SENTRY_DSN: https://[email protected]/4504819859914752
+  RUNTIME__LOG_LEVEL: ERROR
+  DLT_SECRETS_TOML: ${{ secrets.DLT_SECRETS_TOML }}
+
+  ACTIVE_DESTINATIONS: "[\"clickhouse\"]"
+  ALL_FILESYSTEM_DRIVERS: "[\"memory\"]"
+
+jobs:
+  get_docs_changes:
+    name: docs changes
+    uses: ./.github/workflows/get_docs_changes.yml
+    if: ${{ !github.event.pull_request.head.repo.fork || contains(github.event.pull_request.labels.*.name, 'ci from fork')}}
+
+  run_loader:
+    name: test | clickhouse tests
+    needs: get_docs_changes
+    if: needs.get_docs_changes.outputs.changes_outside_docs == 'true'
+    defaults:
+      run:
+        shell: bash
+    runs-on: "ubuntu-latest"
+
+    steps:
+
+      - name: Check out
+        uses: actions/checkout@master
+
+      - name: Setup Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.10.x"
+
+      - name: Install Poetry
+        uses: snok/[email protected]
+        with:
+          virtualenvs-create: true
+          virtualenvs-in-project: true
+          installer-parallel: true
+
+      - name: Load cached venv
+        id: cached-poetry-dependencies
+        uses: actions/cache@v3
+        with:
+          path: .venv
+          key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('**/poetry.lock') }}-gcp
+
+      - name: Install dependencies
+        run: poetry install --no-interaction -E clickhouse --with providers -E parquet --with sentry-sdk --with pipeline
+
+      - name: create secrets.toml
+        run: pwd && echo "$DLT_SECRETS_TOML" > tests/.dlt/secrets.toml
+
+      - run: |
+          poetry run pytest tests/load -m "essential"
+        name: Run essential tests Linux
+        if: ${{ ! (contains(github.event.pull_request.labels.*.name, 'ci full') || github.event_name == 'schedule')}}
+
+      - run: |
+          poetry run pytest tests/load
+        name: Run all tests Linux
+        if: ${{ contains(github.event.pull_request.labels.*.name, 'ci full') || github.event_name == 'schedule'}}

dlt/common/configuration/providers/airflow.py

@@ -14,7 +14,6 @@ def name(self) -> str:
 
     def _look_vault(self, full_key: str, hint: type) -> str:
         """Get Airflow Variable with given `full_key`, return None if not found"""
-
         from airflow.models import Variable
 
         with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):

dlt/common/configuration/specs/base_configuration.py

@@ -197,7 +197,7 @@ def default_factory(att_value=att_value):  # type: ignore[no-untyped-def]
         synth_init = init and ((not base_params or base_params.init) and has_default_init)
         if synth_init != init and has_default_init:
             warnings.warn(
-                f"__init__ method will not be generated on {cls.__name__} because bas class didn't"
+                f"__init__ method will not be generated on {cls.__name__} because base class didn't"
                 " synthesize __init__. Please correct `init` flag in confispec decorator. You are"
                 " probably receiving incorrect __init__ signature for type checking"
             )

dlt/common/data_writers/escape.py

@@ -150,10 +150,47 @@ def escape_databricks_literal(v: Any) -> Any:
         return _escape_extended(json.dumps(v), prefix="'", escape_dict=DATABRICKS_ESCAPE_DICT)
     if isinstance(v, bytes):
         return f"X'{v.hex()}'"
-    if v is None:
-        return "NULL"
+    return "NULL" if v is None else str(v)
 
-    return str(v)
+
+# https://github.com/ClickHouse/ClickHouse/blob/master/docs/en/sql-reference/syntax.md#string
+CLICKHOUSE_ESCAPE_DICT = {
+    "'": "''",
+    "\\": "\\\\",
+    "\n": "\\n",
+    "\t": "\\t",
+    "\b": "\\b",
+    "\f": "\\f",
+    "\r": "\\r",
+    "\0": "\\0",
+    "\a": "\\a",
+    "\v": "\\v",
+}
+
+CLICKHOUSE_ESCAPE_RE = _make_sql_escape_re(CLICKHOUSE_ESCAPE_DICT)
+
+
+def escape_clickhouse_literal(v: Any) -> Any:
+    if isinstance(v, str):
+        return _escape_extended(
+            v, prefix="'", escape_dict=CLICKHOUSE_ESCAPE_DICT, escape_re=CLICKHOUSE_ESCAPE_RE
+        )
+    if isinstance(v, (datetime, date, time)):
+        return f"'{v.isoformat()}'"
+    if isinstance(v, (list, dict)):
+        return _escape_extended(
+            json.dumps(v),
+            prefix="'",
+            escape_dict=CLICKHOUSE_ESCAPE_DICT,
+            escape_re=CLICKHOUSE_ESCAPE_RE,
+        )
+    if isinstance(v, bytes):
+        return f"'{v.hex()}'"
+    return "NULL" if v is None else str(v)
+
+
+def escape_clickhouse_identifier(v: str) -> str:
+    return "`" + v.replace("`", "``").replace("\\", "\\\\") + "`"
 
 
 def format_datetime_literal(v: pendulum.DateTime, precision: int = 6, no_tz: bool = False) -> str:
@@ -176,3 +213,11 @@ def format_bigquery_datetime_literal(
     """Returns BigQuery-adjusted datetime literal by prefixing required `TIMESTAMP` indicator."""
     # https://cloud.google.com/bigquery/docs/reference/standard-sql/lexical#timestamp_literals
     return "TIMESTAMP " + format_datetime_literal(v, precision, no_tz)
+
+
+def format_clickhouse_datetime_literal(
+    v: pendulum.DateTime, precision: int = 6, no_tz: bool = False
+) -> str:
+    """Returns clickhouse compatibel function"""
+    datetime = format_datetime_literal(v, precision, True)
+    return f"toDateTime64({datetime}, {precision}, '{v.tzinfo}')"

dlt/common/destination/capabilities.py

@@ -53,6 +53,7 @@ class DestinationCapabilitiesContext(ContainerInjectableContext):
     insert_values_writer_type: str = "default"
     supports_multiple_statements: bool = True
     supports_clone_table: bool = False
+
     """Destination supports CREATE TABLE ... CLONE ... statements"""
     max_table_nesting: Optional[int] = None  # destination can overwrite max table nesting
 

dlt/common/time.py

@@ -133,7 +133,6 @@ def ensure_pendulum_time(value: Union[str, datetime.time]) -> pendulum.Time:
     Returns:
         A pendulum.Time object
     """
-
     if isinstance(value, datetime.time):
         if isinstance(value, pendulum.Time):
             return value

dlt/destinations/__init__.py

@@ -14,6 +14,7 @@
 from dlt.destinations.impl.synapse.factory import synapse
 from dlt.destinations.impl.databricks.factory import databricks
 from dlt.destinations.impl.dremio.factory import dremio
+from dlt.destinations.impl.clickhouse.factory import clickhouse
 
 
 __all__ = [
@@ -32,5 +33,6 @@
     "synapse",
     "databricks",
     "dremio",
+    "clickhouse",
     "destination",
 ]

dlt/destinations/adapters.py

@@ -4,5 +4,12 @@
 from dlt.destinations.impl.qdrant import qdrant_adapter
 from dlt.destinations.impl.bigquery import bigquery_adapter
 from dlt.destinations.impl.synapse import synapse_adapter
+from dlt.destinations.impl.clickhouse import clickhouse_adapter
 
-__all__ = ["weaviate_adapter", "qdrant_adapter", "bigquery_adapter", "synapse_adapter"]
+__all__ = [
+    "weaviate_adapter",
+    "qdrant_adapter",
+    "bigquery_adapter",
+    "synapse_adapter",
+    "clickhouse_adapter",
+]

dlt/destinations/impl/clickhouse/__init__.py

@@ -0,0 +1,53 @@
+import sys
+
+from dlt.common.pendulum import pendulum
+from dlt.common.arithmetics import DEFAULT_NUMERIC_PRECISION, DEFAULT_NUMERIC_SCALE
+from dlt.common.data_writers.escape import (
+    escape_clickhouse_identifier,
+    escape_clickhouse_literal,
+    format_clickhouse_datetime_literal,
+)
+from dlt.common.destination import DestinationCapabilitiesContext
+from dlt.destinations.impl.clickhouse.clickhouse_adapter import clickhouse_adapter
+
+
+def capabilities() -> DestinationCapabilitiesContext:
+    caps = DestinationCapabilitiesContext()
+    caps.preferred_loader_file_format = "jsonl"
+    caps.supported_loader_file_formats = ["parquet", "jsonl"]
+    caps.preferred_staging_file_format = "jsonl"
+    caps.supported_staging_file_formats = ["parquet", "jsonl"]
+
+    caps.format_datetime_literal = format_clickhouse_datetime_literal
+    caps.escape_identifier = escape_clickhouse_identifier
+    caps.escape_literal = escape_clickhouse_literal
+
+    # https://stackoverflow.com/questions/68358686/what-is-the-maximum-length-of-a-column-in-clickhouse-can-it-be-modified
+    caps.max_identifier_length = 255
+    caps.max_column_identifier_length = 255
+
+    # ClickHouse has no max `String` type length.
+    caps.max_text_data_type_length = sys.maxsize
+
+    caps.schema_supports_numeric_precision = True
+    # Use 'Decimal128' with these defaults.
+    # https://clickhouse.com/docs/en/sql-reference/data-types/decimal
+    caps.decimal_precision = (DEFAULT_NUMERIC_PRECISION, DEFAULT_NUMERIC_SCALE)
+    # Use 'Decimal256' with these defaults.
+    caps.wei_precision = (76, 0)
+    caps.timestamp_precision = 6
+
+    # https://clickhouse.com/docs/en/operations/settings/settings#max_query_size
+    caps.is_max_query_length_in_bytes = True
+    caps.max_query_length = 262144
+
+    # ClickHouse has limited support for transactional semantics, especially for `ReplicatedMergeTree`,
+    # the default ClickHouse Cloud engine. It does, however, provide atomicity for individual DDL operations like `ALTER TABLE`.
+    # https://clickhouse-driver.readthedocs.io/en/latest/dbapi.html#clickhouse_driver.dbapi.connection.Connection.commit
+    # https://clickhouse.com/docs/en/guides/developer/transactional#transactions-commit-and-rollback
+    caps.supports_transactions = False
+    caps.supports_ddl_transactions = False
+
+    caps.supports_truncate_command = True
+
+    return caps