From bc7c82dd464602df3dea98ce627409e912c434c4 Mon Sep 17 00:00:00 2001 From: Kaushik Iska Date: Mon, 19 Oct 2026 21:31:38 +0000 Subject: [PATCH] Filter records before building per-record sink futures ReorderSink and BufferingDecoderSink built and awaited one handle() future per batch record before checking wants(). The futures embed the commit and decode paths' state, so each was a large memmove even for records the sink skips: ReorderSink wants only xact records, about 15% of the lifecycle benchmark's WAL. Checking first cut the memmove share of the serial queue worker from 7.2% to 4.9%. Co-Authored-By: Claude Opus 5.5 --- src/emit/pipeline/reorder.rs | 2 +- src/xact/xact_buffer.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/emit/pipeline/reorder.rs b/src/emit/pipeline/reorder.rs index fd868fca..2c1c46a2 100644 --- a/src/emit/pipeline/reorder.rs +++ b/src/emit/pipeline/reorder.rs @@ -1565,7 +1565,7 @@ impl RecordSink for ReorderSink { _pre: &'a mut [Option], ) -> Pin> + Send + 'a>> { Box::pin(async move { - for record in batch { + for record in batch.iter().filter(|r| Self::wants(r)) { self.handle(record).await?; } self.out.ship().await diff --git a/src/xact/xact_buffer.rs b/src/xact/xact_buffer.rs index 6e5a0a39..db56082d 100644 --- a/src/xact/xact_buffer.rs +++ b/src/xact/xact_buffer.rs @@ -3447,7 +3447,7 @@ impl RecordSink for BufferingDecoderSink { ) -> Pin> + Send + 'a>> { Box::pin(async move { - for (i, record) in batch.iter().enumerate() { + for (i, record) in batch.iter().enumerate().filter(|(_, r)| Self::wants(r)) { let pre = pre.get_mut(i).and_then(Option::take); self.handle(record, pre).await?; }