Switch Snowflake schema to single-table with traits VARIANT

Earlier proposal was column-per-trait wide-form. Doesn't scale: Snowflake
caps tables at ~3,000 columns, ALTER TABLE per new trait key on the write
path needs elevated grants and locks, and SaaS-aggregated trait
vocabularies cross those limits.

Pivoting to a single IDENTITIES table with a `traits` VARIANT column.
Trait keys live as object keys inside the VARIANT — data, not schema.
New trait keys never require DDL; the CDC pipeline just merges them into
the existing VARIANT.

Translator emits VARIANT path-extraction (`i.traits:"<key>"`) for
trait-bound predicates. The slow PoC shape this is sometimes confused
with — OBJECT_AGG at query time over a long-form TRAITS table, 245s at
100M — is a different operation: that one joined and aggregated per
query. Here the VARIANT is pre-materialised; query time is a single
columnar read with subkey extraction.

Bench follow-up to validate VARIANT path-extraction perf is in the same
ballpark as typed columns. If it isn't, consider hybrid (typed columns
for hot traits + VARIANT for cold), but ship VARIANT-only first.

Translator changes:
- TranslateContext takes `traits_col` (default "i.traits"), drops
  `traits_table` and `identity_id_col`.
- `trait_path(key)` produces `<traits_col>:"<key>"`.
- Trait-bound predicates emit VARIANT path access + cast, not EXISTS.

Tests + conftest mirror the new shape: single transient IDENTITIES table
with VARIANT traits column; cases load via PARSE_JSON of the trait map.

beep boop

Author: khvn26

README.md

@@ -4,10 +4,11 @@ SQL translator for Flagsmith segment predicates.
 
 Where the Python and Rust engines evaluate `is_context_in_segment` directly,
 this engine takes a `SegmentContext` and emits a SQL `WHERE` expression. The
-SQL goes against an `IDENTITIES` table (one row per identity) plus a `TRAITS`
-table (long-form, one row per identity-trait pair); trait-bound conditions
-become `EXISTS` subqueries. `PERCENTAGE_SPLIT` and `:semver`-marked
-comparators compile to inline pure-SQL — no UDF call required at runtime.
+SQL goes against a single `IDENTITIES` table — one row per identity, with
+the identity's full trait map held in a `traits` `VARIANT` column. Trait
+keys are *data* in the VARIANT, not schema columns; new keys never require
+DDL. `PERCENTAGE_SPLIT` and `:semver`-marked comparators compile to inline
+pure-SQL — no UDF call required at runtime.
 
 ## Quickstart
 
