Projection Partitioning

By default every projection writes into a single PostgreSQL table. For most applications that is fine. When a single projection becomes a write bottleneck — a high-cardinality leaderboard, a per-entity feed, a counter table getting millions of updates per minute — Causet gives you two tools:

Logical sharding with shard() — add a partition_id column computed at write time so queries can filter to a single shard instead of scanning the whole table.
Projection type metadata — declare type:, processing:, checkpointing:, and retry: on a projection so the worker treats it differently from lower-priority ones.

Most Causet applications never need projection partitioning. Start without it. Add it when a specific projection’s write latency or query scan time is measurably problematic.

The four partitioning concepts

Do not conflate these — they operate at different layers:

Concept	What it affects	Changes table layout?
Kafka partitioning	Event ordering and consumer parallelism	No
Logical sharding (`shard()`)	A `partition_id` column inside one table	Additive column only
Native Postgres partitioning	One parent table → child partition tables	Yes — requires rebuild
Database sharding	Rows distributed across separate databases	Yes — requires data migration

Start with logical sharding. It is additive (a new column), does not require a table rewrite, and is reversible.

Logical sharding with `shard()`

The shard(expr, N) function in a projection’s derive: block computes a deterministic integer [0, N) from any expression using Murmur3 hashing. The resulting shard_id column lets queries filter to a single bucket instead of scanning every row.

projections:
  show_attendance:
    source_events: [SHOW_CHECK_IN_CREATED]
    target:
      table: show_attendance
      primary_key: [show_id]
    fields:
      show_id:      TEXT
      shard_id:     INT
      checkin_count: BIGINT
      updated_at:   BIGINT
    derive:
      show_id:   event.show_id
      shard_id:  shard(event.show_id, 32)    # 32 logical shards
      updated_at: event.ts
    aggregates:
      SHOW_CHECK_IN_CREATED:
        checkin_count: { op: add, by: 1, floor: 0 }
    indexes:
      - columns: [shard_id, show_id]

The matching query filters by shard_id:

queries:
  show_attendance_by_show:
    from: show_attendance
    input:
      show_id: { type: string, required: true }
    where:
      show_id: { eq: input.show_id }
    limit: 1

⚠️

The shard_id value for a given key is immutable after deploy. The shard count (N) is baked into the hash. Changing N from 32 to 64 means every row gets a different shard_id — you must rebuild the projection (see Rebuilds) before queries using the old count will be correct.

Choosing a shard count

Pick a power of two. The right count depends on your write volume:

Writes/sec on this projection	Suggested shard count
< 1,000/s	None (no sharding needed)
1,000–10,000/s	8–16
10,000–100,000/s	32–64
> 100,000/s	64–256

Choose generously — you cannot change the count without a rebuild.

Hot aggregate pattern (counter sharding)

Row-lock contention is the real bottleneck for high-write counters. When many events update the same row (e.g., a global checkin_count for a popular show), Postgres serializes those writes.

The solution: shard the counter rows and sum at read time.

projections:
  show_attendance_sharded:
    source_events: [SHOW_CHECK_IN_CREATED]
    target:
      table: show_attendance_sharded
      primary_key: [show_id, shard_id]       # one row per (show, shard)
    fields:
      show_id:       TEXT
      shard_id:      INT
      checkin_count: BIGINT
    derive:
      show_id:  event.show_id
      shard_id: shard(event.show_id, 16)     # 16 independent counter rows per show
    aggregates:
      SHOW_CHECK_IN_CREATED:
        checkin_count: { op: add, by: 1, floor: 0 }
    indexes:
      - columns: [show_id]

The query SUMs across all shards for a given show_id:

queries:
  show_total_checkins:
    from: show_attendance_sharded
    input:
      show_id: { type: string, required: true }
    where:
      show_id: { eq: input.show_id }
    group_by:
      - show_id
    aggregate:
      total_checkins:
        sum: checkin_count
    limit: 1

This eliminates the single hot-row lock. Each write goes to one of 16 shard rows; the read sums them. The write throughput scales linearly with shard count.

Projection type metadata

The type: field on a projection tells the worker how to treat it — processing mode, batch size, retry behavior, and priority. All fields are optional; absent fields use defaults that reproduce today’s behavior.

projections:
  show_attendance:
    type: hot                    # hot | read_model | analytics | search
    source_events: [SHOW_CHECK_IN_CREATED]
    # ... target, fields, derive, aggregates ...
 
    processing:
      mode: batch                # realtime | batch
      batch_size: 100
      flush_interval_ms: 500
      max_concurrency: 8
 
    checkpointing:
      enabled: true
      strategy: per_projection_partition
 
    retry:
      max_attempts: 10
      backoff: exponential       # none | fixed | exponential
      dead_letter: true
 
    priority: p0                 # p0 | p1 | p2

`type` — projection tier

Type	Processing	Batch size	Retry	Priority	Use case
`hot`	batch	100	dead_letter	p0	High-write counters, feeds, leaderboards
`read_model`	realtime	1	—	p1	Standard read models (default)
`analytics`	scheduled (5m)	—	—	p2	Aggregates not needed in real time
`search`	batch	250	—	p2	Search index tables
(absent)	realtime	1	no DLQ	p1	Current default behavior

