reconcile 1.4.1 work with main (#151)

* Initial work on lw delete example

* LW delete example checkpoint

* Expand high volume lw delete example, fix source schema issue

Author: genzgd

dbt/adapters/clickhouse/__version__.py

@@ -1 +1 @@
-version = '1.4.0'
+version = '1.4.1'

dbt/adapters/clickhouse/impl.py

@@ -150,7 +150,7 @@ def s3source_clause(
         if path:
             if bucket and path and not bucket.endswith('/') and not bucket.startswith('/'):
                 path = f'/{path}'
-            url = f'{url}{path}'
+            url = f'{url}{path}'.replace('//', '/')
         if not url.startswith('http'):
             url = f'https://{url}'
         access = ''

dbt/adapters/clickhouse/relation.py

@@ -1,8 +1,10 @@
 from dataclasses import dataclass, field
-from typing import Optional
+from typing import Any, Optional, Type
 
-from dbt.adapters.base.relation import BaseRelation, Policy
+from dbt.adapters.base.relation import BaseRelation, Policy, Self
+from dbt.contracts.graph.nodes import SourceDefinition
 from dbt.exceptions import DbtRuntimeError
+from dbt.utils import deep_merge
 
 
 @dataclass
@@ -47,3 +49,27 @@ def matches(
         if schema:
             raise DbtRuntimeError(f'Passed unexpected schema value {schema} to Relation.matches')
         return self.database == database and self.identifier == identifier
+
+    @classmethod
+    def create_from_source(cls: Type[Self], source: SourceDefinition, **kwargs: Any) -> Self:
+        source_quoting = source.quoting.to_dict(omit_none=True)
+        source_quoting.pop("column", None)
+        quote_policy = deep_merge(
+            cls.get_default_quote_policy().to_dict(omit_none=True),
+            source_quoting,
+            kwargs.get("quote_policy", {}),
+        )
+
+        # If the database is set, and the source schema is "defaulted" to the source.name, override the
+        # schema with the database instead, since that's presumably what's intended for clickhouse
+        schema = source.schema
+        if schema == source.source_name and source.database:
+            schema = source.database
+
+        return cls.create(
+            database=source.database,
+            schema=schema,
+            identifier=source.identifier,
+            quote_policy=quote_policy,
+            **kwargs,
+        )

dbt/include/clickhouse/macros/adapters.sql

@@ -107,7 +107,7 @@
 {% macro exchange_tables_atomic(old_relation, target_relation, obj_types='TABLES') %}
 
   {%- if adapter.get_clickhouse_cluster_name() is not none and obj_types == 'TABLES' %}
-    {% do run_query("SYSTEM SYNC REPLICA "+target_relation.identifier) %}
+    {% do run_query("SYSTEM SYNC REPLICA "+ target_relation.identifier + on_cluster_clause()) %}
   {%- endif %}
 
   {%- call statement('exchange_tables_atomic') -%}

examples/taxis/.gitignore

@@ -0,0 +1,4 @@
+
+target/
+dbt_packages/
+logs/

examples/taxis/README.md

@@ -0,0 +1,87 @@
+# Taxis Large Incremental Model Example/Test
+
+This is an example designed to test large incremental materializations.  It builds a `taxis_inc` model in the
+`taxis_dbt` database that uses randomized keys to increase on each subsequent run.
+
+## Create the source data
+
+Use this SQL to create and populate the "source" data from the ClickHouse taxis example dataset.
+
+```sql
+
+CREATE DATABASE taxis;
+
+CREATE TABLE taxis.trips (
+    trip_id             UInt32,
+    pickup_datetime     DateTime,
+    dropoff_datetime    DateTime,
+    pickup_longitude    Nullable(Float64),
+    pickup_latitude     Nullable(Float64),
+    dropoff_longitude   Nullable(Float64),
+    dropoff_latitude    Nullable(Float64),
+    passenger_count     UInt8,
+    trip_distance       Float32,
+    fare_amount         Float32,
+    extra               Float32,
+    tip_amount          Float32,
+    tolls_amount        Float32,
+    total_amount        Float32,
+    payment_type        LowCardinality(String),
+    pickup_ntaname      LowCardinality(String),
+    dropoff_ntaname     LowCardinality(String)
+)
+ENGINE = MergeTree
+ORDER BY trip_id;
+  
+SET input_format_skip_unknown_fields = 1;
+
+INSERT INTO taxis.trips
+SELECT
+    trip_id,
+    pickup_datetime,
+    dropoff_datetime,
+    pickup_longitude,
+    pickup_latitude,
+    dropoff_longitude,
+    dropoff_latitude,
+    passenger_count,
+    trip_distance,
+    fare_amount,
+    extra,
+    tip_amount,
+    tolls_amount,
+    total_amount,
+    payment_type,
+    pickup_ntaname,
+    dropoff_ntaname
+FROM s3(
+    'https://datasets-documentation.s3.eu-west-3.amazonaws.com/nyc-taxi/trips_{0..10}.gz',
+    'TabSeparatedWithNames'
+);
+```
+
+## Create a dbt profile entry
+
+Use the following profile to create the associated dbt profile in the  dbt_profiles.yml in ~/.dbt
+```yml
+taxis:
+  outputs:
+
+    dev:
+      type: clickhouse
+      threads: 4 
+      host: localhost
+      port: 8123 
+      user: dbt_test
+      password: dbt_password
+      use_lw_deletes: true
+      schema: taxis_dbt 
+
+  target: dev
+
+```
+
+## Run the model
+
+`dbt run` in this directory should execute the model.  Each run will create a somewhat larger dataset (by adding
+additional random trip_ids).

examples/taxis/analyses/.gitkeep

@@ -0,0 +1,48 @@
+
+# Name your project! Project names should contain only lowercase characters
+# and underscores. A good package name should reflect your organization's
+# name or the intended use of these models
+name: 'taxis'
+version: '1.0.0'
+config-version: 2
+
+# This setting configures which "profile" dbt uses for this project.
+profile: 'taxis'
+
+# These configurations specify where dbt should look for different types of files.
+# The `model-paths` config, for example, states that models in this project can be
+# found in the "models/" directory. You probably won't need to change these!
+model-paths: ["models"]
+analysis-paths: ["analyses"]
+test-paths: ["tests"]
+seed-paths: ["seeds"]
+macro-paths: ["macros"]
+snapshot-paths: ["snapshots"]
+
+target-path: "target"  # directory which will store compiled SQL files
+clean-targets:         # directories to be removed by `dbt clean`
+  - "target"
+  - "dbt_packages"
+
+vars:
+  taxi_s3:
+    bucket: 'datasets-documentation.s3.eu-west-3.amazonaws.com/nyc-taxi'
+    fmt: 'TabSeparatedWithNames'
+    structure:
+      - trip_id             UInt32
+      - pickup_datetime     DateTime
+      - dropoff_datetime    DateTime
+      - pickup_longitude    Nullable(Float64)
+      - pickup_latitude     Nullable(Float64)
+      - dropoff_longitude   Nullable(Float64)
+      - dropoff_latitude    Nullable(Float64)
+      - passenger_count     UInt8
+      - trip_distance       Float32
+      - fare_amount         Float32
+      - extra               Float32
+      - tip_amount          Float32
+      - tolls_amount        Float32
+      - total_amount        Float32
+      - payment_type        LowCardinality(String)
+      - pickup_ntaname      LowCardinality(String)
+      - dropoff_ntaname     LowCardinality(String)

examples/taxis/dbt_project.yml

@@ -0,0 +1,17 @@
+version: 2
+
+models:
+  - name: trips_inc
+    description: NY Taxi dataset from S3
+    config:
+      materialized: incremental
+      order_by: rand_trip_id
+      unique_key: rand_trip_id
+
+  - name: trips_rand
+    description: Random indexes to apply to incremental materialization
+    config:
+      materialized: incremental
+      order_by: date_time
+      uniq_id: date_time
+      incremental_strategy: append