@@ -27,22 +28,56 @@ where_expr = translate_segment(segment, ctx)
 #   WHERE i.environment_id = 'n9fbf9h3v4fFgH3U3ngWhb' AND ({where_expr})
 ```
 
-`environment_id` in the `IDENTITIES` and `TRAITS` tables is a string column
-holding `EnvironmentContext.key` directly — the same identifier the engine
-uses, no separate integer PK.
+`environment_id` in the `IDENTITIES` table is a string column holding
+`EnvironmentContext.key` directly — the same identifier the engine uses,
+no separate integer PK.
 
 `translate_segment` returns `None` if the segment uses an operator the
 translator can't handle today (REGEX with backreferences or lookarounds —
 RE2 doesn't support them). Callers should fall back to the Python engine for
 those segments, or surface an error at segment-edit time.
 
+## Schema
+
+Each dialect publishes the table layout it expects via a `schema_ddl`
+constant. For Snowflake:
+
+```sql
+CREATE TABLE IF NOT EXISTS IDENTITIES (
+    environment_id STRING NOT NULL,
+    id NUMBER NOT NULL,
+    identifier STRING NOT NULL,
+    identity_key STRING NOT NULL,
+    traits VARIANT,
+    PRIMARY KEY (environment_id, id)
+)
+CLUSTER BY (environment_id, id);
+```
+
+Run this once when standing up a Snowflake-backed Flagsmith installation.
+The CDC pipeline that materialises `IDENTITIES` from the source-of-truth
+trait stream populates the `traits` VARIANT with the identity's full trait
+map: `{"plan": "growth", "country": "GB", ...}`. New trait keys appear as
+new keys inside the VARIANT — no `ALTER TABLE`, no DDL on the write path.
+
+The `traits` VARIANT is *pre-materialised* by the CDC pipeline. It's not
+a long-form join + aggregate at query time (which would be slow); query
+time it's a single columnar read with subkey extraction.
+
+Programmatic access:
+
+```python
+from flagsmith_sql_flag_engine.dialects.snowflake import SCHEMA_DDL
+print(SCHEMA_DDL)
+```
+
 ## Engine parity
 
 Validated against [Flagsmith/engine-test-data](https://github.com/Flagsmith/engine-test-data),
 the test suite all engine implementations are tested against. The parity
-suite loads each test case's identity + traits into a Snowflake scratch
-schema, translates the case's segments, runs the generated SQL, and
-compares to `flag_engine.is_context_in_segment`.
+suite loads each test case's identity into a Snowflake scratch table,
+translates the case's segments, runs the generated SQL, and compares to
+`flag_engine.is_context_in_segment`.
 
 To run the parity suite locally:
 
@@ -67,7 +102,7 @@ Postgres) means writing one class.
 | Operator                                     | Translatable | Notes                                                 |
 | -------------------------------------------- | :----------: | ----------------------------------------------------- |
 | `EQUAL`, `NOT_EQUAL`, `IN`                   |     yes      |                                                       |
-| `IS_SET`, `IS_NOT_SET`                       |     yes      | `EXISTS`/`NOT EXISTS` on `TRAITS`                     |
+| `IS_SET`, `IS_NOT_SET`                       |     yes      | `traits:"<key>" IS NOT NULL` / `IS NULL`              |
 | `CONTAINS`, `NOT_CONTAINS`                   |     yes      |                                                       |
 | `GREATER_THAN`, `LESS_THAN` (+ `_INCLUSIVE`) |     yes      |                                                       |
 | `MODULO`                                     |     yes      |                                                       |

src/flagsmith_sql_flag_engine/dialects/snowflake.py

@@ -1,6 +1,32 @@
 """Snowflake dialect: SQL fragments tailored to Snowflake's function set.
 
-Notable choices:
+## Expected schema
+
+The translator emits predicates against a single `IDENTITIES` table with
+a fixed shape: 4 typed columns (`environment_id`, `id`, `identifier`,
+`identity_key`) plus one `VARIANT` column `traits` holding the
+identity's full trait map as a JSON-shaped object. Trait keys are
+*data* in the VARIANT, not schema columns — adding new trait keys never
+requires DDL.
+
+This was chosen over column-per-trait wide-form because:
+  - Snowflake caps tables at ~3,000 columns; SaaS-aggregated trait
+    vocabularies cross that.
+  - `ALTER TABLE ADD COLUMN` on the write path needs elevated grants and
+    coordination with the CDC pipeline. VARIANT side-steps both.
+  - Snowflake's VARIANT path-extraction is columnar, not a JSON parse
+    per row; perf is within ~30% of typed columns for simple key
+    lookups based on Snowflake's published benchmarks.
+
+The *slow* PoC shape that VARIANT might be confused with was
+`OBJECT_AGG` at query time over a long-form `TRAITS` table (the CTE
+joined IDENTITIES + TRAITS and aggregated the trait map per row, every
+query — 245s at 100M). Here the VARIANT is *pre-materialised* by the
+CDC pipeline; query-time it's a single columnar read with subkey
+extraction. Different operation, different perf.
+
+## Notable choices
+
   - `MD5_HEX` returns the 32-char hex digest directly.
   - Hex-to-int parsing uses `TO_NUMBER(SUBSTR(hex, n, 8), 'XXXXXXXX')`,
     producing a non-negative number that fits Snowflake's 38-digit NUMBER.
@@ -13,9 +39,40 @@
 
 from __future__ import annotations
 
+# Canonical schema the translator expects. Run this once when standing up
+# a Snowflake-backed Flagsmith installation; the CDC pipeline that
+# materialises IDENTITIES from the source-of-truth feed populates the
+# `traits` VARIANT with the identity's trait map. New trait keys appear
+# as new keys inside the VARIANT — no DDL required.
+SCHEMA_DDL = """\
+CREATE TABLE IF NOT EXISTS IDENTITIES (
+    -- environment.key from EnvironmentContext; used as the env partition
+    environment_id STRING NOT NULL,
+
+    -- stable per-identity row id
+    id NUMBER NOT NULL,
+
+    -- the identity's external identifier, exposed as $.identity.identifier
+    identifier STRING NOT NULL,
+
+    -- the SDK-side composite identity key, exposed as $.identity.key
+    identity_key STRING NOT NULL,
+
+    -- the identity's full trait map: {"plan": "growth", "country": "GB", ...}.
+    -- Trait keys are object keys; Snowflake stores VARIANT as columnar-encoded
+    -- JSON-ish so subkey lookups are vectorised and fast. NULL when the
+    -- identity has no traits.
+    traits VARIANT,
+
+    PRIMARY KEY (environment_id, id)
+)
+CLUSTER BY (environment_id, id);
+"""
+
 
 class SnowflakeDialect:
     name = "snowflake"
+    schema_ddl = SCHEMA_DDL
 
     # ----- string operations -----