Replace RowBinary serializer with clickhouse-crate inserter sink [EAP-584]

[phacops]

Overview

This PR replaces the custom RowBinary serialization pipeline for eap_items with a new EapItemsInserterSink that uses the official clickhouse crate's Inserter API. This eliminates the need for custom serialization code and improves memory efficiency by streaming rows directly to ClickHouse.

Key Changes

New Inserter Sink (inserter_sink.rs)

• Implements a long-lived actor task that owns a clickhouse::Inserter<EAPItemRow> instance
• Rows are serialized and written to the inserter's byte buffer immediately upon arrival, then dropped
• Peak memory is bounded by the inserter's buffer size, not by row count
• Flush boundaries are managed by the inserter's configured batch settings (max_rows, max_bytes, max_period)
• Only successfully flushed rows have their Kafka offsets committed downstream
• Failed flushes are never pushed, ensuring batch replay on restart (same durability as the old writer)

Processor Changes (eap_items.rs)

• Renamed process_message_row_binary to process_message_eap_row for clarity
• Returns TypedInsertBatch<EAPItemRow> instead of pre-encoded bytes
• Removed inline RowBinary encoding; the inserter sink handles serialization
• Includes source proto length as byte-size estimate for batch sizing

Type System (types.rs)

• Added TypedInsertBatch<R> to represent rows that will be serialized downstream
• Carries optional row, timestamps, metrics, and byte estimates
• Supports skip semantics (row == None) for messages that don't produce inserts

Pipeline Wiring (factory_v2.rs, processor.rs)

• Added make_rust_processor_typed to build pipelines for typed-row processors
• When use_row_binary is enabled, eap_items now uses the inserter sink path instead of the Reduce + ClickhouseWriterStep pair
• Assertion ensures RowBinary mode is only used with EAPItemsProcessor

Removed Code

• Deleted custom rowbinary/ser.rs serializer module (replaced by clickhouse crate)
• Removed InsertFormat enum and related writer step configuration
• Removed custom UUID serialization adapter (using clickhouse::serde::uuid instead)

Dependencies

• Added clickhouse crate (v0.15) with default features disabled
• Updated Rust toolchain from 1.88.0 to 1.94.1

Benefits

1. Reduced Memory Footprint: Rows are serialized and dropped immediately rather than accumulating in a batch
2. Official Support: Uses the upstream clickhouse crate's battle-tested inserter instead of custom serialization
3. Simplified Code: Eliminates ~400 lines of custom RowBinary serialization logic
4. Better Durability: Explicit flush tracking ensures offsets are only committed after successful writes
5. Schema Validation: The clickhouse crate validates rows against the table schema at serialization time

Testing

Existing unit tests for the RowBinary serializer are removed as that functionality is now provided by the upstream crate. The inserter sink's behavior is covered by integration with the processor pipeline and the existing Arroyo strategy test infrastructure.

https://claude.ai/code/session_011Bz3aya6Udg2sd9Sc9uuHf

[linear-code]

EAP-584

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit ab7d324. Configure here.}