Realtime Cohort Calculation (MVP)

MVP

The MVP is complete when we can calculate and query cohort membership for behavioral cohorts such as:

person performed event X at least 5 times in the last 30 days AND
person has a posthog.com email address AND
(person’s first name is Max OR Maxine) AND
person is in Cohort ABC

or:

person has not performed event X in the last 30 days AND
person belongs to Cohort B (a realtime cohort)

Goal:
Feature Flags and Workflows teams can query the cohort_membership table directly for membership status.

Additional Requirements

• Handle person merges (currently via ClickHouse query)

• Handle duplicate events (currently via ClickHouse logic)

• Implement backfilling (no solution yet)

• Migrate existing cohorts that can be behavioral

• Support “has ever performed X” feature

• Support “has performed X for the first time” feature

• Backfilling!

---

Architecture (Flow)

┌───────────────────────────┐
│ Django App                │
│ - Saves Cohorts           │
│ - Generates Bytecode &    │
│   conditionHash           │
└───────────────────────────┘


┌───────────────────────────────────────────────┐
│ Kafka: events_clickhouse_json                 │
│ (Raw events from ingestion pipeline)          │
└────────────┬──────────────────────────────────┘
             │
             ▼
┌───────────────────────────────────────────────┐
│ Event Ingestion Service (Node)                │
│ - Loads cohorts for team                      │
│ - Evaluates events via hogvm                  │
│ - Matches against cohort bytecode             │
│ - Publishes matches to Kafka                  │
└────────────┬──────────────────────────────────┘
             │ matched events
             ▼
┌───────────────────────────────────────────────┐
│ Kafka: pre-calculated-events topic            │
└────────────┬──────────────────────────────────┘
             │
             ▼
┌───────────────────────────────────────────────┐
│ ClickHouse                                   │
│ - Stores pre-calculated-events               │
│ - Queried by Temporal for behavioral checks  │
└───────────────────────────────────────────────┘


              ┌───────────────────────────────────────────────┐
              │ Temporal Cohort Workflow (Periodic Trigger)   │
              │ - Initiates cohort recalculation              │
              │ - Queries:                                    │
              │    • Behavioral (ClickHouse)                  │
              │    • Person                         │
              │    • inCohort/notInCohort           │
              │ - Publishes membership changes                │
              └────────────┬──────────────────────────────────┘
                           │ membership changes
                           ▼
┌───────────────────────────────────────────────┐
│ Kafka: cohort-membership-changed              │
└────────────┬──────────────────────────────────┘
             │
┌────────────┴──────────────────────────────────┐
│ Cohort Membership Consumer                    │
│ - Writes to Postgres (lookup)                 │
│ - Publishes to Kafka (trigger topic)          │
└────────────┬──────────────────────────────────┘
             │
     ┌───────┴────────────────────┬───────────────────────────┐
     │                            │                           │
     ▼                            ▼                           ▼
┌────────────────────────────┐  ┌────────────────────────────┐
│ Postgres: cohort_membership │  │ Kafka: cohort-membership- │
│ - Queried by Flags,         │  │ changed-trigger            │
│   Workflows, APIs           │  │ (for dependent systems)    │
└────────────────────────────┘  └────────────────────────────┘

---

Bytecode Generation for Cohorts (Django)

Events

To determine whether an event should be ingested into the pre-calculated-events table (for later cohort membership calculations), cohorts must generate bytecode that can be evaluated in real time using hogvm.

Purpose

• The bytecode serves as an event matcher, e.g. “has performed event X”.
• The temporal aspect (e.g. “at least 3 times in the last 30 days”) is handled later when querying the pre-calculated-events table.
• Queries use a conditionHash derived from the bytecode.

Handling Negations

• Negated filters (e.g. “has not performed event X”) generate the same bytecode as their non-negated counterpart.
• This ensures events are ingested for absence checks (e.g. “has not performed event X in the last 30 days”).

Cohort Dependencies

• The is(not)InCohort condition is supported only for realtime cohorts.
• A cohort is marked as realtime when all its filters produce valid bytecode that can be evaluated by hogvm.

Persons

The pre-calculated-events table is used only for event matching, enabling queries like “has performed event X in the last 30 days.”

Person Property Conditions

Behavioral cohorts can include person property conditions, such as:

person.property.email contains "@posthog.com"

These person-based conditions are evaluated during cohort membership calculation within temporal workflows that query the person table.

Potential Optimization

If this approach is too slow, we can apply the same pre-calculation pattern used for events.
To prepare for that, we already:

• Generate bytecode for person-property evaluations.
• Compute a conditionHash to allow potential lookup in a future pre-calculated-person-properties table.

Current State

Currently, the person-property bytecode is not actively used — it exists for future optimization.

Missing Work

