Add production S3-to-ClickHouse modeling examples

rill-modeling-examples/connectors/clickhouse.yaml

@@ -1,6 +1,7 @@
-# Scenario: Read-only ClickHouse source connector shared by the ClickHouse examples.
+# Scenario: Read-write ClickHouse OLAP connector used by the S3-to-ClickHouse examples.
+# The ClickHouse user needs CREATE, INSERT, ALTER, and DROP privileges for model tables.
 # Use a native-protocol DSN such as clickhouse://user:password@host:9440/database?secure=true.
 type: connector
 driver: clickhouse
-mode: read
+mode: readwrite
 dsn: "{{ .env.CLICKHOUSE_DSN }}"

rill-modeling-examples/models/clickhouse/00_full_refresh.yaml

@@ -1,37 +1,52 @@
-# Scenario: establish the simplest non-incremental baseline for a small or moderately sized source table.
-# Source and output: Exports data from ClickHouse into a materialized DuckDB table.
-# Flow: every refresh re-runs the full source query and replaces the materialized DuckDB output.
-# Use when: a full scan is affordable and consistent full-history results matter more than incremental cost.
-# Trade-off: there is no partition bookkeeping, so runtime and warehouse cost grow with source history.
-# Development: the template limits the source query to one recent day without changing production SQL.
-# Model YAML reference: https://docs.rilldata.com/reference/project-files/models
-# Incremental models guide: https://docs.rilldata.com/developers/build/models/incremental-models
-# Partitioned models guide: https://docs.rilldata.com/developers/build/models/partitioned-models
-# Runtime note: Current Rill releases do not support direct ClickHouse-to-DuckDB execution; this disabled spec is conceptual.
+# Scenario: full-refresh S3-to-ClickHouse baseline for a bounded Parquet dataset.
+# ClickHouse executes s3() and materializes the result as a MergeTree table.
 type: model
 name: clickhouse_00_full_refresh
 connector: clickhouse
-materialize: true  # Persist the cross-connector result for fast downstream queries.
+materialize: true
 
 refresh:
-  cron: "0 4 * * *"  # daily at 04:00; interpreted in time_zone (production by default).
-  time_zone: UTC  # Prevent host-local timezone differences from shifting the schedule.
-  # disable: true  # Uncomment to keep this resource inactive; it ships commented so a copied example runs as-is.
+  cron: "0 4 * * *"
+  time_zone: UTC
+  # disable: true
 
 sql: |
   SELECT
-    event_id,
-    event_time,
-    updated_at,
-    user_id,
-    region,
-    event_type,
-    revenue
-  FROM analytics.events
+    CAST(event_id AS UUID) AS event_id,
+    CAST(event_time AS DateTime64(3, 'UTC')) AS event_time,
+    toDate(event_time) AS event_date,
+    CAST(updated_at AS DateTime64(3, 'UTC')) AS updated_at,
+    CAST(user_id AS UInt64) AS user_id,
+    CAST(region AS String) AS region,
+    CAST(event_type AS String) AS event_type,
+    CAST(revenue AS Decimal(18, 4)) AS revenue
+  FROM s3(
+    's3://example-bucket/analytics/events/dt=*/hour=*/*.parquet',
+    '{{ .env.AWS_ACCESS_KEY_ID }}',
+    '{{ .env.AWS_SECRET_ACCESS_KEY }}',
+    'Parquet'
+  )
   {{ if dev }}
   WHERE event_time >= now() - INTERVAL 1 DAY
   {{ end }}
 
 output:
-  connector: duckdb
-  materialize: true
+  connector: clickhouse
+  partition_by: toYYYYMM(event_date)
+  order_by: (event_date, region, event_type, user_id, event_time, event_id)
+  primary_key: (event_date, region, event_type, user_id)
+  ttl: event_time + INTERVAL 365 DAY DELETE
+  query_settings: use_structure_from_insertion_table_in_table_functions = 0
+  columns: |
+    (
+      event_id UUID,
+      event_time DateTime64(3, 'UTC'),
+      event_date Date,
+      updated_at DateTime64(3, 'UTC'),
+      user_id UInt64,
+      region LowCardinality(String),
+      event_type LowCardinality(String),
+      revenue Decimal(18, 4),
+      INDEX idx_event_id event_id TYPE bloom_filter(0.01) GRANULARITY 4,
+      INDEX idx_user_id user_id TYPE bloom_filter(0.01) GRANULARITY 4
+    )

rill-modeling-examples/models/clickhouse/01_hourly_partition_overwrite_watermark.yaml

