spiral-sdk 0.1.9

//! SDK table execution over any scandal.
//!
//! The SDK host resolves exact versions and registers prepared versions. Resolver-enabled
//! native hosts receive logical resource roots and obtain scoped storage internally. Explicit
//! transports keep the SDK binding cache. Each host and cache belongs to one caller.

use std::collections::{HashMap, VecDeque};
use std::mem::take;
use std::ops::Range;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use arrow_array::{
    Array, ArrayRef, BinaryArray, RecordBatch, RecordBatchReader, StringArray, StructArray,
    UInt64Array, new_null_array,
};
use arrow_schema::{DataType, Schema, SchemaRef};
use arrow_select::concat::concat;
use arrow_select::interleave::interleave;
use async_trait::async_trait;
use bytes::{Bytes, BytesMut};
use futures::future::BoxFuture;
use futures::stream::{BoxStream, FuturesOrdered};
use futures::{FutureExt as _, StreamExt as _, TryStreamExt as _, future, stream};
use parking_lot::Mutex;
use reqwest::Url;
use tokio::sync::{Mutex as AsyncMutex, OnceCell, mpsc};
use uuid::Uuid;

use crate::api::tables::objects::TableId;
use crate::api::tables::{
    CreateTableRequest, Empty, SignedAccess, StorageAccess, StorageAccessRequest,
    StorageBinding as TablesBinding, StorageMode, WriteAttempt, WriteScope,
};
use crate::auth::Token;
use crate::registration::register;
use crate::resources::arbor::Client as ArborClient;
use crate::scan::{
    ScanInfo, ScanSource, ScanSplit, SplitFilter, Take, WorkBatch, WorkItem, WorkStream,
};
use crate::scandal::api::{
    AppendRowsRequest, AttachColumnsRequest, AttachedColumns, ChangePlan, DeleteBase,
    DeleteRowsRequest, EditBase, PatchRowsRequest, PlanChangesRequest, RowIdRange, SchemaChange,
};
use crate::scandal::blob::{BlobPart, BlobUpload};
use crate::scandal::emit::{
    Aggregate, AggregateFunction, BoundEmit, CountAggregate, Emit, NamedAggregate, ROWS_COLUMN,
};
use crate::scandal::error::ScandalError;
use crate::scandal::ordered::merge_ordered;
use crate::scandal::types::{
    FragmentReceipt, MAX_TAKE_POSITIONS, MAX_TAKES, PlannedSplit, PreparedWriteReceipt, RowRange,
    Scan as ScandalScan, ScanMetricsSnapshot, ScanOptions, Split, SplitTake, TableVersion,
    TableWriteResult,
};
use crate::scandal::{
    RecordBatchStream, Scandal, SplitBatch, StorageBinding, TableBinding, WriteBase,
};
use crate::{
    BatchStream, Blob, BlobStat, BlobStream, Error, Scan, ScanMetrics, SortColumn, TableIdentity,
    WriteInput,
};

/// The most payload bytes one batch of a blob pack carries; a larger payload uploads on its own.
const PACK_BATCH_BYTES: usize = 32 << 20;

/// One batch of `schema`'s single `spql.Blob` column holding `payloads` inline.
fn inline_blobs(
    schema: &SchemaRef,
    payloads: Vec<(Bytes, Option<String>)>,
) -> Result<RecordBatch, Error> {
    let DataType::Struct(fields) = schema.field(0).data_type() else {
        unreachable!("a blob field is a struct");
    };
    let rows = payloads.len();
    let columns = fields
        .iter()
        .map(|field| -> ArrayRef {
            match field.name().as_str() {
                "bytes" => Arc::new(BinaryArray::from_iter_values(
                    payloads.iter().map(|(payload, _)| payload.as_ref()),
                )),
                "mime_type" => Arc::new(StringArray::from_iter(
                    payloads.iter().map(|(_, media_type)| media_type.as_deref()),
                )),
                _ => new_null_array(field.data_type(), rows),
            }
        })
        .collect();
    let blobs = StructArray::try_new(fields.clone(), columns, None).map_err(Error::arrow)?;
    RecordBatch::try_new(Arc::clone(schema), vec![Arc::new(blobs)]).map_err(Error::arrow)
}

/// Scans and writes executed by `scandal`, with the SDK's Tables client as host.
pub(crate) struct ScandalBackend {
    scandal: Arc<dyn Scandal>,
    arbor: ArborClient,
    access: TableAccess,
}

impl ScandalBackend {
    pub(crate) fn new(scandal: Arc<dyn Scandal>, arbor: ArborClient) -> Self {
        let resolves_resources = scandal.resolves_resources();
        Self {
            scandal,
            access: TableAccess::new(arbor.clone()).resolving(resolves_resources),
            arbor,
        }
    }

    pub(crate) fn scandal(&self) -> Arc<dyn Scandal> {
        Arc::clone(&self.scandal)
    }
}

/// The longest a split read may take. Scandal checks a binding's credential only as its
/// storage requests are signed, so a split must finish before the credential it opened with
/// expires; splits are assumed to finish within this budget.
const SPLIT_READ_BUDGET: Duration = Duration::from_secs(5 * 60);

fn read_access() -> StorageAccessRequest {
    StorageAccessRequest {
        mode: StorageMode::ReadOnly,
        write: None,
    }
}

fn write_access(scope: WriteScope) -> StorageAccessRequest {
    StorageAccessRequest {
        mode: StorageMode::ReadWrite,
        write: Some(scope),
    }
}

/// Older stacks expose the SPFS capability through storage access alone.
async fn table_storage_binding(
    arbor: &ArborClient,
    table: TableId,
    request: &StorageAccessRequest,
) -> Result<TablesBinding, Error> {
    if let Some(binding) = arbor.storage_binding(table.clone(), request).await? {
        return Ok(binding);
    }
    let access = arbor.storage_access(table, request).await?;
    let expires_at = Token::from(access.spfs_token.0.clone())
        .remaining_lifetime()
        .ok()
        .and_then(|remaining| SystemTime::now().checked_add(remaining))
        .and_then(|expiry| expiry.duration_since(UNIX_EPOCH).ok())
        .map_or(0, |expiry| expiry.as_secs());
    Ok(TablesBinding {
        root_uri: format!("spfsu://{}/", access.spfs_alias),
        mode: request.mode.clone(),
        expires_at: expires_at.into(),
        access: StorageAccess::Signed(SignedAccess {
            endpoint: String::new(),
            token: access.spfs_token,
            direct_read: None,
        }),
    })
}

/// Resolve a storage binding for one table root with the caller's Tables credentials.
async fn binding(
    resolves_resources: bool,
    arbor: &ArborClient,
    table: Uuid,
    root: String,
    request: StorageAccessRequest,
) -> Result<StorageBinding, Error> {
    if resolves_resources && !root.starts_with("file:") {
        return Ok(StorageBinding::resource(table));
    }
    Ok(expiring_binding(arbor, table, root, request).await?.0)
}

/// A storage binding and how long its credential remains valid; `None` for local files,
/// which need no credential.
async fn expiring_binding(
    arbor: &ArborClient,
    table: Uuid,
    root: String,
    request: StorageAccessRequest,
) -> Result<(StorageBinding, Option<Duration>), Error> {
    if Url::parse(&root).map_err(Error::invalid)?.scheme() == "file" {
        let binding = StorageBinding {
            root_uri: root,
            root_alias: None,
            credential: None,
        };
        return Ok((binding, None));
    }
    let binding = table_storage_binding(arbor, TableId(table), &request).await?;
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default();
    let lifetime = Duration::from_secs(binding.expires_at.0).saturating_sub(now);
    Ok((StorageBinding::from_storage(root, binding)?, Some(lifetime)))
}

/// The storage root of a Table in the local directory `root`, which needs no credential.
fn local_root(root: &Url) -> Result<String, Error> {
    if root.scheme() != "file" || !root.path().ends_with('/') {
        return Err(Error::invalid(format!(
            "a local Table root is a file:// directory URL ending in `/`, not {root}"
        )));
    }
    Ok(root.to_string())
}

/// Bindings of one root for one kind of access, reused while they outlast a split read or a
/// part upload.
struct Bindings {
    resolves_resources: bool,
    arbor: ArborClient,
    table: Uuid,
    root: String,
    request: StorageAccessRequest,
    /// The current binding and the instant after which a split could outlive it, if any.
    current: AsyncMutex<Option<(StorageBinding, Option<Instant>)>>,
}

impl Bindings {
    fn new(arbor: ArborClient, table: Uuid, root: String, request: StorageAccessRequest) -> Self {
        Self {
            resolves_resources: false,
            arbor,
            table,
            root,
            request,
            current: AsyncMutex::new(None),
        }
    }

    fn resolving(mut self, resolving: bool) -> Self {
        self.resolves_resources = resolving;
        self
    }

    /// Forget `failed` if it is still the current binding, so the next request fetches a fresh
    /// one. A binding another request has already renewed stays.
    async fn discard(&self, failed: &StorageBinding) {
        let mut current = self.current.lock().await;
        if current
            .as_ref()
            .is_some_and(|(binding, _)| binding == failed)
        {
            *current = None;
        }
    }

    /// Pass on the result of a request made with `binding`, first discarding the binding when
    /// a fresh one might cure the failure, as a split read does before it reads again.
    async fn discard_if_failed<T>(
        &self,
        binding: &StorageBinding,
        result: Result<T, ScandalError>,
    ) -> Result<T, Error> {
        if let Err(error) = &result
            && retryable(error)
        {
            self.discard(binding).await;
        }
        Ok(result?)
    }

    /// A binding whose credential outlasts a split read or part upload started now.
    async fn for_split(&self) -> Result<StorageBinding, Error> {
        if self.resolves_resources && !self.root.starts_with("file:") {
            return Ok(StorageBinding::resource(self.table));
        }
        let mut current = self.current.lock().await;
        if let Some((binding, usable_until)) = current.as_ref()
            && usable_until.is_none_or(|until| Instant::now() < until)
        {
            return Ok(binding.clone());
        }
        let (binding, lifetime) = expiring_binding(
            &self.arbor,
            self.table,
            self.root.clone(),
            self.request.clone(),
        )
        .await?;
        *current = Some((binding.clone(), usable_until(lifetime)));
        Ok(binding)
    }
}

