diff --git a/rust/lance-index/src/scalar/inverted/index.rs b/rust/lance-index/src/scalar/inverted/index.rs
index d08dacd26e7..95eb3d815d4 100644
--- a/rust/lance-index/src/scalar/inverted/index.rs
+++ b/rust/lance-index/src/scalar/inverted/index.rs
@@ -4318,6 +4318,9 @@ pub async fn flat_bm25_search_stream_with_metrics(
     elapsed_compute: Option<Time>,
 ) -> DataFusionResult<SendableRecordBatchStream> {
     let mut tokenizer = tokenizer;
+
+    // Pre-await synchronous work: query tokenization + chunk-stream setup.
+    let pre_await_start = std::time::Instant::now();
     let query_tokens = Arc::new(collect_query_tokens(&query, &mut tokenizer));
 
     let input_schema = input.schema();
@@ -4336,6 +4339,9 @@ pub async fn flat_bm25_search_stream_with_metrics(
         ACCUMULATE_BYTES,
         SLICE_BYTES,
     );
+    if let Some(t) = &elapsed_compute {
+        t.add_duration(pre_await_start.elapsed());
+    }
 
     // Phase 2 - For each row we need to know the total number of tokens and the count of each
     // of the query tokens.  For example, if the query is "book" and the row is "the book shop"
@@ -4351,13 +4357,10 @@ pub async fn flat_bm25_search_stream_with_metrics(
     .await?;
 
     // Phase 3 - Calculate final scores (this is fairly cheap, probably don't need to parallelize).
-    // Synchronous and single-threaded, so we time it from this thread.
-    let scoring_start = std::time::Instant::now();
+    // All post-await work is synchronous; time the scorer + score + slicing loop together.
+    let post_await_start = std::time::Instant::now();
     let scorer = initialize_scorer(base_scorer.as_ref(), query_tokens.as_ref(), &counted_input);
     let scores = flat_bm25_score(query_tokens.as_ref(), &counted_input, &scorer)?;
-    if let Some(t) = &elapsed_compute {
-        t.add_duration(scoring_start.elapsed());
-    }
 
     // Finally we emit batches according to the target batch size
     let num_out_batches = scores.num_rows().div_ceil(target_batch_size);
@@ -4367,6 +4370,9 @@ pub async fn flat_bm25_search_stream_with_metrics(
         let len = (scores.num_rows() - start).min(target_batch_size);
         batches.push(Ok(scores.slice(start, len)));
     }
+    if let Some(t) = &elapsed_compute {
+        t.add_duration(post_await_start.elapsed());
+    }
     Ok(Box::pin(RecordBatchStreamAdapter::new(
         FTS_SCHEMA.clone(),
         stream::iter(batches),
diff --git a/rust/lance/src/io/exec/fts.rs b/rust/lance/src/io/exec/fts.rs
index 093ca6cb585..ddc9d5dc3ac 100644
--- a/rust/lance/src/io/exec/fts.rs
+++ b/rust/lance/src/io/exec/fts.rs
@@ -34,7 +34,7 @@ use lance_datafusion::utils::{ExecutionPlanMetricsSetExt, MetricsExt, PARTITIONS
 use lance_table::format::IndexMetadata;
 
 use super::PreFilterSource;
-use super::utils::{IndexMetrics, InstrumentedChildInputStream, build_prefilter};
+use super::utils::{IndexMetrics, build_prefilter};
 use crate::index::scalar::inverted::{load_segment_details, load_segments};
 use crate::{Dataset, index::DatasetIndexInternalExt};
 use lance_index::metrics::MetricsCollector;
@@ -695,11 +695,6 @@ impl FlatMatchFilterExec {
         metrics_set: ExecutionPlanMetricsSet,
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let metrics = Arc::new(FtsIndexMetrics::new(&metrics_set, partition));
-        let elapsed_compute = metrics.baseline_metrics.elapsed_compute().clone();
-        // Time the one-shot setup (tokenizer load + query tokenization) so it's
-        // attributed to this node's elapsed_compute. The helper itself only
-        // times per-batch work.
-        let setup_start = std::time::Instant::now();
         let column = query
             .column
             .clone()
@@ -720,44 +715,45 @@ impl FlatMatchFilterExec {
             None => Self::load_tokenizer(&dataset, &column, &metrics.index_metrics).await?,
         };
         let query_tokens = Arc::new(collect_query_tokens(&query.terms, &mut tokenizer));
-        elapsed_compute.add_duration(setup_start.elapsed());
 
-        let helper = InstrumentedChildInputStream::new(
-            input,
-            schema,
-            move |batch| {
-                // Clone per-batch so the work runs *inside* the async block
-                // (i.e., during the helper's timed in_flight poll, not during
-                // its untimed input-pulling phase).
-                let column = column.clone();
-                let query_tokens = query_tokens.clone();
-                let mut tokenizer = tokenizer.box_clone();
-                async move {
-                    let text_column = batch.column_by_name(&column).ok_or_else(|| {
-                        DataFusionError::Execution(format!("Column {} not found in batch", column,))
-                    })?;
-                    let predicate = match text_column.data_type() {
-                        DataType::Utf8 => {
-                            Self::find_matches::<i32>(text_column, &mut tokenizer, &query_tokens)
-                        }
-                        DataType::LargeUtf8 => {
-                            Self::find_matches::<i64>(text_column, &mut tokenizer, &query_tokens)
-                        }
-                        _ => {
-                            return Err(DataFusionError::Execution(format!(
-                                "Column {} is not a string",
-                                column,
-                            )));
-                        }
-                    };
-                    Ok(arrow::compute::filter_record_batch(&batch, &predicate)?)
-                }
-            },
-            1,
-            partition,
-            &metrics_set,
-        );
-        Ok(Box::pin(helper))
+        let baseline = BaselineMetrics::new(&metrics_set, partition);
+        let elapsed_compute = baseline.elapsed_compute().clone();
+        let stream = input.then(move |batch_result| {
+            let column = column.clone();
+            let query_tokens = query_tokens.clone();
+            let mut tokenizer = tokenizer.box_clone();
+            let elapsed_compute = elapsed_compute.clone();
+            async move {
+                let batch = batch_result?;
+                let _t = elapsed_compute.timer();
+                let text_column = batch.column_by_name(&column).ok_or_else(|| {
+                    DataFusionError::Execution(format!("Column {} not found in batch", column,))
+                })?;
+                let predicate = match text_column.data_type() {
+                    DataType::Utf8 => {
+                        Self::find_matches::<i32>(text_column, &mut tokenizer, &query_tokens)
+                    }
+                    DataType::LargeUtf8 => {
+                        Self::find_matches::<i64>(text_column, &mut tokenizer, &query_tokens)
+                    }
+                    _ => {
+                        return Err(DataFusionError::Execution(format!(
+                            "Column {} is not a string",
+                            column,
+                        )));
+                    }
+                };
+                Ok(arrow::compute::filter_record_batch(&batch, &predicate)?)
+            }
+        });
+        let stream = stream.map(move |batch| {
+            let poll = baseline.record_poll(std::task::Poll::Ready(Some(batch)));
+            match poll {
+                std::task::Poll::Ready(Some(b)) => b,
+                _ => unreachable!("record_poll preserves Ready(Some) input"),
+            }
+        });
+        Ok(Box::pin(RecordBatchStreamAdapter::new(schema, stream)))
     }
 }
 
@@ -1010,11 +1006,8 @@ impl ExecutionPlan for FlatMatchQueryExec {
         // so it can attribute the spawn_cpu tokenize work and synchronous
         // scoring back onto this node's `elapsed_compute`. Sharing the same
         // `Time` handle that's already inside the FtsIndexMetrics avoids
-        // registering a duplicate metric. Cloned once for use during setup
-        // timing (below) and again moved into the async block for the
-        // streaming-phase call.
+        // registering a duplicate metric.
         let elapsed_compute = metrics.baseline_metrics.elapsed_compute().clone();
-        let elapsed_compute_for_stream = elapsed_compute.clone();
 
         let column = query.column.ok_or(DataFusionError::Execution(format!(
             "column not set for MatchQuery {}",
@@ -1024,9 +1017,6 @@ impl ExecutionPlan for FlatMatchQueryExec {
             document_input(self.unindexed_input.execute(partition, context)?, &column)?;
 
         let stream = stream::once(async move {
-            // Time the one-shot setup (load segments / open indices / build
-            // scorer / acquire tokenizer) and attribute it to elapsed_compute.
-            let setup_start = std::time::Instant::now();
             let segments = match preset_segments {
                 Some(segments) => Some(segments),
                 None => load_segments(&ds, &column).await?,
@@ -1061,7 +1051,6 @@ impl ExecutionPlan for FlatMatchQueryExec {
                     preset_base_scorer.map(|s| (*s).clone()),
                 ),
             };
-            elapsed_compute.add_duration(setup_start.elapsed());
 
             flat_bm25_search_stream_with_metrics(
                 unindexed_input,
@@ -1070,7 +1059,7 @@ impl ExecutionPlan for FlatMatchQueryExec {
                 tokenizer,
                 base_scorer,
                 target_batch_size,
-                Some(elapsed_compute_for_stream),
+                Some(elapsed_compute),
             )
             .await
         })
diff --git a/rust/lance/src/io/exec/knn.rs b/rust/lance/src/io/exec/knn.rs
index 49ee7be86bc..4e29f5041fb 100644
--- a/rust/lance/src/io/exec/knn.rs
+++ b/rust/lance/src/io/exec/knn.rs
@@ -65,8 +65,7 @@ use crate::{Error, Result};
 use lance_arrow::*;
 
 use super::utils::{
-    FilteredRowIdsToPrefilter, IndexMetrics, InstrumentedChildInputStream, PreFilterSource,
-    SelectionVectorToPrefilter,
+    FilteredRowIdsToPrefilter, IndexMetrics, PreFilterSource, SelectionVectorToPrefilter,
 };
 
 pub struct AnnPartitionMetrics {
@@ -253,43 +252,35 @@ impl ExecutionPlan for KNNVectorDistanceExec {
         context: Arc<datafusion::execution::context::TaskContext>,
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let input_stream = self.input.execute(partition, context)?;
-        let input_schema = input_stream.schema();
         let key = self.query.clone();
         let column = self.column.clone();
         let dt = self.distance_type;
         let schema = self.schema();
 
         // Empty batches don't have a vector column to score; filter them out
-        // before reaching the helper so the transform always sees real work.
-        let filtered_input = Box::pin(RecordBatchStreamAdapter::new(
-            input_schema,
-            input_stream.try_filter(|batch| future::ready(batch.num_rows() > 0)),
-        )) as SendableRecordBatchStream;
-
-        // Mirror of the helper's elapsed_compute counter; used to attribute
-        // wall-clock from the spawn_blocking distance kernel back onto the
-        // node's `elapsed_compute` metric.
-        let elapsed_compute = BaselineMetrics::new(&self.metrics, partition)
-            .elapsed_compute()
-            .clone();
-
-        let stream = InstrumentedChildInputStream::new(
-            filtered_input,
-            schema,
-            move |batch| {
+        // before the transform so it always sees real work.
+        let filtered_input = input_stream.try_filter(|batch| future::ready(batch.num_rows() > 0));
+
+        let baseline = BaselineMetrics::new(&self.metrics, partition);
+        let elapsed_compute = baseline.elapsed_compute().clone();
+
+        let stream = filtered_input
+            .map(move |batch_result| {
                 let key = key.clone();
                 let column = column.clone();
                 let elapsed_compute = elapsed_compute.clone();
                 async move {
+                    let batch = batch_result?;
                     // Time around the .await to capture the spawn_blocking
                     // distance work, which otherwise runs while this future is
-                    // Pending and is missed by the helper's own poll timer.
-                    let start = std::time::Instant::now();
+                    // Pending and is missed by a poll-time timer.
+                    let start = Instant::now();
                     let batch = compute_distance(key, dt, &column, batch)
                         .await
                         .map_err(|e| DataFusionError::External(Box::new(e)))?;
                     elapsed_compute.add_duration(start.elapsed());
 
+                    let _t = elapsed_compute.timer();
                     let distances = batch[DIST_COL].as_primitive::<Float32Type>();
                     let mask = BooleanArray::from_iter(
                         distances
@@ -299,12 +290,17 @@ impl ExecutionPlan for KNNVectorDistanceExec {
                     arrow::compute::filter_record_batch(&batch, &mask)
                         .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
                 }
-            },
-            get_num_compute_intensive_cpus(),
-            partition,
-            &self.metrics,
-        );
-        Ok(Box::pin(stream) as SendableRecordBatchStream)
+            })
+            .buffer_unordered(get_num_compute_intensive_cpus());
+
+        let stream = stream.map(move |batch| {
+            let poll = baseline.record_poll(std::task::Poll::Ready(Some(batch)));
+            match poll {
+                std::task::Poll::Ready(Some(b)) => b,
+                _ => unreachable!("record_poll preserves Ready(Some) input"),
+            }
+        });
+        Ok(Box::pin(RecordBatchStreamAdapter::new(schema, stream)))
     }
 
     fn partition_statistics(&self, partition: Option<usize>) -> DataFusionResult<Statistics> {
diff --git a/rust/lance/src/io/exec/rowids.rs b/rust/lance/src/io/exec/rowids.rs
index cb8280ea874..837d0b81fa3 100644
--- a/rust/lance/src/io/exec/rowids.rs
+++ b/rust/lance/src/io/exec/rowids.rs
@@ -10,7 +10,7 @@ use datafusion::common::ColumnStatistics;
 use datafusion::common::stats::Precision;
 use datafusion::error::{DataFusionError, Result};
 use datafusion::execution::SendableRecordBatchStream;
-use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
+use datafusion::physical_plan::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
 use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties};
 use datafusion_physical_expr::EquivalenceProperties;
 use datafusion_physical_plan::Statistics;
@@ -29,8 +29,6 @@ use crate::Dataset;
 use crate::dataset::rowids::get_row_id_index;
 use crate::utils::future::SharedPrerequisite;
 
-use super::utils::InstrumentedChildInputStream;
-
 /// Add a `_rowaddr` column to a stream of record batches that have a `_rowid`.
 ///
 /// It's generally more efficient to scan the `_rowaddr` column, but this can be
@@ -191,33 +189,41 @@ impl AddRowAddrExec {
         let rowid_pos = self.rowid_pos;
         let rowaddr_pos = self.rowaddr_pos;
         let output_schema = self.output_schema.clone();
-        let stream = InstrumentedChildInputStream::new(
-            input_stream,
+        let baseline = BaselineMetrics::new(&self.metrics, partition);
+        let elapsed_compute = baseline.elapsed_compute().clone();
+        let stream = input_stream.then(move |batch_result| {
+            let output_schema = output_schema.clone();
+            let index_prereq = index_prereq.clone();
+            let elapsed_compute = elapsed_compute.clone();
+            async move {
+                let batch = batch_result?;
+                index_prereq.wait_ready().await?;
+                let row_id_index = index_prereq.get_ready();
+                let index_ref = row_id_index.as_deref();
+
+                let _t = elapsed_compute.timer();
+                let row_addr = Self::compute_row_addrs(batch.column(rowid_pos), index_ref)?;
+
+                let mut columns = Vec::with_capacity(batch.num_columns() + 1);
+                let existing_columns = batch.columns();
+                columns.extend_from_slice(&existing_columns[..rowaddr_pos]);
+                columns.push(row_addr);
+                columns.extend_from_slice(&existing_columns[rowaddr_pos..]);
+
+                Ok(RecordBatch::try_new(output_schema.clone(), columns)?)
+            }
+        });
+        let stream = stream.map(move |batch| {
+            let poll = baseline.record_poll(std::task::Poll::Ready(Some(batch)));
+            match poll {
+                std::task::Poll::Ready(Some(b)) => b,
+                _ => unreachable!("record_poll preserves Ready(Some) input"),
+            }
+        });
+        Ok(Box::pin(RecordBatchStreamAdapter::new(
             self.output_schema.clone(),
-            move |batch| {
-                let output_schema = output_schema.clone();
-                let index_prereq = index_prereq.clone();
-                async move {
-                    index_prereq.wait_ready().await?;
-                    let row_id_index = index_prereq.get_ready();
-                    let index_ref = row_id_index.as_deref();
-
-                    let row_addr = Self::compute_row_addrs(batch.column(rowid_pos), index_ref)?;
-
-                    let mut columns = Vec::with_capacity(batch.num_columns() + 1);
-                    let existing_columns = batch.columns();
-                    columns.extend_from_slice(&existing_columns[..rowaddr_pos]);
-                    columns.push(row_addr);
-                    columns.extend_from_slice(&existing_columns[rowaddr_pos..]);
-
-                    Ok(RecordBatch::try_new(output_schema.clone(), columns)?)
-                }
-            },
-            1,
-            partition,
-            &self.metrics,
-        );
-        Ok(Box::pin(stream))
+            stream,
+        )))
     }
 }
 
diff --git a/rust/lance/src/io/exec/scalar_index.rs b/rust/lance/src/io/exec/scalar_index.rs
index f6fd1d0a197..00157a01b6a 100644
--- a/rust/lance/src/io/exec/scalar_index.rs
+++ b/rust/lance/src/io/exec/scalar_index.rs
@@ -3,9 +3,7 @@
 
 use std::sync::{Arc, LazyLock};
 
-use super::utils::{
-    IndexMetrics, InstrumentedChildInputStream, InstrumentedRecordBatchStreamAdapter,
-};
+use super::utils::{IndexMetrics, InstrumentedRecordBatchStreamAdapter};
 use crate::{
     Dataset,
     dataset::rowids::load_row_id_sequences,
@@ -22,7 +20,7 @@ use datafusion::{
     physical_plan::{
         DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
         execution_plan::{Boundedness, EmissionType},
-        metrics::{ExecutionPlanMetricsSet, MetricsSet},
+        metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet},
         stream::RecordBatchStreamAdapter,
     },
     scalar::ScalarValue,
@@ -293,12 +291,6 @@ impl MapIndexExec {
         index_metrics: Arc<IndexMetrics>,
         metrics_set: ExecutionPlanMetricsSet,
     ) -> datafusion::error::Result<datafusion::physical_plan::SendableRecordBatchStream> {
-        // Time the one-shot setup (fragment bitmap + deletion mask) so it's
-        // attributed to this node's elapsed_compute. The helper itself only
-        // times per-batch work.
-        let elapsed_compute = datafusion::physical_plan::metrics::MetricBuilder::new(&metrics_set)
-            .elapsed_compute(partition);
-        let setup_start = std::time::Instant::now();
         let fragment_bitmap = scalar_index_fragment_bitmap(&dataset, &column_name, &index_name)
             .await?
             .ok_or_else(|| {
@@ -313,17 +305,22 @@ impl MapIndexExec {
         } else {
             None
         };
-        elapsed_compute.add_duration(setup_start.elapsed());
 
-        let helper = InstrumentedChildInputStream::new(
-            input,
-            INDEX_LOOKUP_SCHEMA.clone(),
-            move |batch| {
-                let column_name = column_name.clone();
-                let index_name = index_name.clone();
-                let dataset = dataset.clone();
-                let deletion_mask = deletion_mask.clone();
-                let metrics = index_metrics.clone();
+        let baseline = BaselineMetrics::new(&metrics_set, partition);
+        let elapsed_compute = baseline.elapsed_compute().clone();
+        let stream = input.then(move |batch_result| {
+            let column_name = column_name.clone();
+            let index_name = index_name.clone();
+            let dataset = dataset.clone();
+            let deletion_mask = deletion_mask.clone();
+            let metrics = index_metrics.clone();
+            let elapsed_compute = elapsed_compute.clone();
+            async move {
+                let batch = batch_result?;
+                // Timer spans `map_batch`'s `.await` on purpose: that await is
+                // the per-batch sargable index evaluation, which is the work
+                // we want attributed here.
+                let _t = elapsed_compute.timer();
                 Self::map_batch(
                     column_name,
                     index_name,
@@ -332,12 +329,20 @@ impl MapIndexExec {
                     batch,
                     metrics,
                 )
-            },
-            1,
-            partition,
-            &metrics_set,
-        );
-        Ok(Box::pin(helper))
+                .await
+            }
+        });
+        let stream = stream.map(move |batch| {
+            let poll = baseline.record_poll(std::task::Poll::Ready(Some(batch)));
+            match poll {
+                std::task::Poll::Ready(Some(b)) => b,
+                _ => unreachable!("record_poll preserves Ready(Some) input"),
+            }
+        });
+        Ok(Box::pin(RecordBatchStreamAdapter::new(
+            INDEX_LOOKUP_SCHEMA.clone(),
+            stream,
+        )))
     }
 
     async fn map_batch(
diff --git a/rust/lance/src/io/exec/utils.rs b/rust/lance/src/io/exec/utils.rs
index 004b1a95b5f..ea312296fbb 100644
--- a/rust/lance/src/io/exec/utils.rs
+++ b/rust/lance/src/io/exec/utils.rs
@@ -9,9 +9,8 @@ use lance_index::metrics::MetricsCollector;
 use lance_io::scheduler::ScanScheduler;
 use lance_table::format::IndexMetadata;
 use pin_project::pin_project;
-use std::pin::Pin;
 use std::sync::{Arc, Mutex};
-use std::task::{Context, Poll};
+use std::task::Poll;
 
 use arrow::array::AsArray;
 use arrow_array::{RecordBatch, UInt64Array};
@@ -24,14 +23,12 @@ use datafusion::physical_plan::metrics::{
 use datafusion::physical_plan::{
     DisplayAs, DisplayFormatType, ExecutionPlan, RecordBatchStream, SendableRecordBatchStream,
 };
-use futures::stream::FuturesUnordered;
 use futures::{Stream, StreamExt, TryStreamExt};
 use lance_core::error::{CloneableResult, Error};
 use lance_core::utils::futures::{Capacity, SharedStreamExt};
 use lance_core::{ROW_ID, Result};
 use lance_index::prefilter::FilterLoader;
 use lance_select::{RowAddrMask, RowAddrTreeMap};
-use std::future::Future;
 
 use crate::Dataset;
 use crate::index::prefilter::DatasetPreFilter;
@@ -298,130 +295,6 @@ where
     }
 }
 
-/// Stream wrapper for an `ExecutionPlan` node that pulls from a child input and
-/// applies a per-batch async transform.
-///
-/// `elapsed_compute` measures only the time spent driving the transform
-/// futures — never the time spent polling the child input — so wrapping a
-/// chain of nodes does not double-count child CPU. `output_rows` and
-/// `output_batches` are recorded as the transform produces batches.
-///
-/// `concurrency` caps how many transform futures may be in flight at once.
-/// Use `1` for sequential transforms; larger values parallelize per-batch
-/// work (e.g., KNN distance computation).
-///
-/// For leaf nodes (no child input), use [`InstrumentedRecordBatchStreamAdapter`]
-/// instead.
-pub struct InstrumentedChildInputStream<F, Fut> {
-    schema: SchemaRef,
-    input: SendableRecordBatchStream,
-    transform: F,
-    concurrency: usize,
-    in_flight: FuturesUnordered<Fut>,
-    input_done: bool,
-    baseline_metrics: BaselineMetrics,
-    batch_count: Count,
-}
-
-impl<F, Fut> InstrumentedChildInputStream<F, Fut>
-where
-    F: FnMut(RecordBatch) -> Fut,
-    Fut: Future<Output = DataFusionResult<RecordBatch>>,
-{
-    pub fn new(
-        input: SendableRecordBatchStream,
-        schema: SchemaRef,
-        transform: F,
-        concurrency: usize,
-        partition: usize,
-        metrics: &ExecutionPlanMetricsSet,
-    ) -> Self {
-        assert!(concurrency >= 1, "concurrency must be >= 1");
-        let batch_count = Count::new();
-        MetricBuilder::new(metrics)
-            .with_partition(partition)
-            .build(MetricValue::OutputBatches(batch_count.clone()));
-        Self {
-            schema,
-            input,
-            transform,
-            concurrency,
-            in_flight: FuturesUnordered::new(),
-            input_done: false,
-            baseline_metrics: BaselineMetrics::new(metrics, partition),
-            batch_count,
-        }
-    }
-}
-
-impl<F, Fut> Stream for InstrumentedChildInputStream<F, Fut>
-where
-    F: FnMut(RecordBatch) -> Fut + Unpin,
-    Fut: Future<Output = DataFusionResult<RecordBatch>>,
-{
-    type Item = DataFusionResult<RecordBatch>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let this = self.get_mut();
-
-        // Fill in-flight transforms up to `concurrency` from the input.
-        // Polling the input does NOT count toward `elapsed_compute`.
-        while !this.input_done && this.in_flight.len() < this.concurrency {
-            match this.input.poll_next_unpin(cx) {
-                Poll::Ready(Some(Ok(batch))) => {
-                    this.in_flight.push((this.transform)(batch));
-                }
-                Poll::Ready(Some(Err(e))) => {
-                    return Poll::Ready(Some(Err(e)));
-                }
-                Poll::Ready(None) => {
-                    this.input_done = true;
-                }
-                Poll::Pending => break,
-            }
-        }
-
-        // Drive in-flight transforms; their poll time IS counted.
-        if !this.in_flight.is_empty() {
-            let timer = this.baseline_metrics.elapsed_compute().timer();
-            let poll = this.in_flight.poll_next_unpin(cx);
-            timer.done();
-            match poll {
-                Poll::Ready(Some(result)) => {
-                    if result.is_ok() {
-                        this.batch_count.add(1);
-                    }
-                    return this.baseline_metrics.record_poll(Poll::Ready(Some(result)));
-                }
-                // Unreachable: `FuturesUnordered::poll_next` returns
-                // `Ready(None)` only when empty, and we just checked
-                // `!is_empty` above. A panic here is preferable to a
-                // silent infinite loop if the invariant ever breaks.
-                Poll::Ready(None) => {
-                    unreachable!("FuturesUnordered yielded None while non-empty")
-                }
-                Poll::Pending => return Poll::Pending,
-            }
-        }
-
-        if this.input_done {
-            return Poll::Ready(None);
-        }
-
-        Poll::Pending
-    }
-}
-
-impl<F, Fut> RecordBatchStream for InstrumentedChildInputStream<F, Fut>
-where
-    F: FnMut(RecordBatch) -> Fut + Unpin,
-    Fut: Future<Output = DataFusionResult<RecordBatch>>,
-{
-    fn schema(&self) -> SchemaRef {
-        self.schema.clone()
-    }
-}
-
 impl ExecutionPlan for ReplayExec {
     fn name(&self) -> &str {
         "ReplayExec"
@@ -566,136 +439,7 @@ mod tests {
     use lance_datafusion::exec::OneShotExec;
     use lance_datagen::{BatchCount, RowCount, array};
 
-    use super::{InstrumentedChildInputStream, ReplayExec};
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn instrumented_child_input_stream_excludes_child_poll_time() {
-        use std::sync::atomic::{AtomicUsize, Ordering};
-        use std::task::Poll;
-        use std::time::Duration;
-
-        use arrow_array::Int32Array;
-        use arrow_schema::{DataType, Field, Schema};
-        use datafusion::physical_plan::SendableRecordBatchStream;
-        use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
-
-        let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
-        let n_batches: usize = 3;
-        // child_delay is intentionally several times larger than transform_delay
-        // so the assertion tolerates significant `std::thread::sleep` overshoot
-        // on busy CI runners (we've seen ~2-3x overshoot on macOS Actions).
-        let child_delay = Duration::from_millis(150);
-        let transform_delay = Duration::from_millis(30);
-
-        let counter = Arc::new(AtomicUsize::new(0));
-        let s = schema.clone();
-        let child = futures::stream::poll_fn(move |_cx| {
-            let n = counter.fetch_add(1, Ordering::SeqCst);
-            if n >= n_batches {
-                return Poll::Ready(None);
-            }
-            std::thread::sleep(child_delay);
-            let batch = arrow_array::RecordBatch::try_new(
-                s.clone(),
-                vec![Arc::new(Int32Array::from(vec![n as i32]))],
-            )
-            .unwrap();
-            Poll::Ready(Some(Ok(batch)))
-        });
-        let child: SendableRecordBatchStream =
-            Box::pin(RecordBatchStreamAdapter::new(schema.clone(), child));
-
-        let metrics = ExecutionPlanMetricsSet::new();
-        let stream = InstrumentedChildInputStream::new(
-            child,
-            schema,
-            move |batch| async move {
-                std::thread::sleep(transform_delay);
-                Ok(batch)
-            },
-            1,
-            0,
-            &metrics,
-        );
-
-        let batches: Vec<_> = stream.try_collect().await.unwrap();
-        assert_eq!(batches.len(), n_batches);
-
-        let elapsed_ns = metrics
-            .clone_inner()
-            .elapsed_compute()
-            .expect("elapsed_compute should be recorded");
-        let elapsed = Duration::from_nanos(elapsed_ns as u64);
-
-        // Expect ~ transform_delay * n. The upper bound is set generously to
-        // absorb sleep overshoot on slow CI (~4-5x per call) while still
-        // cleanly rejecting any version that double-counts child poll time,
-        // which would yield ~ (transform_delay + child_delay) * n.
-        let upper = Duration::from_millis(400);
-        assert!(
-            elapsed >= transform_delay * (n_batches as u32 - 1),
-            "elapsed_compute={:?} too low; transform time was not measured",
-            elapsed,
-        );
-        assert!(
-            elapsed < upper,
-            "elapsed_compute={:?} >= {:?}; child input time was double-counted",
-            elapsed,
-            upper,
-        );
-    }
-
-    #[tokio::test]
-    async fn instrumented_child_input_stream_propagates_child_error() {
-        use std::sync::atomic::{AtomicUsize, Ordering};
-        use std::task::Poll;
-
-        use arrow_array::Int32Array;
-        use arrow_schema::{DataType, Field, Schema};
-        use datafusion::error::DataFusionError;
-        use datafusion::physical_plan::SendableRecordBatchStream;
-        use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
-
-        let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
-        let s = schema.clone();
-        let step = Arc::new(AtomicUsize::new(0));
-        // Yield one OK batch, then an Err, then None.
-        let child = futures::stream::poll_fn(move |_cx| {
-            let n = step.fetch_add(1, Ordering::SeqCst);
-            match n {
-                0 => {
-                    let batch = arrow_array::RecordBatch::try_new(
-                        s.clone(),
-                        vec![Arc::new(Int32Array::from(vec![1]))],
-                    )
-                    .unwrap();
-                    Poll::Ready(Some(Ok(batch)))
-                }
-                1 => Poll::Ready(Some(Err(DataFusionError::Execution("boom".into())))),
-                _ => Poll::Ready(None),
-            }
-        });
-        let child: SendableRecordBatchStream =
-            Box::pin(RecordBatchStreamAdapter::new(schema.clone(), child));
-
-        let metrics = ExecutionPlanMetricsSet::new();
-        let stream = InstrumentedChildInputStream::new(
-            child,
-            schema,
-            move |batch| async move { Ok(batch) },
-            1,
-            0,
-            &metrics,
-        );
-
-        let mut stream = Box::pin(stream);
-        let first = stream.next().await.expect("first item present");
-        assert!(first.is_ok(), "expected first batch ok, got {:?}", first);
-
-        let second = stream.next().await.expect("error item present");
-        let err = second.expect_err("expected propagated error");
-        assert!(err.to_string().contains("boom"), "got {}", err);
-    }
+    use super::ReplayExec;
 
     #[tokio::test]
     async fn test_replay() {