delta-arrow-reader 0.7.0

//! Object-store metering for direct Parquet data-file reads.

use std::{
    fmt, io,
    ops::Range,
    sync::{Arc, Mutex},
    task::Poll,
    time::{Duration, Instant},
};

use async_trait::async_trait;
use bytes::Bytes;
use futures_util::{StreamExt, stream, stream::BoxStream};
use object_store::{
    CopyOptions, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload,
    OBJECT_STORE_COALESCE_DEFAULT, ObjectMeta, ObjectStore, ObjectStoreScheme, PutMultipartOptions,
    PutOptions, PutPayload, PutResult, RenameOptions, Result, path::Path,
};
use tokio::sync::Semaphore;
use tracing::Instrument;
use url::Url;

use super::cost_model::{
    FallbackEstimator, TransportEstimate, bandwidth_delay_bytes, duration_cost_bytes, range_bytes,
    request_waves,
};
use super::network_profile::{
    ActiveRead, CompletedRangeRead, CurrentTransportEstimate, ParquetRangeReadEstimator,
    RangeReadCalibration, RangeReadTiming,
};
use super::range_planning::{
    ChosenRangePlan, RANGE_READ_PERMITS, RangePlanDecision, choose_range_plan, execute_range_plan,
    merge_ranges,
};
use crate::{
    DeltaScanMetrics,
    reader::{
        ParquetRangeReadPolicy,
        options::{MAX_CONCURRENT_PARQUET_RANGE_READS, MAX_SHARED_RANGE_READS},
    },
};

const RANGE_PLANNING_DIAGNOSTIC_TARGET: &str =
    "delta_arrow_reader::diagnostics::parquet_range_planning";

pub(crate) struct MeteredParquetObjectStore {
    inner: Arc<dyn ObjectStore>,
    metrics: DeltaScanMetrics,
    multi_range_read_strategy: MultiRangeReadStrategy,
    range_read_estimator: Arc<ParquetRangeReadEstimator>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum MultiRangeReadStrategy {
    /// Pass the original range list to the store's `get_ranges` implementation.
    UseStoreImplementation,
    /// Read only the normalized ranges requested by Parquet.
    ReadExactRanges,
    /// Merge ranges separated by at most one MiB before reading them.
    MergeRangesWithinOneMegabyte,
    /// Choose a physical range plan from recent remote transport observations.
    ChooseAutomatically,
}

impl MultiRangeReadStrategy {
    pub(crate) fn for_policy(policy: ParquetRangeReadPolicy, table_url: &Url) -> Self {
        match policy {
            ParquetRangeReadPolicy::Automatic => Self::for_table_url(table_url),
            ParquetRangeReadPolicy::ExactRanges => Self::ReadExactRanges,
            ParquetRangeReadPolicy::MergeRangesWithinOneMegabyte => {
                Self::MergeRangesWithinOneMegabyte
            }
            ParquetRangeReadPolicy::StoreImplementation => Self::UseStoreImplementation,
        }
    }

    /// Uses the store implementation for local, memory, and custom URL schemes. Built-in remote
    /// stores choose their physical range plan automatically.
    pub(crate) fn for_table_url(table_url: &Url) -> Self {
        match ObjectStoreScheme::parse(table_url) {
            Ok((ObjectStoreScheme::Local | ObjectStoreScheme::Memory, _)) | Err(_) => {
                Self::UseStoreImplementation
            }
            Ok(_) => Self::ChooseAutomatically,
        }
    }
}

impl MeteredParquetObjectStore {
    pub(crate) fn new(
        inner: Arc<dyn ObjectStore>,
        metrics: DeltaScanMetrics,
        multi_range_read_strategy: MultiRangeReadStrategy,
    ) -> Self {
        Self {
            inner,
            metrics,
            multi_range_read_strategy,
            range_read_estimator: Arc::new(ParquetRangeReadEstimator::default()),
        }
    }

    /// Reuses transport measurements collected by other readers in the same store context.
    pub(crate) fn with_range_read_estimator(
        mut self,
        range_read_estimator: Arc<ParquetRangeReadEstimator>,
    ) -> Self {
        self.range_read_estimator = range_read_estimator;
        self
    }

    pub(super) fn range_read_estimator(&self) -> &Arc<ParquetRangeReadEstimator> {
        &self.range_read_estimator
    }

    async fn get_with_metrics(
        &self,
        location: &Path,
        options: GetOptions,
        active_read: Option<ActiveRead>,
    ) -> Result<GetResult> {
        let should_meter_payload = !options.head;
        if should_meter_payload {
            if options.range.is_some() {
                self.metrics.record_parquet_data_file_range_get_operation();
            } else {
                self.metrics.record_parquet_data_file_full_get_operation();
            }
        }

        let result = self
            .inner
            .get_opts(location, options)
            .instrument(tracing::debug_span!(
                target: "delta_arrow_reader::profile",
                "Object store transport"
            ))
            .await?;
        if should_meter_payload {
            Ok(track_response_body(
                result,
                self.metrics.clone(),
                active_read,
            ))
        } else {
            Ok(result)
        }
    }

    /// Reads one physical range and measures request latency separately from payload delivery.
    async fn read_range_with_timing(
        &self,
        location: &Path,
        range: Range<u64>,
        budget: &Semaphore,
    ) -> Result<(Bytes, CompletedRangeRead)> {
        // Hold capacity only while this request (including retries) is in flight.
        // Queueing time must not become a network-latency sample.
        let available = budget.try_acquire().ok();
        let queued = available.is_none();
        let _permit = match available {
            Some(permit) => permit,
            None => budget
                .acquire()
                .await
                .map_err(|error| object_store::Error::Generic {
                    store: "delta-arrow-reader",
                    source: Box::new(error),
                })?,
        };
        let expected_bytes = range.end - range.start;
        let request_started = Instant::now();
        // The surrounding range plan already tracks activity for these requests.
        let result = self
            .get_with_metrics(location, GetOptions::new().with_range(Some(range)), None)
            .await?;
        let payload_started = Instant::now();
        let bytes = result.bytes().await?;
        if u64::try_from(bytes.len()).unwrap_or(u64::MAX) != expected_bytes {
            return Err(object_store::Error::Generic {
                store: "delta-arrow-reader",
                source: io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "object store returned a byte range with an unexpected length",
                )
                .into(),
            });
        }
        let payload_finished = Instant::now();
        let completed_read = CompletedRangeRead {
            request_latency: payload_started.saturating_duration_since(request_started),
            payload_started,
            payload_finished,
            bytes_received: bytes.len(),
            queued,
        };
        Ok((bytes, completed_read))
    }

    /// Returns the available evidence and robust estimate from the same estimator view.
    fn current_transport_estimate(&self) -> CurrentTransportEstimate {
        self.range_read_estimator.current_transport_estimate()
    }

    pub(super) fn request_overhead_bytes(&self) -> Option<u128> {
        let capacity = self
            .range_read_estimator
            .current_transport_estimate()
            .estimate?
            .shared_throughput_bytes_per_second;
        self.request_overhead_at_capacity(capacity)
    }

    /// Uses the caller's bandwidth so a retained profile keeps consistent cost units.
    pub(super) fn request_overhead_at_capacity(&self, capacity: u64) -> Option<u128> {
        // Request processing is shared too; do not discount its cost per reader.
        Some(duration_cost_bytes(
            self.range_read_estimator.request_overhead()?,
            capacity,
        ))
    }

    /// Returns matching bandwidth and request samples from the last fully sampled warmup.
    pub(super) fn range_read_calibration(&self) -> Option<Arc<RangeReadCalibration>> {
        Some(Arc::clone(self.fallback_cost_estimator()?.calibration()))
    }

    /// File readers share one cache per complete calibration; passive observations do not replace it.
    pub(super) fn fallback_cost_estimator(&self) -> Option<Arc<FallbackEstimator>> {
        self.range_read_estimator.fallback_cost_estimator()
    }

    #[cfg(test)]
    pub(super) fn seed_range_read_calibration(&self, calibration: RangeReadCalibration) {
        self.range_read_estimator
            .seed_range_read_calibration(calibration);
    }

    pub(super) fn transport_estimate(&self) -> Option<TransportEstimate> {
        self.current_transport_estimate().estimate
    }

    #[cfg(test)]
    pub(super) fn seed_request_overhead(&self, overhead: Duration) {
        self.range_read_estimator.seed_request_overhead(overhead);
    }

    #[cfg(test)]
    pub(super) fn seed_transport_estimate(&self, estimate: TransportEstimate) {
        self.range_read_estimator.seed_transport_estimate(estimate);
    }

    #[cfg(test)]
    fn record_completed_range_reads(&self, completed_reads: &[CompletedRangeRead]) {
        self.range_read_estimator
            .record_transport_sample(completed_reads);
    }

    /// Records the exact request and chosen physical plan before its reads start.
    fn record_chosen_range_plan_metrics(&self, plan: &ChosenRangePlan) {
        self.metrics
            .record_parquet_data_file_exact_ranges_requested(
                plan.exact_range_count,
                plan.exact_bytes,
            );
        self.metrics.record_parquet_data_file_physical_range_plan(
            plan.physical_ranges.len(),
            plan.planned_bytes,
            MAX_CONCURRENT_PARQUET_RANGE_READS,
        );
        match plan.decision {
            RangePlanDecision::ColdStart => self
                .metrics
                .record_parquet_data_file_cold_start_range_plan(),
            RangePlanDecision::CostBasedExact => self
                .metrics
                .record_parquet_data_file_cost_based_exact_range_plan(),
            RangePlanDecision::CostBasedMerged => self
                .metrics
                .record_parquet_data_file_cost_based_merged_range_plan(),
        }
    }

    /// Reads one physical plan and restores the caller's original ranges and order.
    async fn read_physical_ranges(
        &self,
        location: &Path,
        requested_ranges: &[Range<u64>],
        physical_ranges: &[Range<u64>],
    ) -> Result<(Vec<Bytes>, Duration)> {
        self.read_ranges_with_concurrency(
            location,
            requested_ranges,
            physical_ranges,
            MAX_CONCURRENT_PARQUET_RANGE_READS,
            &RANGE_READ_PERMITS,
        )
        .await
        .map(|(bytes, timing)| (bytes, timing.elapsed))
    }