/// The instant after which a split read started with a credential of `lifetime` could outlive
/// it. `None` means the binding never needs replacing: local files carry no credential, and a
/// credential can outlast every instant the clock can represent.
fn usable_until(lifetime: Option<Duration>) -> Option<Instant> {
    lifetime
        .and_then(|lifetime| Instant::now().checked_add(lifetime.saturating_sub(SPLIT_READ_BUDGET)))
}

/// What every request for a client's tables reuses: each exact version's storage root, and one
/// read-only binding per table root. A read made with the binding an earlier read used lets
/// scandal reuse the version it opened and the requests it signed then, so reading a version
/// again costs no setup requests. A client acts as one identity, so its bindings serve all its
/// reads; it keeps a small entry per version and per table it has read.
struct TableAccess {
    resolves_resources: bool,
    arbor: ArborClient,
    /// The storage root of each exact version; a version never changes. Concurrent first reads
    /// of a version share one lookup.
    roots: Mutex<HashMap<TableIdentity, Arc<OnceCell<String>>>>,
    /// The read-only bindings of each table root.
    reads: Mutex<HashMap<(Uuid, String), Arc<Bindings>>>,
}

impl TableAccess {
    fn new(arbor: ArborClient) -> Self {
        Self {
            resolves_resources: false,
            arbor,
            roots: Mutex::default(),
            reads: Mutex::default(),
        }
    }

    fn resolving(mut self, resolving: bool) -> Self {
        self.resolves_resources = resolving;
        self
    }

    /// The storage root of the exact version `table`.
    async fn root(&self, table: TableIdentity) -> Result<String, Error> {
        let root = Arc::clone(self.roots.lock().entry(table).or_default());
        root.get_or_try_init(|| async {
            let version = self
                .arbor
                .get_version(TableId(table.resource_id), table.version_id)
                .await?;
            Ok::<_, Error>(version.root_uri)
        })
        .await
        .cloned()
    }

    /// Record that the exact version `table` is stored at the local directory `root`, so its
    /// reads resolve no root through the control plane.
    fn add_local_root(&self, table: TableIdentity, root: &Url) -> Result<(), Error> {
        let root = local_root(root)?;
        let cell = Arc::clone(self.roots.lock().entry(table).or_default());
        // A version's storage never moves, so another root for it is a mistake.
        if cell.set(root.clone()).is_err() && cell.get() != Some(&root) {
            return Err(Error::invalid(format!(
                "version {} is stored at another root",
                table.version_id
            )));
        }
        Ok(())
    }

    /// The read-only bindings of the exact version `table`'s storage.
    async fn reads(&self, table: TableIdentity) -> Result<Arc<Bindings>, Error> {
        let root = self.root(table).await?;
        let mut reads = self.reads.lock();
        let bindings =
            reads
                .entry((table.resource_id, root))
                .or_insert_with_key(|(table_id, root)| {
                    Arc::new(
                        Bindings::new(self.arbor.clone(), *table_id, root.clone(), read_access())
                            .resolving(self.resolves_resources),
                    )
                });
        Ok(Arc::clone(bindings))
    }
}

fn scandal_scan(table: TableIdentity, options: &ScanOptions) -> Result<ScandalScan, Error> {
    options.stream_count()?;
    if options.split_token.is_some() {
        return Err(Error::invalid(
            "scandal reads splits from its scan plan, not split tokens",
        ));
    }
    Ok(ScandalScan {
        table: TableVersion {
            resource_id: table.resource_id,
            version_id: table.version_id,
        },
        columns: options.columns.clone(),
        projection: options.projection.clone(),
        filter: options.filter.clone(),
        limit: options.limit.map(Into::into),
        batch_rows: options
            .batch_rows
            .map(|rows| u32::try_from(rows).map_err(Error::invalid))
            .transpose()?,
        collect_metrics: options.collect_metrics.then_some(true),
        emit: options.emit.clone(),
        row_range: options
            .row_range
            .map(|range| RowRange::new(range.start..range.end)),
        row_id_column: options.row_id_column.clone(),
        row_ids: options
            .row_ids
            .as_ref()
            .map(|ranges| ranges.iter().map(row_id_range).collect()),
    })
}

/// The contract form of the row IDs `range`.
fn row_id_range(range: &Range<u64>) -> RowIdRange {
    RowIdRange {
        start: range.start.into(),
        end: range.end.into(),
    }
}

impl ScandalBackend {
    pub(crate) async fn open_scan(
        &self,
        table: TableIdentity,
        options: ScanOptions,
    ) -> Result<Scan, Error> {
        let scan = scandal_scan(table, &options)?;
        let bindings = self.access.reads(table).await?;
        let storage = bindings.for_split().await?;
        let plan = bindings
            .discard_if_failed(&storage, self.scandal.plan_scan(&scan, &storage).await)
            .await?;
        let info = ScanInfo {
            statistics: plan.info.statistics,
            write_layout: plan.info.write_layout,
            schema: serde_json::from_value(plan.info.schema.0).map_err(Error::arrow)?,
            filter_clauses: plan.info.filter_clauses,
        };
        let lanes = options.stream_count()?;
        let ordered =
            !options.partitioned || options.limit.is_some() || options.row_range.is_some();
        let schema = Arc::new(info.schema.clone());
        let read = match &options.emit {
            None => SplitReads::Rows {
                lanes: if ordered { 1 } else { lanes },
            },
            Some(emit) => SplitReads::Emit {
                emit: emit.bind_results(&schema)?,
                lanes: options.max_streams.unwrap_or(DEFAULT_EMIT_LANES),
            },
        };
        let scan_splits = plan
            .splits
            .iter()
            .map(|planned| ScanSplit {
                fragment: planned.split.fragment.into(),
                rows: planned.split.rows.range(),
                estimated_rows: planned.estimated_rows.map(Into::into),
                estimated_bytes: planned.estimated_bytes.map(Into::into),
                fragment_id: planned.fragment_id,
            })
            .collect();
        let source = ScandalScanSource {
            scandal: Arc::clone(&self.scandal),
            identity: table,
            bindings,
            scan,
            schema,
            scan_splits,
            splits: plan.splits,
            plan_metrics: plan.metrics,
            read,
        };
        Scan::new(table, options, info, Arc::new(source))
    }

    pub(crate) async fn read_blob(
        &self,
        table: TableIdentity,
        blob: &Blob,
        range: Option<Range<u64>>,
    ) -> Result<BlobStream, Error> {
        if blob.table_id != table.resource_id {
            return Err(Error::invalid("blob belongs to a different Table"));
        }
        let bindings = self.access.reads(table).await?;
        let storage = bindings.for_split().await?;
        let bytes = bindings
            .discard_if_failed(
                &storage,
                self.scandal.read_blob(blob, range, &storage).await,
            )
            .await?;
        Ok(bytes.map_err(Error::from).boxed())
    }

    pub(crate) async fn stat_blob(
        &self,
        table: TableIdentity,
        blob: &Blob,
    ) -> Result<BlobStat, Error> {
        let bindings = self.access.reads(table).await?;
        let storage = bindings.for_split().await?;
        bindings
            .discard_if_failed(&storage, self.scandal.stat_blob(blob, &storage).await)
            .await
    }
}

/// One scan's splits, read through scandal in stored order or concurrently.
struct ScandalScanSource {
    scandal: Arc<dyn Scandal>,
    identity: TableIdentity,
    /// The read-only bindings of the version's storage root, shared with every other read of the
    /// table.
    bindings: Arc<Bindings>,
    scan: ScandalScan,
    schema: SchemaRef,
    /// `splits` as the SDK's public type.
    scan_splits: Vec<ScanSplit>,
    splits: Vec<PlannedSplit>,
    /// The plan's pruning counters, when the scan collects metrics.
    plan_metrics: Option<ScanMetricsSnapshot>,
    read: SplitReads,
}

/// Splits an emit reads at once unless the scan sets `max_streams`.
const DEFAULT_EMIT_LANES: usize = 4;

impl ScandalScanSource {
    /// Reads of `scan` over this source's splits, without counters or narrowing.
    fn split_reader(&self, scan: ScandalScan) -> SplitReader {
        SplitReader {
            page_rows: None,
            scandal: Arc::clone(&self.scandal),
            scan,
            bindings: Arc::clone(&self.bindings),
            counters: None,
            narrowing: None,
        }
    }

    /// The split a validated work item reads: its stored rows of its split's fragment.
    fn work_split(&self, item: WorkItem) -> Split {
        Split {
            fragment: self.splits[item.split].split.fragment,
            rows: RowRange::new(item.rows),
        }
    }
}

/// The rows an aggregate without groups counted over one split.
fn split_rows(results: &[RecordBatch]) -> Result<u64, Error> {
    results.iter().try_fold(0u64, |rows, batch| {
        let counted = batch
            .column_by_name(ROWS_COLUMN)
            .and_then(|column| column.as_any().downcast_ref::<UInt64Array>())
            .ok_or_else(|| Error::arrow(format!("count results lack a `{ROWS_COLUMN}` column")))?;
        Ok(rows + counted.values().iter().sum::<u64>())
    })
}

/// How a scan reads its splits.
enum SplitReads {
    /// Rows, reading `lanes` splits at once; one keeps stored order and lets a limit span
    /// splits.
    Rows { lanes: usize },
    /// Each split's results whole, `lanes` at once, merged into one result.
    Emit { emit: BoundEmit, lanes: usize },
}

#[async_trait]
impl ScanSource for ScandalScanSource {
    async fn table_binding(&self) -> Result<Option<TableBinding>, Error> {
        Ok(Some(TableBinding {
            table_id: self.identity.resource_id,
            storage: Some(self.bindings.for_split().await?),
        }))
    }

