feat(property-vals): add two-stage aggregation to collapse pod amplification

[andyzzhao]

Problem

property-vals-rs aggregates (team, type, key, value) tuples in each pod's local HashMap and flushes to clickhouse_property_values every 30s. The input topic team_event_partitioned_events_json keys by team_id-event-random(0..6), which spreads a single team's events across most of the 128 partitions. With N pods, the same low-cardinality tuple like (team=2, $browser, Chrome) lives in N different pods' HashMaps, so each pod emits a record for it at flush. ClickHouse ends up consuming N records for what should be one logical update per window.

When I collasped the number of replicas in prod from 12 to 1, the records-per-event ratio from 0.17 to 0.092, a ~46% reduction. That portion of clickhouse_property_values traffic is entirely cross-pod duplication, not unique data.

Changes

Add a second aggregation pass so duplicates collapse before they reach clickhouse_property_values.

flowchart TB
      E[team_event_partitioned_events_json<br/>128 partitions, keyed team-event-random]

      subgraph POD1["pod 1"]
        S1_1["events/groups worker<br/>fan-out + local HashMap"]
        S2_1["merger<br/>merge cross-pod copies"]
      end
      subgraph POD2["pod 2"]
        S1_2["events/groups worker"]
        S2_2["merger"]
      end
      subgraph PODN["pod ...N"]
        S1_N["events/groups worker"]
        S2_N["merger"]
      end

      INT["property_vals_intermediate<br/>NEW, 64 partitions<br/>keyed hash team:type:key:value"]
      OUT["clickhouse_property_values<br/>1 record per tuple per window"]
      CH[(ClickHouse<br/>kafka_property_values MV<br/>AggregatingMergeTree sum)]

      E --> S1_1 & S1_2 & S1_N
      S1_1 & S1_2 & S1_N -->|partial aggregates<br/>routed by tuple hash| INT
      INT -->|each partition owned<br/>by one merger pod| S2_1 & S2_2 & S2_N
      S2_1 & S2_2 & S2_N -->|collapsed| OUT
      OUT --> CH

      style INT fill:#90EE90,stroke:#006400,color:#000
      style S1_1 fill:#FFE4B5,color:#000
      style S1_2 fill:#FFE4B5,color:#000
      style S1_N fill:#FFE4B5,color:#000
      style S2_1 fill:#FFB6C1,color:#000
      style S2_2 fill:#FFB6C1,color:#000
      style S2_N fill:#FFB6C1,color:#000

Loading

The events and groups workers are existing. Their only behavior change is the destination topic and the partition key. The merger is a new consumer group in the same binary; each pod owns a partition slice and re-aggregates the per-pod copies the events/groups workers produced for the same tuple before forwarding to the final output topic.

Implementation notes:

• fan_out/fan_out_group now return Vec<(TupleKey, u64)>. The events/groups workers still emit count=1 per tuple; the wider signature lets the same worker_loop handle the merger, where the count comes from the input message.
• New PropertyValueMessage type for the merger's input. A round-trip test pins the schema to the producer's wire format so future drift fails loudly instead of silently dropping records.
• Producer partition key changes from team:key to team:type:key:value so the same tuple always lands on the same intermediate partition.
• Defaults: INTERMEDIATE_TOPIC=property_vals_intermediate, MERGER_CONSUMER_GROUP=clickhouse-property-vals-rs-merger.

How did you test this code?

I'm an agent and have not run the binary or the prod stack manually.

• All 42 unit + property tests in property-vals-rs pass locally: cargo test -p property-vals-rs
• cargo clippy -p property-vals-rs --no-deps --all-targets is clean
• cargo fmt -p property-vals-rs is a no-op
• A new wire-format round-trip test pins PropertyValueMessage deserialization against the producer's Outgoing serialization to catch silent schema drift in either direction.

Publish to changelog?

no

🤖 Agent context

Co-authored with Claude. Earlier in the same session we measured the 46%/54% split between pod-amplification cost and unique-singleton cost by scaling property-vals-rs from 12 pods to 1 and comparing the flush_tuples_sum / events_received ratio. That experiment is what motivated investing in the two-pass architecture as opposed to alternatives like an MV-side GROUP BY (which only helps if the CH-side bottleneck is INSERT/merge, not consume — and the data showed DelayedInserts=0, RejectedInserts=0, idle merge pool, while KafkaRowsRead was plateaued at ~62k/replica during the lag spike).

Considered and rejected:

• MV-level GROUP BY on kafka_property_values: only helps stages 4–5 of the CH pipeline; the bottleneck data showed stages 1–2 (consume + parse) were where the ceiling lived.
• Upstream re-partition (key the input topic by team_id alone): would create hot partitions for team 2.
• In-process consistent-hash routing via gRPC: lower latency but rebuilds Kafka's partition-assignment/rebalance/durability machinery in user code. Not worth the complexity given Kafka topics are cheap.

A singleton emission gate (drop tuples with count=1 at flush time) is a complementary follow-up — it addresses the other ~54% of the cost, which is unique high-cardinality identifiers that don't aggregate.

[greptile-apps]

Prompt To Fix All With AI

Fix the following 2 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 2
rust/property-vals-rs/src/types.rs:63-71
The canonical string representations for `PropertyType` ("event", "person", "group_N") are expressed twice: once in `as_kafka_key_segment` and once in the `Serialize` impl. The two impls must stay in sync, but nothing enforces that. `Serialize` can delegate to `as_kafka_key_segment` to make it OnceAndOnlyOnce.

```suggestion
impl Serialize for PropertyType {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.serialize_str(&self.as_kafka_key_segment())
    }
}
```

### Issue 2 of 2
rust/property-vals-rs/src/fan_out.rs:421-428
**Round-trip test uses a local copy of `Outgoing`, not the real one**

The test defines its own `Outgoing` struct identical to `producer.rs::Outgoing`. If a field is renamed in the actual producer struct (e.g., `property_count` → `count`), the real producer starts emitting different JSON while this test's local copy still serializes the old field name — the test stays green but the merger silently drops the count from every message.

Making `Outgoing` `pub(crate)` in `producer.rs` and importing it here would let the test reference the real struct, so any drift between the two sides fails the build rather than the test.

_{Reviews (1): Last reviewed commit: "chore(property-vals): update last stage-..." | Re-trigger Greptile}

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-05-28 18:55 UTC	Run
prod-us	✅ Deployed	2026-05-28 19:10 UTC	Run
prod-eu	✅ Deployed	2026-05-28 19:15 UTC	Run