    /// Reads a partial-page plan without coalescing again. Return complete physical
    /// buffers so page reconstruction can reuse bytes read while filling gaps.
    /// Elapsed time is returned only when it can be sampled without contention.
    pub(super) async fn read_partial_ranges(
        &self,
        location: &Path,
        requested_ranges: &[Range<u64>],
        physical_ranges: &[Range<u64>],
        budget: &Semaphore,
    ) -> Result<(Vec<Bytes>, Option<Duration>)> {
        let (bytes, timing) = self
            .read_partial_ranges_with_timing(location, requested_ranges, physical_ranges, budget)
            .await?;
        Ok((bytes, timing.can_sample.then_some(timing.elapsed)))
    }

    pub(super) async fn read_partial_ranges_with_timing(
        &self,
        location: &Path,
        requested_ranges: &[Range<u64>],
        physical_ranges: &[Range<u64>],
        budget: &Semaphore,
    ) -> Result<(Vec<Bytes>, RangeReadTiming)> {
        let store = Self {
            inner: Arc::clone(
                self.range_read_estimator
                    .partial_range_store()
                    .unwrap_or(&self.inner),
            ),
            metrics: self.metrics.clone(),
            multi_range_read_strategy: self.multi_range_read_strategy,
            range_read_estimator: Arc::clone(&self.range_read_estimator),
        };
        let concurrency = physical_ranges.len().min(MAX_SHARED_RANGE_READS);
        let exact_ranges = merge_ranges(requested_ranges, 0);
        self.metrics
            .record_parquet_data_file_exact_ranges_requested(
                exact_ranges.len(),
                range_bytes(&exact_ranges),
            );
        self.metrics.record_parquet_data_file_physical_range_plan(
            physical_ranges.len(),
            range_bytes(physical_ranges),
            concurrency,
        );
        let (bytes, timing) = store
            .read_ranges_with_concurrency(
                location,
                physical_ranges,
                physical_ranges,
                concurrency,
                budget,
            )
            .await?;
        self.range_read_estimator
            .record_request_interval(timing.request_interval);
        Ok((bytes, timing))
    }

    /// Each request acquires shared capacity independently, so a slow request
    /// cannot hold idle slots and another plan can use newly released capacity.
    pub(super) async fn read_ranges_with_concurrency(
        &self,
        location: &Path,
        requested_ranges: &[Range<u64>],
        physical_ranges: &[Range<u64>],
        concurrency: usize,
        budget: &Semaphore,
    ) -> Result<(Vec<Bytes>, RangeReadTiming)> {
        let active_read = self.range_read_estimator.start_read();
        let plan_started = Instant::now();
        let completed_reads = Arc::new(Mutex::new(Vec::with_capacity(physical_ranges.len())));
        let results = execute_range_plan(requested_ranges, physical_ranges, concurrency, |range| {
            let completed_reads = Arc::clone(&completed_reads);
            async move {
                let (bytes, completed_read) =
                    self.read_range_with_timing(location, range, budget).await?;
                completed_reads
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .push(completed_read);
                Ok::<Bytes, object_store::Error>(bytes)
            }
        })
        .await?;
        let observed_plan_time = plan_started.elapsed();
        let mut completed = completed_reads
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let timing = self.range_read_estimator.record_range_plan(
            std::mem::take(&mut completed),
            &active_read,
            observed_plan_time,
            concurrency,
        );
        self.metrics
            .record_parquet_range_successful_plan_time(plan_started.elapsed());
        Ok((results, timing))
    }
}

/// Emits the inputs and outcome of one successfully completed automatic range plan.
fn trace_completed_automatic_range_plan(
    plan: &ChosenRangePlan,
    estimate: CurrentTransportEstimate,
    observed_plan_time: Duration,
) {
    let plan_estimate = plan.transport_estimate;
    tracing::debug!(
        target: RANGE_PLANNING_DIAGNOSTIC_TARGET,
        event = "automatic_parquet_range_plan_completed",
        latency_sample_count = u64::try_from(estimate.latency_sample_count).unwrap_or(u64::MAX),
        throughput_sample_count =
            u64::try_from(estimate.throughput_sample_count).unwrap_or(u64::MAX),
        estimated_request_latency_micros =
            plan_estimate.map(|value| value.request_latency.as_micros()),
        estimated_shared_throughput_bytes_per_second =
            plan_estimate.map(|value| value.shared_throughput_bytes_per_second),
        estimated_bandwidth_delay_bytes = plan_estimate.map(bandwidth_delay_bytes),
        exact_range_count = u64::try_from(plan.exact_range_count).unwrap_or(u64::MAX),
        exact_bytes = plan.exact_bytes,
        exact_request_waves = u64::try_from(request_waves(
            plan.exact_range_count,
            MAX_CONCURRENT_PARQUET_RANGE_READS,
        ))
        .unwrap_or(u64::MAX),
        baseline_range_count = u64::try_from(plan.baseline_range_count).unwrap_or(u64::MAX),
        baseline_bytes = plan.baseline_bytes,
        baseline_request_waves = u64::try_from(request_waves(
            plan.baseline_range_count,
            MAX_CONCURRENT_PARQUET_RANGE_READS,
        ))
        .unwrap_or(u64::MAX),
        selected_range_count = u64::try_from(plan.physical_ranges.len()).unwrap_or(u64::MAX),
        selected_bytes = plan.planned_bytes,
        selected_request_waves = u64::try_from(request_waves(
            plan.physical_ranges.len(),
            MAX_CONCURRENT_PARQUET_RANGE_READS,
        ))
        .unwrap_or(u64::MAX),
        baseline_predicted_cost_bytes = plan.baseline_predicted_cost_bytes,
        selected_predicted_cost_bytes = plan.selected_predicted_cost_bytes,
        observed_selected_plan_micros = observed_plan_time.as_micros(),
        decision = plan.decision.as_str(),
        "Automatic Parquet range plan completed"
    );
}

impl fmt::Debug for MeteredParquetObjectStore {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("MeteredParquetObjectStore")
    }
}

impl fmt::Display for MeteredParquetObjectStore {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("MeteredParquetObjectStore")
    }
}

#[async_trait]
impl ObjectStore for MeteredParquetObjectStore {
    async fn put_opts(
        &self,
        location: &Path,
        payload: PutPayload,
        options: PutOptions,
    ) -> Result<PutResult> {
        self.inner.put_opts(location, payload, options).await
    }

    async fn put_multipart_opts(
        &self,
        location: &Path,
        options: PutMultipartOptions,
    ) -> Result<Box<dyn MultipartUpload>> {
        self.inner.put_multipart_opts(location, options).await
    }

    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
        let active_read = (!options.head).then(|| self.range_read_estimator.start_read());
        self.get_with_metrics(location, options, active_read).await
    }

    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
        if ranges.is_empty() {
            return Ok(Vec::new());
        }

        match self.multi_range_read_strategy {
            MultiRangeReadStrategy::UseStoreImplementation => {
                let _active_read = self.range_read_estimator.start_read();
                let plan_started = Instant::now();
                let exact_ranges = merge_ranges(ranges, 0);
                self.metrics
                    .record_parquet_data_file_exact_ranges_requested(
                        exact_ranges.len(),
                        range_bytes(&exact_ranges),
                    );
                self.metrics
                    .record_parquet_data_file_store_delegated_range_plan();
                self.metrics.record_parquet_data_file_range_get_operation();
                let results = self
                    .inner
                    .get_ranges(location, ranges)
                    .instrument(tracing::debug_span!(
                        target: "delta_arrow_reader::profile",
                        "Object store transport"
                    ))
                    .await?;
                for bytes in &results {
                    self.metrics
                        .record_parquet_data_file_bytes_received(bytes.len());
                }
                self.metrics
                    .record_parquet_range_successful_plan_time(plan_started.elapsed());
                Ok(results)
            }
            strategy @ (MultiRangeReadStrategy::ReadExactRanges
            | MultiRangeReadStrategy::MergeRangesWithinOneMegabyte) => {
                let exact_ranges = merge_ranges(ranges, 0);
                let max_gap = if strategy == MultiRangeReadStrategy::MergeRangesWithinOneMegabyte {
                    OBJECT_STORE_COALESCE_DEFAULT
                } else {
                    0
                };
                let physical_ranges = merge_ranges(ranges, max_gap);
                self.metrics
                    .record_parquet_data_file_exact_ranges_requested(
                        exact_ranges.len(),
                        range_bytes(&exact_ranges),
                    );
                self.metrics.record_parquet_data_file_physical_range_plan(
                    physical_ranges.len(),
                    range_bytes(&physical_ranges),
                    MAX_CONCURRENT_PARQUET_RANGE_READS,
                );
                let (results, _) = self
                    .read_physical_ranges(location, ranges, &physical_ranges)
                    .await?;
                Ok(results)
            }
            MultiRangeReadStrategy::ChooseAutomatically => {
                let estimate = self.current_transport_estimate();
                let plan = choose_range_plan(ranges, estimate.estimate);
                self.record_chosen_range_plan_metrics(&plan);
                let (results, observed_plan_time) = self
                    .read_physical_ranges(location, ranges, &plan.physical_ranges)
                    .await?;
                trace_completed_automatic_range_plan(&plan, estimate, observed_plan_time);
                Ok(results)
            }
        }
    }

    fn delete_stream(
        &self,
        locations: BoxStream<'static, Result<Path>>,
    ) -> BoxStream<'static, Result<Path>> {
        self.inner.delete_stream(locations)
    }

    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
        self.inner.list(prefix)
    }

    fn list_with_offset(
        &self,
        prefix: Option<&Path>,
        offset: &Path,
    ) -> BoxStream<'static, Result<ObjectMeta>> {
        self.inner.list_with_offset(prefix, offset)
    }

    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
        self.inner.list_with_delimiter(prefix).await
    }

    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
        self.inner.copy_opts(from, to, options).await
    }

    async fn rename_opts(&self, from: &Path, to: &Path, options: RenameOptions) -> Result<()> {
        self.inner.rename_opts(from, to, options).await
    }
}