    fn estimated_rows(&self) -> Option<u64> {
        self.splits.iter().try_fold(0, |rows: u64, planned| {
            Some(rows.saturating_add(planned.estimated_rows?.into()))
        })
    }

    fn splits(&self) -> &[ScanSplit] {
        &self.scan_splits
    }

    async fn read_work(&self, work: Vec<WorkItem>) -> Result<WorkStream, Error> {
        let reader = Arc::new(self.split_reader(self.scan.clone()));
        let splits = work
            .into_iter()
            .map(|item| self.work_split(item))
            .enumerate()
            .collect::<Vec<_>>();
        let batches = stream::iter(splits)
            .map(move |(index, split)| {
                let batches = reader
                    .read_split_batches(split)
                    .map_ok(move |read| WorkBatch {
                        batch: read.batch,
                        item: index,
                        next_row: read.next_row,
                    })
                    .boxed();
                let (opening, opened) = started(batches).remote_handle();
                tokio::spawn(opening);
                opened
            })
            .buffered(ORDERED_READ_AHEAD)
            .try_flatten();
        Ok(WorkStream::new(Arc::clone(&self.schema), batches))
    }

    async fn take(&self, takes: Vec<Take>) -> Result<BatchStream, Error> {
        let counters = self
            .scan
            .collects_metrics()
            .then(|| Arc::new(Mutex::new(ScanMetricsSnapshot::default())));
        let reader = Arc::new(TakeReader {
            scandal: Arc::clone(&self.scandal),
            scan: self.scan.clone(),
            bindings: Arc::clone(&self.bindings),
            counters: counters.clone(),
        });
        let takes = takes.into_iter().map(|take| SplitTake {
            split: self.splits[take.split].split.clone(),
            positions: take.positions.into_iter().map(Into::into).collect(),
        });
        let batches = stream::iter(take_requests(takes))
            .map(move |takes| Arc::clone(&reader).read(takes))
            .flatten()
            .map_ok(|batch| batch.batch)
            .boxed();
        // A position past its split fails the call, not the stream.
        let batches = started(batches).await?;
        let stream = BatchStream::from_stream(Arc::clone(&self.schema), batches);
        Ok(match counters {
            Some(counters) => stream.with_metrics(ScanMetrics::from_snapshot_fn(move || {
                Ok(counters.lock().clone())
            })),
            None => stream,
        })
    }

    async fn count_rows(&self, work: &[WorkItem]) -> Result<Vec<u64>, Error> {
        // Rows pass or fail the filter whatever the scan outputs, so counting computes none of
        // its outputs.
        let counting = ScandalScan {
            columns: None,
            projection: Some(Vec::new()),
            emit: Some(Emit::Aggregate(Aggregate {
                groups: Vec::new(),
                aggregates: vec![NamedAggregate {
                    name: "rows".into(),
                    function: AggregateFunction::Count(CountAggregate { column: None }),
                }],
                max_groups: None,
            })),
            ..self.scan.clone()
        };
        let reader = Arc::new(self.split_reader(counting));
        let splits = work
            .iter()
            .map(|item| self.work_split(item.clone()))
            .collect::<Vec<_>>();
        let counts = stream::iter(splits).map(move |split| {
            let reader = Arc::clone(&reader);
            async move { split_rows(&reader.read_results(split).await?) }
        });
        counts.buffered(DEFAULT_EMIT_LANES).try_collect().await
    }

    fn ordered_partitions(&self) -> Option<(usize, Vec<SortColumn>)> {
        if self.scan.emit.is_some() || (self.scan.limit.is_some() && self.splits.len() != 1) {
            return None;
        }
        let mut ordering = self.splits.first()?.ordering.clone()?;
        ordering.truncate(
            ordering
                .iter()
                .take_while(|key| {
                    key.path.len() == 1
                        && key
                            .transform
                            .as_deref()
                            .is_none_or(|transform| transform == "identity")
                        && self.schema.index_of(&key.path[0]).is_ok()
                })
                .count(),
        );
        for split in &self.splits[1..] {
            let other = split.ordering.as_ref()?;
            let common = ordering
                .iter()
                .zip(other)
                .take_while(|(left, right)| left == right)
                .count();
            ordering.truncate(common);
        }
        (!ordering.is_empty()).then_some((self.splits.len(), ordering))
    }

    fn read_partitions(
        &self,
        narrowing: Option<Arc<dyn SplitFilter>>,
        lanes: usize,
        batch_rows: usize,
    ) -> Option<Vec<BatchStream>> {
        self.ordered_partitions()?;
        let counters = self
            .plan_metrics
            .clone()
            .map(|metrics| Arc::new(Mutex::new(metrics)));
        let mut scan = self.scan.clone();
        let fragments_per_lane = self.splits.len().div_ceil(lanes);
        scan.batch_rows =
            Some(u32::try_from(batch_rows.div_ceil(fragments_per_lane).max(1)).unwrap_or(u32::MAX));
        let reader = Arc::new(SplitReader {
            page_rows: (fragments_per_lane > 1)
                .then(|| u64::from(scan.batch_rows.expect("ordered input batch size"))),
            scandal: Arc::clone(&self.scandal),
            scan,
            bindings: Arc::clone(&self.bindings),
            counters: counters.clone(),
            narrowing,
        });
        let metrics = counters
            .map(|counters| ScanMetrics::from_snapshot_fn(move || Ok(counters.lock().clone())))
            .unwrap_or_default();
        let mut groups = (0..lanes).map(|_| Vec::new()).collect::<Vec<_>>();
        for (index, planned) in self.splits.iter().enumerate() {
            let batches =
                Arc::clone(&reader).read_in_order(VecDeque::from([planned.split.clone()]));
            groups[index % lanes].push(
                batches
                    .map_err(|error| ScandalError::Internal {
                        message: error.to_string(),
                    })
                    .boxed(),
            );
        }
        let (_, ordering) = self.ordered_partitions()?;
        Some(
            groups
                .into_iter()
                .map(|mut streams| {
                    let batches = if streams.len() == 1 {
                        streams.pop().expect("one stream")
                    } else {
                        merge_ordered(Arc::clone(&self.schema), streams, &ordering, batch_rows)
                            .unwrap_or_else(|error| stream::once(async { Err(error) }).boxed())
                    };
                    BatchStream::from_stream(Arc::clone(&self.schema), batches.map_err(Error::from))
                        .with_metrics(metrics.clone())
                })
                .collect(),
        )
    }

    async fn read(&self, narrowing: Option<Arc<dyn SplitFilter>>) -> Result<BatchStream, Error> {
        // Each execution counts afresh: the plan's pruning, then each split read's reads.
        let counters = self
            .plan_metrics
            .clone()
            .map(|metrics| Arc::new(Mutex::new(metrics)));
        let splits = self
            .splits
            .iter()
            .map(|planned| planned.split.clone())
            .collect::<VecDeque<_>>();
        let reader = Arc::new(SplitReader {
            page_rows: None,
            scandal: Arc::clone(&self.scandal),
            scan: self.scan.clone(),
            bindings: Arc::clone(&self.bindings),
            counters: counters.clone(),
            narrowing,
        });
        let batches = match &self.read {
            SplitReads::Rows { lanes: 1 } => reader.read_in_order(splits),
            SplitReads::Rows { lanes } => reader.read_unordered(splits, *lanes),
            SplitReads::Emit { emit, lanes } => reader.read_merged(splits, emit.clone(), *lanes),
        };
        let stream = BatchStream::from_stream(Arc::clone(&self.schema), batches);
        Ok(match counters {
            Some(counters) => stream.with_metrics(ScanMetrics::from_snapshot_fn(move || {
                Ok(counters.lock().clone())
            })),
            None => stream,
        })
    }
}

/// Times a split is reopened after failures without progress before its read fails.
const SPLIT_RETRIES: usize = 3;

/// One split's batches and the stored row each reached.
type SplitBatches = BoxStream<'static, Result<SplitBatch, ScandalError>>;

/// Where a split read stands: the stored rows still to read and the output rows delivered.
struct SplitProgress {
    split: Split,
    delivered: u64,
    failures: usize,
    batches: Option<SplitBatches>,
    /// The binding the latest read of the split opened with.
    binding: Option<StorageBinding>,
}

/// What every split read of one scan execution shares.
struct SplitReader {
    /// Close each response before yielding a page when a merge needs more fragments than lanes.
    page_rows: Option<u64>,
    scandal: Arc<dyn Scandal>,
    scan: ScandalScan,
    bindings: Arc<Bindings>,
    counters: Option<Arc<Mutex<ScanMetricsSnapshot>>>,
    /// Narrows each row split as it opens. Emit reads are never narrowed: a filter on their
    /// results says nothing about their input rows.
    narrowing: Option<Arc<dyn SplitFilter>>,
}

impl SplitReader {
    /// The request a row split opens with: at most `limit` rows, narrowed by the split filter as
    /// it stands now.
    fn request(&self, limit: Option<u64>) -> ScandalScan {
        let narrowing = self
            .narrowing
            .as_ref()
            .and_then(|narrowing| narrowing.current());
        // Narrowing only lets a split skip work, so one too large to join the scan's filter is
        // left out.
        let filter = match (&self.scan.filter, narrowing) {
            (Some(filter), Some(narrowing)) => {
                Some(filter.and(&narrowing).unwrap_or_else(|_| filter.clone()))
            }
            (filter, narrowing) => filter.clone().or(narrowing),
        };
        ScandalScan {
            limit: limit.map(Into::into),
            filter,
            ..self.scan.clone()
        }
    }

