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<Predecoded>],
     ) -> Pin<Box<dyn Future<Output = Result<(), SinkError>> + 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<Box<dyn std::future::Future<Output = std::result::Result<(), SinkError>> + 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?;
             }