fn track_response_body(
    result: GetResult,
    metrics: DeltaScanMetrics,
    mut active_read: Option<ActiveRead>,
) -> GetResult {
    let GetResult {
        payload,
        meta,
        range,
        attributes,
    } = result;

    let payload = match payload {
        GetResultPayload::Stream(payload) => payload,
        #[cfg(not(target_arch = "wasm32"))]
        GetResultPayload::File(file, path) => {
            let local_result = GetResult {
                payload: GetResultPayload::File(file, path),
                meta: meta.clone(),
                range: range.clone(),
                attributes: attributes.clone(),
            };
            stream::once(async move { local_result.bytes().await }).boxed()
        }
    };

    // Headers alone do not end a read; hold its activity until the body finishes.
    let mut payload = Some(payload);
    let payload = stream::poll_fn(move |cx| {
        let Some(stream) = payload.as_mut() else {
            return Poll::Ready(None);
        };
        let next = stream.poll_next_unpin(cx);
        match &next {
            Poll::Ready(Some(Ok(bytes))) => {
                metrics.record_parquet_data_file_bytes_received(bytes.len());
            }
            Poll::Ready(_) => {
                payload = None;
                active_read.take();
            }
            Poll::Pending => {}
        }
        next
    })
    .boxed();

    GetResult {
        payload: GetResultPayload::Stream(payload),
        meta,
        range,
        attributes,
    }
}

#[cfg(test)]
mod tests {
    use std::{
        collections::BTreeMap,
        fmt,
        fs::File,
        io,
        ops::Range,
        path::PathBuf,
        sync::{
            Arc, Mutex, Once,
            atomic::{AtomicU64, Ordering},
        },
        time::{Duration, Instant},
    };

    use async_trait::async_trait;
    use bytes::Bytes;
    use chrono::{DateTime, Utc};
    use futures_util::{StreamExt, stream, stream::BoxStream};
    use object_store::{
        Attributes, CopyOptions, Error, GetOptions, GetResult, GetResultPayload, ListResult,
        MultipartUpload, ObjectMeta, ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions,
        PutPayload, PutResult, RenameOptions, Result,
        memory::InMemory,
        path::Path,
        throttle::{ThrottleConfig, ThrottledStore},
    };
    use tracing::{
        Event, Level, Metadata, Subscriber,
        field::{Field as TracingField, Visit},
        span::{Attributes as TracingAttributes, Id, Record},
        subscriber::{Interest, with_default},
    };

    use super::{
        ChosenRangePlan, CurrentTransportEstimate, MeteredParquetObjectStore,
        MultiRangeReadStrategy, ParquetRangeReadEstimator, RANGE_PLANNING_DIAGNOSTIC_TARGET,
        RangePlanDecision, TransportEstimate,
    };
    use crate::{
        DeltaScanMetrics, ParquetReaderBackend,
        reader::{ParquetRangeReadPolicy, metrics::DeltaScanMetricsConfig},
    };

    use super::super::network_profile::{
        SMALL_READ_CONCURRENCY, WARMUP_READ_BYTES,
        tests::{completed_read, completed_read_at_throughput},
        warmup,
    };

    static TRACING_TEST_LOCK: Mutex<()> = Mutex::new(());
    static TRACING_TEST_GLOBAL_SUBSCRIBER: Once = Once::new();

    #[derive(Clone, Default)]
    struct EventFields(Arc<Mutex<Vec<BTreeMap<String, String>>>>);

    impl Subscriber for EventFields {
        fn register_callsite(&self, metadata: &'static Metadata<'static>) -> Interest {
            if metadata.target() == RANGE_PLANNING_DIAGNOSTIC_TARGET
                && *metadata.level() == Level::DEBUG
            {
                Interest::always()
            } else {
                Interest::sometimes()
            }
        }

        fn enabled(&self, metadata: &Metadata<'_>) -> bool {
            metadata.target() == RANGE_PLANNING_DIAGNOSTIC_TARGET
                && *metadata.level() == Level::DEBUG
        }

        fn new_span(&self, _attributes: &TracingAttributes<'_>) -> Id {
            Id::from_u64(1)
        }

        fn record(&self, _span: &Id, _values: &Record<'_>) {}

        fn record_follows_from(&self, _span: &Id, _follows: &Id) {}

        fn event(&self, event: &Event<'_>) {
            let mut fields = event
                .metadata()
                .fields()
                .iter()
                .map(|field| (field.name().to_owned(), "<empty>".to_owned()))
                .collect();
            event.record(&mut FieldVisitor(&mut fields));
            self.0
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .push(fields);
        }

        fn enter(&self, _span: &Id) {}

        fn exit(&self, _span: &Id) {}
    }

    struct FieldVisitor<'fields>(&'fields mut BTreeMap<String, String>);

    impl Visit for FieldVisitor<'_> {
        fn record_debug(&mut self, field: &TracingField, value: &dyn fmt::Debug) {
            self.0.insert(field.name().to_owned(), format!("{value:?}"));
        }

        fn record_str(&mut self, field: &TracingField, value: &str) {
            self.0.insert(field.name().to_owned(), value.to_owned());
        }

        fn record_u64(&mut self, field: &TracingField, value: u64) {
            self.0.insert(field.name().to_owned(), value.to_string());
        }