    /// Read one split in stored order.
    fn read_split(
        self: &Arc<Self>,
        split: Split,
    ) -> BoxStream<'static, Result<RecordBatch, Error>> {
        match self.page_rows {
            Some(rows) => self.read_split_paged(split, rows),
            None => self
                .read_split_batches(split)
                .map_ok(|batch| batch.batch)
                .boxed(),
        }
    }

    fn read_split_paged(
        self: &Arc<Self>,
        split: Split,
        page_rows: u64,
    ) -> BoxStream<'static, Result<RecordBatch, Error>> {
        let reader = Arc::new(Self {
            page_rows: None,
            scan: ScandalScan {
                limit: Some(page_rows.into()),
                ..self.scan.clone()
            },
            scandal: Arc::clone(&self.scandal),
            bindings: Arc::clone(&self.bindings),
            counters: self.counters.clone(),
            narrowing: self.narrowing.clone(),
        });
        stream::try_unfold(split, move |mut split| {
            let reader = Arc::clone(&reader);
            async move {
                if split.rows.start >= split.rows.end {
                    return Ok(None);
                }
                let batches = reader
                    .read_split_batches(split.clone())
                    .try_collect::<Vec<_>>()
                    .await?;
                let Some(last) = batches.last() else {
                    return Ok(None);
                };
                if last.next_row <= split.rows.start.0 {
                    return Err(Error::invalid("ordered page did not advance"));
                }
                let delivered = batches
                    .iter()
                    .map(|batch| batch.batch.num_rows() as u64)
                    .sum::<u64>();
                split.rows.start = if delivered < page_rows {
                    split.rows.end
                } else {
                    last.next_row.into()
                };
                let batches = stream::iter(batches.into_iter().map(|batch| Ok(batch.batch)));
                Ok(Some((batches, split)))
            }
        })
        .try_flatten()
        .boxed()
    }

    fn read_split_batches(
        self: &Arc<Self>,
        split: Split,
    ) -> BoxStream<'static, Result<SplitBatch, Error>> {
        let progress = SplitProgress {
            split,
            delivered: 0,
            failures: 0,
            batches: None,
            binding: None,
        };
        let reader = Arc::clone(self);
        stream::try_unfold(progress, move |mut progress| {
            let reader = Arc::clone(&reader);
            async move {
                loop {
                    // An open read runs to its end, which may carry its counters.
                    let next = match progress.batches.as_mut() {
                        Some(batches) => batches.try_next().await,
                        None => {
                            let limit = reader
                                .scan
                                .limit
                                .map(|limit| limit.0 - progress.delivered.min(limit.0));
                            if limit == Some(0)
                                || progress.split.rows.start >= progress.split.rows.end
                            {
                                return Ok(None);
                            }
                            let scan = reader.request(limit);
                            let storage = reader.bindings.for_split().await?;
                            let opened = reader
                                .scandal
                                .read_scan(&scan, &progress.split, &storage)
                                .await;
                            progress.binding = Some(storage);
                            match opened {
                                Ok(read) => {
                                    progress.batches = Some(read.batches);
                                    continue;
                                }
                                Err(error) => Err(error),
                            }
                        }
                    };
                    match next {
                        Ok(Some(batch)) => {
                            if let (Some(counters), Some(metrics)) =
                                (&reader.counters, &batch.metrics)
                            {
                                counters.lock().add(metrics);
                            }
                            if batch.batch.num_rows() == 0 {
                                continue;
                            }
                            progress.split.rows.start = batch.next_row.into();
                            progress.delivered += batch.batch.num_rows() as u64;
                            progress.failures = 0;
                            return Ok(Some((batch, progress)));
                        }
                        Ok(None) => return Ok(None),
                        Err(error) if retryable(&error) && progress.failures < SPLIT_RETRIES => {
                            progress.failures += 1;
                            progress.batches = None;
                            if let Some(failed) = progress.binding.take() {
                                reader.bindings.discard(&failed).await;
                            }
                        }
                        Err(error) => return Err(error.into()),
                    }
                }
            }
        })
        .boxed()
    }

    /// Read `lanes` splits at once, delivering rows as they arrive.
    fn read_unordered(
        self: Arc<Self>,
        splits: VecDeque<Split>,
        lanes: usize,
    ) -> BoxStream<'static, Result<RecordBatch, Error>> {
        stream::iter(splits)
            .map(move |split| self.read_split(split))
            .flatten_unordered(lanes)
            .boxed()
    }

    /// Read splits so rows arrive in stored order, opening the next splits while one is
    /// delivered. Under a limit, the read ends once the limit's rows are delivered and stops
    /// the splits opened ahead; its counters then cover only the splits it read to their end.
    fn read_in_order(
        self: Arc<Self>,
        splits: VecDeque<Split>,
    ) -> BoxStream<'static, Result<RecordBatch, Error>> {
        let limit = self.scan.limit.map(u64::from);
        let batches = self.read_ahead_in_order(splits);
        match limit {
            Some(limit) => first_rows(batches, limit),
            None => batches,
        }
    }

    /// Read splits in stored order while the next splits open.
    ///
    /// Each split opens on its own task, so it progresses while the current split is
    /// delivered, and stops when the read is dropped. Each reads at most the scan's limit.
    fn read_ahead_in_order(
        self: Arc<Self>,
        splits: VecDeque<Split>,
    ) -> BoxStream<'static, Result<RecordBatch, Error>> {
        stream::iter(splits)
            .map(move |split| {
                let (opening, opened) = started(self.read_split(split)).remote_handle();
                tokio::spawn(opening);
                opened
            })
            .buffered(ORDERED_READ_AHEAD)
            .try_flatten()
            .boxed()
    }

    /// Read every split's emit results, `lanes` splits at once, and merge them in split order.
    fn read_merged(
        self: Arc<Self>,
        splits: VecDeque<Split>,
        emit: BoundEmit,
        lanes: usize,
    ) -> BoxStream<'static, Result<RecordBatch, Error>> {
        let results = stream::iter(splits)
            .map(move |split| Arc::clone(&self).read_results(split))
            .buffered(lanes)
            .try_concat();
        stream::once(async move { Ok(emit.merge(&results.await?)?) }).boxed()
    }

    /// One split's emit results. They are complete only at the end of the read, so a failed
    /// read is read again from the split's start, with a fresh binding.
    async fn read_results(self: Arc<Self>, split: Split) -> Result<Vec<RecordBatch>, Error> {
        let mut failures = 0;
        loop {
            let storage = self.bindings.for_split().await?;
            let read = match self.scandal.read_scan(&self.scan, &split, &storage).await {
                Ok(read) => read.batches.try_collect::<Vec<_>>().await,
                Err(error) => Err(error),
            };
            match read {
                Ok(batches) => {
                    let mut results = Vec::with_capacity(batches.len());
                    for batch in batches {
                        if let (Some(counters), Some(metrics)) = (&self.counters, &batch.metrics) {
                            counters.lock().add(metrics);
                        }
                        if batch.batch.num_rows() > 0 {
                            results.push(batch.batch);
                        }
                    }
                    return Ok(results);
                }
                Err(error) if retryable(&error) && failures < SPLIT_RETRIES => {
                    failures += 1;
                    self.bindings.discard(&storage).await;
                }
                Err(error) => return Err(error.into()),
            }
        }
    }
}

/// Failures that reading the rest of a split again can cure: a lost replica or connection,
/// a busy replica, or a credential that lapsed.
fn retryable(error: &ScandalError) -> bool {
    matches!(
        error,
        ScandalError::Internal { .. }
            | ScandalError::Unavailable { .. }
            | ScandalError::Unauthenticated { .. }
    )
}

/// `takes` as requests within one request's bounds, in order, a take with more positions than
/// a request holds divided across requests.
fn take_requests(takes: impl Iterator<Item = SplitTake>) -> Vec<Vec<SplitTake>> {
    let mut requests: Vec<Vec<SplitTake>> = Vec::new();
    let mut positions = MAX_TAKE_POSITIONS;
    for take in takes {
        let mut rest = take.positions.as_slice();
        loop {
            if positions == MAX_TAKE_POSITIONS
                || requests
                    .last()
                    .is_none_or(|request| request.len() == MAX_TAKES)
            {
                requests.push(Vec::new());
                positions = 0;
            }
            let count = rest.len().min(MAX_TAKE_POSITIONS - positions);
            let (now, later) = rest.split_at(count);
            if let Some(request) = requests.last_mut() {
                request.push(SplitTake {
                    split: take.split.clone(),
                    positions: now.to_vec(),
                });
            }
            positions += count;
            rest = later;
            if rest.is_empty() {
                break;
            }
        }
    }
    requests
}

/// `takes` without their first `delivered` positions, counted across them.
fn after_delivered(takes: &[SplitTake], mut delivered: u64) -> Vec<SplitTake> {
    let mut rest = Vec::with_capacity(takes.len());
    for take in takes {
        let skipped = usize::try_from(delivered)
            .unwrap_or(usize::MAX)
            .min(take.positions.len());
        delivered -= skipped as u64;
        if skipped < take.positions.len() {
            rest.push(SplitTake {
                split: take.split.clone(),
                positions: take.positions[skipped..].to_vec(),
            });
        }
    }
    rest
}

/// What every take request of one scan shares.
struct TakeReader {
    scandal: Arc<dyn Scandal>,
    scan: ScandalScan,
    bindings: Arc<Bindings>,
    counters: Option<Arc<Mutex<ScanMetricsSnapshot>>>,
}

/// Where a take request stands: its positions not yet delivered.
struct TakeProgress {
    takes: Vec<SplitTake>,
    /// Positions of `takes` that the open read delivered.
    delivered: u64,
    failures: usize,
    batches: Option<SplitBatches>,
    binding: Option<StorageBinding>,
}