@@ -1,83 +1,119 @@
-# Scenario: the recommended incremental pattern: hourly partitions replaced idempotently, with automatic late-data handling.
-# Source and output: Exports data from ClickHouse into a materialized DuckDB table.
-# Flow: partition SQL returns each recent closed hour with MAX(updated_at); an advancing watermark re-queues that hour.
-# Recommended: partition_overwrite for idempotency, change_mode patch for cheap definition changes, and a watermark for late data.
-# Strategy: partition_overwrite replaces every row for the hour, so a retried or re-queued hour never duplicates rows.
-# Change behavior: patch adopts a new definition immediately instead of triggering an automatic full-history rebuild.
-# Watermark: LEAST(now, hour + 3h) equals now while the hour is inside its lateness window, then freezes at a constant.
-# Late data: the hour is therefore re-queued on every refresh for three hours, and is never re-queued automatically after that.
-# Development: only the last few closed hours are discovered, keeping local scans small.
-# Model YAML reference: https://docs.rilldata.com/reference/project-files/models
-# Incremental models guide: https://docs.rilldata.com/developers/build/models/incremental-models
-# Partitioned models guide: https://docs.rilldata.com/developers/build/models/partitioned-models
-# Runtime note: Current Rill releases do not support direct ClickHouse-to-DuckDB execution; this disabled spec is conceptual.
-# Retry error references: https://clickhouse.com/docs/concepts/features/configuration/settings/memory-overcommit and https://clickhouse.com/docs/reference/system-tables/errors
-# Driver error format: https://github.com/ClickHouse/clickhouse-go/blob/v2.41.0/lib/proto/exception.go
+# Scenario: recommended hourly S3 ingestion with idempotent partition replacement.
+# The physical ClickHouse partition matches the Rill overwrite unit. A 30-day TTL
+# bounds the table to about 720 active hourly partitions.
 type: model
 name: clickhouse_01_hourly_partition_overwrite_watermark
 connector: clickhouse
-incremental: true  # Track completed partitions and process only new or re-queued partitions.
-change_mode: patch  # Recommended: adopt a new definition immediately without an automatic history rebuild.
-materialize: true  # Persist the cross-connector result for fast downstream queries.
-partitions_watermark: watermark_interval  # Any expression works, not just a column: this one advances with the clock, then freezes.
-partitions_concurrency: 4  # Maximum partitions processed in parallel; tune to source capacity.
-timeout: 20m  # Abort one model execution after this duration.
+incremental: true
+change_mode: patch
+materialize: true
+partitions_watermark: updated_on
+partitions_concurrency: 4
+timeout: 20m
 
 refresh:
-  cron: "5 * * * *"  # five minutes after every hour; interpreted in time_zone (production by default).
-  time_zone: UTC  # Prevent host-local timezone differences from shifting the schedule.
-  # disable: true  # Uncomment to keep this resource inactive; it ships commented so a copied example runs as-is.
+  cron: "5 * * * *"
+  time_zone: UTC
+  # disable: true
 
-partitions:  # Each result row defines one independently tracked partition.
-  connector: clickhouse
-  sql: |
-    SELECT
-      toStartOfHour(event_time) AS partition_hour,
-      least(now(), toStartOfHour(event_time) + INTERVAL 3 HOUR) AS watermark_interval
-    FROM analytics.events
-    WHERE event_time >= now() - INTERVAL 48 HOUR
-      AND event_time < toStartOfHour(now())
-    GROUP BY 1
-
-dev:  # Override partition discovery locally to keep development scans small.
-  change_mode: reset  # Rebuild only the small development slice after edits.
-  partitions:
-    connector: clickhouse
-    sql: |
-      SELECT
-        toStartOfHour(event_time) AS partition_hour,
-        least(now(), toStartOfHour(event_time) + INTERVAL 3 HOUR) AS watermark_interval
-      FROM analytics.events
-      WHERE event_time >= now() - INTERVAL 3 HOUR
-        AND event_time < toStartOfHour(now())
-      GROUP BY 1
+# Production partition discovery scans all hourly prefixes. Object modification
+# time is the watermark, so rewriting a prefix safely re-queues that hour.
+partitions:
+  glob:
+    connector: s3
+    path: s3://example-bucket/analytics/events/dt=*/hour=*
+    partition: directory
 
 sql: |
+  WITH parseDateTime64BestEffort(
+    concat(
+      extract('{{ .partition.uri }}', 'dt=([0-9]{4}-[0-9]{2}-[0-9]{2})'),
+      ' ',
+      extract('{{ .partition.uri }}', 'hour=([0-9]{2})'),
+      ':00:00'
+    ),
+    3,
+    'UTC'
+  ) AS source_hour
   SELECT