        fn record_u128(&mut self, field: &TracingField, value: u128) {
            self.0.insert(field.name().to_owned(), value.to_string());
        }
    }

    fn capture_range_planning_events<T>(
        run: impl FnOnce() -> T,
    ) -> (T, Vec<BTreeMap<String, String>>) {
        let _lock = TRACING_TEST_LOCK
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let events = Arc::new(Mutex::new(Vec::new()));
        let subscriber = EventFields(Arc::clone(&events));
        TRACING_TEST_GLOBAL_SUBSCRIBER.call_once(|| {
            let _ = tracing::subscriber::set_global_default(EventFields::default());
        });
        let result = with_default(subscriber, || {
            tracing::callsite::rebuild_interest_cache();
            run()
        });
        tracing::callsite::rebuild_interest_cache();
        let captured = events
            .lock()
            .map(|events| events.clone())
            .unwrap_or_default();
        (result, captured)
    }

    fn event_field<'event>(
        event: &'event BTreeMap<String, String>,
        field: &str,
    ) -> Option<&'event str> {
        event.get(field).map(String::as_str)
    }

    fn assert_event_fields(event: &BTreeMap<String, String>, expected_fields: &[(&str, &str)]) {
        for (field, expected) in expected_fields {
            assert_eq!(event_field(event, field), Some(*expected), "{field}");
        }
    }

    fn direct_metrics() -> DeltaScanMetrics {
        DeltaScanMetrics::new(DeltaScanMetricsConfig {
            snapshot_version: 1,
            parquet_backend: ParquetReaderBackend::Direct,
            scan_partitions_planned: 1,
            files_planned: 1,
            add_actions_excluded_during_planning: Some(0),
            estimated_input_rows: Some(1),
            estimated_input_bytes: Some(1),
        })
    }

    #[tokio::test]
    async fn network_warmup_bounds_reads_and_handles_cancellation() -> Result<()> {
        use super::super::tests::{GateRequest, GatedObjectStore};

        // Concurrent tests must not consume this calibration's request slots.
        let budget = tokio::sync::Semaphore::new(super::MAX_SHARED_RANGE_READS);
        let memory = Arc::new(InMemory::new());
        let path = Path::from("data.parquet");
        let size = 8 * 1024 * 1024;
        memory.put(&path, Bytes::from(vec![0; size]).into()).await?;
        let ordinary_store = GatedObjectStore::new(
            Arc::new(ThrottledStore::new(
                memory.clone(),
                ThrottleConfig {
                    wait_get_per_call: Duration::from_millis(1),
                    wait_get_per_byte: Duration::from_nanos(20),
                    ..Default::default()
                },
            )),
            GateRequest::Range(0),
        );
        let partial_store = GatedObjectStore::new(
            Arc::new(ThrottledStore::new(
                memory,
                ThrottleConfig {
                    wait_get_per_call: Duration::from_millis(10),
                    wait_get_per_byte: Duration::from_nanos(20),
                    ..Default::default()
                },
            )),
            GateRequest::Range(0),
        );
        let store = MeteredParquetObjectStore::new(
            ordinary_store.clone(),
            direct_metrics(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        let store = store.with_range_read_estimator(Arc::new(
            ParquetRangeReadEstimator::default()
                .for_network_warmup()
                .with_partial_range_store(partial_store.clone()),
        ));
        assert!(!warmup(&store, &[], &budget).await?);
        assert!(!warmup(&store, &[(path.clone(), 1)], &budget).await?);
        assert!(store.range_read_calibration().is_none());
        let files = [(path.clone(), size as u64)];
        assert!(warmup(&store, &files, &budget).await?);
        let counts = store.metrics.snapshot();
        assert_eq!(counts.parquet_data_file_range_get_operations, Some(3_180));
        assert_eq!(counts.parquet_data_file_bytes_received, Some(25_559_040));
        assert_eq!(ordinary_store.range_call_count(), 12);
        assert_eq!(partial_store.range_call_count(), 3_168);
        let profile = store.current_transport_estimate();
        assert_eq!(profile.latency_sample_count, 3);
        assert_eq!(profile.throughput_sample_count, 3);
        let estimate = store.transport_estimate().ok_or_else(|| Error::Generic {
            store: "test",
            source: "missing warmup estimate".into(),
        })?;
        assert!(store.request_overhead_bytes().is_some());
        let calibration = store.range_read_calibration().unwrap();
        let costs = store.fallback_cost_estimator().unwrap();
        assert_eq!(
            calibration.bytes_per_second,
            estimate.shared_throughput_bytes_per_second
        );
        assert_eq!(calibration.ordinary_request_latencies.len(), 12);
        assert_eq!(
            calibration.partial_request_latencies.len(),
            3 * 2 * super::MAX_SHARED_RANGE_READS
        );
        assert_eq!(
            calibration.partial_request_durations.len(),
            calibration.partial_request_latencies.len()
        );
        let small = calibration.small_reads.as_ref().unwrap();
        assert_eq!(small.latencies.len(), 3 * SMALL_READ_CONCURRENCY);
        assert_eq!(small.durations.len(), small.latencies.len());
        assert!(
            small
                .latencies
                .iter()
                .all(|latency| *latency >= Duration::from_millis(10))
        );
        assert!(
            small
                .durations
                .iter()
                .zip(&small.latencies)
                .all(|(duration, latency)| {
                    *duration >= *latency + Duration::from_nanos(20 * 4096)
                })
        );
        assert!(
            calibration
                .ordinary_request_latencies
                .iter()
                .all(|latency| *latency >= Duration::from_millis(1))
        );
        assert!(
            calibration
                .partial_request_latencies
                .iter()
                .all(|latency| *latency >= Duration::from_millis(10))
        );
        assert!(
            calibration
                .partial_request_durations
                .iter()
                .zip(&calibration.partial_request_latencies)
                .all(|(duration, latency)| *duration >= *latency + Duration::from_nanos(20 * 4096))
        );
        // Planning and request overhead both use shared capacity, regardless of
        // how many files are reading. Overlapping plans cannot recalibrate it.
        let overhead = store.request_overhead_bytes().unwrap();
        let next_profile = store.range_read_estimator.for_network_warmup();
        assert!(next_profile.current_transport_estimate().estimate.is_none());
        assert!(next_profile.fallback_cost_estimator().is_none());
        let first = store.range_read_estimator.start_read();
        let second = next_profile.start_read();
        let shared = store.transport_estimate().unwrap();
        assert_eq!(shared, estimate);
        assert!(!first.can_sample());
        assert!(!second.can_sample());
        assert_eq!(overhead, store.request_overhead_bytes().unwrap());
        let before = store.current_transport_estimate();
        let ranges = std::iter::once(0..4096).collect::<Vec<_>>();
        let (_, observed) = store
            .read_partial_ranges(&path, &ranges, &ranges, &tokio::sync::Semaphore::new(1))
            .await?;
        assert!(observed.is_none());
        assert_eq!(store.current_transport_estimate(), before);
        assert_eq!(overhead, store.request_overhead_bytes().unwrap());
        drop(second);
        assert!(
            !first.can_sample(),
            "an overlap remains disqualifying after the other read ends"
        );
        drop(first);
        assert!(store.range_read_estimator.start_read().can_sample());
        assert!(next_profile.start_read().can_sample());
        assert_eq!(store.transport_estimate(), Some(estimate));
        let ranges = vec![0..4096; 2 * super::MAX_SHARED_RANGE_READS];
        store
            .read_partial_ranges(&path, &ranges, &ranges, &budget)
            .await?;
        assert!(Arc::ptr_eq(
            &calibration,
            &store.range_read_calibration().unwrap()
        ));
        let changed = TransportEstimate {
            shared_throughput_bytes_per_second: estimate.shared_throughput_bytes_per_second + 1,
            ..estimate
        };
        for _ in 0..3 {
            store.seed_transport_estimate(changed);
        }
        assert_eq!(store.transport_estimate(), Some(changed));
        assert_eq!(
            store.range_read_calibration().unwrap().bytes_per_second,
            estimate.shared_throughput_bytes_per_second
        );
        assert!(Arc::ptr_eq(
            &costs,
            &store.fallback_cost_estimator().unwrap()
        ));
        // Transport evidence alone is incomplete when all small-request
        // observations overlap another read and must be discarded.
        let contended = MeteredParquetObjectStore::new(
            Arc::clone(&store.inner),
            direct_metrics(),
            MultiRangeReadStrategy::ChooseAutomatically,
        )
        .with_range_read_estimator(Arc::new(
            ParquetRangeReadEstimator::default().for_network_warmup(),
        ));
        contended.seed_transport_estimate(estimate);
        let _peer = contended.range_read_estimator.start_read();
        assert!(!warmup(&contended, &files, &budget).await?);
        assert!(contended.request_overhead_bytes().is_none());
        assert!(contended.range_read_calibration().is_none());
        // Cancelling a stalled warmup must not leave a usable partial profile.
        let stalled = MeteredParquetObjectStore::new(
            Arc::new(ThrottledStore::new(
                InMemory::new(),
                ThrottleConfig {
                    wait_get_per_call: Duration::from_secs(1),
                    ..Default::default()
                },
            )),
            direct_metrics(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        assert!(
            tokio::time::timeout(Duration::from_millis(1), warmup(&stalled, &files, &budget))
                .await
                .is_err()
        );
        assert!(stalled.transport_estimate().is_none());
        assert!(stalled.range_read_calibration().is_none());
        assert!(stalled.range_read_estimator.start_read().can_sample());
        assert!(
            warmup(
                &store,
                &[(Path::from("missing.parquet"), size as u64)],
                &budget
            )
            .await
            .is_err()
        );
        assert!(Arc::ptr_eq(
            &calibration,
            &store.range_read_calibration().unwrap()
        ));
        Ok(())
    }

    #[tokio::test]
    async fn cost_cache_follows_complete_calibration() -> Result<()> {
        use super::super::tests::{GateRequest, GatedObjectStore};

        let memory = Arc::new(InMemory::new());
        let path = Path::from("data.parquet");
        let size = WARMUP_READ_BYTES;
        memory.put(&path, vec![0; size as usize].into()).await?;
        let remote: Arc<dyn ObjectStore> = Arc::new(ThrottledStore::new(
            memory,
            ThrottleConfig {
                wait_get_per_call: Duration::from_millis(1),
                wait_get_per_byte: Duration::from_nanos(20),
                ..Default::default()
            },
        ));
        // Hold the first ordinary request of the second warmup so readers can observe it.
        let ordinary = GatedObjectStore::new(remote.clone(), GateRequest::Range(13));
        let estimator = Arc::new(
            ParquetRangeReadEstimator::default()
                .for_network_warmup()
                .with_partial_range_store(remote),
        );
        let reader = || {
            MeteredParquetObjectStore::new(
                ordinary.clone(),
                direct_metrics(),
                MultiRangeReadStrategy::ChooseAutomatically,
            )
            .with_range_read_estimator(Arc::clone(&estimator))
        };
        let first = reader();
        let second = reader();
        let budget = tokio::sync::Semaphore::new(super::MAX_SHARED_RANGE_READS);
        let files = [(path, size)];
        assert!(first.fallback_cost_estimator().is_none());
        assert!(warmup(&first, &files, &budget).await?);
        let old_costs = first.fallback_cost_estimator().unwrap();
        let old_calibration = first.range_read_calibration().unwrap();
        assert!(Arc::ptr_eq(old_costs.calibration(), &old_calibration));
        assert!(Arc::ptr_eq(
            &old_costs,
            &second.fallback_cost_estimator().unwrap()
        ));
        let counts = first.metrics.snapshot();
        assert_eq!(counts.parquet_data_file_range_get_operations, Some(3_180));
        assert_eq!(counts.parquet_data_file_bytes_received, Some(25_559_040));

        let mut warming = Box::pin(warmup(&first, &files, &budget));
        tokio::time::timeout(Duration::from_secs(5), async {
            tokio::select! {
                _ = ordinary.wait_started() => {}
                result = &mut warming => panic!("warmup finished before cancellation: {result:?}"),
            }
        })
        .await
        .unwrap();
        assert!(Arc::ptr_eq(
            &old_costs,
            &second.fallback_cost_estimator().unwrap()
        ));
        drop(warming);
        assert!(ordinary.was_cancelled());
        assert!(Arc::ptr_eq(
            &old_costs,
            &second.fallback_cost_estimator().unwrap()
        ));
        assert_eq!(budget.available_permits(), super::MAX_SHARED_RANGE_READS);

        assert!(warmup(&first, &files, &budget).await?);
        let new_costs = second.fallback_cost_estimator().unwrap();
        assert!(!Arc::ptr_eq(&old_costs, &new_costs));
        assert!(Arc::ptr_eq(
            &new_costs,
            &first.fallback_cost_estimator().unwrap()
        ));
        assert!(Arc::ptr_eq(
            new_costs.calibration(),
            &second.range_read_calibration().unwrap()
        ));
        assert!(!Arc::ptr_eq(new_costs.calibration(), &old_calibration));
        assert!(Arc::ptr_eq(old_costs.calibration(), &old_calibration));
        // Holding old measurements alone must not keep their discarded result cache alive.
        let released = Arc::downgrade(&old_costs);
        drop(old_costs);
        assert!(released.upgrade().is_none());
        Ok(())
    }

    #[tokio::test]
    async fn interrupted_warmup_discards_completed_request_samples() -> Result<()> {
        use super::super::tests::{GateRequest, GatedObjectStore};

        let memory = Arc::new(InMemory::new());
        let path = Path::from("data.parquet");
        let size = WARMUP_READ_BYTES;
        memory.put(&path, vec![0; size as usize].into()).await?;
        for (interrupted_request, completed_waves) in [
            (2 * SMALL_READ_CONCURRENCY + 1, 0),
            (
                3 * SMALL_READ_CONCURRENCY + 4 * super::MAX_SHARED_RANGE_READS + 1,
                2,
            ),
        ] {
            for fail in [false, true] {
                let partial_store = GatedObjectStore::new(
                    memory.clone(),
                    GateRequest::Range(if fail { 0 } else { interrupted_request }),
                );
                if fail {
                    partial_store.fail_range(interrupted_request);
                }
                let store = MeteredParquetObjectStore::new(
                    memory.clone(),
                    direct_metrics(),
                    MultiRangeReadStrategy::ChooseAutomatically,
                )
                .with_range_read_estimator(Arc::new(
                    ParquetRangeReadEstimator::default()
                        .for_network_warmup()
                        .with_partial_range_store(partial_store.clone()),
                ));
                store.seed_transport_estimate(TransportEstimate {
                    request_latency: Duration::from_millis(200),
                    shared_throughput_bytes_per_second: 20 * 1024 * 1024,
                });
                let budget = tokio::sync::Semaphore::new(super::MAX_SHARED_RANGE_READS);
                let files = [(path.clone(), size)];
                let mut warming = Box::pin(warmup(&store, &files, &budget));
                let interrupted = async {
                    if fail {
                        assert!(warming.as_mut().await.is_err());
                    } else {
                        tokio::select! {
                            _ = partial_store.wait_started() => {}
                            result = &mut warming => {
                                panic!("warmup finished before cancellation: {result:?}");
                            }
                        }
                    }
                };
                tokio::time::timeout(Duration::from_secs(5), interrupted)
                    .await
                    .expect("warmup reaches the interrupted sample round");
                drop(warming);
                if !fail {
                    assert!(partial_store.was_cancelled());
                }
                assert_eq!(
                    store.range_read_estimator.request_overhead_sample_count(),
                    completed_waves
                );
                assert!(store.range_read_calibration().is_none());
                assert_eq!(budget.available_permits(), super::MAX_SHARED_RANGE_READS);
                assert!(store.range_read_estimator.start_read().can_sample());
            }
        }
        Ok(())
    }

    async fn memory_store(metrics: DeltaScanMetrics) -> Result<MeteredParquetObjectStore> {
        memory_store_with_strategy(metrics, MultiRangeReadStrategy::ChooseAutomatically).await
    }

    async fn memory_store_with_strategy(
        metrics: DeltaScanMetrics,
        strategy: MultiRangeReadStrategy,
    ) -> Result<MeteredParquetObjectStore> {
        let inner = Arc::new(InMemory::new());
        inner
            .put(
                &Path::from("data.parquet"),
                PutPayload::from_static(b"0123456789abcdef"),
            )
            .await?;
        Ok(MeteredParquetObjectStore::new(inner, metrics, strategy))
    }

    #[test]
    fn cold_start_completion_reports_plan_shape_without_an_estimate()
    -> std::result::Result<(), Box<dyn std::error::Error>> {
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;
        let (result, events) = capture_range_planning_events(|| {
            runtime.block_on(async {
                memory_store(direct_metrics())
                    .await?
                    .get_ranges(&Path::from("data.parquet"), &[0..4, 8..12])
                    .await
            })
        });
        result?;

        assert_eq!(events.len(), 1);
        let event = &events[0];
        assert_event_fields(
            event,
            &[
                ("event", "automatic_parquet_range_plan_completed"),
                ("latency_sample_count", "0"),
                ("throughput_sample_count", "0"),
                ("estimated_request_latency_micros", "<empty>"),
                ("estimated_shared_throughput_bytes_per_second", "<empty>"),
                ("estimated_bandwidth_delay_bytes", "<empty>"),
                ("exact_range_count", "2"),
                ("exact_bytes", "8"),
                ("exact_request_waves", "1"),
                ("baseline_range_count", "2"),
                ("baseline_bytes", "8"),
                ("baseline_request_waves", "1"),
                ("selected_range_count", "2"),
                ("selected_bytes", "8"),
                ("selected_request_waves", "1"),
                ("baseline_predicted_cost_bytes", "<empty>"),
                ("selected_predicted_cost_bytes", "<empty>"),
                ("decision", "cold_start"),
            ],
        );
        assert_ne!(
            event_field(event, "observed_selected_plan_micros"),
            Some("<empty>")
        );
        Ok(())
    }

    #[test]
    fn warmed_completions_report_the_estimate_and_exact_candidate_scores()
    -> std::result::Result<(), Box<dyn std::error::Error>> {
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;
        let location =
            Path::from("private-bucket/credential-secret/schema-secret/predicate-secret.parquet");
        let requested_ranges = (0..=10)
            .map(|index| {
                let start = index * 1_000;
                start..start + 100
            })
            .collect::<Vec<_>>();

        for (bytes_received, throughput, expected_fields) in [
            (
                1024 * 1024,
                1_000,
                [
                    ("estimated_shared_throughput_bytes_per_second", "1000"),
                    ("estimated_bandwidth_delay_bytes", "100"),
                    ("selected_range_count", "11"),
                    ("selected_bytes", "1100"),
                    ("selected_request_waves", "2"),
                    ("baseline_predicted_cost_bytes", "1300"),
                    ("selected_predicted_cost_bytes", "1300"),
                    ("decision", "cost_based_exact"),
                ],
            ),
            (
                10_000_000,
                1_000_000_000,
                [
                    ("estimated_shared_throughput_bytes_per_second", "1000000000"),
                    ("estimated_bandwidth_delay_bytes", "100000000"),
                    ("selected_range_count", "10"),
                    ("selected_bytes", "2000"),
                    ("selected_request_waves", "1"),
                    ("baseline_predicted_cost_bytes", "200001100"),
                    ("selected_predicted_cost_bytes", "100002000"),
                    ("decision", "cost_based_merged"),
                ],
            ),
        ] {
            let (result, events) = capture_range_planning_events(|| {
                runtime.block_on(async {
                    let inner = Arc::new(InMemory::new());
                    inner
                        .put(&location, PutPayload::from(vec![0_u8; 10_100]))
                        .await?;
                    let store = MeteredParquetObjectStore::new(
                        inner,
                        direct_metrics(),
                        MultiRangeReadStrategy::ChooseAutomatically,
                    );
                    let started = Instant::now();
                    for _ in 0..3 {
                        store.record_completed_range_reads(&[completed_read_at_throughput(
                            started,
                            100,
                            bytes_received,
                            throughput,
                        )]);
                    }
                    store.get_ranges(&location, &requested_ranges).await
                })
            });
            assert!(result?.iter().all(|bytes| bytes.len() == 100));
            assert_eq!(events.len(), 1);
            let event = &events[0];
            assert_event_fields(
                event,
                &[
                    ("latency_sample_count", "3"),
                    ("throughput_sample_count", "3"),
                    ("estimated_request_latency_micros", "100000"),
                    ("exact_range_count", "11"),
                    ("exact_bytes", "1100"),
                    ("exact_request_waves", "2"),
                    ("baseline_range_count", "11"),
                    ("baseline_bytes", "1100"),
                    ("baseline_request_waves", "2"),
                ],
            );
            assert_event_fields(event, &expected_fields);
            assert_ne!(
                event_field(event, "observed_selected_plan_micros"),
                Some("<empty>")
            );

            let diagnostic = format!("{event:?}");
            for private_value in [
                "private-bucket",
                "credential-secret",
                "schema-secret",
                "predicate-secret",
            ] {
                assert!(!diagnostic.contains(private_value));
            }
        }
        Ok(())
    }

    #[test]
    fn range_strategy_follows_table_store_kind()
    -> std::result::Result<(), Box<dyn std::error::Error>> {
        for location in ["file:///tmp/table", "memory:///table", "hdfs://host/table"] {
            assert_eq!(
                MultiRangeReadStrategy::for_table_url(&url::Url::parse(location)?),
                MultiRangeReadStrategy::UseStoreImplementation,
                "{location}"
            );
        }
        for location in [
            "s3://bucket/table",
            "gs://bucket/table",
            "https://example.com/table",
        ] {
            assert_eq!(
                MultiRangeReadStrategy::for_table_url(&url::Url::parse(location)?),
                MultiRangeReadStrategy::ChooseAutomatically,
                "{location}"
            );
        }
        Ok(())
    }

    #[test]
    fn diagnostic_policy_names_map_directly_to_range_strategies()
    -> std::result::Result<(), Box<dyn std::error::Error>> {
        let remote_url = url::Url::parse("https://example.com/table")?;
        for (policy, strategy) in [
            (
                ParquetRangeReadPolicy::Automatic,
                MultiRangeReadStrategy::ChooseAutomatically,
            ),
            (
                ParquetRangeReadPolicy::ExactRanges,
                MultiRangeReadStrategy::ReadExactRanges,
            ),
            (
                ParquetRangeReadPolicy::MergeRangesWithinOneMegabyte,
                MultiRangeReadStrategy::MergeRangesWithinOneMegabyte,
            ),
            (
                ParquetRangeReadPolicy::StoreImplementation,
                MultiRangeReadStrategy::UseStoreImplementation,
            ),
        ] {
            assert_eq!(
                MultiRangeReadStrategy::for_policy(policy, &remote_url),
                strategy
            );
        }
        Ok(())
    }

    #[tokio::test]
    async fn bounded_and_unbounded_gets_record_exact_operations_and_bytes() -> Result<()> {
        let range_metrics = direct_metrics();
        let range_store = memory_store(range_metrics.clone()).await?;
        let bytes = range_store
            .get_range(&Path::from("data.parquet"), 2..7)
            .await?;
        assert_eq!(bytes.as_ref(), b"23456");
        let snapshot = range_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(1));
        assert_eq!(snapshot.parquet_data_file_full_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(5));

        let full_metrics = direct_metrics();
        let full_store = memory_store(full_metrics.clone()).await?;
        let bytes = full_store
            .get(&Path::from("data.parquet"))
            .await?
            .bytes()
            .await?;
        assert_eq!(bytes.as_ref(), b"0123456789abcdef");
        let snapshot = full_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_full_get_operations, Some(1));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(16));
        Ok(())
    }

    #[tokio::test]
    async fn head_and_failed_gets_preserve_attempt_metrics() -> Result<()> {
        let head_metrics = direct_metrics();
        let head_store = memory_store(head_metrics.clone()).await?;
        let options = GetOptions::new().with_range(Some(1_u64..4)).with_head(true);
        let _result = head_store
            .get_opts(&Path::from("data.parquet"), options)
            .await?;
        assert!(head_store.range_read_estimator.start_read().can_sample());
        let snapshot = head_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_full_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(0));

        let failure_metrics = direct_metrics();
        let failure_store = MeteredParquetObjectStore::new(
            Arc::new(InMemory::new()),
            failure_metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        let result = failure_store
            .get_opts(
                &Path::from("missing.parquet"),
                GetOptions::new().with_range(Some(0_u64..4)),
            )
            .await;
        assert!(result.is_err());
        assert!(failure_store.range_read_estimator.start_read().can_sample());
        let snapshot = failure_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(1));
        assert_eq!(snapshot.parquet_data_file_full_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(0));

        Ok(())
    }

    #[tokio::test]
    async fn multi_range_strategies_preserve_results_and_expose_metrics() -> Result<()> {
        let automatic_metrics = direct_metrics();
        let automatic_store = memory_store(automatic_metrics.clone()).await?;
        let bytes = automatic_store
            .get_ranges(&Path::from("data.parquet"), &[0..4, 8..12])
            .await?;
        assert_eq!(bytes[0].as_ref(), b"0123");
        assert_eq!(bytes[1].as_ref(), b"89ab");
        let snapshot = automatic_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_exact_ranges_requested, Some(2));
        assert_eq!(
            snapshot.parquet_data_file_exact_range_bytes_requested,
            Some(8)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_requests_planned,
            Some(2)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_bytes_planned,
            Some(8)
        );
        assert_eq!(snapshot.parquet_data_file_cold_start_range_plans, Some(1));
        assert_eq!(
            snapshot.parquet_data_file_cost_based_exact_range_plans,
            Some(0)
        );
        assert_eq!(
            snapshot.parquet_data_file_cost_based_merged_range_plans,
            Some(0)
        );
        assert_eq!(
            snapshot.parquet_data_file_store_delegated_range_plans,
            Some(0)
        );
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(2));
        assert_eq!(snapshot.parquet_data_file_full_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(8));
        let diagnostic = automatic_metrics.parquet_range_planning_diagnostic_snapshot();
        assert_eq!(diagnostic.max_concurrent_physical_range_requests, 10);
        assert_eq!(diagnostic.physical_range_request_waves_planned, 1);

        let delegated_metrics = direct_metrics();
        let delegated_store = memory_store_with_strategy(
            delegated_metrics.clone(),
            MultiRangeReadStrategy::UseStoreImplementation,
        )
        .await?;
        let bytes = delegated_store
            .get_ranges(&Path::from("data.parquet"), &[0..4, 8..12])
            .await?;
        assert_eq!(bytes[0].as_ref(), b"0123");
        assert_eq!(bytes[1].as_ref(), b"89ab");
        let snapshot = delegated_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_exact_ranges_requested, Some(2));
        assert_eq!(
            snapshot.parquet_data_file_exact_range_bytes_requested,
            Some(8)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_requests_planned,
            Some(0)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_bytes_planned,
            Some(0)
        );
        assert_eq!(snapshot.parquet_data_file_cold_start_range_plans, Some(0));
        assert_eq!(
            snapshot.parquet_data_file_cost_based_exact_range_plans,
            Some(0)
        );
        assert_eq!(
            snapshot.parquet_data_file_cost_based_merged_range_plans,
            Some(0)
        );
        assert_eq!(
            snapshot.parquet_data_file_store_delegated_range_plans,
            Some(1)
        );
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(1));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(8));
        assert_eq!(
            delegated_metrics
                .parquet_range_planning_diagnostic_snapshot()
                .physical_range_request_waves_planned,
            0
        );
        Ok(())
    }

    #[tokio::test]
    async fn diagnostic_range_strategies_preserve_bytes_and_apply_their_plans() -> Result<()> {
        let requested_ranges = [10..14, 0..4, 2..6, 10..14];
        let expected = [b"abcd".as_slice(), b"0123", b"2345", b"abcd"];

        for (strategy, expected_planned_requests, expected_planned_bytes) in [
            (MultiRangeReadStrategy::ReadExactRanges, 2, 10),
            (MultiRangeReadStrategy::MergeRangesWithinOneMegabyte, 1, 14),
            (MultiRangeReadStrategy::ChooseAutomatically, 2, 10),
            (MultiRangeReadStrategy::UseStoreImplementation, 0, 0),
        ] {
            let metrics = direct_metrics();
            let store = memory_store_with_strategy(metrics.clone(), strategy).await?;
            let actual = store
                .get_ranges(&Path::from("data.parquet"), &requested_ranges)
                .await?;
            assert_eq!(
                actual.iter().map(Bytes::as_ref).collect::<Vec<_>>(),
                expected,
                "{strategy:?}"
            );
            let snapshot = metrics.snapshot();
            assert_eq!(
                snapshot.parquet_data_file_physical_range_requests_planned,
                Some(expected_planned_requests),
                "{strategy:?}"
            );
            assert_eq!(
                snapshot.parquet_data_file_physical_range_bytes_planned,
                Some(expected_planned_bytes),
                "{strategy:?}"
            );
        }
        Ok(())
    }

    #[tokio::test]
    async fn partial_reads_reuse_released_capacity_and_cancel_queued_requests() -> Result<()> {
        use super::super::tests::{GateRequest, GatedObjectStore};
        use futures_util::poll;
        use tokio::sync::Semaphore;

        let inner = Arc::new(InMemory::new());
        let path = Path::from("data.parquet");
        inner
            .put(&path, Bytes::from_static(b"01234567").into())
            .await?;
        let gated = GatedObjectStore::new(inner, GateRequest::Range(1));
        let ordinary = Arc::new(InMemory::new());
        ordinary
            .put(&path, Bytes::from_static(b"ordinary").into())
            .await?;
        let metrics = direct_metrics();
        let store = MeteredParquetObjectStore::new(
            ordinary,
            metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        )
        .with_range_read_estimator(Arc::new(
            ParquetRangeReadEstimator::default()
                .for_network_warmup()
                .with_partial_range_store(gated.clone()),
        ));
        let ranges = [0..1, 2..3, 4..5, 6..7];
        let budget = Semaphore::new(4);
        let other_reads = budget.try_acquire_many(3).expect("free capacity");
        let mut reading = Box::pin(store.read_partial_ranges(&path, &ranges, &ranges, &budget));
        assert!(poll!(&mut reading).is_pending());
        assert_eq!(budget.available_permits(), 0);
        assert_eq!(
            metrics.snapshot().parquet_data_file_range_get_operations,
            Some(1)
        );

        // The first request stays blocked. The other three must start as soon
        // as capacity returns, without restarting or replanning the round.
        drop(other_reads);
        assert!(poll!(&mut reading).is_pending());
        assert_eq!(
            metrics.snapshot().parquet_data_file_range_get_operations,
            Some(4)
        );
        assert_eq!(
            budget.available_permits(),
            3,
            "completed requests release their slots"
        );
        drop(reading);
        assert!(gated.was_cancelled());
        assert_eq!(budget.available_permits(), 4);

        // Cancelling a plan queued behind other reads must also remove its waiters.
        let held = budget.try_acquire_many(4).expect("all capacity returned");
        let mut queued = Box::pin(store.read_partial_ranges(&path, &ranges, &ranges, &budget));
        assert!(poll!(&mut queued).is_pending());
        drop(queued);
        drop(held);
        assert_eq!(budget.available_permits(), 4);
        assert_eq!(
            metrics.snapshot().parquet_data_file_range_get_operations,
            Some(4)
        );
        // Ordinary reads keep their pooled transport; partial reads keep the
        // transport whose costs were measured, including after cancellation.
        assert_eq!(
            store.get_range(&path, 0..1).await?,
            Bytes::from_static(b"o")
        );
        let ranges = std::iter::once(2..3).collect::<Vec<_>>();
        let (bytes, _) = store
            .read_partial_ranges(&path, &ranges, &ranges, &budget)
            .await?;
        assert_eq!(bytes, [Bytes::from_static(b"2")]);
        Ok(())
    }

    #[tokio::test]
    async fn partial_reads_report_requests_bytes_and_concurrency_limit() -> Result<()> {
        let inner = Arc::new(InMemory::new());
        let path = Path::from("data.parquet");
        inner.put(&path, vec![42_u8; 64].into()).await?;
        let metrics = direct_metrics();
        let store = MeteredParquetObjectStore::new(
            inner,
            metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        let ranges: Vec<_> = (0..32).map(|i| i * 2..i * 2 + 1).collect();
        let budget = tokio::sync::Semaphore::new(32);
        let (data, _) = store
            .read_partial_ranges(&path, &ranges, &ranges, &budget)
            .await?;
        assert_eq!(data, vec![Bytes::from_static(&[42]); 32]);
        let snapshot = metrics.snapshot();
        assert_eq!(
            snapshot.parquet_data_file_physical_range_requests_planned,
            Some(32)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_bytes_planned,
            Some(32)
        );
        let diagnostic = metrics.parquet_range_planning_diagnostic_snapshot();
        assert_eq!(diagnostic.max_concurrent_physical_range_requests, 32);
        assert_eq!(diagnostic.physical_range_request_waves_planned, 1);
        // The cache must receive the complete physical buffer, while requested
        // byte metrics retain the smaller logical ranges before gap merging.
        let (data, _) = store
            .read_partial_ranges(&path, &[0..1, 3..4], std::slice::from_ref(&(0..4)), &budget)
            .await?;
        assert_eq!(data, vec![Bytes::from_static(&[42; 4])]);
        let snapshot = metrics.snapshot();
        assert_eq!(
            snapshot.parquet_data_file_exact_range_bytes_requested,
            Some(34)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_bytes_planned,
            Some(36)
        );
        Ok(())
    }

    #[test]
    fn chosen_range_plan_metrics_distinguish_decisions() {
        let metrics = direct_metrics();
        let store = MeteredParquetObjectStore::new(
            Arc::new(InMemory::new()),
            metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        for (physical_ranges, planned_bytes, decision) in [
            (vec![0..2, 4..6, 8..10], 6, RangePlanDecision::ColdStart),
            (
                vec![0..2, 4..6, 8..10],
                6,
                RangePlanDecision::CostBasedExact,
            ),
            (vec![0..6, 8..10], 8, RangePlanDecision::CostBasedMerged),
        ] {
            store.record_chosen_range_plan_metrics(&ChosenRangePlan {
                exact_range_count: 3,
                exact_bytes: 6,
                baseline_range_count: 3,
                baseline_bytes: 6,
                physical_ranges,
                planned_bytes,
                transport_estimate: None,
                baseline_predicted_cost_bytes: None,
                selected_predicted_cost_bytes: None,
                decision,
            });
        }

        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_exact_ranges_requested, Some(9));
        assert_eq!(
            snapshot.parquet_data_file_exact_range_bytes_requested,
            Some(18)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_requests_planned,
            Some(8)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_bytes_planned,
            Some(20)
        );
        assert_eq!(snapshot.parquet_data_file_cold_start_range_plans, Some(1));
        assert_eq!(
            snapshot.parquet_data_file_cost_based_exact_range_plans,
            Some(1)
        );
        assert_eq!(
            snapshot.parquet_data_file_cost_based_merged_range_plans,
            Some(1)
        );
    }

    #[tokio::test]
    async fn separate_stores_share_only_an_explicitly_reused_estimator() -> Result<()> {
        let inner = Arc::new(InMemory::new());
        inner
            .put(
                &Path::from("data.parquet"),
                PutPayload::from_static(b"0123456789abcdef"),
            )
            .await?;
        let estimator = Arc::new(ParquetRangeReadEstimator::default());
        let first = MeteredParquetObjectStore::new(
            inner.clone(),
            direct_metrics(),
            MultiRangeReadStrategy::ChooseAutomatically,
        )
        .with_range_read_estimator(Arc::clone(&estimator));
        let second = MeteredParquetObjectStore::new(
            inner.clone(),
            direct_metrics(),
            MultiRangeReadStrategy::ChooseAutomatically,
        )
        .with_range_read_estimator(Arc::clone(&estimator));
        let started = Instant::now();

        first.record_completed_range_reads(&[completed_read(started, 10, 1024 * 1024)]);
        first.record_completed_range_reads(&[completed_read(started, 20, 2 * 1024 * 1024)]);
        assert_eq!(
            first.current_transport_estimate(),
            CurrentTransportEstimate {
                latency_sample_count: 2,
                throughput_sample_count: 2,
                estimate: None,
            }
        );
        second.record_completed_range_reads(&[completed_read(started, 30, 3 * 1024 * 1024)]);
        assert!(second.current_transport_estimate().estimate.is_some());

        let warm_metrics = direct_metrics();
        let warm = MeteredParquetObjectStore::new(
            inner.clone(),
            warm_metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        )
        .with_range_read_estimator(estimator);
        let ranges = [0..4, 8..12];
        let warm_bytes = warm
            .get_ranges(&Path::from("data.parquet"), &ranges)
            .await?;
        let snapshot = warm_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_cold_start_range_plans, Some(0));
        assert_eq!(
            snapshot
                .parquet_data_file_cost_based_exact_range_plans
                .unwrap_or_default()
                + snapshot
                    .parquet_data_file_cost_based_merged_range_plans
                    .unwrap_or_default(),
            1
        );

        let cold_metrics = direct_metrics();
        let cold = MeteredParquetObjectStore::new(
            inner,
            cold_metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        let cold_bytes = cold
            .get_ranges(&Path::from("data.parquet"), &ranges)
            .await?;
        assert_eq!(cold_bytes, warm_bytes);
        assert_eq!(
            cold_metrics
                .snapshot()
                .parquet_data_file_cold_start_range_plans,
            Some(1)
        );
        Ok(())
    }

    #[tokio::test]
    async fn successful_tiny_plans_update_only_latency_evidence() -> Result<()> {
        let inner = InMemory::new();
        inner
            .put(
                &Path::from("data.parquet"),
                PutPayload::from_static(b"0123456789abcdef"),
            )
            .await?;
        let throttled = ThrottledStore::new(
            inner,
            ThrottleConfig {
                wait_get_per_call: Duration::from_millis(1),
                wait_get_per_byte: Duration::from_micros(10),
                ..Default::default()
            },
        );
        let store = MeteredParquetObjectStore::new(
            Arc::new(throttled),
            direct_metrics(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        for _ in 0..3 {
            let results = store
                .get_ranges(&Path::from("data.parquet"), &[0..4, 8..12])
                .await?;
            assert_eq!(results[0].as_ref(), b"0123");
            assert_eq!(results[1].as_ref(), b"89ab");
        }
        assert_eq!(
            store.current_transport_estimate(),
            CurrentTransportEstimate {
                latency_sample_count: 3,
                throughput_sample_count: 0,
                estimate: None,
            }
        );

        let failed_metrics = direct_metrics();
        let failed_store = MeteredParquetObjectStore::new(
            Arc::new(InMemory::new()),
            failed_metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        let started = Instant::now();
        for _ in 0..2 {
            failed_store.record_completed_range_reads(&[completed_read(started, 10, 1_000)]);
        }
        assert!(
            failed_store
                .get_ranges(&Path::from("missing.parquet"), &[0..4, 8..12])
                .await
                .is_err()
        );
        assert_eq!(
            failed_store.current_transport_estimate(),
            CurrentTransportEstimate {
                latency_sample_count: 2,
                throughput_sample_count: 0,
                estimate: None,
            }
        );
        let snapshot = failed_metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_exact_ranges_requested, Some(2));
        assert_eq!(
            snapshot.parquet_data_file_exact_range_bytes_requested,
            Some(8)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_requests_planned,
            Some(2)
        );
        assert_eq!(
            snapshot.parquet_data_file_physical_range_bytes_planned,
            Some(8)
        );
        assert_eq!(snapshot.parquet_data_file_cold_start_range_plans, Some(1));
        Ok(())
    }

    #[test]
    fn failed_automatic_plan_emits_no_completion_and_adds_no_sample() {
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("test runtime");
        let ((result, estimate), events) = capture_range_planning_events(|| {
            runtime.block_on(async {
                let store = MeteredParquetObjectStore::new(
                    Arc::new(InMemory::new()),
                    direct_metrics(),
                    MultiRangeReadStrategy::ChooseAutomatically,
                );
                let started = Instant::now();
                for _ in 0..2 {
                    store.record_completed_range_reads(&[completed_read(started, 10, 1_000)]);
                }
                let result = store
                    .get_ranges(&Path::from("private-missing.parquet"), &[0..4, 8..12])
                    .await;
                (result, store.current_transport_estimate())
            })
        });

        assert!(result.is_err());
        assert_eq!(
            estimate,
            CurrentTransportEstimate {
                latency_sample_count: 2,
                throughput_sample_count: 0,
                estimate: None,
            }
        );
        assert!(events.is_empty());
    }

    #[test]
    fn cancelled_partial_automatic_plan_emits_no_completion_and_adds_no_sample()
    -> std::result::Result<(), Box<dyn std::error::Error>> {
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;
        let (result, events) = capture_range_planning_events(|| {
            runtime.block_on(async {
                let first = PutPayload::from_static(b"abc")
                    .into_iter()
                    .next()
                    .ok_or_else(missing_chunk)?;
                let payload = stream::iter([Ok(first)])
                    .chain(stream::pending::<Result<Bytes>>())
                    .boxed();
                let metrics = direct_metrics();
                let store = Arc::new(MeteredParquetObjectStore::new(
                    Arc::new(ScriptedGetStore::new(test_get_result(
                        GetResultPayload::Stream(payload),
                        0..8,
                    ))),
                    metrics.clone(),
                    MultiRangeReadStrategy::ChooseAutomatically,
                ));
                let started = Instant::now();
                for _ in 0..2 {
                    store.record_completed_range_reads(&[completed_read(started, 10, 1_000)]);
                }

                let task_store = Arc::clone(&store);
                let task = tokio::spawn(async move {
                    task_store
                        .get_ranges(&Path::from("data.parquet"), &[0..4, 4..8])
                        .await
                });
                tokio::time::timeout(Duration::from_secs(5), async {
                    while metrics.snapshot().parquet_data_file_bytes_received != Some(3) {
                        tokio::task::yield_now().await;
                    }
                })
                .await?;

                task.abort();
                assert!(task.await.is_err_and(|error| error.is_cancelled()));
                assert_eq!(
                    store.current_transport_estimate(),
                    CurrentTransportEstimate {
                        latency_sample_count: 2,
                        throughput_sample_count: 0,
                        estimate: None,
                    }
                );
                Ok::<_, Box<dyn std::error::Error>>(())
            })
        });
        result?;
        assert!(events.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn truncated_automatic_plan_does_not_update_transport_estimates()
    -> std::result::Result<(), Box<dyn std::error::Error>> {
        let truncated = PutPayload::from_static(b"abc")
            .into_iter()
            .next()
            .ok_or_else(missing_chunk)?;
        let metrics = direct_metrics();
        let store = MeteredParquetObjectStore::new(
            Arc::new(ScriptedGetStore::new(test_get_result(
                GetResultPayload::Stream(stream::iter([Ok(truncated)]).boxed()),
                0..8,
            ))),
            metrics.clone(),
            MultiRangeReadStrategy::ChooseAutomatically,
        );
        let started = Instant::now();
        for _ in 0..2 {
            store.record_completed_range_reads(&[completed_read(started, 10, 1_000)]);
        }

        let error = store
            .get_ranges(&Path::from("data.parquet"), &[0..4, 4..8])
            .await
            .expect_err("truncated range unexpectedly succeeded");
        assert!(error.to_string().contains("unexpected length"));
        assert_eq!(metrics.snapshot().parquet_data_file_bytes_received, Some(3));
        assert_eq!(
            store.current_transport_estimate(),
            CurrentTransportEstimate {
                latency_sample_count: 2,
                throughput_sample_count: 0,
                estimate: None,
            }
        );
        Ok(())
    }

    #[tokio::test]
    async fn store_multi_range_handles_empty_and_failed_calls() -> Result<()> {
        let metrics = direct_metrics();
        let store = MeteredParquetObjectStore::new(
            Arc::new(InMemory::new()),
            metrics.clone(),
            MultiRangeReadStrategy::UseStoreImplementation,
        );

        assert!(
            store
                .get_ranges(&Path::from("missing.parquet"), &[])
                .await?
                .is_empty()
        );
        assert_eq!(
            metrics.snapshot().parquet_data_file_range_get_operations,
            Some(0)
        );
        assert_eq!(
            metrics
                .snapshot()
                .parquet_data_file_store_delegated_range_plans,
            Some(0)
        );

        assert!(
            store
                .get_ranges(&Path::from("missing.parquet"), &[0..4, 8..12])
                .await
                .is_err()
        );
        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_exact_ranges_requested, Some(2));
        assert_eq!(
            snapshot.parquet_data_file_exact_range_bytes_requested,
            Some(8)
        );
        assert_eq!(
            snapshot.parquet_data_file_store_delegated_range_plans,
            Some(1)
        );
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(1));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(0));
        Ok(())
    }

    #[tokio::test]
    async fn stream_delivery_records_only_successful_consumed_chunks() -> Result<()> {
        let first = PutPayload::from_static(b"abc")
            .into_iter()
            .next()
            .ok_or_else(missing_chunk)?;
        let second = PutPayload::from_static(b"defgh")
            .into_iter()
            .next()
            .ok_or_else(missing_chunk)?;
        let success_metrics = direct_metrics();
        let success_store = scripted_store(
            test_get_result(
                GetResultPayload::Stream(stream::iter(vec![Ok(first), Ok(second)]).boxed()),
                0..8,
            ),
            success_metrics.clone(),
        );
        let result = success_store.get(&Path::from("data.parquet")).await?;
        assert_eq!(result.meta.location, Path::from("data.parquet"));
        assert_eq!(result.meta.e_tag.as_deref(), Some("opaque-etag"));
        assert_eq!(result.range, 0..8);
        assert!(!success_store.range_read_estimator.start_read().can_sample());
        let mut payload = result.into_stream();
        let mut bytes = Vec::new();
        while let Some(chunk) = payload.next().await {
            bytes.extend_from_slice(&chunk?);
        }
        assert_eq!(bytes, b"abcdefgh");
        assert!(success_store.range_read_estimator.start_read().can_sample());
        assert_eq!(
            success_metrics.snapshot().parquet_data_file_bytes_received,
            Some(8)
        );

        let dropped_metrics = direct_metrics();
        let chunks = PutPayload::from_static(b"abc")
            .into_iter()
            .chain(PutPayload::from_static(b"defgh").into_iter())
            .map(Ok)
            .collect::<Vec<_>>();
        let dropped_store = scripted_store(
            test_get_result(GetResultPayload::Stream(stream::iter(chunks).boxed()), 0..8),
            dropped_metrics.clone(),
        );
        let mut payload = dropped_store
            .get(&Path::from("data.parquet"))
            .await?
            .into_stream();
        assert_eq!(
            payload
                .next()
                .await
                .transpose()?
                .ok_or_else(missing_chunk)?
                .as_ref(),
            b"abc"
        );
        assert!(!dropped_store.range_read_estimator.start_read().can_sample());
        drop(payload);
        assert!(dropped_store.range_read_estimator.start_read().can_sample());
        assert_eq!(
            dropped_metrics.snapshot().parquet_data_file_bytes_received,
            Some(3)
        );

        let error_metrics = direct_metrics();
        let error = Error::Generic {
            store: "test",
            source: io::Error::other("payload failure").into(),
        };
        let successful = PutPayload::from_static(b"abc")
            .into_iter()
            .next()
            .ok_or_else(missing_chunk)?;
        let error_store = scripted_store(
            test_get_result(
                GetResultPayload::Stream(stream::iter(vec![Ok(successful), Err(error)]).boxed()),
                0..8,
            ),
            error_metrics.clone(),
        );
        let mut payload = error_store
            .get(&Path::from("data.parquet"))
            .await?
            .into_stream();
        assert!(payload.next().await.transpose()?.is_some());
        assert!(!error_store.range_read_estimator.start_read().can_sample());
        assert!(payload.next().await.transpose().is_err());
        assert!(error_store.range_read_estimator.start_read().can_sample());
        assert!(payload.next().await.is_none());
        assert_eq!(
            error_metrics.snapshot().parquet_data_file_bytes_received,
            Some(3)
        );
        Ok(())
    }

    #[tokio::test]
    async fn cancelling_a_pending_body_releases_read_activity() -> Result<()> {
        let (retained, mut released) = tokio::sync::oneshot::channel::<()>();
        let payload = stream::once(async move {
            let _retained = retained;
            futures_util::future::pending::<Result<Bytes>>().await
        })
        .boxed();
        let store = scripted_store(
            test_get_result(GetResultPayload::Stream(payload), 0..8),
            direct_metrics(),
        );
        let result = store.get(&Path::from("data.parquet")).await?;
        let mut body = Box::pin(result.bytes());
        assert!(futures_util::poll!(&mut body).is_pending());
        assert!(!store.range_read_estimator.start_read().can_sample());
        drop(body);
        assert!(store.range_read_estimator.start_read().can_sample());
        assert_eq!(
            released.try_recv(),
            Err(tokio::sync::oneshot::error::TryRecvError::Closed)
        );
        Ok(())
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn local_file_payload_stays_lazy_and_yields_one_large_chunk()
    -> std::result::Result<(), Box<dyn std::error::Error>> {
        let file = TemporaryTestFile::new(&vec![7_u8; 20_000])?;
        let range = 100_u64..16_500;

        let dropped_metrics = direct_metrics();
        let dropped_store =
            scripted_store(file.get_result(range.clone())?, dropped_metrics.clone());
        let result = dropped_store
            .get_opts(
                &Path::from("data.parquet"),
                GetOptions::new().with_range(Some(range.clone())),
            )
            .await?;
        assert!(!dropped_store.range_read_estimator.start_read().can_sample());
        drop(result);
        assert!(dropped_store.range_read_estimator.start_read().can_sample());
        assert_eq!(
            dropped_metrics.snapshot().parquet_data_file_bytes_received,
            Some(0)
        );

        let delivered_metrics = direct_metrics();
        let delivered_store =
            scripted_store(file.get_result(range.clone())?, delivered_metrics.clone());
        let mut payload = delivered_store
            .get_opts(
                &Path::from("data.parquet"),
                GetOptions::new().with_range(Some(range.clone())),
            )
            .await?
            .into_stream();
        let bytes = payload
            .next()
            .await
            .transpose()?
            .ok_or_else(missing_chunk)?;
        assert_eq!(bytes.len(), usize::try_from(range.end - range.start)?);
        assert!(payload.next().await.is_none());
        assert!(
            delivered_store
                .range_read_estimator
                .start_read()
                .can_sample()
        );
        assert_eq!(
            delivered_metrics
                .snapshot()
                .parquet_data_file_bytes_received,
            Some(range.end - range.start)
        );
        Ok(())
    }

    #[tokio::test]
    async fn delegated_operations_and_diagnostics_do_not_leak_or_meter() -> Result<()> {
        let metrics = direct_metrics();
        let store = MeteredParquetObjectStore::new(
            Arc::new(InMemory::new()),
            metrics.clone(),
            MultiRangeReadStrategy::UseStoreImplementation,
        );
        let first = Path::from("first.parquet");
        let second = Path::from("second.parquet");
        let third = Path::from("third.parquet");

        store.put(&first, PutPayload::from_static(b"data")).await?;
        assert!(store.list(None).next().await.transpose()?.is_some());
        store.copy(&first, &second).await?;
        store.rename(&second, &third).await?;
        store.delete(&first).await?;
        store.delete(&third).await?;
        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.parquet_data_file_range_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_full_get_operations, Some(0));
        assert_eq!(snapshot.parquet_data_file_bytes_received, Some(0));
        assert_eq!(snapshot.estimated_parquet_task_bytes_admitted, Some(0));
        assert_eq!(format!("{store:?}"), "MeteredParquetObjectStore");

        let redacted_metrics = direct_metrics();
        let redacted_store = scripted_store(
            test_get_result(GetResultPayload::Stream(stream::empty().boxed()), 0..0),
            redacted_metrics.clone(),
        );
        let location = Path::from("private/user-password-secret-token.parquet");
        let mut options = GetOptions::new().with_range(Some(987_654_321_u64..987_654_999_u64));
        options.if_match = Some("secret-conditional-header".to_owned());
        options.version = Some("secret-object-version".to_owned());
        let _result = redacted_store.get_opts(&location, options).await?;
        let diagnostics = format!(
            "{redacted_store:?} {redacted_store} {:?}",
            redacted_metrics.snapshot()
        );
        for secret in [
            "user-password-secret-token",
            "secret-conditional-header",
            "secret-object-version",
            "987654321",
            "987654999",
        ] {
            assert!(!diagnostics.contains(secret));
        }
        Ok(())
    }

    fn missing_chunk() -> Error {
        Error::Generic {
            store: "test",
            source: io::Error::other("missing test chunk").into(),
        }
    }

    fn scripted_store(result: GetResult, metrics: DeltaScanMetrics) -> MeteredParquetObjectStore {
        MeteredParquetObjectStore::new(
            Arc::new(ScriptedGetStore::new(result)),
            metrics,
            MultiRangeReadStrategy::UseStoreImplementation,
        )
    }

    fn test_get_result(payload: GetResultPayload, range: Range<u64>) -> GetResult {
        GetResult {
            payload,
            meta: ObjectMeta {
                location: Path::from("data.parquet"),
                last_modified: DateTime::<Utc>::UNIX_EPOCH,
                size: range.end,
                e_tag: Some("opaque-etag".to_owned()),
                version: Some("opaque-version".to_owned()),
            },
            range,
            attributes: Attributes::new(),
        }
    }

    struct ScriptedGetStore {
        result: Mutex<Option<GetResult>>,
        delegate: InMemory,
    }

    impl ScriptedGetStore {
        fn new(result: GetResult) -> Self {
            Self {
                result: Mutex::new(Some(result)),
                delegate: InMemory::new(),
            }
        }
    }

    impl fmt::Debug for ScriptedGetStore {
        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            formatter.write_str("ScriptedGetStore")
        }
    }

    impl fmt::Display for ScriptedGetStore {
        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            formatter.write_str("ScriptedGetStore")
        }
    }

    #[async_trait]
    impl ObjectStore for ScriptedGetStore {
        async fn put_opts(
            &self,
            location: &Path,
            payload: PutPayload,
            options: PutOptions,
        ) -> Result<PutResult> {
            self.delegate.put_opts(location, payload, options).await
        }

        async fn put_multipart_opts(
            &self,
            location: &Path,
            options: PutMultipartOptions,
        ) -> Result<Box<dyn MultipartUpload>> {
            self.delegate.put_multipart_opts(location, options).await
        }

        async fn get_opts(&self, _location: &Path, _options: GetOptions) -> Result<GetResult> {
            self.result
                .lock()
                .map_err(|_| missing_chunk())?
                .take()
                .ok_or_else(missing_chunk)
        }

        fn delete_stream(
            &self,
            locations: BoxStream<'static, Result<Path>>,
        ) -> BoxStream<'static, Result<Path>> {
            self.delegate.delete_stream(locations)
        }

        fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
            self.delegate.list(prefix)
        }

        fn list_with_offset(
            &self,
            prefix: Option<&Path>,
            offset: &Path,
        ) -> BoxStream<'static, Result<ObjectMeta>> {
            self.delegate.list_with_offset(prefix, offset)
        }

        async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
            self.delegate.list_with_delimiter(prefix).await
        }

        async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
            self.delegate.copy_opts(from, to, options).await
        }

        async fn rename_opts(&self, from: &Path, to: &Path, options: RenameOptions) -> Result<()> {
            self.delegate.rename_opts(from, to, options).await
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    struct TemporaryTestFile {
        path: PathBuf,
    }

    #[cfg(not(target_arch = "wasm32"))]
    impl TemporaryTestFile {
        fn new(contents: &[u8]) -> io::Result<Self> {
            static NEXT_FILE_ID: AtomicU64 = AtomicU64::new(0);
            let id = NEXT_FILE_ID.fetch_add(1, Ordering::Relaxed);
            let path = std::env::temp_dir().join(format!(
                "delta-arrow-reader-metered-object-store-{}-{id}",
                std::process::id()
            ));
            std::fs::write(&path, contents)?;
            Ok(Self { path })
        }

        fn get_result(&self, range: Range<u64>) -> io::Result<GetResult> {
            Ok(test_get_result(
                GetResultPayload::File(File::open(&self.path)?, self.path.clone()),
                range,
            ))
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    impl Drop for TemporaryTestFile {
        fn drop(&mut self) {
            let _ = std::fs::remove_file(&self.path);
        }
    }
}