impl TakeReader {
    /// Read one request's takes in request order. After a retryable failure, the positions not
    /// yet delivered are taken again with a fresh binding, so no row is delivered twice or
    /// skipped.
    fn read(
        self: Arc<Self>,
        takes: Vec<SplitTake>,
    ) -> BoxStream<'static, Result<SplitBatch, Error>> {
        let progress = TakeProgress {
            takes,
            delivered: 0,
            failures: 0,
            batches: None,
            binding: None,
        };
        stream::try_unfold(progress, move |mut progress| {
            let reader = Arc::clone(&self);
            async move {
                loop {
                    let next = match progress.batches.as_mut() {
                        Some(batches) => batches.try_next().await,
                        None => {
                            if progress.takes.iter().all(|take| take.positions.is_empty()) {
                                return Ok(None);
                            }
                            let storage = reader.bindings.for_split().await?;
                            let opened = reader
                                .scandal
                                .take_rows(&reader.scan, &progress.takes, &storage)
                                .await;
                            progress.binding = Some(storage);
                            match opened {
                                Ok(read) => {
                                    progress.batches = Some(read.batches);
                                    progress.delivered = 0;
                                    continue;
                                }
                                Err(error) => Err(error),
                            }
                        }
                    };
                    match next {
                        Ok(Some(batch)) => {
                            if let (Some(counters), Some(metrics)) =
                                (&reader.counters, &batch.metrics)
                            {
                                counters.lock().add(metrics);
                            }
                            if batch.batch.num_rows() == 0 {
                                continue;
                            }
                            progress.delivered = batch.next_row;
                            progress.failures = 0;
                            return Ok(Some((batch, progress)));
                        }
                        Ok(None) => return Ok(None),
                        Err(error) if retryable(&error) && progress.failures < SPLIT_RETRIES => {
                            progress.failures += 1;
                            progress.takes = after_delivered(&progress.takes, progress.delivered);
                            progress.batches = None;
                            if let Some(failed) = progress.binding.take() {
                                reader.bindings.discard(&failed).await;
                            }
                        }
                        Err(error) => return Err(error.into()),
                    }
                }
            }
        })
        .boxed()
    }
}

/// Splits an ordered read opens ahead of the one it delivers, so that their storage requests
/// overlap. Each holds at most its first batch until its turn. Under a limit, the splits opened
/// ahead of the one that meets it read rows the limit leaves out, which with a selective filter
/// can be most of each split.
const ORDERED_READ_AHEAD: usize = 8;

/// Wait for a split read's first batch, so a split opened ahead is under way before its turn.
async fn started<T: Send + 'static>(
    mut batches: BoxStream<'static, Result<T, Error>>,
) -> Result<BoxStream<'static, Result<T, Error>>, Error> {
    Ok(match batches.try_next().await? {
        Some(first) => stream::once(future::ready(Ok(first)))
            .chain(batches)
            .boxed(),
        None => stream::empty().boxed(),
    })
}

/// The first `limit` rows of `batches`. `batches` is dropped as soon as the last of them is
/// read, so reads still under way stop before those rows are delivered.
fn first_rows(
    batches: BoxStream<'static, Result<RecordBatch, Error>>,
    limit: u64,
) -> BoxStream<'static, Result<RecordBatch, Error>> {
    stream::try_unfold(
        ((limit > 0).then_some(batches), limit),
        |(batches, remaining)| async move {
            let Some(mut batches) = batches else {
                return Ok(None);
            };
            let Some(batch) = batches.try_next().await? else {
                return Ok(None);
            };
            let rows = (batch.num_rows() as u64).min(remaining);
            let remaining = remaining - rows;
            Ok(Some((
                batch.slice(0, rows as usize),
                ((remaining > 0).then_some(batches), remaining),
            )))
        },
    )
    .boxed()
}

impl ScandalBackend {
    pub(crate) async fn write_blob(
        &self,
        table: TableIdentity,
        media_type: &str,
        max_bytes: u64,
        input: BlobStream,
    ) -> Result<Blob, Error> {
        let storage = binding(
            self.scandal.resolves_resources(),
            &self.arbor,
            table.resource_id,
            self.access.root(table).await?,
            write_access(WriteScope::Publication(Empty {})),
        )
        .await?;
        let input = input
            .map_err(|error| ScandalError::InvalidArgument {
                message: error.to_string(),
            })
            .boxed();
        Ok(self
            .scandal
            .write_blob(
                &TableBinding {
                    table_id: table.resource_id,
                    storage: Some(storage),
                },
                media_type,
                max_bytes,
                input,
            )
            .await?)
    }

    /// Record that the exact version `table` is stored at the local directory `root`.
    pub(crate) fn add_local_root(&self, table: TableIdentity, root: &Url) -> Result<(), Error> {
        self.access.add_local_root(table, root)
    }

    /// Begin the first version of a new Table stored at the local directory `root`.
    pub(crate) fn begin_local_create(
        &self,
        root: &Url,
        schema: &arrow_schema::Schema,
    ) -> Result<AttemptWrite, Error> {
        // Arbor names Tables with random UUIDs and their versions with time-ordered ones.
        Ok(self.attempt(
            Uuid::new_v4(),
            local_root(root)?,
            None,
            WriteBase::genesis(schema)?,
        ))
    }

    pub(crate) async fn create(
        &self,
        project: &str,
        input: WriteInput,
    ) -> Result<TableWriteResult, Error> {
        let mut write = self.begin_create(project, &input.schema()).await?;
        write.append(input).await?;
        register(&self.arbor, write.prepare().await?).await
    }

    /// Create a Table in `project` and begin its first version, with rows of `schema`, as one
    /// fresh write attempt.
    pub(crate) async fn begin_create(
        &self,
        project: &str,
        schema: &arrow_schema::Schema,
    ) -> Result<AttemptWrite, Error> {
        let created = self
            .arbor
            .create(&CreateTableRequest {
                project_id: project.into(),
            })
            .await?;
        Ok(self.attempt(
            created.table_id.0,
            created.root_uri,
            None,
            WriteBase::genesis(schema)?,
        ))
    }

    pub(crate) async fn write(
        &self,
        identity: TableIdentity,
        input: WriteInput,
        overwrite: bool,
    ) -> Result<TableWriteResult, Error> {
        let mut write = self.begin_write(identity, overwrite).await?;
        write.append(input).await?;
        register(&self.arbor, write.prepare().await?).await
    }

    pub(crate) async fn delete(
        &self,
        identity: TableIdentity,
        filter: Option<crate::Expression>,
        expected_rows: u64,
    ) -> Result<TableWriteResult, Error> {
        let mut write = self.begin_write(identity, false).await?;
        write.base = WriteBase::Delete(DeleteBase {
            version_id: identity.version_id,
            filter,
            attempt: write.attempt,
            storage: Some(write.staging.for_split().await?),
            expected_rows: expected_rows.into(),
        });
        register(&self.arbor, write.prepare().await?).await
    }

    /// Begin an edit of `identity` as one fresh write attempt: a successor that may take
    /// appended rows, patches, deletions and column files, after `schema_changes` apply to its
    /// schema.
    pub(crate) async fn begin_edit(
        &self,
        identity: TableIdentity,
        schema_changes: Vec<SchemaChange>,
    ) -> Result<AttemptWrite, Error> {
        let base = WriteBase::Edit(EditBase {
            version_id: identity.version_id,
            schema_changes: (!schema_changes.is_empty()).then_some(schema_changes),
        });
        Ok(self.attempt(
            identity.resource_id,
            self.access.root(identity).await?,
            Some(identity.version_id),
            base,
        ))
    }

    /// How version `to` of a Table changed since its version `from`, from metadata only.
    pub(crate) async fn plan_changes(
        &self,
        to: TableIdentity,
        from: Uuid,
        columns: Option<Vec<String>>,
    ) -> Result<ChangePlan, Error> {
        let bindings = self.access.reads(to).await?;
        let storage = bindings.for_split().await?;
        let request = PlanChangesRequest {
            table: to.resource_id,
            from,
            to: to.version_id,
            columns,
            storage: Some(storage.clone()),
        };
        bindings
            .discard_if_failed(&storage, self.scandal.plan_changes(&request).await)
            .await
    }

    /// Begin a successor of `identity` as one fresh write attempt.
    pub(crate) async fn begin_write(
        &self,
        identity: TableIdentity,
        overwrite: bool,
    ) -> Result<AttemptWrite, Error> {
        let base = if overwrite {
            WriteBase::overwrite(identity.version_id)
        } else {
            WriteBase::append(identity.version_id)
        };
        Ok(self.attempt(
            identity.resource_id,
            self.access.root(identity).await?,
            Some(identity.version_id),
            base,
        ))
    }

    fn attempt(
        &self,
        table_id: Uuid,
        root: String,
        base_version: Option<Uuid>,
        base: WriteBase,
    ) -> AttemptWrite {
        let attempt = Uuid::now_v7();
        AttemptWrite {
            scandal: Arc::clone(&self.scandal),
            arbor: self.arbor.clone(),
            staging: Arc::new(
                Bindings::new(
                    self.arbor.clone(),
                    table_id,
                    root.clone(),
                    write_access(WriteScope::Attempt(WriteAttempt {
                        attempt_id: attempt,
                    })),
                )
                .resolving(self.scandal.resolves_resources()),
            ),
            table_id,
            root,
            base_version,
            base,
            attempt,
            receipts: Vec::new(),
        }
    }
}

/// Rows an edit writes beside appended rows, each addressed by its row ID.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum RowEdit {
    /// New values recorded beside the data.
    Patch,
    /// Deletions, recorded beside the data.
    Delete,
}

/// Parts of one blob upload in flight at once.
const PARTS_IN_FLIGHT: usize = 4;

/// Times one part is sent again after a failure before its upload fails.
const PART_RETRIES: usize = 3;

/// One write attempt of a table: blobs uploaded into its directory and rows appended there,
/// prepared as one version. The attempt's binding creates files only in its directory, and
/// publication's only the version's metadata.
pub(crate) struct AttemptWrite {
    scandal: Arc<dyn Scandal>,
    arbor: ArborClient,
    staging: Arc<Bindings>,
    table_id: Uuid,
    root: String,
    base_version: Option<Uuid>,
    base: WriteBase,
    attempt: Uuid,
    receipts: Vec<FragmentReceipt>,
}

impl AttemptWrite {
    pub(crate) fn attempt(&self) -> Uuid {
        self.attempt
    }

    pub(crate) fn table_id(&self) -> Uuid {
        self.table_id
    }

