diff --git a/rill-modeling-examples/README.md b/rill-modeling-examples/README.md index f6628e8..caf28af 100644 --- a/rill-modeling-examples/README.md +++ b/rill-modeling-examples/README.md @@ -1,6 +1,6 @@ # Rill model patterns -A source-by-source gallery of YAML-only Rill model patterns for Snowflake, BigQuery, ClickHouse, and S3. The warehouse examples use a small `analytics.events` fact table, and the S3 examples use an equivalent Parquet prefix, with this logical shape: +A gallery of YAML-only Rill model patterns for Snowflake, BigQuery, S3-to-DuckDB, and S3-to-ClickHouse. The warehouse examples use a small `analytics.events` fact table. Both S3-backed directories use the equivalent Parquet prefix with this logical shape: | Column | Purpose | | -------------------------------------------- | -------------------------------------- | @@ -9,46 +9,78 @@ A source-by-source gallery of YAML-only Rill model patterns for Snowflake, BigQu | `updated_at` | Last source-side update timestamp | | `user_id`, `region`, `event_type`, `revenue` | Example dimensions and measure | -The S3 examples assume the layout `s3://example-bucket/analytics/events/dt=YYYY-MM-DD/hour=HH/*.parquet`. Replace the bucket and prefix with your own before running anything. +The S3 examples assume the layout `s3://example-bucket/analytics/events/dt=YYYY-MM-DD/hour=HH/*.parquet`. The ClickHouse examples execute ClickHouse's native `s3()` table function and materialize the selected rows into ClickHouse. Replace the bucket and prefix with your own before running anything. -Every model carries a `# disable: true` line inside its `refresh` block, commented out. A copied example therefore runs as-is on its cron once credentials are present; uncomment that line whenever you want a resource to stay inactive. Because the schedules are live, do not point the repository root at real credentials unless you intend the models to run. +Every model carries a `# disable: true` line inside its `refresh` block, commented out. A copied example therefore reconciles when its project starts and can run later on its schedule once credentials are present. Starting this gallery root can reconcile many models immediately, independently of their cron schedules. Do not start the gallery root or give it real credentials; copy only the example you intend to use into a scratch project, and uncomment `disable: true` until you are ready to ingest data. -Current Rill releases do not provide a direct ClickHouse-source-to-DuckDB-output executor. The ClickHouse models document the requested target shape and are marked conceptual; do not enable them without first replacing that connector path with a supported architecture. BigQuery, Snowflake, and S3 exports use supported paths into DuckDB. +The project default remains DuckDB for the other galleries, so every ClickHouse model explicitly sets `connector: clickhouse` for SQL execution and `output.connector: clickhouse` for materialization. The ClickHouse connector is read-write because Rill creates, inserts into, alters, and replaces model tables. + +In each model, the top-level `partitions` and `sql` are the production defaults. The canonical scenarios 01 and 02 also repeat them under explicit `prod.partitions` and `prod.sql` blocks to demonstrate environment overrides. A nested `dev` block narrows ingestion to a single date or fewer buckets. ## The recommended pattern -For a new incremental model, start from scenario **01** and change only the partition grain: +For a new S3-to-ClickHouse incremental model, start from scenario **01** or **02**. The example below uses daily replacement units: ```yaml incremental: true change_mode: patch # Adopt definition changes without an automatic full rebuild. -partitions_watermark: watermark_interval # Re-queue a partition while it is still inside its lateness window. +partitions_watermark: updated_on # Re-queue a prefix when its S3 objects change. partitions: - sql: | - SELECT - DATE(event_time) AS partition_date, - LEAST(NOW(), TIMESTAMP(DATE(event_time)) + INTERVAL 1 DAY) AS watermark_interval - FROM events - GROUP BY 1 + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=* + partition: directory + +sql: | + SELECT ... + FROM s3( + '{{ .partition.uri }}/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) output: + connector: clickhouse incremental_strategy: partition_overwrite # Idempotent: a rerun replaces the partition. partition_by: 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 + ) ``` The three properties solve three different problems, and they are designed to be used together: - **`partition_overwrite` gives idempotency.** Rerunning a partition replaces its rows instead of adding to them, so a failed run, a manual retry, and a targeted backfill all converge on the same output. This is the property that makes a pipeline safe to operate. - **`change_mode: patch` keeps definition changes cheap.** New logic takes effect on the next run without automatically rescanning history. Combined with `partition_overwrite`, you can then backfill exactly the partitions that need the new logic. -- **`partitions_watermark` handles late data.** It names any timestamp the partition query returns, not necessarily a column. Whenever the value advances, Rill marks that already-loaded partition pending again. +- **`partitions_watermark` handles changed objects.** Glob discovery returns `updated_on`; whenever it advances, Rill marks the already-loaded S3 prefix pending again. + +For ClickHouse, `output.partition_by` defines both the physical MergeTree partition and the replacement boundary used by `partition_overwrite`. Those boundaries must match the data produced by one Rill partition run. An hourly run must not emit only one hour into a monthly replacement partition, because that would replace the rest of the month. + +The overwrite examples derive `event_hour` or `event_date` from the S3 prefix and filter out rows whose `event_time` does not belong to that prefix. This prevents a misplaced row from replacing a different ClickHouse partition. In production, route rejected rows to a quarantine process or fail the upstream partition build rather than silently discarding them. -The gallery uses `LEAST(NOW(), + INTERVAL n)` rather than a bare `MAX(updated_at)`, because it makes the lateness policy explicit: +- Scenario 01 uses hourly physical partitions and a 30-day TTL, bounding active partitions to roughly 720. +- Scenarios 02, 04, 05, and 06 use daily physical partitions and a 365-day TTL, bounding active partitions to roughly 365. +- Merge scenarios use monthly physical partitions because they insert versions instead of replacing an incomplete physical partition. -- While the partition is younger than the interval, the expression evaluates to `NOW()`, which advances on every discovery pass, so the partition is reprocessed on every refresh. With `partition_overwrite` that is safe, just not free. -- Once the interval has elapsed, the expression freezes at a constant and the partition is never re-queued again. +The `order_by` and primary key assume queries filter first by date, then by region/event type, and sometimes by user. Adapt them to actual query logs before creating production tables because a MergeTree sorting key cannot be changed in place. The bloom-filter indexes demonstrate exact lookups outside the leading key prefix; validate that they skip enough granules on real data before retaining them. -That is a **declared** lateness window: it needs no reliable `updated_at` in the source and no `MAX()` aggregation to detect change, at the cost of reprocessing partitions that did not change and going blind to corrections arriving after the window. `MAX(updated_at)` is the data-driven alternative — it reprocesses only what actually changed and catches arbitrarily late corrections, but depends on the source maintaining that column. Scenario 03 keeps `MAX(updated_at)` so both are visible, and the S3 examples combine the two with `GREATEST(updated_on, LEAST(...))`. +The `query_settings` value makes ClickHouse infer the schema of `s3()` from the Parquet files rather than from the insertion table. This is necessary because the output schema contains derived columns such as `event_date` that do not exist in the files. Note that Rill spells the strategy `partition_overwrite`; `partition_override` is not a valid value. @@ -61,7 +93,7 @@ Each source directory contains the same nine scenarios. Filenames omit the redun | 00 | Full-refresh baseline | Non-incremental materialization and a dev limit | | 01 | **Hourly partition overwrite with watermark** | **The recommended pattern: `partition_overwrite` + `patch` + watermark, one hour per partition** | | 02 | Daily partition overwrite with watermark | The same recommended pattern at a daily grain | -| 03 | Late hourly updates with merge | `merge` by `unique_key` when no output column identifies a whole partition | +| 03 | Late hourly updates with merge | Versioned `ReplacingMergeTree` ingestion when rows are updated independently | | 04 | Patch schema evolution | `change_mode: patch`; new logic applies without a history rebuild | | 05 | Reset schema evolution | `change_mode: reset`; definition changes rebuild history | | 06 | Schema hook and retries | Idempotent additive `pre_exec`, output field, post-check, exponential backoff | @@ -86,15 +118,17 @@ Every schedule has an explicit `time_zone`. Cron schedules do not run in local d ## Incremental strategies - `partition_overwrite` replaces every output row belonging to the configured `output.partition_by` value. It is the preferred strategy, and the default in this gallery, because retrying a day, hour, week, or month remains idempotent. -- `merge` updates matching rows and inserts new rows using `output.unique_key`. Use it when the output has no single column that identifies a complete partition. A unique key must be stable, non-null, and genuinely unique; a weak key can update the wrong row, leave duplicates, or produce nondeterministic results. A composite identity can be expressed as multiple key columns. Merge never deletes, so a row dropped from a reprocessed source window stays in the output. +- For ClickHouse, `merge` requires a `ReplacingMergeTree` engine. The ClickHouse executor does not use `output.unique_key`; deduplication is determined by the complete sorting key. These examples use `ReplacingMergeTree(updated_at)` with `(event_date, event_id)`, so `event_time` must remain immutable for an event. Replacement is eventually applied by background merges, and queries requiring immediate deduplication may need `FINAL` selectively. - `append` is deliberately not demonstrated. It inserts without matching or deleting, so a rerun of the same input duplicates rows. Use `partition_overwrite` instead; it is idempotent even for data you believe is immutable. ## Watermarks and late data `partitions_watermark` names a timestamp returned by the partition discovery query. The watermark is excluded from the partition identity. When its value advances, Rill marks that existing partition pending again. -- For warehouse sources, the discovery query returns `MAX(updated_at)` per day or hour and the model reloads the complete affected window. -- For S3, glob partition discovery already exposes each object's `updated_on`, so `partitions_watermark: updated_on` re-queues a prefix whose files were rewritten. +- Warehouse examples discover watermarks with SQL. +- S3 glob discovery exposes `updated_on`, so `partitions_watermark: updated_on` re-queues a prefix whose objects were rewritten. + +An object-listing watermark is not a deletion manifest. Deleting an object may not advance the directory watermark, and a reprocessed prefix that becomes completely empty produces no temporary ClickHouse partition to replace. If deletes must propagate, publish a monotonically increasing manifest/version for each prefix or run an explicit destination-partition cleanup workflow. The watermark cannot be assigned the value `manual`; it is a column name, not an execution mode. @@ -107,8 +141,6 @@ For sources whose updates can arrive after a long delay, expand the discovery qu `change_mode: manual` also exists in Rill, but this gallery does not demonstrate it: it pauses reconciliation until an operator chooses a refresh, which is useful for approval gates but obscures the pattern the examples are teaching. -See [Model behavior and operator control](docs/model-behavior.md) for a focused comparison of watermarks, change modes, refresh schedules, incremental strategies, and partitions versus state. - ## Running an example Create a scratch Rill project containing only one connector and one model. This keeps unrelated models from scanning warehouses or producing missing-credential errors: @@ -117,30 +149,30 @@ Create a scratch Rill project containing only one connector and one model. This example_dir="$(mktemp -d /tmp/rill-model-example.XXXXXX)" mkdir -p "$example_dir/connectors" "$example_dir/models" cp rill.yaml .env.example "$example_dir/" -cp connectors/snowflake.yaml "$example_dir/connectors/" -cp models/snowflake/01_hourly_partition_overwrite_watermark.yaml "$example_dir/models/" +cp connectors/s3.yaml connectors/clickhouse.yaml "$example_dir/connectors/" +cp models/clickhouse/01_hourly_partition_overwrite_watermark.yaml "$example_dir/models/" cd "$example_dir" cp .env.example .env -# Populate SNOWFLAKE_DSN and adjust analytics.events if necessary. +# Populate CLICKHOUSE_DSN and AWS credentials, then replace the example S3 path. # The cron is live; uncomment refresh.disable if you want to inspect the model without running it. rill start . ``` -Substitute the connector and model paths for BigQuery, ClickHouse, S3, or another scenario. Do not enable models in the repository root; it is intentionally kept as an inactive reference project. +Substitute connector and model paths for another scenario. Do not run models from the gallery root; it contains many live example resources and is intended for browsing and copying. Useful commands: ```bash -rill project partitions --local --model snowflake_01_hourly_partition_overwrite_watermark -rill project refresh --local --model snowflake_01_hourly_partition_overwrite_watermark -rill project refresh --local --model snowflake_01_hourly_partition_overwrite_watermark --full +rill project partitions --local --model clickhouse_01_hourly_partition_overwrite_watermark +rill project refresh --local --model clickhouse_01_hourly_partition_overwrite_watermark +rill project refresh --local --model clickhouse_01_hourly_partition_overwrite_watermark --full # Copy a key from the partitions listing, then target only that partition: -rill project refresh --local --model snowflake_01_hourly_partition_overwrite_watermark --partition +rill project refresh --local --model clickhouse_01_hourly_partition_overwrite_watermark --partition ``` ## Validation -Run `./scripts/check_examples.sh`. It checks YAML parsing, unique resource names, the commented-out `refresh.disable` convention, source-specific connector names, the recommended-pattern properties on scenarios 01 and 02, the absence of `change_mode: manual` and `incremental_strategy: append`, the partitions-or-state split, retry behavior, and schema-hook intent. The checks are static; running a model still requires real credentials and an `analytics.events` table or Parquet prefix. +Run `./scripts/check_examples.sh`. It checks that all nine ClickHouse examples read through `s3()`, write to ClickHouse, define ordering, TTL, schema, and indexes, and that the recommended incremental examples use S3 partitions with `partition_overwrite`. The checks are static; running a model still requires real credentials and the example Parquet prefix. ## Canonical references @@ -149,3 +181,6 @@ Run `./scripts/check_examples.sh`. It checks YAML parsing, unique resource names - [Partitioned models](https://docs.rilldata.com/developers/build/models/partitioned-models) - [Incremental models](https://docs.rilldata.com/developers/build/models/incremental-models) - [Connector YAML reference](https://docs.rilldata.com/reference/project-files/connectors) +- [ClickHouse primary-key guidance](https://clickhouse.com/docs/best-practices/choosing-a-primary-key) +- [ClickHouse partitioning guidance](https://clickhouse.com/docs/best-practices/choosing-a-partitioning-key) +- [ClickHouse skipping indexes](https://clickhouse.com/docs/best-practices/use-data-skipping-indices-where-appropriate) diff --git a/rill-modeling-examples/connectors/clickhouse.yaml b/rill-modeling-examples/connectors/clickhouse.yaml index 6da16dd..af5598d 100644 --- a/rill-modeling-examples/connectors/clickhouse.yaml +++ b/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 }}" diff --git a/rill-modeling-examples/models/clickhouse/00_full_refresh.yaml b/rill-modeling-examples/models/clickhouse/00_full_refresh.yaml index 3608ca2..ef2d70e 100644 --- a/rill-modeling-examples/models/clickhouse/00_full_refresh.yaml +++ b/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 + ) diff --git a/rill-modeling-examples/models/clickhouse/01_hourly_partition_overwrite_watermark.yaml b/rill-modeling-examples/models/clickhouse/01_hourly_partition_overwrite_watermark.yaml index 4197469..9751f54 100644 --- a/rill-modeling-examples/models/clickhouse/01_hourly_partition_overwrite_watermark.yaml +++ b/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 + ) diff --git a/rill-modeling-examples/models/clickhouse/02_daily_partition_overwrite_watermark.yaml b/rill-modeling-examples/models/clickhouse/02_daily_partition_overwrite_watermark.yaml index 976c21e..ddfd1a6 100644 --- a/rill-modeling-examples/models/clickhouse/02_daily_partition_overwrite_watermark.yaml +++ b/rill-modeling-examples/models/clickhouse/02_daily_partition_overwrite_watermark.yaml @@ -1,68 +1,103 @@ -# Scenario: apply the recommended incremental pattern at a daily grain for large fact tables. -# Source and output: Exports data from ClickHouse into a materialized DuckDB table. -# Flow: partition SQL returns each recent date with MAX(updated_at); an advancing watermark re-queues that date. -# Recommended: the same partition_overwrite, change_mode patch, and watermark combination as scenario 01, one day per partition. -# Strategy: partition_overwrite replaces the whole day, so reprocessing a corrected day stays idempotent. -# Change behavior: patch keeps existing days untouched; backfill explicitly when a change must apply to history. -# Watermark: LEAST(now, date + 1 day) equals now while the day is inside its lateness window, then freezes at a constant. -# Late data: the day is therefore re-queued on every refresh until it is a day old, and is never re-queued automatically after that. -# Development: only the last two days are discovered, avoiding an accidental year-long local scan. -# 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: daily S3 ingestion with partition_overwrite and object-update watermarks. +# The ClickHouse partition is one day, exactly matching each Rill replacement unit. type: model name: clickhouse_02_daily_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: 30m # Abort one model execution after this duration. +incremental: true +change_mode: patch +materialize: true +partitions_watermark: updated_on +partitions_concurrency: 4 +timeout: 30m refresh: - cron: "0 2 * * *" # daily at 02: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 2 * * *" + time_zone: UTC + # disable: true -partitions: # Each result row defines one independently tracked partition. - connector: clickhouse +# Production partition discovery: one Hive-style dt directory per run unit. +partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=* + partition: directory + +sql: | + WITH toDate( + extract('{{ .partition.uri }}', 'dt=([0-9]{4}-[0-9]{2}-[0-9]{2})') + ) AS source_date + SELECT + CAST(event_id AS UUID) AS event_id, + CAST(event_time AS DateTime64(3, 'UTC')) AS event_time, + source_date 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( + '{{ .partition.uri }}/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) + WHERE CAST(event_time AS DateTime64(3, 'UTC')) >= toDateTime(source_date, 'UTC') + AND CAST(event_time AS DateTime64(3, 'UTC')) < toDateTime(source_date, 'UTC') + INTERVAL 1 DAY + +prod: + partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=* + partition: directory sql: | + WITH toDate( + extract('{{ .partition.uri }}', 'dt=([0-9]{4}-[0-9]{2}-[0-9]{2})') + ) AS source_date SELECT - toDate(event_time) AS partition_date, - least(now(), toDateTime(toDate(event_time)) + INTERVAL 1 DAY) AS watermark_interval - FROM analytics.events - WHERE event_time >= today() - INTERVAL 30 DAY - GROUP BY 1 + CAST(event_id AS UUID) AS event_id, + CAST(event_time AS DateTime64(3, 'UTC')) AS event_time, + source_date 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( + '{{ .partition.uri }}/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) + WHERE CAST(event_time AS DateTime64(3, 'UTC')) >= toDateTime(source_date, 'UTC') + AND CAST(event_time AS DateTime64(3, 'UTC')) < toDateTime(source_date, 'UTC') + INTERVAL 1 DAY -dev: # Override partition discovery locally to keep development scans small. - change_mode: reset # Rebuild only the small development slice after edits. +dev: + change_mode: reset partitions: - connector: clickhouse - sql: | - SELECT - toDate(event_time) AS partition_date, - least(now(), toDateTime(toDate(event_time)) + INTERVAL 1 DAY) AS watermark_interval - FROM analytics.events - WHERE event_time >= today() - INTERVAL 2 DAY - GROUP BY 1 - -sql: | - SELECT - event_id, - event_time, - toDate(event_time) AS event_date, - updated_at, - user_id, - region, - event_type, - revenue - FROM analytics.events - WHERE event_time >= toDateTime('{{ .partition.partition_date }}') - AND event_time < toDateTime('{{ .partition.partition_date }}') + INTERVAL 1 DAY + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=2026-08-27 + partition: directory output: - connector: duckdb - incremental_strategy: partition_overwrite # Replace the complete output partition; idempotent for retries. - partition_by: event_date # Column identifying the output rows that partition_overwrite replaces. + connector: clickhouse + incremental_strategy: partition_overwrite + partition_by: 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 + ) diff --git a/rill-modeling-examples/models/clickhouse/03_hourly_merge_late.yaml b/rill-modeling-examples/models/clickhouse/03_hourly_merge_late.yaml index b7c096d..023d456 100644 --- a/rill-modeling-examples/models/clickhouse/03_hourly_merge_late.yaml +++ b/rill-modeling-examples/models/clickhouse/03_hourly_merge_late.yaml @@ -1,82 +1,71 @@ -# Scenario: handle late arrivals when the output has no single column that identifies a complete partition. -# Source and output: Exports data from ClickHouse into a materialized DuckDB table. -# Flow: MAX(updated_at) is returned beside each hour; an advancing watermark re-queues that existing hour. -# Strategy: merge upserts rows by event_id instead of replacing every row of an output partition. -# Prefer scenario 01: partition_overwrite is idempotent by construction; choose merge only when a partition column is unavailable. -# Unique key: event_id must be stable, non-null, and unique, or merge can update the wrong row or leave duplicates. -# Change behavior: patch adopts compatible new logic without rebuilding prior hours. -# Watermark contrast: this example deliberately keeps the data-driven MAX(updated_at) form; scenarios 01, 02, and 04 use the clock-driven LEAST(now, start + interval) window. -# 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: late S3 updates merged into ClickHouse by stable event identity. +# ReplacingMergeTree(updated_at) keeps the newest version for an event. event_time +# must be immutable because event_date participates in the deduplication key. type: model name: clickhouse_03_hourly_merge_late 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: changed_at # Timestamp column from partition SQL; advances re-queue an existing partition. -partitions_concurrency: 6 # 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, - max(updated_at) AS changed_at - FROM analytics.events - WHERE event_time >= now() - INTERVAL 168 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, - max(updated_at) AS changed_at - FROM analytics.events - WHERE event_time >= now() - INTERVAL 6 HOUR - AND event_time < toStartOfHour(now()) - GROUP BY 1 +partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=*/hour=* + partition: directory sql: | SELECT - event_id, - event_time, - 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, + 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( + '{{ .partition.uri }}/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) -retry: # Retries apply to transient execution failures, not invalid SQL or schema errors. - attempts: 4 # Total attempts, including retries after the first failure. - delay: 15s # 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. +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: merge # Upsert by unique_key; use when no single output column identifies a partition. - unique_key: # Required for merge; columns must form a stable, non-null, truly unique row identity. - - event_id + connector: clickhouse + incremental_strategy: merge + engine: ReplacingMergeTree(updated_at) + partition_by: toYYYYMM(event_date) + order_by: (event_date, event_id) + primary_key: (event_date, event_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 + ) diff --git a/rill-modeling-examples/models/clickhouse/04_schema_patch.yaml b/rill-modeling-examples/models/clickhouse/04_schema_patch.yaml index 9806692..b4fe714 100644 --- a/rill-modeling-examples/models/clickhouse/04_schema_patch.yaml +++ b/rill-modeling-examples/models/clickhouse/04_schema_patch.yaml @@ -1,66 +1,77 @@ -# Scenario: demonstrate an additive, backward-compatible schema change without rebuilding historical output. -# Source and output: Exports data from ClickHouse into a materialized DuckDB table. -# Flow: recent days are re-queued by the watermark and new rows include the added traffic_source column. -# Watermark: LEAST(now, date + 1 day) equals now while the day is inside its lateness window, then freezes at a constant. -# Change behavior: patch switches future runs to the new logic while preserving already-loaded history. -# Trade-off: historical rows are not recomputed; backfill affected partitions explicitly when consistency is needed. -# Strategy: merge uses event_id to update existing events and insert newly arrived events. -# Compare: scenario 05 shows the reset alternative for changes that must apply to all history. -# 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: add traffic_source without automatically rebuilding historical data. +# Re-run selected S3 date partitions when historical consistency is required. type: model name: clickhouse_04_schema_patch 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. +incremental: true +change_mode: patch +materialize: true +partitions_watermark: updated_on refresh: - cron: "0 6 * * 1-5" # weekdays at 06: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 6 * * 1-5" + time_zone: UTC + # disable: true -partitions: # Each result row defines one independently tracked partition. - connector: clickhouse - sql: | - SELECT - toDate(event_time) AS partition_date, - least(now(), toDateTime(toDate(event_time)) + INTERVAL 1 DAY) AS watermark_interval - FROM analytics.events - WHERE event_time >= today() - INTERVAL 14 DAY - 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 - toDate(event_time) AS partition_date, - least(now(), toDateTime(toDate(event_time)) + INTERVAL 1 DAY) AS watermark_interval - FROM analytics.events - WHERE event_time >= today() - INTERVAL 1 DAY - GROUP BY 1 +partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=* + partition: directory sql: | + WITH toDate( + extract('{{ .partition.uri }}', 'dt=([0-9]{4}-[0-9]{2}-[0-9]{2})') + ) AS source_date SELECT - event_id, - event_time, - updated_at, + CAST(event_id AS UUID) AS event_id, + CAST(event_time AS DateTime64(3, 'UTC')) AS event_time, + source_date AS event_date, + CAST(updated_at AS DateTime64(3, 'UTC')) AS updated_at, CAST(coalesce(traffic_source, 'unknown') AS String) AS traffic_source, - user_id, - region, - event_type, - revenue - FROM analytics.events - WHERE event_time >= toDateTime('{{ .partition.partition_date }}') - AND event_time < toDateTime('{{ .partition.partition_date }}') + INTERVAL 1 DAY + 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 }}/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) + WHERE CAST(event_time AS DateTime64(3, 'UTC')) >= toDateTime(source_date, 'UTC') + AND CAST(event_time AS DateTime64(3, 'UTC')) < toDateTime(source_date, 'UTC') + INTERVAL 1 DAY + +dev: + change_mode: reset + partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=2026-08-27 + partition: directory output: - connector: duckdb - incremental_strategy: merge # Upsert by unique_key; use when no single output column identifies a partition. - unique_key: # Required for merge; columns must form a stable, non-null, truly unique row identity. - - event_id + connector: clickhouse + incremental_strategy: partition_overwrite + partition_by: 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 + pre_exec: | + ALTER TABLE IF EXISTS clickhouse_04_schema_patch + ADD COLUMN IF NOT EXISTS traffic_source LowCardinality(String) DEFAULT 'unknown' AFTER updated_at + columns: | + ( + event_id UUID, + event_time DateTime64(3, 'UTC'), + event_date Date, + updated_at DateTime64(3, 'UTC'), + traffic_source LowCardinality(String), + 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 + ) diff --git a/rill-modeling-examples/models/clickhouse/05_schema_reset.yaml b/rill-modeling-examples/models/clickhouse/05_schema_reset.yaml index 892f6c4..f9c52e9 100644 --- a/rill-modeling-examples/models/clickhouse/05_schema_reset.yaml +++ b/rill-modeling-examples/models/clickhouse/05_schema_reset.yaml @@ -1,53 +1,70 @@ -# Scenario: demonstrate a breaking schema or semantic change that must be consistent across all historical rows. -# Source and output: Exports data from ClickHouse into a materialized DuckDB table. -# Flow: monthly partitions bound each source query, but a definition change causes the output to be rebuilt. -# Change behavior: reset automatically drops and recreates the model after its specification changes. -# Use when: renaming columns, changing types, or changing business logic makes mixed historical schemas unsafe. -# Trade-off: reset is operationally simple but can trigger an expensive full-history source scan. -# Compare: scenario 04 uses patch when mixed historical logic is acceptable. -# 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: reset the ClickHouse table when a breaking schema change must apply to all history. type: model name: clickhouse_05_schema_reset connector: clickhouse -incremental: true # Track completed partitions and process only new or re-queued partitions. -change_mode: reset # Definition changes automatically rebuild all historical output. -materialize: true # Persist the cross-connector result for fast downstream queries. +incremental: true +change_mode: reset +materialize: true +partitions_watermark: updated_on refresh: - cron: "0 3 1 * *" # monthly on day 1 at 03: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 3 1 * *" + time_zone: UTC + # disable: true -partitions: # Each result row defines one independently tracked partition. - connector: clickhouse - sql: | - SELECT DISTINCT toStartOfMonth(event_time) AS partition_month - FROM analytics.events - -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 toStartOfMonth(today()) AS partition_month +partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=* + partition: directory sql: | + WITH toDate( + extract('{{ .partition.uri }}', 'dt=([0-9]{4}-[0-9]{2}-[0-9]{2})') + ) AS source_date SELECT - event_id, - event_time, - toStartOfMonth(event_time) AS event_month, - updated_at, - CAST(revenue AS Decimal(18, 4)) AS revenue_amount, - user_id, - region, - event_type - FROM analytics.events - WHERE event_time >= toDateTime('{{ .partition.partition_month }}') - AND event_time < toDateTime('{{ .partition.partition_month }}') + INTERVAL 1 MONTH + CAST(event_id AS UUID) AS event_id, + CAST(event_time AS DateTime64(3, 'UTC')) AS event_time, + source_date 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_amount + FROM s3( + '{{ .partition.uri }}/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) + WHERE CAST(event_time AS DateTime64(3, 'UTC')) >= toDateTime(source_date, 'UTC') + AND CAST(event_time AS DateTime64(3, 'UTC')) < toDateTime(source_date, 'UTC') + INTERVAL 1 DAY + +dev: + partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=2026-08-27 + partition: directory output: - connector: duckdb - incremental_strategy: partition_overwrite # Replace the complete output partition; idempotent for retries. - partition_by: event_month # Column identifying the output rows that partition_overwrite replaces. + connector: clickhouse + incremental_strategy: partition_overwrite + partition_by: 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_amount 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 + ) diff --git a/rill-modeling-examples/models/clickhouse/06_hooks_retry.yaml b/rill-modeling-examples/models/clickhouse/06_hooks_retry.yaml index 2ef1844..dda111e 100644 --- a/rill-modeling-examples/models/clickhouse/06_hooks_retry.yaml +++ b/rill-modeling-examples/models/clickhouse/06_hooks_retry.yaml @@ -1,74 +1,91 @@ -# Scenario: add a source field before ingestion, export it to DuckDB, and retry transient ClickHouse failures. -# Source and output: Exports data from ClickHouse into a materialized DuckDB table. -# Flow: after an initial output exists, output.pre_exec adds ingestion_source; the query populates it; post_exec updates statistics. -# Retries: exponential backoff spaces repeated attempts; if_error_matches replaces, not extends, Rill defaults. -# Safety: ADD COLUMN IF NOT EXISTS is idempotent because retries may execute output.pre_exec more than once. -# Strategy: partition_overwrite keeps a retried day free of duplicate output rows. -# 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: idempotent ClickHouse output hooks plus retries for transient S3/network failures. type: model name: clickhouse_06_hooks_retry 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_concurrency: 3 # Maximum partitions processed in parallel; tune to source capacity. -timeout: 25m # Abort one model execution after this duration. +incremental: true +change_mode: patch +materialize: true +partitions_watermark: updated_on +partitions_concurrency: 3 +timeout: 25m refresh: - cron: "0 2 * * *" # daily at 02: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 2 * * *" + time_zone: UTC + # disable: true -partitions: # Each result row defines one independently tracked partition. - connector: clickhouse - sql: | - SELECT DISTINCT toDate(event_time) AS partition_date - FROM analytics.events - WHERE event_time >= today() - INTERVAL 30 DAY - -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 today() AS partition_date +partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=* + partition: directory sql: | + WITH toDate( + extract('{{ .partition.uri }}', 'dt=([0-9]{4}-[0-9]{2}-[0-9]{2})') + ) AS source_date SELECT - event_id, - event_time, - toDate(event_time) AS event_date, - updated_at, - user_id, - region, - event_type, - 'clickhouse' AS ingestion_source, - revenue - FROM analytics.events - WHERE event_time >= toDateTime('{{ .partition.partition_date }}') - AND event_time < toDateTime('{{ .partition.partition_date }}') + INTERVAL 1 DAY + CAST(event_id AS UUID) AS event_id, + CAST(event_time AS DateTime64(3, 'UTC')) AS event_time, + source_date 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('s3' AS String) AS ingestion_source, + CAST(revenue AS Decimal(18, 4)) AS revenue + FROM s3( + '{{ .partition.uri }}/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) + WHERE CAST(event_time AS DateTime64(3, 'UTC')) >= toDateTime(source_date, 'UTC') + AND CAST(event_time AS DateTime64(3, 'UTC')) < toDateTime(source_date, 'UTC') + INTERVAL 1 DAY -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. +dev: + change_mode: reset + partitions: + glob: + connector: s3 + path: s3://example-bucket/analytics/events/dt=2026-08-27 + partition: directory + +retry: + attempts: 5 + delay: 10s + exponential_backoff: true + if_error_matches: + - ".*HTTP 5[0-9]{2}.*" + - ".*SlowDown.*" + - ".*RequestTimeout.*" + - ".*code: 209, message:.*" + - ".*code: 210, message:.*" + - ".*code: 745, message:.*" output: - connector: duckdb - incremental_strategy: partition_overwrite # Replace the complete output partition; idempotent for retries. - partition_by: event_date # Column identifying the output rows that partition_overwrite replaces. + connector: clickhouse + incremental_strategy: partition_overwrite + partition_by: 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 pre_exec: | ALTER TABLE IF EXISTS clickhouse_06_hooks_retry - ADD COLUMN IF NOT EXISTS ingestion_source VARCHAR DEFAULT 'unknown' - post_exec: | - ANALYZE clickhouse_06_hooks_retry + ADD COLUMN IF NOT EXISTS ingestion_source LowCardinality(String) DEFAULT 'unknown' AFTER event_type + post_exec: SELECT 1 FROM clickhouse_06_hooks_retry LIMIT 1 + 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), + ingestion_source 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 + ) diff --git a/rill-modeling-examples/models/clickhouse/07_hash_buckets.yaml b/rill-modeling-examples/models/clickhouse/07_hash_buckets.yaml index 20241f7..76c70ba 100644 --- a/rill-modeling-examples/models/clickhouse/07_hash_buckets.yaml +++ b/rill-modeling-examples/models/clickhouse/07_hash_buckets.yaml @@ -1,63 +1,85 @@ -# Scenario: split a very large table by stable hash buckets when time windows are skewed or poorly distributed. -# Source and output: Exports data from ClickHouse into a materialized DuckDB table. -# Flow: each bucket is a partition and MAX(updated_at) re-queues buckets containing changed events. -# Strategy: merge upserts bucket results by event_id, so reprocessing a bucket updates instead of duplicates. -# Trade-off: watermark discovery groups the source by hash and may be costlier than native time partition pruning. -# Prefer scenario 01 or 02 when the source has a usable time column; hash buckets are a fallback for skewed data. -# Development: two buckets provide representative behavior while reducing local scan and write volume. -# Watermark contrast: hash buckets have no time key to add an interval to, so this example must use the data-driven MAX(updated_at) form. -# 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: partition overwrite by stable hash bucket when time prefixes are highly skewed. +# Each bucket query scans the S3 prefix, so prefer time partitions when available. type: model name: clickhouse_07_hash_buckets 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: changed_at # Timestamp column from partition SQL; advances re-queue an existing partition. -partitions_concurrency: 8 # Maximum partitions processed in parallel; tune to source capacity. +incremental: true +change_mode: patch +materialize: true +partitions_concurrency: 4 refresh: - cron: "0 1 * * *" # daily at 01: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 1 * * *" + time_zone: UTC + # disable: true -partitions: # Each result row defines one independently tracked partition. +# Production SQL-based partition discovery creates sixteen stable buckets. +partitions: connector: clickhouse - sql: | - SELECT - cityHash64(toString(event_id)) % 16 AS bucket_id, - max(updated_at) AS changed_at - FROM analytics.events - 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 - cityHash64(toString(event_id)) % 2 AS bucket_id, - max(updated_at) AS changed_at - FROM analytics.events - GROUP BY 1 + sql: SELECT number AS bucket_id FROM numbers(16) 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(cityHash64(toString(event_id)) % 16 AS UInt8) AS bucket_id, + 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' + ) WHERE cityHash64(toString(event_id)) % 16 = {{ .partition.bucket_id }} +dev: + change_mode: reset + partitions: + connector: clickhouse + sql: SELECT number AS bucket_id FROM numbers(2) + sql: | + SELECT + 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(cityHash64(toString(event_id)) % 2 AS UInt8) AS bucket_id, + CAST(revenue AS Decimal(18, 4)) AS revenue + FROM s3( + 's3://example-bucket/analytics/events/dt=2026-08-27/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) + WHERE cityHash64(toString(event_id)) % 2 = {{ .partition.bucket_id }} + output: - connector: duckdb - incremental_strategy: merge # Upsert by unique_key; use when no single output column identifies a partition. - unique_key: # Required for merge; columns must form a stable, non-null, truly unique row identity. - - event_id + connector: clickhouse + incremental_strategy: partition_overwrite + partition_by: bucket_id + order_by: (bucket_id, event_date, region, event_type, user_id, event_time, event_id) + primary_key: (bucket_id, 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), + bucket_id UInt8, + 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 + ) diff --git a/rill-modeling-examples/models/clickhouse/08_state_incremental.yaml b/rill-modeling-examples/models/clickhouse/08_state_incremental.yaml index 8df37f7..8a15482 100644 --- a/rill-modeling-examples/models/clickhouse/08_state_incremental.yaml +++ b/rill-modeling-examples/models/clickhouse/08_state_incremental.yaml @@ -1,65 +1,84 @@ -# Scenario: show dbt-style state-based incremental ingestion and why partitions are the preferred alternative. -# Source and output: Exports data from ClickHouse into a materialized DuckDB table. -# Flow: Rill stores the state query result after each run; the next run filters the source with that stored watermark. -# Strategy: merge by event_id, because a state filter cannot guarantee that a row is loaded exactly once. -# Trade-off: state is not idempotent. A failed or partially applied run cannot be replayed the way a partition can. -# Constraint: state and partitions are mutually exclusive, so there is no per-partition retry or targeted backfill. -# Prefer scenario 01 or 02 unless the source genuinely has no column that can define partitions. -# Boundary: the filter is strictly greater than the stored value, so updated_at must be monotonic and set on every change. -# 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: state-based S3 ingestion into ClickHouse when no usable partition exists. +# This scans the S3 prefix to find rows newer than the stored watermark; prefer +# scenarios 01 or 02 for idempotent retries and targeted backfills. type: model name: clickhouse_08_state_incremental connector: clickhouse -incremental: true # Reuse the stored state instead of rescanning the full source on every run. -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. -timeout: 30m # Abort one model execution after this duration. +incremental: true +change_mode: patch +materialize: true +timeout: 30m refresh: - cron: "0 3 * * *" # daily at 03: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 3 * * *" + time_zone: UTC + # disable: true -state: # Evaluated after each successful run; the result is available as .state in the next run. - connector: duckdb # State is computed against the DuckDB output table, not the source warehouse. - sql: SELECT MAX(updated_at) AS max_updated_at FROM clickhouse_08_state_incremental +state: + connector: clickhouse + sql: SELECT max(updated_at) AS max_updated_at FROM clickhouse_08_state_incremental 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 incremental }} - WHERE updated_at > parseDateTimeBestEffort('{{ .state.max_updated_at }}') + WHERE updated_at >= parseDateTime64BestEffort('{{ .state.max_updated_at }}', 3) {{ end }} -dev: # Bound the local initial load; without partitions the first run would otherwise scan all history. - change_mode: reset # Rebuild only the small development slice after edits. +dev: + change_mode: reset sql: | SELECT - event_id, - event_time, - updated_at, - user_id, - region, - event_type, - revenue - FROM analytics.events - WHERE event_time >= now() - INTERVAL 2 DAY + 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=2026-08-27/hour=*/*.parquet', + '{{ .env.AWS_ACCESS_KEY_ID }}', + '{{ .env.AWS_SECRET_ACCESS_KEY }}', + 'Parquet' + ) {{ if incremental }} - AND updated_at > parseDateTimeBestEffort('{{ .state.max_updated_at }}') + WHERE updated_at >= parseDateTime64BestEffort('{{ .state.max_updated_at }}', 3) {{ end }} output: - connector: duckdb - incremental_strategy: merge # Upsert by unique_key; use when no single output column identifies a partition. - unique_key: # Required for merge; columns must form a stable, non-null, truly unique row identity. - - event_id + connector: clickhouse + incremental_strategy: merge + engine: ReplacingMergeTree(updated_at) + partition_by: toYYYYMM(event_date) + order_by: (event_date, event_id) + primary_key: (event_date, event_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 + ) diff --git a/rill-modeling-examples/rill.yaml b/rill-modeling-examples/rill.yaml index 81b61f7..ed179bf 100644 --- a/rill-modeling-examples/rill.yaml +++ b/rill-modeling-examples/rill.yaml @@ -1,6 +1,6 @@ -# Scenario: Safe project shell for browsing the model gallery. -# The individual model resources remain disabled until explicitly enabled. +# Scenario: Metadata for the model gallery. +# Starting this directory reconciles its resources; inspect or copy examples without starting the gallery root. compiler: rillv1 title: Rill Model Patterns -description: Partition-based and state-based model examples for Snowflake, BigQuery, ClickHouse, and S3 +description: Partition-based and state-based model examples, including production S3-to-ClickHouse ingestion olap_connector: duckdb diff --git a/rill-modeling-examples/scripts/check_examples.sh b/rill-modeling-examples/scripts/check_examples.sh new file mode 100755 index 0000000..563251e --- /dev/null +++ b/rill-modeling-examples/scripts/check_examples.sh @@ -0,0 +1,96 @@ +#!/usr/bin/env bash + +set -euo pipefail + +project_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +clickhouse_dir="$project_dir/models/clickhouse" + +fail() { + echo "check_examples: $1" >&2 + exit 1 +} + +[[ -d "$clickhouse_dir" ]] || fail "missing models/clickhouse" + +ruby -e ' + require "yaml" + ARGV.each do |file| + YAML.safe_load(File.read(file), [], [], false, filename: file) + end +' "$project_dir/rill.yaml" "$project_dir"/connectors/*.yaml "$clickhouse_dir"/*.yaml \ + || fail "YAML parsing failed" + +if rg -n "connector: duckdb|ClickHouse into a materialized DuckDB|ClickHouse-to-DuckDB" "$clickhouse_dir" "$project_dir/README.md"; then + fail "found obsolete ClickHouse-to-DuckDB configuration or documentation" +fi + +rg -q '^mode: readwrite$' "$project_dir/connectors/clickhouse.yaml" || fail "ClickHouse connector must be readwrite" + +model_count=0 +while IFS= read -r model; do + model_count=$((model_count + 1)) + rg -q '^connector: clickhouse' "$model" || fail "$model must execute SQL in ClickHouse" + rg -q 'FROM s3\(' "$model" || fail "$model must read Parquet through ClickHouse s3()" + rg -q '^output:$' "$model" || fail "$model must define output properties" + rg -q '^ connector: clickhouse$' "$model" || fail "$model must write to ClickHouse" + rg -q '^ order_by:' "$model" || fail "$model must define output.order_by" + rg -q '^ ttl:' "$model" || fail "$model must define output.ttl" + rg -q '^ query_settings: use_structure_from_insertion_table_in_table_functions = 0$' "$model" \ + || fail "$model must infer the structure from S3 instead of the insertion table" + rg -q '^ columns: \|' "$model" || fail "$model must define an explicit ClickHouse schema" + rg -q 'INDEX idx_event_id' "$model" || fail "$model must define the event_id skipping index" +done < <(find "$clickhouse_dir" -maxdepth 1 -type f -name '*.yaml' | sort) + +[[ "$model_count" -eq 9 ]] || fail "expected 9 ClickHouse model examples, found $model_count" + +for scenario in 01_hourly_partition_overwrite_watermark 02_daily_partition_overwrite_watermark; do + model="$clickhouse_dir/$scenario.yaml" + rg -q '^incremental: true' "$model" || fail "$model must be incremental" + rg -q '^ connector: s3$' "$model" || fail "$model must discover S3 partitions" + rg -q '^ incremental_strategy: partition_overwrite' "$model" || fail "$model must use partition_overwrite" + rg -q '^ partition_by:' "$model" || fail "$model must define the replacement and physical partition key" +done + +ruby -e ' + require "yaml" + ARGV.each do |file| + model = YAML.safe_load(File.read(file), [], [], false, filename: file) + raise "#{file}: missing prod.partitions" unless model.dig("prod", "partitions") + raise "#{file}: missing prod.sql" unless model.dig("prod", "sql") + raise "#{file}: prod.partitions drifted from the production default" unless model.dig("prod", "partitions") == model["partitions"] + raise "#{file}: prod.sql drifted from the production default" unless model.dig("prod", "sql") == model["sql"] + end +' "$clickhouse_dir/01_hourly_partition_overwrite_watermark.yaml" \ + "$clickhouse_dir/02_daily_partition_overwrite_watermark.yaml" \ + || fail "canonical models must define explicit production partitions and SQL" + +hourly="$clickhouse_dir/01_hourly_partition_overwrite_watermark.yaml" +daily="$clickhouse_dir/02_daily_partition_overwrite_watermark.yaml" +rg -q '^ partition_by: event_hour$' "$hourly" || fail "hourly overwrite must replace hourly ClickHouse partitions" +rg -q '^ ttl: event_time \+ INTERVAL 30 DAY DELETE$' "$hourly" || fail "hourly partition count must be TTL-bounded" +rg -q '^ partition_by: event_date$' "$daily" || fail "daily overwrite must replace daily ClickHouse partitions" +rg -q "extract\('\{\{ \.partition\.uri \}\}', 'dt=" "$hourly" || fail "hourly partition key must come from the S3 path" +rg -q "extract\('\{\{ \.partition\.uri \}\}', 'dt=" "$daily" || fail "daily partition key must come from the S3 path" + +for scenario in 04_schema_patch 05_schema_reset 06_hooks_retry; do + model="$clickhouse_dir/$scenario.yaml" + rg -q "extract\('\{\{ \.partition\.uri \}\}', 'dt=" "$model" || fail "$model partition key must come from the S3 path" +done + +for scenario in 03_hourly_merge_late 08_state_incremental; do + model="$clickhouse_dir/$scenario.yaml" + rg -q '^ engine: ReplacingMergeTree\(updated_at\)$' "$model" || fail "$model must use a versioned ReplacingMergeTree" +done + +rg -q "updated_at >= parseDateTime64BestEffort" "$clickhouse_dir/08_state_incremental.yaml" \ + || fail "state ingestion must overlap tied watermark values" +if rg -q '^ unique_key:' "$clickhouse_dir/03_hourly_merge_late.yaml" "$clickhouse_dir/08_state_incremental.yaml"; then + fail "ClickHouse merge examples must not imply that unique_key controls ReplacingMergeTree deduplication" +fi + +rg -q "ADD COLUMN IF NOT EXISTS traffic_source.*AFTER updated_at" "$clickhouse_dir/04_schema_patch.yaml" \ + || fail "schema patch must add traffic_source in SELECT insertion order" +rg -q "ADD COLUMN IF NOT EXISTS ingestion_source.*AFTER event_type" "$clickhouse_dir/06_hooks_retry.yaml" \ + || fail "hook example must add ingestion_source in SELECT insertion order" + +echo "check_examples: all static checks passed"