## Problem Siphon replicates GitLab.com PostgreSQL tables into ClickHouse through a native consumer, but the only path into Snowflake today is indirect: GCS Parquet consumer → PySpark compaction on Dataproc → Iceberg tables in Snowflake Open Catalog. That path works for the 10-table pilot but carries a JVM/Spark job outside the Siphon binary, hours of latency, and a second schema-management workflow (`iceberg_schema_generator`). The original Siphon design document names a Snowflake consumer as the second first-party consumer next to ClickHouse. We want a **native Snowflake consumer**, shipped inside the single `siphon` binary and Docker image, that follows the ClickHouse consumer's operational pattern: tables pre-exist, the consumer appends versioned rows, deletes are tombstones, dedup happens in the target, and schema is validated rather than inferred. ## Goals - `type: snowflake` consumer selectable in the unified config, wired like `clickhouse` / `postgresql` / `gcs_parquet`. - Same durability contract as the other consumers: a NATS message is ACKed only after the write is durable in Snowflake; any error NAKs and redelivers; every statement is idempotent under redelivery. - Same metadata contract: `_siphon_replicated_at` (version) and `_siphon_deleted` (tombstone) on every row. - Authoritative types from the packet (`column_type_oids`, shipped since !`08310f2`) and from Snowflake `INFORMATION_SCHEMA`, never inferred from values. - Schema drift detected and reported with an explicit policy; reconciliation of target schema against source. - Key-pair (JWT) authentication only, per the Data Team's service-user standard. - Reviewable in small MRs (each child task below is one MR). ## Non-goals (this issue) - Replacing the Iceberg/Open Catalog pipeline. Both can coexist; a follow-up decides retirement. - Snowpipe Streaming or a MERGE-based "current state" write mode (Phase 2, see below). - Red-data masking policies. Out of scope for the consumer; the DDL generator leaves a hook. ## How CDC lands in Snowflake without ReplacingMergeTree ClickHouse dedups on read/merge via `ReplacingMergeTree(_siphon_replicated_at, _siphon_deleted)`. Snowflake has no engine-level equivalent, so the plan splits the two halves that the engine bundles: ### Recommended: append-only log table + Dynamic Table for current state | Layer | Object | Written by | Purpose | |---|---|---|---| | Log | `<TABLE>` (regular table, `CLUSTER BY (<pk>)`) | Siphon consumer, INSERT only | Every event version, including tombstones. Full change history, dbt-snapshot friendly. | | Current | `<TABLE>_CURRENT` (Dynamic Table, `TARGET_LAG` configurable) | Snowflake, incremental refresh | `QUALIFY ROW_NUMBER() OVER (PARTITION BY <pk> ORDER BY _SIPHON_REPLICATED_AT DESC) = 1` then `WHERE NOT _SIPHON_DELETED`. The ReplacingMergeTree `FINAL` equivalent. | Why this shape: - **Faithful to the ClickHouse pattern.** Consumer logic is INSERT + tombstone INSERT, exactly what `clickhouse_delete_query.go` does today. No target-side UPDATE/DELETE/MERGE in the hot path. - **Idempotent under redelivery.** Re-inserting the same rows produces duplicate log rows with identical version; the Dynamic Table collapses them. Same crash-safety story as ClickHouse. - **Snapshots, CDC, and future correction packages all reduce to inserts** (correction = range tombstone + insert, mirroring #300). - **History is preserved**, which the Data Team wants for SCD modelling and which ClickHouse loses on merge. - **Schema evolution is cheap.** Snowflake docs confirm a Dynamic Table defined with `SELECT *` picks up an `ADD COLUMN` on its base table incrementally, with NULL for existing rows and no reinitialization. `ALTER TABLE ADD COLUMN` on the log is metadata-only. - **Per-table latency knob.** `TARGET_LAG` per Dynamic Table; a plain VIEW can replace the Dynamic Table for small tables that need zero lag. ### Alternatives considered | Option | Verdict | Why | |---|---|---| | `MERGE INTO` current-state table per package (like the PG sink's `ON CONFLICT`) | Phase 2 `write_mode: merge` | Simplest TOAST handling, no history. But MERGE rewrites micro-partitions per package on large tables, Snowflake PKs are not enforced so the consumer must dedup source keys itself, and a nondeterministic MERGE errors. Costly at CDC cadence. | | Snowpipe Streaming (high-performance architecture, REST API) | Phase 2 backend | Serverless per-GB pricing beats a 24/7 warehouse for continuous small batches, offset tokens map to NATS sequence for exactly-once. But SDKs are Java/Python/Node only; Go would use the REST API with JWT and channel management. Design the writer behind an interface so this slots in. | | GCS Parquet consumer + Snowpipe auto-ingest + Dynamic Table | Rejected as the primary path | Almost no new Go code, but keeps a second storage hop and doesn't give us the in-binary consumer the design calls for. Worth remembering as a fallback. | | Hybrid Tables (enforced PK, row store) | Rejected | OLTP pricing and limits; not for analytics tables of this size. | | Streams + Tasks on the log table | Rejected | Dynamic Tables are the current Snowflake-recommended replacement and need no orchestration. | ### Write path details - **Driver:** `github.com/snowflakedb/gosnowflake/v2` (pure Go, CGO-free, distroless-compatible). Multi-row inserts via array binding; the driver switches to a temp-stage upload above `CLIENT_STAGE_ARRAY_BINDING_THRESHOLD`, which is what we want for snapshot packages. Requires `CREATE STAGE` on the schema. - **Dispatch shape:** per-message `EventHandler` only (no cross-message buffering). One package = one or a few INSERT statements = one ACK. The existing `keepAlive`/`InProgress` heartbeat covers multi-second statements and warehouse resume. - **Within-package dedup:** last event per PK wins, using the shared `dedupLatestPerPK` (`events_dedup_core.go`), because every row in a package shares one `_siphon_replicated_at` and `ROW_NUMBER` ties would be nondeterministic. Observable final state is identical to ClickHouse. - **Order of operations:** snapshot rows → inserts/updates → tombstones. Chunked through the shared `executeWithAdaptiveBatch`. - **TOAST-marked updates:** an unchanged TOAST column carries no value. Inserting NULL would corrupt the current row (the well-known partial-update pitfall with `ROW_NUMBER` dedup). Port the ClickHouse fetch-and-merge approach: `INSERT INTO log SELECT COALESCE(new, latest) ... FROM VALUES ... LEFT JOIN (latest row per PK from the **log**, not the lagging Dynamic Table)`. Until that MR lands the handler must return an explicit error on TOAST markers, never write NULL. Tables with `REPLICA IDENTITY FULL` never hit this path. - **Identifiers:** emitted upper-cased and double-quoted (`"ISSUES"."AUTHOR_ID"`) so analysts can use unquoted names; the metadata provider matches case-insensitively. - **Stream target:** `Stream.Target` accepts `DB.SCHEMA.TABLE`, `SCHEMA.TABLE`, or `TABLE`, defaulting missing parts from `snowflake.database` / `snowflake.schema`. Extend `parseStreamTarget`. ## Table contract (mirrors the ClickHouse rules in README) - Log table has every replicated PG column plus `_SIPHON_REPLICATED_AT TIMESTAMP_NTZ(6) NOT NULL` and `_SIPHON_DELETED BOOLEAN NOT NULL DEFAULT FALSE`. - Log table declares `PRIMARY KEY (<pg pk columns>)` (informational in Snowflake; the consumer reads it via `SHOW PRIMARY KEYS`). No PK → hard error, same as ClickHouse. - Extra target columns are allowed; missing non-PK source columns are allowed (column intersection, `buildColumnList` analogue). Missing PK column → hard error. - Dynamic Table `<TABLE>_CURRENT` is `SELECT * FROM <TABLE> QUALIFY ... = 1` filtered on `NOT _SIPHON_DELETED`; generated, never hand-written. ### PostgreSQL → Snowflake type mapping (from `column_type_oids`) | PG | pb | Snowflake log column | Note | |---|---|---|---| | smallint/integer/bigint | Int16/Int64 | `NUMBER(38,0)` | | | numeric(p,s) | String | `NUMBER(p,s)` | precision from PG catalog in the DDL generator; the consumer binds the string, Snowflake casts. No typmod in the packet, so `evolve` refuses to auto-add numeric columns. | | real/double precision | Double | `FLOAT` | | | boolean | Boolean | `BOOLEAN` | | | text/varchar/citext/inet/cidr/uuid/tsvector/ranges/enums/interval | String | `VARCHAR` | uuid could be `VARCHAR(36)` | | timestamp / timestamptz | Timestamp | `TIMESTAMP_NTZ(6)` in UTC | same choice as ClickHouse `DateTime64(6,'UTC')` | | date | Timestamp | `DATE` | | | time | Timestamp | `TIME(6)` | | | bytea | Bytes | `BINARY` | | | json/jsonb | String | `VARCHAR` in log; `TRY_PARSE_JSON` → `VARIANT` in the Dynamic Table | binding VARCHAR into VARIANT fails in `INSERT ... VALUES`; keep the hot path bind-only | | int/text arrays | *ArrayValue or PG literal string | `VARCHAR` (JSON text) in log; `PARSE_JSON` → `ARRAY` in the Dynamic Table | same reason | ## Schema changes and validation Two distinct concerns, both explicitly designed: **1. Drift handling in the consumer (per package).** Compare `pkg.Columns`/`column_type_oids` with the cached target schema. | Situation | Behaviour | |---|---| | Target has a column the source lacks | allowed, left NULL (as ClickHouse) | | Source has a column the target lacks | `on_schema_drift: warn` (default: drop column, count `siphon_snowflake_consumer_schema_drift_total{kind="missing_in_target"}`, log once per version_hash) · `fail` (NAK, alert) · `evolve` (additive `ALTER TABLE ADD COLUMN` from the OID map, then continue; the Dynamic Table absorbs it) | | Type mismatch on a shared column | warn + metric; never auto-alter | | PK column missing or ignored | hard error, always | Operator runbook stays the same as ClickHouse: apply target DDL **before** the PostgreSQL migration when `on_schema_drift` is not `evolve`. **2. Reconciliation (scheduled, outside the hot path).** A Snowflake schema reconciler, modelled on `pkg/clickhouse_reconciler`'s gocron scheduling, that per configured table: - validates log-table columns/types/PK against the expected schema (from the source PG catalog or the latest packet), reports mismatches as metrics + log; - checks the Dynamic Table exists, its columns match the log, and its refresh is healthy (`SHOW DYNAMIC TABLES` / `DYNAMIC_TABLE_REFRESH_HISTORY`: state, actual lag); - exposes a Snowflake target adapter for the row-level completeness checker in #276 (counts and PK-range checksums against the **log with QUALIFY**, tolerating Dynamic Table lag). The existing `pkg/clickhouse_reconciler` is a DEFAULT-expression staleness scanner and does not transfer; nothing in it is reused beyond the scheduler shape. ## Testing strategy There is no Snowflake container. Follow the PostgreSQL sink precedent: - SQL builders, type mapping, identifier quoting, dedup, presence handling: plain table-driven Go tests in `package consumers` / `package snowflake` asserting emitted SQL and args. No network. - Handler dispatch, ACK/NAK propagation, drift policy, metrics: tests against a fake `Executor`/`MetadataProvider` interface (like `mockFileSystem`). - DDL generator: Ginkgo spec against the suite's live PostgreSQL container. - Live end-to-end: Ginkgo specs under `Label("snowflake")` that **skip when `SIPHON_SNOWFLAKE_*` credentials are absent** (nothing in CI passes a label filter, so an unconditional spec would break `consumer_tests`). A manual/scheduled CI job runs them against a dev account with masked key-pair variables. ## Packaging and release - Same binary, same Dockerfile (`CGO_ENABLED=0`, distroless base already ships CA certs). New dependency: `gosnowflake/v2`. Verify OCSP egress from the cluster (or set `ocsp_fail_open`), and that the `fips_build` job's binary can perform RSA-JWT auth under `GODEBUG=fips140=on`. - `schema_drift_check` requires regenerating `schemas/config.schema.json` in the config MR. - `schema generate-values` (Helm fragment generator) learns the `snowflake` consumer type so table onboarding stays SSOT-driven. - Grafana: reuse the consumer dashboard with the new metric names. ## Dependencies - Snowflake dev/staging **service user, role, warehouse, database/schema** provisioned via `gitlab-data/snowflake-infrastructure` (Terraform), key-pair in the Data Team vault, masked CI variables in this project. - Grants: `USAGE` on warehouse/db/schema, `CREATE TABLE`/`CREATE DYNAMIC TABLE`/`CREATE STAGE` on the schema, `INSERT`/`SELECT` on tables; `ALTER` only when `evolve` is enabled. - #276 for row-level completeness; #298/#299/#300 for correction packages (the Snowflake handler drops them via `skipUnappliedCorrection` until its own task lands). ## Open decisions (call out in review) 1. Dynamic Table vs VIEW default for `_CURRENT`, and default `TARGET_LAG`. 2. `timestamptz` as `TIMESTAMP_NTZ(6)` UTC (ClickHouse parity) vs `TIMESTAMP_TZ`. 3. Whether `evolve` is allowed in production or stays a staging convenience. 4. Log-table retention (time travel / a scheduled compaction that drops superseded versions older than N days) — needed eventually, not for MVP. 5. Warehouse sizing and auto-suspend: per-package INSERTs keep an XS warehouse awake; Phase 2 Snowpipe Streaming is the cost lever. ## Work breakdown (one MR per task, child tasks below) **Wave 1 (independent, start together)** 1. #305 Config + driver dependency + connection/auth (`SnowflakeConfig`, `pkg/snowflake/connection.go`, schema JSON, sample config). 2. #306 Type mapping and identifier quoting (`pkg/snowflake/types.go`, table-driven tests). 3. #307 Design doc `docs/snowflake_consumer.md` + README/AGENTS entries. **Wave 2 (after 1 and 2)** 4. #308 Metadata provider (`INFORMATION_SCHEMA.COLUMNS`, `SHOW PRIMARY KEYS`, TTL cache, `Executor` interface + fake). 5. #309 SQL builders: append INSERT with array binding, tombstone rows, column intersection, within-package dedup; TOAST markers → explicit error. 6. #310 Handler + `cmd/main.go` dispatch + Prometheus metrics + testdata config. **Wave 3** 7. #311 DDL generator: `schema generate-snowflake-ddl` from the PG catalog (log table + Dynamic Table). 8. #312 TOAST alignment (fetch-and-merge INSERT ... SELECT from the log). 9. #313 Schema drift detection and `on_schema_drift` policy. 10. #314 Live integration suite (`Label("snowflake")`, env-gated) + CI job + Snowflake provisioning. 11. #315 Snowflake schema reconciler + Dynamic Table health + #276 adapter. 12. #316 Correction packages for Snowflake (range tombstone + insert, mirrors #300). 13. #317 Packaging verification: image, OCSP, FIPS, Helm values generator, dashboard. **Phase 2 (separate issues once MVP is in staging)**: `write_mode: merge`; Snowpipe Streaming backend; masking-policy hook in the DDL generator; retire or keep the Iceberg path. ## References - Design doc: https://handbook.gitlab.com/handbook/engineering/architecture/design-documents/siphon/ - Pilot epic: https://gitlab.com/gitlab-data/analytics/-/work_items/26384 - Data Team key-pair runbook: https://gitlab.com/gitlab-data/runbooks/-/blob/main/snowflake_api_user/key-pair-authentication.md - Snowflake: [Dynamic Tables modify/schema changes](https://docs.snowflake.com/en/user-guide/dynamic-tables/modify), [Design patterns (CDC dedup)](https://docs.snowflake.com/en/user-guide/dynamic-tables/design-patterns), [Go driver v2](https://pkg.go.dev/github.com/snowflakedb/gosnowflake/v2), [Snowpipe Streaming high-performance](https://docs.snowflake.com/en/user-guide/snowpipe-streaming-high-performance-overview)