    /// Upload `input` as one blob of this attempt, in parts sent at once and each retried on its
    /// own. A failed upload is aborted, so its parts are not kept.
    pub(crate) async fn upload_blob(
        &self,
        media_type: &str,
        input: BlobStream,
    ) -> Result<Blob, Error> {
        let table = TableBinding {
            table_id: self.table_id,
            storage: Some(self.staging.for_split().await?),
        };
        let upload = self
            .scandal
            .start_blob_upload(&table, self.attempt, media_type, None)
            .await?;
        let uploaded = self.upload_parts(&upload, input).await;
        if uploaded.is_err()
            && let Err(error) = self.abort(&upload).await
        {
            // The parts stay under the attempt's directory, which is reclaimed with it.
            tracing::warn!(%error, "aborting a failed blob upload failed");
        }
        uploaded
    }

    /// Store `payloads` as blobs of this attempt and return their `spql.Blob` column in order.
    /// Payloads of up to [`PACK_BATCH_BYTES`] are packed in one request; a larger one uploads on
    /// its own.
    pub(crate) async fn pack_blobs(
        &self,
        payloads: Vec<(Bytes, Option<String>)>,
    ) -> Result<ArrayRef, Error> {
        if payloads.is_empty() {
            return Ok(Blob::array(&[])?);
        }
        let (uploaded, packed): (Vec<_>, Vec<_>) = payloads
            .into_iter()
            .enumerate()
            .partition(|(_, (payload, _))| payload.len() > PACK_BATCH_BYTES);
        // Each row's blob as (source column, row of that column).
        let mut rows = vec![(0, 0); uploaded.len() + packed.len()];
        let mut columns = Vec::new();
        if !packed.is_empty() {
            for (row, (position, _)) in packed.iter().enumerate() {
                rows[*position] = (columns.len(), row);
            }
            columns.push(
                self.pack_inline(packed.into_iter().map(|(_, payload)| payload).collect())
                    .await?,
            );
        }
        for (position, (payload, media_type)) in uploaded {
            let media_type = media_type.as_deref().unwrap_or("application/octet-stream");
            let input = stream::once(future::ready(Ok(payload))).boxed();
            let blob = self.upload_blob(media_type, input).await?;
            rows[position] = (columns.len(), 0);
            columns.push(Blob::array(&[blob])?);
        }
        let columns: Vec<&dyn Array> = columns.iter().map(AsRef::as_ref).collect();
        interleave(&columns, &rows).map_err(Error::arrow)
    }

    /// Pack inline payloads through scandal, in batches of at most [`PACK_BATCH_BYTES`].
    async fn pack_inline(&self, payloads: Vec<(Bytes, Option<String>)>) -> Result<ArrayRef, Error> {
        let schema = Arc::new(Schema::new(vec![Blob::field("blob")]));
        let mut batches = Vec::new();
        let mut batch = Vec::new();
        let mut batch_bytes = 0;
        for payload in payloads {
            if batch_bytes + payload.0.len() > PACK_BATCH_BYTES && !batch.is_empty() {
                batches.push(inline_blobs(&schema, take(&mut batch))?);
                batch_bytes = 0;
            }
            batch_bytes += payload.0.len();
            batch.push(payload);
        }
        batches.push(inline_blobs(&schema, batch)?);
        let table = TableBinding {
            table_id: self.table_id,
            storage: Some(self.staging.for_split().await?),
        };
        let packed = self
            .scandal
            .pack_blobs(
                &table,
                Some(self.attempt),
                Arc::clone(&schema),
                stream::iter(batches.into_iter().map(Ok)).boxed(),
            )
            .await?;
        let columns: Vec<&dyn Array> = packed
            .iter()
            .map(|batch| batch.column(0).as_ref())
            .collect();
        concat(&columns).map_err(Error::arrow)
    }

    async fn abort(&self, upload: &BlobUpload) -> Result<(), Error> {
        let storage = self.staging.for_split().await?;
        Ok(self.scandal.abort_blob_upload(upload, &storage).await?)
    }

    async fn upload_parts(&self, upload: &BlobUpload, input: BlobStream) -> Result<Blob, Error> {
        let parts = fixed_parts(input, upload.part_bytes.0 as usize)
            .zip(stream::iter(1u32..))
            .map(|(bytes, part)| async move { self.upload_part(upload, part, bytes?).await })
            .buffered(PARTS_IN_FLIGHT)
            .try_collect::<Vec<_>>()
            .await?;
        if parts.is_empty() {
            return Err(Error::invalid("a blob upload holds at least one byte"));
        }
        let storage = self.staging.for_split().await?;
        Ok(self
            .scandal
            .complete_blob_upload(upload, &parts, &storage)
            .await?)
    }

    async fn upload_part(
        &self,
        upload: &BlobUpload,
        part: u32,
        bytes: Bytes,
    ) -> Result<BlobPart, Error> {
        let mut failures = 0;
        loop {
            let storage = self.staging.for_split().await?;
            let body = stream::once(future::ready(Ok(bytes.clone()))).boxed();
            match self
                .scandal
                .upload_blob_part(upload, part, body, &storage)
                .await
            {
                Ok(stored) => return Ok(stored),
                Err(error) if retryable(&error) && failures < PART_RETRIES => {
                    failures += 1;
                    if matches!(error, ScandalError::Unauthenticated { .. }) {
                        self.staging.discard(&storage).await;
                    }
                }
                Err(error) => return Err(error.into()),
            }
        }
    }

    /// Append `input` as fragments under this attempt.
    pub(crate) async fn append(&mut self, input: WriteInput) -> Result<(), Error> {
        let staging = TableBinding {
            table_id: self.table_id,
            storage: Some(self.staging.for_split().await?),
        };
        let request = AppendRowsRequest {
            table: staging,
            base: self.base_version,
            attempt: Some(self.attempt),
            properties: Some(input.properties),
            schema_changes: self.schema_changes(),
        };
        let mut writes = FuturesOrdered::new();
        for partition in input.partitions {
            let request = request.clone();
            let scandal = Arc::clone(&self.scandal);
            writes.push_back(async move {
                scandal
                    .append_partition(&request, partition.schema(), reader_stream(partition))
                    .await
            });
        }
        let receipts = writes.try_collect::<Vec<_>>().await?;
        self.receipts.extend(receipts.into_iter().flatten());
        Ok(())
    }

    /// Make this write an edit of its base version, so that it takes patches and deletions
    /// beside appended rows. An append base becomes an edit base without schema
    /// changes; an edit base stays as it is.
    pub(crate) fn edit_base(&mut self) -> Result<(), Error> {
        match &self.base {
            WriteBase::Edit(_) => Ok(()),
            WriteBase::Append(base) => {
                self.base = WriteBase::Edit(EditBase {
                    version_id: base.version_id,
                    schema_changes: None,
                });
                Ok(())
            }
            _ => Err(Error::invalid(
                "only a write that keeps its base version's rows can edit them",
            )),
        }
    }

    /// The schema changes this write applies, which every request writing its parts carries.
    fn schema_changes(&self) -> Option<Vec<SchemaChange>> {
        match &self.base {
            WriteBase::Edit(edit) => edit.schema_changes.clone(),
            _ => None,
        }
    }

    /// Write `input` as one part of an edit: rows addressed by row ID in `row_id_column`. The
    /// request borrows nothing from this write, so parts can run beside each other; keep the
    /// receipts it returns with [`Self::add_receipts`].
    pub(crate) fn edit_rows(
        &self,
        edit: RowEdit,
        row_id_column: &str,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> BoxFuture<'static, Result<Vec<FragmentReceipt>, Error>> {
        let scandal = Arc::clone(&self.scandal);
        let staging = Arc::clone(&self.staging);
        let (table_id, attempt, base) = (self.table_id, Some(self.attempt), self.base_version);
        let schema_changes = self.schema_changes();
        let row_id_column = row_id_column.to_owned();
        async move {
            let base = base.ok_or_else(|| Error::invalid("a new table has no rows to edit"))?;
            let table = TableBinding {
                table_id,
                storage: Some(staging.for_split().await?),
            };
            let receipts = match edit {
                RowEdit::Patch => {
                    let request = PatchRowsRequest {
                        attempt,
                        base,
                        row_id_column,
                        schema_changes,
                        table,
                    };
                    scandal.patch_rows(&request, schema, input).await
                }
                RowEdit::Delete => {
                    let request = DeleteRowsRequest {
                        attempt,
                        base,
                        row_id_column,
                        table,
                    };
                    scandal.delete_rows(&request, schema, input).await
                }
            };
            Ok(receipts?)
        }
        .boxed()
    }

    /// Attach values of the columns this edit adds, for every row that fragment `fragment_id`
    /// of the base version keeps, in stored order, as column files of the fragment. Like
    /// [`Self::edit_rows`], the request borrows nothing from this write.
    pub(crate) fn attach_columns(
        &self,
        fragment_id: Uuid,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> BoxFuture<'static, Result<AttachedColumns, Error>> {
        let scandal = Arc::clone(&self.scandal);
        let staging = Arc::clone(&self.staging);
        let (table_id, attempt, base) = (self.table_id, Some(self.attempt), self.base_version);
        let schema_changes = self.schema_changes();
        async move {
            let base =
                base.ok_or_else(|| Error::invalid("a new table has no rows to attach to"))?;
            let request = AttachColumnsRequest {
                table: TableBinding {
                    table_id,
                    storage: Some(staging.for_split().await?),
                },
                base,
                fragment_id,
                attempt,
                schema_changes,
            };
            Ok(scandal.attach_columns(&request, schema, input).await?)
        }
        .boxed()
    }

    /// Keep the receipts of parts written beside this write, for its version.
    pub(crate) fn add_receipts(&mut self, receipts: impl IntoIterator<Item = FragmentReceipt>) {
        self.receipts.extend(receipts);
    }

