moraine 0.7.0

//! Typed reads over the `inline/*` keyspace: chunks, tombstones, and
//! per-schema-version schema records. Mirrors `store::read`'s decode-only
//! contract — no DuckLake interpretation here.

pub(crate) mod stream;

use std::collections::{BTreeSet, HashMap};

use bytes::Bytes;

use crate::{
    error::{Error, Result},
    store::{
        handle::{ReadHandle, ScanShape},
        key::{
            InlineKey, InlineOperation, InlineOperationKind, Key, inline_chunk_locator_suffix,
            inline_chunk_locator_table_prefix, inline_chunk_range_table_prefix,
            inline_live_table_prefix, inline_row_tombstone_table_prefix,
            inline_schema_dropped_prefix, inline_schema_dropped_table_prefix, inline_schema_prefix,
            inline_schema_table_prefix,
        },
        proto::{
            InlineChunkRangeValue, InlineChunkValue, InlineFileDeleteTableValue,
            InlineFileDeleteValue, InlineInlineDeleteValue, InlineSchemaDroppedValue,
            InlineSchemaValue,
        },
        read::{read_singleton, scan_decode, scan_keys},
        value,
    },
};

/// One directory entry identifying an immutable inline chunk and its row
/// range.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct InlineChunkLocator {
    operation: InlineOperation,
    row_id_start: u64,
    row_id_end: u64,
}

impl InlineChunkLocator {
    /// The locator a chunk-directory write names: the key carries the
    /// chunk's identity and the end of its range, the value the start.
    pub(crate) fn from_directory(
        operation: InlineOperation,
        row_id_start: u64,
        row_id_end: u64,
    ) -> Result<Self> {
        if row_id_start > row_id_end {
            return Err(Error::Corruption(format!(
                "inline chunk directory names an inverted range {row_id_start}..={row_id_end}"
            )));
        }
        Ok(Self {
            operation,
            row_id_start,
            row_id_end,
        })
    }

    /// Metadata for a nonempty immutable insert chunk.
    pub(crate) fn from_chunk(
        operation: InlineOperation,
        chunk: &InlineChunkValue,
    ) -> Result<Option<Self>> {
        let Some(last) = chunk.row_count.checked_sub(1) else {
            return Ok(None);
        };
        let row_id_end = chunk
            .row_id_start
            .checked_add(last)
            .ok_or_else(|| Error::Corruption("inline chunk row range overflow".into()))?;
        Ok(Some(Self {
            operation,
            row_id_start: chunk.row_id_start,
            row_id_end,
        }))
    }

    /// The schema version this chunk decodes against; `None` for a
    /// locator naming a non-insert operation.
    pub(crate) fn schema_version(self) -> Option<u64> {
        match self.operation {
            InlineOperation::Insert { schema_version, .. } => Some(schema_version),
            InlineOperation::InlineDelete { .. } | InlineOperation::FileDelete { .. } => None,
        }
    }

    /// The chunk's store key operation.
    pub(crate) fn operation(self) -> InlineOperation {
        self.operation
    }

    /// The chunk's first row id.
    pub(crate) fn row_id_start(self) -> u64 {
        self.row_id_start
    }

    /// The chunk's last row id, inclusive.
    pub(crate) fn row_id_end(self) -> u64 {
        self.row_id_end
    }

    /// How many rows the chunk holds; ranges are validated non-inverted at
    /// decode.
    pub(crate) fn row_count(self) -> u64 {
        self.row_id_end - self.row_id_start + 1
    }
}

/// Resolves the directory entries owning `row_ids` without fetching their
/// chunk bodies. `None` means the directory is absent or incomplete.
pub(crate) async fn find_inline_chunk_locators_for_rows(
    handle: ReadHandle<'_>,
    table_id: u64,
    row_ids: &[u64],
    width_bound: Option<u64>,
) -> Result<Option<Vec<InlineChunkLocator>>> {
    let targets: Vec<u64> = row_ids
        .iter()
        .copied()
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect();
    let Some(first) = targets.first().copied() else {
        return Ok(Some(Vec::new()));
    };

    let prefix = inline_chunk_locator_table_prefix(table_id);
    let start = inline_chunk_locator_suffix(table_id, first);
    // A bounded walk ends within a few entries of the last target; an
    // unbounded one runs to the end of the directory.
    let shape = if width_bound.is_some() {
        ScanShape::Seek
    } else {
        ScanShape::Probe
    };
    let mut iter = handle.scan_prefix(prefix, start.., shape).await?;
    let mut locators = Vec::new();
    let mut found = 0usize;

    // Ranges can overlap — an UPDATE re-inlines a row inside another live
    // chunk's range — so a target is not owned by the first range that
    // covers it, and the walk cannot stop at a match. It can stop once no
    // remaining chunk could reach back to a target: a chunk holding `t`
    // spans at most `width_bound`, so it ends at or before `t + bound`.
    // An absent bound stops nothing.
    let last = targets.last().copied().unwrap_or(first);
    let scan_until = width_bound.map(|bound| last.saturating_add(bound));

    while let Some(entry) = iter.next().await? {
        let (row_id_end, operation) = match Key::decode(&entry.key)? {
            Key::Inline(InlineKey::ChunkLocator {
                table_id: owner,
                row_id_end,
                schema_version,
                begin_snapshot,
                chunk_seq,
            }) if owner == table_id => (
                row_id_end,
                InlineOperation::Insert {
                    table_id,
                    schema_version,
                    begin_snapshot,
                    chunk_seq,
                },
            ),
            other => {
                return Err(Error::Corruption(format!(
                    "non-directory key in inline chunk-range scan: {other:?}"
                )));
            }
        };
        if scan_until.is_some_and(|until| row_id_end > until) {
            break;
        }

        let locator: InlineChunkRangeValue = value::decode_owned(entry.value)?;
        if locator.row_id_start > row_id_end {
            return Err(Error::Corruption(format!(
                "inline chunk directory for table {table_id} has inverted range {}..={row_id_end}",
                locator.row_id_start
            )));
        }

        let low = targets.partition_point(|&target| target < locator.row_id_start);
        let high = targets.partition_point(|&target| target <= row_id_end);
        if low == high {
            continue;
        }
        found += high - low;
        locators.push(InlineChunkLocator {
            operation,
            row_id_start: locator.row_id_start,
            row_id_end,
        });
    }

    // Every target must land in some chunk; a row the directory cannot
    // place means it is absent or incomplete, and the caller falls back.
    if found < targets.len() {
        return Ok(None);
    }

    Ok(Some(locators))
}