Setting type: hot automatically applies the hot defaults in the table above — you only need to override individual fields if you want different values.

`processing`

Field	Description
`mode: realtime`	Apply each event immediately as it arrives (default)
`mode: batch`	Buffer events and flush as a single multi-row UPSERT
`batch_size`	Max events to buffer before flushing (default: 100 for `hot`)
`flush_interval_ms`	Max time to hold a batch before flushing (default: 500ms)
`max_concurrency`	Max parallel write threads for this projection

Batching reduces write amplification on hot projections — 100 events become one UPSERT instead of 100 sequential round-trips.

`checkpointing`

Field	Description
`enabled: true`	Track the last processed offset per projection (default: true)
`strategy: per_projection_partition`	Checkpoint per `(projection_name, partition_id)` — accurate for sharded projections
`strategy: per_tenant_partition`	Legacy default — checkpoint per `(tenant, kafka_partition)` only

`retry`

Field	Description
`max_attempts`	How many times to retry a failing event (default: 3)
`backoff`	`none`, `fixed`, or `exponential`
`dead_letter: true`	Route poison events to the DLQ instead of blocking the partition

🚫

Without dead_letter: true, a poison event can block an entire Kafka partition.

Today’s default behavior swallows projection errors and acks the offset anyway — meaning a failing event causes silent data loss. Setting dead_letter: true routes unrecoverable events to the dead letter queue and advances the offset so other events are not blocked.

`priority`

Value	Meaning
`p0`	Critical — never deprioritized under backpressure
`p1`	Normal (default)
`p2`	Low — first to be paused when the consumer is under load

Under heavy write load, the worker pauses p2 projections first to protect p0 and p1 throughput.

`source` — custom topic and consumer group

By default all projections share the causet.projection-events.v1 Kafka topic and the causet-projection-worker consumer group.

For a critical projection that must be isolated from noisy neighbors, declare a dedicated consumer group:

projections:
  show_attendance:
    type: hot
    source:
      consumer_group: projection.show-attendance.v1    # isolated group
    # ...

⚠️

Each consumer group means an additional lag dashboard to monitor. Only split groups for projections that need true isolation. The operational cost is real.

`stores` — logical store binding

The stores: top-level block declares named logical stores. A projection can then declare store: hot_db to be written to that store instead of the default projections database.

stores:
  hot_projection_db:
    type: postgres
    binding: ${HOT_PROJECTION_DB_URL}
    partitioning:
      strategy: hash
      key: show_id
      partitions: 32
      mode: logical           # logical | postgres_native
 
projections:
  show_attendance:
    type: hot
    store: hot_projection_db
    source_events: [SHOW_CHECK_IN_CREATED]
    # ...

mode controls the partitioning implementation:

Mode	What it does	Table rewrite?
`logical`	Adds a `partition_id` column; one physical table	No
`postgres_native`	Postgres declarative `PARTITION BY`; child tables	Yes — requires rebuild

⚠️

Multi-store support (routing projections to different physical databases) is a future capability, not yet implemented. stores is not in causet-compiler’s set of recognized top-level DSL keys today — including it in an app will fail compilation with an “Unknown top-level key” error, not silently no-op. Don’t add stores: or store: to a real app until this ships. All projections currently share the same database.

Native Postgres partitioning

For the highest-volume tables, native Postgres partitioning (PARTITION BY HASH or PARTITION BY RANGE) can improve query performance by pruning child tables automatically.

projections:
  show_attendance:
    type: hot
    store: hot_projection_db        # store must declare postgres_native mode
    source_events: [SHOW_CHECK_IN_CREATED]
    target:
      table: show_attendance
      primary_key: [show_id]
    # ...

The store’s partitioning config drives DDL generation:

stores:
  hot_projection_db:
    type: postgres
    partitioning:
      strategy: hash
      key: show_id
      partitions: 32
      mode: postgres_native

⚠️

Native partitioning requires a full projection rebuild. It rewrites the physical table using the shadow-table-swap mechanism (_v{n}). Do not use it unless you have measured that query pruning is necessary and you have a rebuild window.

See Rebuilds for the rebuild process.

Migration path: adding sharding to an existing projection

If your projection is already deployed and you want to add shard_id, do this:

Add the shard_id derive and index to your projection DSL and deploy.
Rebuild the projection — replay the source events into a shadow table _v{n+1} that now includes the shard_id column (see Rebuilds for the mechanism; there is no dedicated causet subcommand for this yet, so it runs as an internal replay against the shadow table).
Swap once the rebuild completes.
Update queries to add where: { shard_id: { eq: input.shard_id } } if you want partition-filtered reads.

Until step 4, queries continue to do full-table scans — correct but unoptimized. The shard_id column is available immediately after the swap.

Observability

The worker exposes metrics for monitoring projection health:

Metric	Description
`projection_lag_ms{projection, partition}`	Milliseconds behind the live event stream
`projection_batch_size`	Average events per batch flush
`projection_flush_latency_ms`	Time to write a batch to the DB
`projection_db_write_latency_ms{projection}`	Per-projection DB write time
`projection_dlq_total{projection}`	Events sent to the dead letter queue
`projection_retry_total{projection}`	Retried events
`projection_partition_write_total{projection, partition}`	Writes per shard — alert on skew

A hot partition alert (partition_write_total for one shard >> others) means your partition key is skewed. Consider a higher-cardinality key.

Monitor projection lag and query latency from the Causet Cloud control plane metrics for your application.

Complete example: sharded show leaderboard

A concert app tracks check-ins in real time. At peak (festival weekend), a single popular show can receive thousands of check-ins per second. Without sharding, every write contends on the same show_id counter row. With the hot aggregate pattern, 16 shard rows absorb the writes in parallel and the query SUMs them.

State and events

# states/show.state.causet
state:
  show:
    entity_key: show_id
    fields:
      - name: title
        type: string
        default: ""
      - name: venue_id
        type: string
        default: ""
 
# states/user.state.causet
state:
  user:
    entity_key: user_id
    fields:
      - name: checkins
        type: array
        item_type: string
        default: []

# events/checkin.events.causet
events:
  SHOW_CHECK_IN_CREATED:
    state: user
    entity_expr: event.user_id
    payload:
      user_id:    string
      show_id:    string
      venue_id:   string
      checked_in_at: integer

Action

# actions/checkin.actions.causet
actions:
  CHECK_IN_TO_SHOW:
    state: user
    entity_id_expr: intent.user_id
    input:
      user_id:  { type: string, required: true }
      show_id:  { type: string, required: true }
      venue_id: { type: string, required: true }
    preflight:
      rules:
        - name: reject_duplicate
          when:
            expr: "contains(entity.checkins, event.show_id)"
          then:
            - op: reject
              code: ALREADY_CHECKED_IN
    core:
      rules:
        - name: record_checkin
          when: {}
          then:
            - op: push
              path: /checkins
              value: event.show_id
            - op: emit
              event_type: SHOW_CHECK_IN_CREATED
              payload:
                user_id:       event.user_id
                show_id:       event.show_id
                venue_id:      event.venue_id
                checked_in_at: event.ts

Projection — sharded counter

Each show gets 16 counter rows (one per shard). Writes are distributed across rows; queries sum them.

# projections/show_checkin_counts.projections.causet
projections:
  show_checkin_counts:
    type: hot
    source_events: [SHOW_CHECK_IN_CREATED]
    target:
      table: show_checkin_counts
      primary_key: [show_id, shard_id]      # one row per (show, shard)
    fields:
      show_id:        TEXT
      venue_id:       TEXT
      shard_id:       INT
      checkin_count:  BIGINT
      last_checkin_at: BIGINT
    derive:
      show_id:         event.show_id
      venue_id:        event.venue_id
      shard_id:        shard(event.show_id, 16)
      last_checkin_at: event.ts
    aggregates:
      SHOW_CHECK_IN_CREATED:
        checkin_count: { op: add, by: 1, floor: 0 }
    indexes:
      - columns: [show_id]
      - columns: [venue_id, last_checkin_at]
        direction: desc
    processing:
      mode: batch
      batch_size: 100
      flush_interval_ms: 250
    retry:
      max_attempts: 10
      backoff: exponential
      dead_letter: true
    priority: p0

Queries

Get total check-in count for a show (SUMs across all 16 shards):

# queries/show_checkin_counts.queries.causet
queries:
  show_checkin_count:
    from: show_checkin_counts
    input:
      show_id: { type: string, required: true }
    where:
      show_id: { eq: input.show_id }
    group_by:
      - show_id
    aggregate:
      total_checkins:
        sum: checkin_count
    limit: 1
 
  venue_leaderboard:
    from: show_checkin_counts
    input:
      venue_id: { type: string, required: true }
    where:
      venue_id: { eq: input.venue_id }
    group_by:
      - show_id
      - venue_id
    aggregate:
      total_checkins:
        sum: checkin_count
    order_by:
      total_checkins: desc
    limit: 20

What this gives you

Event arrives: CHECK_IN_TO_SHOW { user_id: "u1", show_id: "s-99" }
                        │
                        ▼
          shard("s-99", 16) → shard_id = 7
                        │
                        ▼
          UPSERT show_checkin_counts
            WHERE show_id = "s-99" AND shard_id = 7
            SET checkin_count = checkin_count + 1
                        │
            (15 other shard rows for "s-99" are untouched)


Query: show_checkin_count?show_id=s-99
                        │
                        ▼
          SELECT show_id, SUM(checkin_count) AS total_checkins
          FROM show_checkin_counts
          WHERE show_id = 's-99'
          GROUP BY show_id
          → { total_checkins: 47832 }

At 10,000 check-ins/second for a single show, writes are spread across 16 rows instead of one. Each row receives ~625 writes/second — well within Postgres single-row throughput. The type: hot + batch config groups writes into 100-event UPSERT batches, further reducing round-trips.

Defining Projections Schema Versioning