    /// Prepare a version on this write's base from every appended fragment.
    pub(crate) async fn prepare(self) -> Result<PreparedWriteReceipt, Error> {
        let publication = TableBinding {
            table_id: self.table_id,
            storage: Some(
                binding(
                    self.scandal.resolves_resources(),
                    &self.arbor,
                    self.table_id,
                    self.root,
                    write_access(WriteScope::Publication(Empty {})),
                )
                .await?,
            ),
        };
        Ok(self
            .scandal
            .finalize_write(&publication, &self.base, &self.receipts)
            .await?)
    }
}

/// `input` in parts of exactly `part_bytes`, and a shorter last part.
fn fixed_parts(input: BlobStream, part_bytes: usize) -> BoxStream<'static, Result<Bytes, Error>> {
    stream::try_unfold(
        (input, BytesMut::new(), false),
        move |(mut input, mut held, ended)| async move {
            while !ended && held.len() < part_bytes {
                match input.try_next().await? {
                    Some(chunk) => held.extend_from_slice(&chunk),
                    None => {
                        let last = held.split().freeze();
                        return Ok((!last.is_empty()).then_some((last, (input, held, true))));
                    }
                }
            }
            if held.is_empty() {
                return Ok(None);
            }
            let part = held.split_to(held.len().min(part_bytes)).freeze();
            Ok(Some((part, (input, held, ended))))
        },
    )
    .boxed()
}

/// `error` as the failure of an input stream that a scandal reads.
pub(crate) fn input_error(error: Error) -> ScandalError {
    match error {
        Error::Scandal(error) => error,
        error => ScandalError::InvalidArgument {
            message: error.to_string(),
        },
    }
}