-    event_id,
-    event_time,
-    toStartOfHour(event_time) AS event_hour,
-    updated_at,
-    user_id,
-    region,
-    event_type,
-    revenue
-  FROM analytics.events
-  WHERE event_time >= parseDateTimeBestEffort('{{ .partition.partition_hour }}')
-    AND event_time < parseDateTimeBestEffort('{{ .partition.partition_hour }}') + INTERVAL 1 HOUR
+    CAST(event_id AS UUID) AS event_id,
+    CAST(event_time AS DateTime64(3, 'UTC')) AS event_time,
+    source_hour AS event_hour,
+    CAST(updated_at AS DateTime64(3, 'UTC')) AS updated_at,
+    CAST(user_id AS UInt64) AS user_id,
+    CAST(region AS String) AS region,
+    CAST(event_type AS String) AS event_type,
+    CAST(revenue AS Decimal(18, 4)) AS revenue
+  FROM s3(
+    '{{ .partition.uri }}/*.parquet',
+    '{{ .env.AWS_ACCESS_KEY_ID }}',
+    '{{ .env.AWS_SECRET_ACCESS_KEY }}',
+    'Parquet'
+  )
+  WHERE CAST(event_time AS DateTime64(3, 'UTC')) >= source_hour
+    AND CAST(event_time AS DateTime64(3, 'UTC')) < source_hour + INTERVAL 1 HOUR
 
-retry:  # Retries apply to transient execution failures, not invalid SQL or schema errors.
-  attempts: 5  # Total attempts, including retries after the first failure.
-  delay: 10s  # Initial wait between attempts.
-  exponential_backoff: true  # Increase the delay after each failed attempt.
-  if_error_matches:  # Explicit matches replace Rill defaults; clickhouse-go exposes codes and messages, not names.
-    - ".*OvercommitTracker.*"  # ClickHouse memory overcommit selected or rejected a query.
-    - ".*code: 202, message:.*"  # TOO_MANY_SIMULTANEOUS_QUERIES: concurrency admission rejected the query.
-    - ".*code: 209, message:.*"  # SOCKET_TIMEOUT: a ClickHouse socket operation timed out.
-    - ".*code: 210, message:.*"  # NETWORK_ERROR: ClickHouse reported a network-layer failure.
-    - ".*code: 745, message:.*"  # SERVER_OVERLOADED: server admission control rejected the query.
+prod:
+  partitions:
+    glob:
+      connector: s3
+      path: s3://example-bucket/analytics/events/dt=*/hour=*
+      partition: directory
+  sql: |
+    WITH parseDateTime64BestEffort(
+      concat(
+        extract('{{ .partition.uri }}', 'dt=([0-9]{4}-[0-9]{2}-[0-9]{2})'),
+        ' ',
+        extract('{{ .partition.uri }}', 'hour=([0-9]{2})'),
+        ':00:00'
+      ),
+      3,
+      'UTC'
+    ) AS source_hour
+    SELECT
+      CAST(event_id AS UUID) AS event_id,
+      CAST(event_time AS DateTime64(3, 'UTC')) AS event_time,
+      source_hour AS event_hour,
+      CAST(updated_at AS DateTime64(3, 'UTC')) AS updated_at,
+      CAST(user_id AS UInt64) AS user_id,
+      CAST(region AS String) AS region,
+      CAST(event_type AS String) AS event_type,
+      CAST(revenue AS Decimal(18, 4)) AS revenue
+    FROM s3(
+      '{{ .partition.uri }}/*.parquet',
+      '{{ .env.AWS_ACCESS_KEY_ID }}',
+      '{{ .env.AWS_SECRET_ACCESS_KEY }}',
+      'Parquet'
+    )
+    WHERE CAST(event_time AS DateTime64(3, 'UTC')) >= source_hour
+      AND CAST(event_time AS DateTime64(3, 'UTC')) < source_hour + INTERVAL 1 HOUR
+
+dev:
+  change_mode: reset
+  partitions:
+    glob:
+      connector: s3
+      path: s3://example-bucket/analytics/events/dt=2026-08-27/hour=*
+      partition: directory
 
 output:
-  connector: duckdb
-  incremental_strategy: partition_overwrite  # Replace the complete output partition; idempotent for retries.
-  partition_by: event_hour  # Column identifying the output rows that partition_overwrite replaces.
+  connector: clickhouse
+  incremental_strategy: partition_overwrite
+  partition_by: event_hour
+  order_by: (event_hour, region, event_type, user_id, event_time, event_id)
+  primary_key: (event_hour, region, event_type, user_id)
+  ttl: event_time + INTERVAL 30 DAY DELETE
+  query_settings: use_structure_from_insertion_table_in_table_functions = 0
+  columns: |
+    (
+      event_id UUID,
+      event_time DateTime64(3, 'UTC'),
+      event_hour DateTime64(3, 'UTC'),
+      updated_at DateTime64(3, 'UTC'),
+      user_id UInt64,
+      region LowCardinality(String),
+      event_type LowCardinality(String),
+      revenue Decimal(18, 4),
+      INDEX idx_event_id event_id TYPE bloom_filter(0.01) GRANULARITY 4,
+      INDEX idx_user_id user_id TYPE bloom_filter(0.01) GRANULARITY 4
+    )