• Generate bytecode for behavioral filters used by pre-calculated-events-consumer @meikelmosby (PR)
• Create unified conditionHash algorithm for writers and temporal worker @meikelmosby (PR)
• Re-save all cohorts to generate byte_code and cohort_type @meikelmosby
  • Write management command for testing @meikelmosby (PR)
  • migrate team2 to use realtime cohorts @meikelmosby
  • migrate rest of posthog over @meikelmosby
  • Write migration for production/hobby deployments @meikelmosby
• Ensure inCohort is implemented and working properly @meikelmosby

---

Event Ingestion Service (Behavioral Event Consumer in Node)

This service consumes events from the events_clickhouse_json topic and loads all cohorts for a given team to evaluate each event against the cohort’s bytecode.

Event Matching and Publishing

• For each cohort, the service checks if the event matches the cohort’s bytecode condition.
• If a match is found, the service publishes an event to a topic that populates the pre-calculated-events table in ClickHouse.

Key Implementation Details

De-duplication of Filters

All filter conditions across cohorts are iterated and deduplicated to avoid duplicate entries, resulting in a unique set of condition hashes per team.

Event-to-Condition Mapping

Each event is matched against all unique filter conditions, writing one entry to the table per matching condition hash.

Exclusion of Person-Property Filters

Person property filters are filtered out — they are not handled by this service.

Missing Work

• use cohorts instead of actions, handle deduplication & person-property condition filtering @meikelmosby (PR)

---

Cohort Membership Calculation (Temporal)

Cohort membership is calculated using Temporal workflows, which evaluate three main components:

1. Behavioral conditions – e.g. “performed event X at least N times over the last M days”
2. Person property conditions – e.g. “email contains @posthog.com”
3. Cohort membership conditions – e.g. “is in Cohort A” or “is not in Cohort B”

Behavioral Conditions

• Loads all realtime (behavioral) cohorts for a team.
• Extracts filter conditions (JSON) and associated conditionHash.
• Constructs a ClickHouse query to evaluate membership against:
  • pre-calculated-events
  • cohort-membership (ClickHouse)
• Determines whether a person has joined or left a cohort based on event history.

Person Property Conditions

• Person property filters (e.g. email contains "@posthog.com") are currently evaluated via Postgres queries.
• If slow, a pre-calculated table similar to pre-calculated-events can be introduced.

Cohort Membership Conditions

• Handles dependencies like isInCohort and isNotInCohort.
• Evaluates via the cohort-membership table in Postgres.
• Only realtime cohorts are eligible for real-time referencing.

Final Evaluation and Publishing

• Combines results from all conditions to determine final membership.
• Publishes changes (join/leave) to the cohort-membership-changed Kafka topic.

Missing Work

• calculate cohort membership based on cohorts and not actions @gustavohstrassburger (with help of @meikelmosby ) (PR)
  • calculate temporal conditions @gustavohstrassburger (with help of @meikelmosby )
  • calculate person conditions @gustavohstrassburger (with help of @meikelmosby )
  • calculate inCohort conditions @gustavohstrassburger (with help of @meikelmosby )
• adjust logic to publish cohort_membership_changed entry to kafka (make sure we set partition key correctly) @gustavohstrassburger (with help of @meikelmosby ) (PR)
• rename cdp-behavioral-events to cdp-precalculated-filters @gustavohstrassburger (PR)

Improvements to handle Team 2:

• add max_execution_time and retries @gustavohstrassburger (PR and PR)
• build precalculated_person_properties and use it in HogQLRealtimeCohortQuery @gustavohstrassburger (PR)
• optimize query by merging sibling property groups @gustavohstrassburger (PR)
• use person_id from precalculated_events and precalculated_person_properties instead of joining with persons @gustavohstrassburger (PR, PR, PR) (reviews pending to merge)
• "update" precalculated tables when persons are merged @gustavohstrassburger

Cohort Membership Consumer

The consumer reads cohort_membership_changes from Kafka and writes them to Postgres for fast lookups (used by Flags, Workflows, etc.). The service also writes to a kafka topic about the cohort-membership change which can be used to power workflows, etc.

Missing Work

• make postgres fully usable @meikelmosby
  • pgbouncer usage
  • ensure migrations work (build on top of this PR)
• publish cohort-membership-change result to kafka @meikelmosby PR
• deploy service (charts) @meikelmosby

---

Misc

ClickHouse

• Create migrations for cohort_membership table, materialized views, and Kafka tables @Daesgar
• Move behavioral cohort computations to dedicated Kubernetes nodes @Daesgar
• optimise query perf for cohort membership calculation @Daesgar

Productionization

• use cohort_membership to update the field count of the table posthog_cohort
• use cohort_membership to match feature flags when using real-time cohorts in conditions
• use cohort_membership to display persons in the cohort details page
• disable former cohort count for realtime cohorts
• Extend Grafana dashboard with all systems and ClickHouse @meikelmosby @Daesgar
• Implement E2E tests
• Write documentation
• Alerts (!!!!) @meikelmosby

Cohort Model Modifications

• Add backfill property to indicate when recalculation is needed
• Add flag for “behavioral” (dynamic-v2) cohorts used by Flags and Workflows
• Track staleness state for updates