/// Pull a synchronous Arrow reader on a blocking worker, one batch ahead.
#[expect(
    clippy::disallowed_methods,
    reason = "The open SDK adapts blocking Arrow callbacks without a Vortex runtime"
)]
pub(crate) fn reader_stream(mut reader: Box<dyn RecordBatchReader + Send>) -> RecordBatchStream {
    let (sender, receiver) = mpsc::channel(1);
    tokio::task::spawn_blocking(move || {
        while !sender.is_closed() {
            let Some(batch) = reader.next() else { break };
            let batch = batch.map_err(|error| ScandalError::InvalidArgument {
                message: error.to_string(),
            });
            let failed = batch.is_err();
            if sender.blocking_send(batch).is_err() || failed {
                break;
            }
        }
    });
    stream::unfold(receiver, |mut receiver| async move {
        receiver.recv().await.map(|batch| (batch, receiver))
    })
    .boxed()
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicI64, Ordering};
    use std::time::SystemTime;

    use arrow_array::Int64Array;
    use arrow_array::cast::AsArray as _;
    use arrow_array::types::Int64Type;
    use arrow_schema::{DataType, Field, Schema};
    use serde_json::json;
    use tokio::sync::Notify;
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    use super::*;
    use crate::ScanRowRange;
    use crate::api::tables::SecretToken;
    use crate::auth::TokenSource;
    use crate::scandal::blob::{Blob, BlobStat};
    use crate::scandal::expression::{Expression as SerializedExpression, NamedExpression};
    use crate::scandal::functions::FunctionInfo;
    use crate::scandal::types::{FragmentReceipt, PreparedWriteReceipt, RowRange, ScanPlan};
    use crate::scandal::vortex::expr::{col, gt, lit};
    use crate::scandal::{ByteStream, Evaluation, SplitRead, StorageCredential};

    #[test]
    fn a_credential_that_outlasts_the_clock_never_needs_replacing() {
        assert_eq!(usable_until(Some(Duration::MAX)), None);
        assert_eq!(usable_until(None), None);
        assert!(usable_until(Some(Duration::from_secs(3600))).is_some());
    }

    const ROOT: &str = "spfsu://4b44-t1/";

    /// Serve `table`'s read binding of [`ROOT`], signed for the next quarter hour, expecting
    /// `fetches` requests for it.
    async fn serve_read_binding(server: &MockServer, table: Uuid, fetches: u64) {
        let expires_at = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            + Duration::from_mins(15);
        Mock::given(method("POST"))
            .and(path(format!("/api/tables/{table}/storage-binding")))
            .respond_with(ResponseTemplate::new(200).set_body_json(TablesBinding {
                root_uri: ROOT.to_owned(),
                mode: StorageMode::ReadOnly,
                expires_at: expires_at.as_secs().into(),
                access: StorageAccess::Signed(SignedAccess {
                    endpoint: String::new(),
                    token: SecretToken("capability".into()),
                    direct_read: None,
                }),
            }))
            .expect(fetches)
            .mount(server)
            .await;
    }

    #[tokio::test]
    async fn resolver_hosts_receive_resource_references_without_sdk_storage_requests() {
        let server = MockServer::start().await;
        let client = crate::Client::builder()
            .api_url(server.uri())
            .build()
            .unwrap();
        let table = Uuid::now_v7();
        let binding = Bindings::new(client.tables().clone(), table, ROOT.into(), read_access())
            .resolving(true);
        assert_eq!(
            binding.for_split().await.unwrap(),
            StorageBinding::resource(table)
        );
        assert!(server.received_requests().await.unwrap().is_empty());
    }

    /// Reads of one version resolve its storage root and fetch its read binding once, and
    /// share that binding.
    #[tokio::test]
    async fn reads_of_one_version_share_its_root_and_binding() {
        let server = MockServer::start().await;
        let table = TableIdentity {
            resource_id: Uuid::new_v4(),
            version_id: Uuid::now_v7(),
        };
        Mock::given(method("GET"))
            .and(path(format!(
                "/api/tables/{}/versions/{}",
                table.resource_id, table.version_id
            )))
            .respond_with(
                ResponseTemplate::new(200).set_body_json(json!({"rootUri": ROOT, "seqNo": "1"})),
            )
            .expect(1)
            .mount(&server)
            .await;
        serve_read_binding(&server, table.resource_id, 1).await;
        let access = TableAccess::new(
            ArborClient::new(&server.uri())
                .unwrap()
                .with_token_source(TokenSource::fixed("test".into())),
        );
        let (first, again) = tokio::join!(access.reads(table), access.reads(table));
        let (first, again) = (first.unwrap(), again.unwrap());
        assert!(Arc::ptr_eq(&first, &again));
        assert_eq!(
            first.for_split().await.unwrap(),
            again.for_split().await.unwrap()
        );
        assert_eq!(access.root(table).await.unwrap(), ROOT);
    }

    /// A failed request discards the binding it used only while that binding is current: the
    /// next request then fetches a fresh one, and a binding already renewed stays.
    #[tokio::test]
    async fn a_failure_discards_only_the_binding_it_used() {
        let server = MockServer::start().await;
        let table = Uuid::new_v4();
        serve_read_binding(&server, table, 2).await;
        let bindings = Bindings::new(
            ArborClient::new(&server.uri())
                .unwrap()
                .with_token_source(TokenSource::fixed("test".into())),
            table,
            ROOT.to_owned(),
            read_access(),
        );
        let current = bindings.for_split().await.unwrap();
        let renewed_since = StorageBinding {
            credential: Some(StorageCredential("an earlier capability".to_owned())),
            ..current.clone()
        };
        bindings.discard(&renewed_since).await;
        assert_eq!(bindings.for_split().await.unwrap(), current);
        bindings.discard(&current).await;
        bindings.for_split().await.unwrap();
    }

    /// Takes beyond one request's bounds go out as several requests in order, and a resumed
    /// request takes only the positions after those delivered.
    #[test]
    fn takes_divide_into_requests_and_resume_after_delivered_positions() {
        let split = |fragment: u64| Split {
            fragment: fragment.into(),
            rows: RowRange::new(0..1),
        };
        let take = |fragment, positions: std::ops::Range<u64>| SplitTake {
            split: split(fragment),
            positions: positions.map(Into::into).collect(),
        };
        let shape = |requests: Vec<Vec<SplitTake>>| {
            requests
                .iter()
                .map(|request| {
                    request
                        .iter()
                        .map(|take| (take.split.fragment.0, take.positions.len()))
                        .collect::<Vec<_>>()
                })
                .collect::<Vec<_>>()
        };
        let large = MAX_TAKE_POSITIONS as u64;
        assert_eq!(
            shape(take_requests(
                [take(0, 0..large + 1), take(1, 0..2)].into_iter()
            )),
            [vec![(0, MAX_TAKE_POSITIONS)], vec![(0, 1), (1, 2)]]
        );
        let many = take_requests((0..=MAX_TAKES as u64).map(|fragment| take(fragment, 0..1)));
        assert_eq!(
            many.iter().map(Vec::len).collect::<Vec<_>>(),
            [MAX_TAKES, 1]
        );
        let rest = after_delivered(&[take(0, 0..2), take(1, 10..13)], 3);
        assert_eq!(rest, [take(1, 11..13)]);
    }

    /// Splits of one row each, holding their index; a split's read finishes only once the
    /// next split's read has started.
    struct ChainedSplits {
        schema: SchemaRef,
        started: Vec<Notify>,
        /// The filter each split's read asked for, with its fragment.
        filters: Mutex<Vec<(u64, Option<SerializedExpression>)>>,
    }

    fn unsupported() -> ScandalError {
        ScandalError::InvalidArgument {
            message: "the chained scandal only reads splits".into(),
        }
    }

    #[async_trait]
    impl Scandal for ChainedSplits {
        async fn list_functions(&self) -> Result<Vec<FunctionInfo>, ScandalError> {
            Err(unsupported())
        }

        async fn plan_scan(
            &self,
            _: &ScandalScan,
            _: &StorageBinding,
        ) -> Result<ScanPlan, ScandalError> {
            Err(unsupported())
        }

        async fn read_scan(
            &self,
            scan: &ScandalScan,
            split: &Split,
            _: &StorageBinding,
        ) -> Result<SplitRead, ScandalError> {
            self.filters
                .lock()
                .push((split.fragment.0, scan.filter.clone()));
            let index = split.fragment.0 as usize;
            self.started[index].notify_one();
            if let Some(next) = self.started.get(index + 1) {
                next.notified().await;
            }
            let batch = RecordBatch::try_new(
                Arc::clone(&self.schema),
                vec![Arc::new(Int64Array::from(vec![index as i64]))],
            )
            .unwrap();
            Ok(SplitRead {
                schema: Arc::clone(&self.schema),
                batches: stream::iter([Ok(SplitBatch {
                    batch,
                    next_row: split.rows.end.0,
                    metrics: None,
                })])
                .boxed(),
            })
        }

        async fn read_blob(
            &self,
            _: &Blob,
            _: Option<Range<u64>>,
            _: &StorageBinding,
        ) -> Result<ByteStream, ScandalError> {
            Err(unsupported())
        }

        async fn take_rows(
            &self,
            _: &ScandalScan,
            _: &[SplitTake],
            _: &StorageBinding,
        ) -> Result<SplitRead, ScandalError> {
            Err(unsupported())
        }

        async fn stat_blob(&self, _: &Blob, _: &StorageBinding) -> Result<BlobStat, ScandalError> {
            Err(unsupported())
        }

        async fn write_blob(
            &self,
            _: &TableBinding,
            _: &str,
            _: u64,
            _: ByteStream,
        ) -> Result<Blob, ScandalError> {
            Err(unsupported())
        }

        async fn pack_blobs(
            &self,
            _: &TableBinding,
            _: Option<Uuid>,
            _: SchemaRef,
            _: RecordBatchStream,
        ) -> Result<Vec<RecordBatch>, ScandalError> {
            Err(unsupported())
        }

        async fn start_blob_upload(
            &self,
            _: &TableBinding,
            _: Uuid,
            _: &str,
            _: Option<u64>,
        ) -> Result<BlobUpload, ScandalError> {
            Err(unsupported())
        }

        async fn upload_blob_part(
            &self,
            _: &BlobUpload,
            _: u32,
            _: ByteStream,
            _: &StorageBinding,
        ) -> Result<BlobPart, ScandalError> {
            Err(unsupported())
        }

        async fn complete_blob_upload(
            &self,
            _: &BlobUpload,
            _: &[BlobPart],
            _: &StorageBinding,
        ) -> Result<Blob, ScandalError> {
            Err(unsupported())
        }

        async fn abort_blob_upload(
            &self,
            _: &BlobUpload,
            _: &StorageBinding,
        ) -> Result<(), ScandalError> {
            Err(unsupported())
        }

        async fn append_rows(
            &self,
            _: &TableBinding,
            _: Option<Uuid>,
            _: Option<Uuid>,
            _: SchemaRef,
            _: RecordBatchStream,
        ) -> Result<Vec<FragmentReceipt>, ScandalError> {
            Err(unsupported())
        }

        async fn finalize_write(
            &self,
            _: &TableBinding,
            _: &WriteBase,
            _: &[FragmentReceipt],
        ) -> Result<PreparedWriteReceipt, ScandalError> {
            Err(unsupported())
        }

        async fn bind_projection(
            &self,
            _: &Schema,
            _: &[NamedExpression],
        ) -> Result<SchemaRef, ScandalError> {
            Err(unsupported())
        }

        async fn evaluate(
            &self,
            _: &[NamedExpression],
            _: SchemaRef,
            _: RecordBatchStream,
            _: &[TableBinding],
        ) -> Result<Evaluation, ScandalError> {
            Err(unsupported())
        }
    }

    /// A scan of a fresh table version.
    fn chained_scan() -> ScandalScan {
        ScandalScan::new(TableVersion {
            resource_id: Uuid::new_v4(),
            version_id: Uuid::now_v7(),
        })
    }

    /// Read four chained one-row splits with `scan`, in stored order unless `lanes` says
    /// otherwise: the fragments delivered, and the filter each split's read asked for.
    async fn read_chained(
        scan: ScandalScan,
        narrowing: Option<Arc<dyn SplitFilter>>,
        lanes: usize,
    ) -> (Vec<i64>, Vec<(u64, Option<SerializedExpression>)>) {
        let splits = 4;
        let scandal = Arc::new(ChainedSplits {
            schema: Arc::new(Schema::new(vec![Field::new(
                "fragment",
                DataType::Int64,
                false,
            )])),
            started: (0..splits).map(|_| Notify::new()).collect(),
            filters: Mutex::default(),
        });
        let reader = Arc::new(SplitReader {
            page_rows: None,
            scandal: Arc::clone(&scandal) as Arc<dyn Scandal>,
            scan,
            bindings: Arc::new(Bindings::new(
                ArborClient::new("http://127.0.0.1:9")
                    .unwrap()
                    .with_token_source(TokenSource::fixed("test".into())),
                Uuid::new_v4(),
                "file:///tables/t1/".to_owned(),
                read_access(),
            )),
            counters: None,
            narrowing,
        });
        let splits = (0..splits as u64)
            .map(|fragment| Split {
                fragment: fragment.into(),
                rows: RowRange::new(0..1),
            })
            .collect();
        let read = match lanes {
            1 => reader.read_in_order(splits),
            lanes => reader.read_unordered(splits, lanes),
        }
        .try_collect::<Vec<_>>();
        let batches = tokio::time::timeout(Duration::from_secs(10), read)
            .await
            .expect("each split's read waits for the next one to start")
            .unwrap();
        let fragments = batches
            .iter()
            .flat_map(|batch| {
                batch
                    .column(0)
                    .as_primitive::<Int64Type>()
                    .values()
                    .to_vec()
            })
            .collect();
        let filters = std::mem::take(&mut *scandal.filters.lock());
        (fragments, filters)
    }

    /// A scan's estimate sums its splits' estimates, saturating, and is unknown when any split
    /// has none.
    #[test]
    fn a_scan_estimates_the_rows_of_all_its_splits() {
        let estimate = |estimates: &[Option<u64>]| {
            let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
            ScandalScanSource {
                scandal: Arc::new(ChainedSplits {
                    schema: Arc::clone(&schema),
                    started: Vec::new(),
                    filters: Mutex::default(),
                }),
                identity: TableIdentity {
                    resource_id: Uuid::new_v4(),
                    version_id: Uuid::now_v7(),
                },
                bindings: Arc::new(Bindings::new(
                    ArborClient::new("http://127.0.0.1:9")
                        .unwrap()
                        .with_token_source(TokenSource::fixed("test".into())),
                    Uuid::new_v4(),
                    "file:///tables/t1/".to_owned(),
                    read_access(),
                )),
                scan: chained_scan(),
                schema,
                scan_splits: Vec::new(),
                splits: (0..)
                    .zip(estimates)
                    .map(|(fragment, rows)| PlannedSplit {
                        ordering: None,
                        split: Split {
                            fragment: fragment.into(),
                            rows: RowRange::new(0..1),
                        },
                        estimated_rows: rows.map(Into::into),
                        estimated_bytes: None,
                        fragment_id: None,
                    })
                    .collect(),
                plan_metrics: None,
                read: SplitReads::Rows { lanes: 1 },
            }
            .estimated_rows()
        };
        assert_eq!(estimate(&[]), Some(0));
        assert_eq!(estimate(&[Some(3), Some(4)]), Some(7));
        assert_eq!(estimate(&[Some(3), None]), None);
        assert_eq!(estimate(&[Some(u64::MAX), Some(1)]), Some(u64::MAX));
    }

    /// An ordered read opens the next splits while it delivers rows in stored order, under a
    /// limit too, and stops at the limit.
    #[tokio::test]
    async fn ordered_reads_open_later_splits_ahead() {
        let (fragments, _) = read_chained(chained_scan(), None, 1).await;
        assert_eq!(fragments, [0, 1, 2, 3]);
        let limited = ScandalScan {
            limit: Some(3.into()),
            ..chained_scan()
        };
        let (fragments, _) = read_chained(limited, None, 1).await;
        assert_eq!(fragments, [0, 1, 2]);
    }

    /// A split filter that is a tighter bound each time it is read.
    struct Tightening(AtomicI64);

    impl SplitFilter for Tightening {
        fn current(&self) -> Option<SerializedExpression> {
            let bound = self.0.fetch_add(1, Ordering::SeqCst);
            Some(SerializedExpression::new(gt(col("fragment"), lit(bound))).unwrap())
        }
    }

    /// Each row split opens with the scan's filter and the split filter as it stands then, in
    /// order and unordered.
    #[tokio::test]
    async fn row_splits_open_narrowed_by_the_split_filter_as_it_stands() {
        let filter = SerializedExpression::new(gt(col("fragment"), lit(-1i64))).unwrap();
        let expected = (0..4_i64)
            .map(|bound| {
                let narrowing = SerializedExpression::new(gt(col("fragment"), lit(bound))).unwrap();
                Some(filter.and(&narrowing).unwrap())
            })
            .collect::<Vec<_>>();
        for lanes in [1, 4] {
            let scan = ScandalScan {
                filter: Some(filter.clone()),
                ..chained_scan()
            };
            let narrowing = Arc::new(Tightening(AtomicI64::new(0)));
            let (_, requested) = read_chained(scan, Some(narrowing), lanes).await;
            assert_eq!(requested.len(), expected.len());
            assert!(
                requested
                    .iter()
                    .all(|(_, filter)| expected.contains(filter))
            );
        }
    }

    #[test]
    fn row_range_is_forwarded_to_self_contained_scandal_request() {
        let table = TableIdentity {
            resource_id: Uuid::new_v4(),
            version_id: Uuid::new_v4(),
        };
        let options = ScanOptions {
            row_range: Some(ScanRowRange { start: 12, end: 24 }),
            ..Default::default()
        };
        let scan = scandal_scan(table, &options).unwrap();
        assert_eq!(scan.row_range.unwrap().range(), 12..24);
        assert_eq!(scan.table.resource_id, table.resource_id);
        assert_eq!(scan.table.version_id, table.version_id);
        assert!(
            scandal_scan(
                table,
                &ScanOptions {
                    row_range: Some(ScanRowRange { start: 24, end: 12 }),
                    ..options
                }
            )
            .is_err()
        );
    }
}