/// Fetches and validates the immutable chunk named by `locator`.
pub(crate) async fn read_inline_chunk_locator(
    handle: ReadHandle<'_>,
    table_id: u64,
    locator: InlineChunkLocator,
) -> Result<(InlineOperation, InlineChunkValue)> {
    let Some(chunk): Option<InlineChunkValue> =
        read_singleton(handle, Key::Inline(InlineKey::Live(locator.operation))).await?
    else {
        return Err(Error::Corruption(format!(
            "inline chunk directory for table {table_id} names a missing chunk"
        )));
    };
    let actual_end = chunk
        .row_count
        .checked_sub(1)
        .and_then(|count| chunk.row_id_start.checked_add(count));
    if chunk.row_id_start != locator.row_id_start || actual_end != Some(locator.row_id_end) {
        return Err(Error::Corruption(format!(
            "inline chunk directory for table {table_id} disagrees with its chunk range"
        )));
    }
    Ok((locator.operation, chunk))
}

/// Every chunk-range locator for `table_id` as a full [`InlineChunkLocator`],
/// ordered by inclusive range end. The directory names its chunks without
/// carrying their bodies, so this scan stays small where
/// [`scan_inline_chunks`] hauls every chunk's Arrow payload.
pub(crate) async fn scan_inline_chunk_locators(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<InlineChunkLocator>> {
    scan_decode(
        handle,
        inline_chunk_locator_table_prefix(table_id),
        ScanShape::Probe,
        |key, bytes| match key {
            Key::Inline(InlineKey::ChunkLocator {
                table_id: owner,
                row_id_end,
                schema_version,
                begin_snapshot,
                chunk_seq,
            }) if owner == table_id => {
                let locator: InlineChunkRangeValue = value::decode_owned(bytes)?;
                if locator.row_id_start > row_id_end {
                    return Err(Error::Corruption(format!(
                        "inline chunk directory for table {table_id} has inverted range \
                         {}..={row_id_end}",
                        locator.row_id_start
                    )));
                }
                Ok(InlineChunkLocator {
                    operation: InlineOperation::Insert {
                        table_id,
                        schema_version,
                        begin_snapshot,
                        chunk_seq,
                    },
                    row_id_start: locator.row_id_start,
                    row_id_end,
                })
            }
            other => Err(Error::Corruption(format!(
                "non-directory key in inline chunk-range scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every chunk-range locator for `table_id` as its range end and the
/// identity of the chunk it names, ordered by inclusive range end. Reads no
/// values: the identity rides in the key.
pub(crate) async fn scan_inline_chunk_ranges(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<(u64, InlineOperation)>> {
    scan_decode(
        handle,
        inline_chunk_locator_table_prefix(table_id),
        ScanShape::Probe,
        |key, _| match key {
            Key::Inline(InlineKey::ChunkLocator {
                table_id: owner,
                row_id_end,
                schema_version,
                begin_snapshot,
                chunk_seq,
            }) if owner == table_id => Ok((
                row_id_end,
                InlineOperation::Insert {
                    table_id,
                    schema_version,
                    begin_snapshot,
                    chunk_seq,
                },
            )),
            other => Err(Error::Corruption(format!(
                "non-directory key in inline chunk-range scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every schema version `table_id` still holds live inline rows under,
/// taken from the chunk directory's keys alone. The directory names no
/// empty chunk, so a version it lists has rows a flush would write out.
pub(crate) async fn scan_inline_live_schema_versions(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<BTreeSet<u64>> {
    let versions = scan_decode(
        handle,
        inline_chunk_locator_table_prefix(table_id),
        ScanShape::Probe,
        |key, _| match key {
            Key::Inline(InlineKey::ChunkLocator { schema_version, .. }) => Ok(schema_version),
            other => Err(Error::Corruption(format!(
                "non-directory key in inline live schema-version scan: {other:?}"
            ))),
        },
    )
    .await?;

    Ok(versions.into_iter().collect())
}

/// Every superseded chunk-range key for `table_id`. Nothing writes these
/// any more; a store carried across the change still holds them, and the
/// directory's repair deletes what it finds.
pub(crate) async fn scan_legacy_inline_chunk_ranges(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<u64>> {
    scan_decode(
        handle,
        inline_chunk_range_table_prefix(table_id),
        ScanShape::Probe,
        |key, _| match key {
            Key::Inline(InlineKey::ChunkRange {
                table_id: owner,
                row_id_end,
            }) if owner == table_id => Ok(row_id_end),
            other => Err(Error::Corruption(format!(
                "non-directory key in superseded chunk-range scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every inlined-insert chunk for `table_id`, across all schema versions,
/// in key order (schema version, then commit snapshot, then chunk
/// sequence).
pub(crate) async fn scan_inline_chunks(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<(InlineOperation, InlineChunkValue)>> {
    scan_decode(
        handle,
        inline_live_table_prefix(InlineOperationKind::Insert, table_id),
        ScanShape::Probe,
        |key, bytes| match key {
            Key::Inline(InlineKey::Live(op @ InlineOperation::Insert { .. })) => {
                Ok((op, value::decode_owned(bytes)?))
            }
            other => Err(Error::Corruption(format!(
                "non-insert key in inline chunk scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every inlined-insert chunk for `table_id` as a body-free
/// [`InlineChunkLocator`], in the same key order [`scan_inline_chunks`]
/// walks. Each chunk's body is decoded and dropped rather than retained,
/// so a caller that fetches bodies per window holds none of them. Empty
/// chunks are skipped: they hold no row, and the directory never names
/// them.
///
/// Walked as bulk: the pass reads every chunk in the table and keeps
/// none, so admitting its blocks would evict the working set to cache
/// what nothing reads again.
pub(crate) async fn scan_inline_chunk_headers(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<InlineChunkLocator>> {
    let headers = scan_decode(
        handle,
        inline_live_table_prefix(InlineOperationKind::Insert, table_id),
        ScanShape::Bulk,
        |key, bytes| match key {
            Key::Inline(InlineKey::Live(op @ InlineOperation::Insert { .. })) => {
                let chunk: InlineChunkValue = value::decode_owned(bytes)?;
                InlineChunkLocator::from_chunk(op, &chunk)
            }
            other => Err(Error::Corruption(format!(
                "non-insert key in inline chunk header scan: {other:?}"
            ))),
        },
    )
    .await?;

    Ok(headers.into_iter().flatten().collect())
}

/// Every inlined-insert-row tombstone, ordered by row id and end snapshot.
pub(crate) async fn scan_inline_deletes(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<(u64, InlineInlineDeleteValue)>> {
    scan_decode(
        handle,
        inline_row_tombstone_table_prefix(table_id),
        ScanShape::Probe,
        |key, bytes| match key {
            Key::Inline(InlineKey::RowTombstone {
                row_id,
                end_snapshot,
                ..
            }) => {
                let value: InlineInlineDeleteValue = value::decode_owned(bytes)?;
                if value.end_snapshot != end_snapshot {
                    return Err(Error::Corruption(
                        "inline tombstone key and value disagree".to_owned(),
                    ));
                }
                Ok((row_id, value))
            }
            other => Err(Error::Corruption(format!(
                "non-tombstone key in inline tombstone scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every inlined Parquet-row delete for `table_id`, keyed by
/// `(data_file_id, row_id)`.
pub(crate) async fn scan_inline_file_deletes(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<(u64, u64, InlineFileDeleteValue)>> {
    scan_decode(
        handle,
        inline_live_table_prefix(InlineOperationKind::FileDelete, table_id),
        ScanShape::Probe,
        |key, bytes| match key {
            Key::Inline(InlineKey::Live(InlineOperation::FileDelete {
                data_file_id,
                row_id,
                ..
            })) => Ok((data_file_id, row_id, value::decode_owned(bytes)?)),
            other => Err(Error::Corruption(format!(
                "non-file_delete key in inline file delete scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Whether `table_id`'s `ducklake_inlined_delete_<table_id>` has been
/// marked as existing.
pub(crate) async fn read_inline_file_delete_table(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<bool> {
    let marker: Option<InlineFileDeleteTableValue> =
        read_singleton(handle, Key::Inline(InlineKey::FileDeleteTable { table_id })).await?;
    Ok(marker.is_some())
}

/// Resolves `schemas` (a table's records in key order) to each version's
/// Arrow bytes, following the one hop a `same_as_version` reference is
/// allowed. A reference naming a missing version, itself, or another
/// reference is corruption, not a fallback.
fn resolve_inline_schemas(
    table_id: u64,
    schemas: &[(u64, InlineSchemaValue)],
) -> Result<Vec<(u64, Bytes)>> {
    let canonical: HashMap<u64, &Bytes> = schemas
        .iter()
        .filter(|(_, value)| value.same_as_version.is_none())
        .map(|(schema_version, value)| (*schema_version, &value.arrow_schema))
        .collect();

    schemas
        .iter()
        .map(|(schema_version, value)| match value.same_as_version {
            None => Ok((*schema_version, value.arrow_schema.clone())),
            Some(target) => canonical
                .get(&target)
                .map(|bytes| (*schema_version, (*bytes).clone()))
                .ok_or_else(|| {
                    Error::Corruption(format!(
                        "inline schema {table_id}/{schema_version} references version {target}, \
                         which carries no schema of its own"
                    ))
                }),
        })
        .collect()
}

/// One table's Arrow IPC schema at `schema_version`, if recorded.
pub(crate) async fn read_inline_schema(
    handle: ReadHandle<'_>,
    table_id: u64,
    schema_version: u64,
) -> Result<Option<Bytes>> {
    let Some(record): Option<InlineSchemaValue> = read_singleton(
        handle,
        Key::Inline(InlineKey::Schema {
            table_id,
            schema_version,
        }),
    )
    .await?
    else {
        return Ok(None);
    };
    let Some(target) = record.same_as_version else {
        return Ok(Some(record.arrow_schema));
    };

    let canonical: Option<InlineSchemaValue> = read_singleton(
        handle,
        Key::Inline(InlineKey::Schema {
            table_id,
            schema_version: target,
        }),
    )
    .await?;
    match canonical {
        Some(value) if value.same_as_version.is_none() => Ok(Some(value.arrow_schema)),
        _ => Err(Error::Corruption(format!(
            "inline schema {table_id}/{schema_version} references version {target}, which carries \
             no schema of its own"
        ))),
    }
}

/// Every schema version recorded for `table_id` with its Arrow bytes,
/// references resolved, in key order.
pub(crate) async fn scan_inline_schemas_resolved(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<(u64, Bytes)>> {
    let schemas = scan_inline_schemas(handle, table_id).await?;
    resolve_inline_schemas(table_id, &schemas)
}

/// Every schema version recorded for `table_id`, in key order.
pub(crate) async fn scan_inline_schemas(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<(u64, InlineSchemaValue)>> {
    scan_decode(
        handle,
        inline_schema_table_prefix(table_id),
        ScanShape::Probe,
        |key, bytes| match key {
            Key::Inline(InlineKey::Schema { schema_version, .. }) => {
                Ok((schema_version, value::decode_owned(bytes)?))
            }
            other => Err(Error::Corruption(format!(
                "non-schema key in inline schema scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every `(table_id, schema_version)` with a schema record, in key order,
/// without decoding the schemas themselves — the registry needs the pair,
/// not the columns.
pub(crate) async fn scan_all_inline_schema_keys(handle: ReadHandle<'_>) -> Result<Vec<(u64, u64)>> {
    scan_keys(
        handle,
        inline_schema_prefix(),
        ScanShape::Probe,
        |key| match key {
            Key::Inline(InlineKey::Schema {
                table_id,
                schema_version,
            }) => Ok((table_id, schema_version)),
            other => Err(Error::Corruption(format!(
                "non-schema key in all-table inline schema key scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Whether `table_id`'s `schema_version` is still registered: it has a
/// schema record and no drop marker. A deregistered version keeps its
/// record so its name goes on resolving, so the record alone does not
/// answer this.
pub(crate) async fn read_inline_schema_registered(
    handle: ReadHandle<'_>,
    table_id: u64,
    schema_version: u64,
) -> Result<bool> {
    let (schema, dropped) = futures::try_join!(
        read_singleton::<InlineSchemaValue>(
            handle,
            Key::Inline(InlineKey::Schema {
                table_id,
                schema_version,
            }),
        ),
        read_singleton::<InlineSchemaDroppedValue>(
            handle,
            Key::Inline(InlineKey::SchemaDropped {
                table_id,
                schema_version,
            }),
        ),
    )?;
    Ok(schema.is_some() && dropped.is_none())
}

/// Every deregistered schema version of `table_id`, in key order. Their
/// `inline/schema` records are retained, so this is what separates a
/// version DuckLake still lists from one it has flushed away.
pub(crate) async fn scan_inline_dropped_schemas(
    handle: ReadHandle<'_>,
    table_id: u64,
) -> Result<Vec<u64>> {
    scan_decode(
        handle,
        inline_schema_dropped_table_prefix(table_id),
        ScanShape::Probe,
        |key, _| match key {
            Key::Inline(InlineKey::SchemaDropped { schema_version, .. }) => Ok(schema_version),
            other => Err(Error::Corruption(format!(
                "non-drop key in inline dropped-schema scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every deregistered `(table_id, schema_version)` across every table, in
/// key order.
pub(crate) async fn scan_all_inline_dropped_schemas(
    handle: ReadHandle<'_>,
) -> Result<Vec<(u64, u64)>> {
    scan_decode(
        handle,
        inline_schema_dropped_prefix(),
        ScanShape::Probe,
        |key, _| match key {
            Key::Inline(InlineKey::SchemaDropped {
                table_id,
                schema_version,
            }) => Ok((table_id, schema_version)),
            other => Err(Error::Corruption(format!(
                "non-drop key in all-table inline dropped-schema scan: {other:?}"
            ))),
        },
    )
    .await
}

/// Every `inline/schema` record across every table, in key order
/// (`table_id`, then `schema_version`). Registration reads take
/// [`scan_all_inline_schema_keys`]; this hauls the bodies too.
#[cfg(test)]
pub(crate) async fn scan_all_inline_schemas(
    handle: ReadHandle<'_>,
) -> Result<Vec<(u64, u64, InlineSchemaValue)>> {
    scan_decode(
        handle,
        inline_schema_prefix(),
        ScanShape::Probe,
        |key, bytes| match key {
            Key::Inline(InlineKey::Schema {
                table_id,
                schema_version,
            }) => Ok((table_id, schema_version, value::decode_owned(bytes)?)),
            other => Err(Error::Corruption(format!(
                "non-schema key in all-table inline schema scan: {other:?}"
            ))),
        },
    )
    .await
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use object_store::memory::InMemory;
    use slatedb::IsolationLevel;

    use super::*;
    use crate::store::open::StoreBuilder;

    /// The range directory resolves a row to exactly its chunk; a table
    /// with no directory reports `None`.
    #[tokio::test]
    async fn chunk_range_directory_selects_owners_and_detects_legacy_chunks() {
        let (db, _) = StoreBuilder::new("t", Arc::new(InMemory::new()))
            .open_writer()
            .await
            .unwrap();
        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();

        for (table_id, row_id_start, row_id_end, chunk_seq) in [
            (7, 0, 9, 0),
            (7, 100, 109, 1),
            (7, 1_000, 1_009, 2),
            (8, 0, 9, 0),
        ] {
            tx.put(
                Key::Inline(InlineKey::Live(InlineOperation::Insert {
                    table_id,
                    schema_version: 3,
                    begin_snapshot: 11,
                    chunk_seq,
                }))
                .encode(),
                value::encode_value(&InlineChunkValue {
                    body: format!("chunk-{table_id}-{chunk_seq}").into_bytes().into(),
                    row_id_start,
                    row_count: row_id_end - row_id_start + 1,
                    data_file_id: None,
                }),
            )
            .unwrap();
            if table_id == 7 {
                tx.put(
                    Key::Inline(InlineKey::ChunkLocator {
                        table_id,
                        row_id_end,
                        schema_version: 3,
                        begin_snapshot: 11,
                        chunk_seq,
                    })
                    .encode(),
                    value::encode_value(&InlineChunkRangeValue {
                        row_id_start,
                        schema_version: 3,
                        begin_snapshot: 11,
                        chunk_seq,
                    }),
                )
                .unwrap();
            }
        }
        tx.commit().await.unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        let selected =
            find_inline_chunk_locators_for_rows(ReadHandle::Tx(&tx), 7, &[105, 1_005], None)
                .await
                .unwrap()
                .unwrap();
        assert_eq!(selected.len(), 2);
        assert_eq!(selected[0].row_id_start, 100);
        assert_eq!(selected[1].row_id_start, 1_000);
        let (_, first) = read_inline_chunk_locator(ReadHandle::Tx(&tx), 7, selected[0])
            .await
            .unwrap();
        assert_eq!(first.body.as_ref(), b"chunk-7-1");
        assert_eq!(
            find_inline_chunk_locators_for_rows(ReadHandle::Tx(&tx), 8, &[5], None)
                .await
                .unwrap(),
            None,
        );

        tx.rollback();
        db.close().await.unwrap();
    }

    /// The header walk names the chunks the body scan walks, with the same
    /// ranges and the same scoping to one table, hauling no body.
    #[tokio::test]
    async fn chunk_headers_name_every_chunk_the_body_scan_walks() {
        let (db, _) = StoreBuilder::new("t", Arc::new(InMemory::new()))
            .open_writer()
            .await
            .unwrap();
        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();

        for (table_id, row_id_start, row_count, chunk_seq) in
            [(7, 0, 10, 0), (7, 10, 5, 1), (7, 40, 1, 2), (8, 0, 3, 0)]
        {
            tx.put(
                Key::Inline(InlineKey::Live(InlineOperation::Insert {
                    table_id,
                    schema_version: 3,
                    begin_snapshot: 11,
                    chunk_seq,
                }))
                .encode(),
                value::encode_value(&InlineChunkValue {
                    body: format!("chunk-{table_id}-{chunk_seq}").into_bytes().into(),
                    row_id_start,
                    row_count,
                    data_file_id: None,
                }),
            )
            .unwrap();
        }
        tx.commit().await.unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        let headers = scan_inline_chunk_headers(ReadHandle::Tx(&tx), 7)
            .await
            .unwrap();
        let walked: Vec<InlineChunkLocator> = scan_inline_chunks(ReadHandle::Tx(&tx), 7)
            .await
            .unwrap()
            .iter()
            .filter_map(|(operation, chunk)| {
                InlineChunkLocator::from_chunk(*operation, chunk).transpose()
            })
            .collect::<Result<Vec<_>>>()
            .unwrap();
        assert_eq!(headers, walked);
        assert_eq!(
            headers
                .iter()
                .map(|locator| (locator.row_id_start(), locator.row_id_end()))
                .collect::<Vec<_>>(),
            [(0, 9), (10, 14), (40, 40)]
        );

        tx.rollback();
        db.close().await.unwrap();
    }

    /// The width bound stops the walk once no remaining chunk could reach
    /// back to a target — but never before an overlapping chunk that does.
    /// An absent bound stops nothing.
    #[tokio::test]
    async fn locator_width_bound_stops_the_walk_without_skipping_an_overlap() {
        let (db, _) = StoreBuilder::new("t", Arc::new(InMemory::new()))
            .open_writer()
            .await
            .unwrap();
        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();

        // A wide chunk, a single-row chunk re-inlined inside it (sharing its
        // end), and a distant chunk the bound should exclude.
        for (row_id_start, row_id_end, chunk_seq) in [(0, 100, 0), (100, 100, 1), (9_000, 9_000, 2)]
        {
            tx.put(
                Key::Inline(InlineKey::Live(InlineOperation::Insert {
                    table_id: 7,
                    schema_version: 3,
                    begin_snapshot: 11,
                    chunk_seq,
                }))
                .encode(),
                value::encode_value(&InlineChunkValue {
                    body: format!("chunk-{chunk_seq}").into_bytes().into(),
                    row_id_start,
                    row_count: row_id_end - row_id_start + 1,
                    data_file_id: None,
                }),
            )
            .unwrap();
            tx.put(
                Key::Inline(InlineKey::ChunkLocator {
                    table_id: 7,
                    row_id_end,
                    schema_version: 3,
                    begin_snapshot: 11,
                    chunk_seq,
                })
                .encode(),
                value::encode_value(&InlineChunkRangeValue {
                    row_id_start,
                    schema_version: 3,
                    begin_snapshot: 11,
                    chunk_seq,
                }),
            )
            .unwrap();
        }
        tx.commit().await.unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();

        // Row 100 sits in both the wide chunk and the re-inlined one; the
        // bound must admit both.
        let bounded =
            find_inline_chunk_locators_for_rows(ReadHandle::Tx(&tx), 7, &[100], Some(100))
                .await
                .unwrap()
                .unwrap();
        let mut starts: Vec<u64> = bounded.iter().map(|locator| locator.row_id_start).collect();
        starts.sort_unstable();
        assert_eq!(
            starts,
            vec![0, 100],
            "the bound skipped a chunk that holds the target"
        );

        // Unbounded agrees, so the bound changed only how far the walk ran.
        let unbounded = find_inline_chunk_locators_for_rows(ReadHandle::Tx(&tx), 7, &[100], None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(unbounded.len(), bounded.len());

        tx.rollback();

        // Proof the bound truncates rather than merely agreeing: an
        // inverted locator far past it is refused by a full walk and never
        // reached by a bounded one.
        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        tx.put(
            Key::Inline(InlineKey::ChunkLocator {
                table_id: 7,
                row_id_end: 50_000,
                schema_version: 3,
                begin_snapshot: 11,
                chunk_seq: 9,
            })
            .encode(),
            value::encode_value(&InlineChunkRangeValue {
                row_id_start: 60_000,
                schema_version: 3,
                begin_snapshot: 11,
                chunk_seq: 9,
            }),
        )
        .unwrap();
        tx.commit().await.unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        assert!(
            find_inline_chunk_locators_for_rows(ReadHandle::Tx(&tx), 7, &[100], None)
                .await
                .is_err(),
            "a full walk must reach the sentinel"
        );
        assert!(
            find_inline_chunk_locators_for_rows(ReadHandle::Tx(&tx), 7, &[100], Some(100))
                .await
                .is_ok(),
            "the bound must stop the walk before the sentinel"
        );

        tx.rollback();
        db.close().await.unwrap();
    }

    /// Seeds inline records for two tables and one that overlaps chunk
    /// sequence numbers across two schema versions, then asserts every
    /// scan returns exactly its table's records, key-ordered, and that
    /// point reads work.
    #[tokio::test]
    #[allow(clippy::too_many_lines)]
    async fn scans_return_inline_records_in_key_order() {
        let (db, _) = StoreBuilder::new("t", Arc::new(InMemory::new()))
            .open_writer()
            .await
            .unwrap();

        let schema_v0 = InlineSchemaValue {
            arrow_schema: b"schema-v0".to_vec().into(),
            same_as_version: None,
        };
        let schema_v1 = InlineSchemaValue {
            arrow_schema: b"schema-v1".to_vec().into(),
            same_as_version: None,
        };

        let chunk_v0_seq0 = InlineChunkValue {
            body: b"chunk-v0-0".to_vec().into(),
            row_id_start: 0,
            row_count: 2,
            data_file_id: None,
        };
        let chunk_v0_seq1 = InlineChunkValue {
            body: b"chunk-v0-1".to_vec().into(),
            row_id_start: 2,
            row_count: 3,
            data_file_id: None,
        };
        let chunk_v1_seq0 = InlineChunkValue {
            body: b"chunk-v1-0".to_vec().into(),
            row_id_start: 5,
            row_count: 1,
            data_file_id: None,
        };
        let other_table_chunk = InlineChunkValue {
            body: b"other-table".to_vec().into(),
            row_id_start: 0,
            row_count: 1,
            data_file_id: None,
        };

        let inline_delete = InlineInlineDeleteValue { end_snapshot: 9 };
        let file_delete = InlineFileDeleteValue { begin_snapshot: 4 };

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        tx.put(
            Key::Inline(InlineKey::Schema {
                table_id: 7,
                schema_version: 0,
            })
            .encode(),
            value::encode_value(&schema_v0),
        )
        .unwrap();
        tx.put(
            Key::Inline(InlineKey::Schema {
                table_id: 7,
                schema_version: 1,
            })
            .encode(),
            value::encode_value(&schema_v1),
        )
        .unwrap();
        tx.put(
            Key::Inline(InlineKey::Live(InlineOperation::Insert {
                table_id: 7,
                schema_version: 0,
                begin_snapshot: 1,
                chunk_seq: 0,
            }))
            .encode(),
            value::encode_value(&chunk_v0_seq0),
        )
        .unwrap();
        tx.put(
            Key::Inline(InlineKey::Live(InlineOperation::Insert {
                table_id: 7,
                schema_version: 0,
                begin_snapshot: 1,
                chunk_seq: 1,
            }))
            .encode(),
            value::encode_value(&chunk_v0_seq1),
        )
        .unwrap();
        tx.put(
            Key::Inline(InlineKey::Live(InlineOperation::Insert {
                table_id: 7,
                schema_version: 1,
                begin_snapshot: 2,
                chunk_seq: 0,
            }))
            .encode(),
            value::encode_value(&chunk_v1_seq0),
        )
        .unwrap();
        tx.put(
            Key::Inline(InlineKey::RowTombstone {
                table_id: 7,
                row_id: 3,
                end_snapshot: 9,
            })
            .encode(),
            value::encode_value(&inline_delete),
        )
        .unwrap();
        tx.put(
            Key::Inline(InlineKey::Live(InlineOperation::FileDelete {
                table_id: 7,
                data_file_id: 5,
                row_id: 1,
            }))
            .encode(),
            value::encode_value(&file_delete),
        )
        .unwrap();
        // A different table's chunk must not leak into table 7's scans.
        tx.put(
            Key::Inline(InlineKey::Live(InlineOperation::Insert {
                table_id: 8,
                schema_version: 0,
                begin_snapshot: 1,
                chunk_seq: 0,
            }))
            .encode(),
            value::encode_value(&other_table_chunk),
        )
        .unwrap();

        tx.commit()
            .await
            .unwrap()
            .unwrap()
            .await_durable()
            .await
            .unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();

        let chunks = scan_inline_chunks(ReadHandle::Tx(&tx), 7).await.unwrap();
        assert_eq!(
            chunks,
            vec![
                (
                    InlineOperation::Insert {
                        table_id: 7,
                        schema_version: 0,
                        begin_snapshot: 1,
                        chunk_seq: 0,
                    },
                    chunk_v0_seq0,
                ),
                (
                    InlineOperation::Insert {
                        table_id: 7,
                        schema_version: 0,
                        begin_snapshot: 1,
                        chunk_seq: 1,
                    },
                    chunk_v0_seq1,
                ),
                (
                    InlineOperation::Insert {
                        table_id: 7,
                        schema_version: 1,
                        begin_snapshot: 2,
                        chunk_seq: 0,
                    },
                    chunk_v1_seq0,
                ),
            ]
        );

        let inline_deletes = scan_inline_deletes(ReadHandle::Tx(&tx), 7).await.unwrap();
        assert_eq!(inline_deletes, vec![(3, inline_delete)]);

        let file_deletes = scan_inline_file_deletes(ReadHandle::Tx(&tx), 7)
            .await
            .unwrap();
        assert_eq!(file_deletes, vec![(5, 1, file_delete)]);

        let schemas = scan_inline_schemas(ReadHandle::Tx(&tx), 7).await.unwrap();
        assert_eq!(
            schemas,
            vec![(0, schema_v0.clone()), (1, schema_v1.clone())]
        );

        assert_eq!(
            read_inline_schema(ReadHandle::Tx(&tx), 7, 0).await.unwrap(),
            Some(schema_v0.arrow_schema)
        );
        assert_eq!(
            read_inline_schema(ReadHandle::Tx(&tx), 7, 1).await.unwrap(),
            Some(schema_v1.arrow_schema)
        );
        assert_eq!(
            read_inline_schema(ReadHandle::Tx(&tx), 7, 2).await.unwrap(),
            None
        );

        let other_table_chunks = scan_inline_chunks(ReadHandle::Tx(&tx), 8).await.unwrap();
        assert_eq!(
            other_table_chunks,
            vec![(
                InlineOperation::Insert {
                    table_id: 8,
                    schema_version: 0,
                    begin_snapshot: 1,
                    chunk_seq: 0,
                },
                other_table_chunk,
            )]
        );

        tx.rollback();
        db.close().await.unwrap();
    }

    /// `scan_all_inline_schemas` returns every table's schema records
    /// together, in `(table_id, schema_version)` order, unlike the
    /// table-scoped `scan_inline_schemas`.
    #[tokio::test]
    async fn scan_all_inline_schemas_covers_every_table() {
        let (db, _) = StoreBuilder::new("t", Arc::new(InMemory::new()))
            .open_writer()
            .await
            .unwrap();

        let table_one_v0 = InlineSchemaValue {
            arrow_schema: b"a-0".to_vec().into(),
            same_as_version: None,
        };
        let table_one_v1 = InlineSchemaValue {
            arrow_schema: b"a-1".to_vec().into(),
            same_as_version: None,
        };
        let table_two_v0 = InlineSchemaValue {
            arrow_schema: b"b-0".to_vec().into(),
            same_as_version: None,
        };

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        for (table_id, schema_version, value) in [
            (1, 0, &table_one_v0),
            (1, 1, &table_one_v1),
            (2, 0, &table_two_v0),
        ] {
            tx.put(
                Key::Inline(InlineKey::Schema {
                    table_id,
                    schema_version,
                })
                .encode(),
                value::encode_value(value),
            )
            .unwrap();
        }
        tx.commit()
            .await
            .unwrap()
            .unwrap()
            .await_durable()
            .await
            .unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        let all = scan_all_inline_schemas(ReadHandle::Tx(&tx)).await.unwrap();
        assert_eq!(
            all,
            vec![
                (1, 0, table_one_v0),
                (1, 1, table_one_v1),
                (2, 0, table_two_v0),
            ]
        );
        tx.rollback();
        db.close().await.unwrap();
    }

    /// A reference resolves to its target's bytes in one hop, and a
    /// reference naming a version that carries none is corruption rather
    /// than a silently empty schema.
    #[tokio::test]
    async fn a_schema_reference_resolves_to_its_target_or_reports_corruption() {
        let (db, _) = StoreBuilder::new("t", Arc::new(InMemory::new()))
            .open_writer()
            .await
            .unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        for (table_id, schema_version, value) in [
            (
                1,
                0,
                InlineSchemaValue {
                    arrow_schema: Bytes::new(),
                    same_as_version: Some(1),
                },
            ),
            (
                1,
                1,
                InlineSchemaValue {
                    arrow_schema: b"columns".to_vec().into(),
                    same_as_version: None,
                },
            ),
            (
                2,
                0,
                InlineSchemaValue {
                    arrow_schema: Bytes::new(),
                    same_as_version: Some(9),
                },
            ),
        ] {
            tx.put(
                Key::Inline(InlineKey::Schema {
                    table_id,
                    schema_version,
                })
                .encode(),
                value::encode_value(&value),
            )
            .unwrap();
        }
        tx.commit()
            .await
            .unwrap()
            .unwrap()
            .await_durable()
            .await
            .unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        assert_eq!(
            scan_inline_schemas_resolved(ReadHandle::Tx(&tx), 1)
                .await
                .unwrap(),
            vec![
                (0, Bytes::from_static(b"columns")),
                (1, Bytes::from_static(b"columns"))
            ]
        );
        assert_eq!(
            read_inline_schema(ReadHandle::Tx(&tx), 1, 0).await.unwrap(),
            Some(Bytes::from_static(b"columns"))
        );
        assert!(
            scan_inline_schemas_resolved(ReadHandle::Tx(&tx), 2)
                .await
                .is_err()
        );
        assert!(read_inline_schema(ReadHandle::Tx(&tx), 2, 0).await.is_err());
        tx.rollback();
        db.close().await.unwrap();
    }

    /// The registration scan reads keys and never the schema bodies
    /// beside them. Proved by divergence: a record whose value cannot be
    /// decoded is still a registration, and only the scan that wants the
    /// columns fails on it.
    #[tokio::test]
    async fn scan_all_inline_schema_keys_never_decodes_a_schema_body() {
        let (db, _) = StoreBuilder::new("t", Arc::new(InMemory::new()))
            .open_writer()
            .await
            .unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        tx.put(
            Key::Inline(InlineKey::Schema {
                table_id: 1,
                schema_version: 0,
            })
            .encode(),
            value::encode_value(&InlineSchemaValue {
                arrow_schema: b"a-0".to_vec().into(),
                same_as_version: None,
            }),
        )
        .unwrap();
        tx.put(
            Key::Inline(InlineKey::Schema {
                table_id: 1,
                schema_version: 1,
            })
            .encode(),
            b"not a value at all",
        )
        .unwrap();
        tx.commit()
            .await
            .unwrap()
            .unwrap()
            .await_durable()
            .await
            .unwrap();

        let tx = db.begin(IsolationLevel::Snapshot).await.unwrap();
        let keys = scan_all_inline_schema_keys(ReadHandle::Tx(&tx))
            .await
            .unwrap();
        assert_eq!(keys, vec![(1, 0), (1, 1)]);
        assert!(scan_all_inline_schemas(ReadHandle::Tx(&tx)).await.is_err());
        tx.rollback();
        db.close().await.unwrap();
    }
}