moraine 0.7.0

//! The inline read seam: serves DuckLake's four inline scan variants over
//! the `inline/*` keyspace and re-exports
//! [`InlineScanKind`](crate::catalog::inline::InlineScanKind) from the
//! otherwise-private `catalog`. Each function here opens a fresh
//! read-only transaction, scans, and rolls back; an
//! [`InlineScanCursor`] instead holds one open across the windows it
//! serves, until it is finished.

use std::collections::HashSet;

use bytes::Bytes;

#[doc(hidden)]
pub use crate::catalog::inline::{InlineBodies, InlineScanKind};
use crate::{
    catalog::{InlineScan, ReadOnlyCatalog},
    error::{Error, Result},
    store::{inline as store_inline, key::InlineOperation},
};

/// One inlined row, referencing its chunk's body by index into its
/// window's deduplicated
/// [`chunk_bodies`](InlineScanRecord::chunk_bodies).
#[doc(hidden)]
pub struct InlineRowRecord {
    /// The row's dense id.
    pub row_id: u64,
    /// The schema version the owning chunk was written under.
    pub schema_version: u64,
    /// The commit snapshot that inserted this row.
    pub begin_snapshot: u64,
    /// The commit snapshot that tombstoned this row, if any.
    pub end_snapshot: Option<u64>,
    /// The owning chunk: an index into the scan's `chunk_bodies`.
    pub chunk_index: u64,
    /// The row's index within its chunk (`0..row_count`).
    pub offset_in_chunk: u64,
}

/// One window of a scan: the rows plus the chunk bodies they reference.
#[doc(hidden)]
pub struct InlineScanRecord {
    /// The window's rows, in the scan variant's order.
    pub rows: Vec<InlineRowRecord>,
    /// The referenced chunks' full Arrow IPC record-batch bodies, each
    /// appearing once, indexed by [`InlineRowRecord::chunk_index`].
    pub chunk_bodies: Vec<Bytes>,
}

/// A resumable inline scan: `table_id`'s rows under `kind`'s variant at
/// `snapshot` (windowed from `start` for the incremental variants),
/// selected once and returned a window at a time — the read model behind
/// `moraine_inline_scan_open`. The selection is served from the
/// chunk-range directory once it is known complete, and each window
/// point-reads only the chunk bodies its own rows reference, so a
/// consumer that drops each window never holds the table's payload.
/// `schema_version`, when set, keeps only rows whose chunk was written
/// under that version, so a caller serving one version's projection never
/// reads another's.
#[doc(hidden)]
pub struct InlineScanCursor {
    scan: InlineScan,
}

impl InlineScanCursor {
    /// Opens the scan, holding one read session until [`Self::finish`].
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying store scan fails or decodes
    /// corrupt bytes.
    pub async fn open(
        catalog: &ReadOnlyCatalog,
        table_id: u64,
        kind: InlineScanKind,
        snapshot: u64,
        start: u64,
        schema_version: Option<u64>,
        bodies: InlineBodies,
    ) -> Result<Self> {
        let scan = catalog
            .open_inline_scan(table_id, kind, snapshot, start, schema_version, bodies)
            .await?;

        Ok(Self { scan })
    }

    /// The next at most `max_rows` rows, with the bodies they reference.
    /// An exhausted scan returns an empty record.
    ///
    /// # Errors
    ///
    /// Returns an error if a chunk body read fails, or if the rows and
    /// chunks disagree.
    pub async fn next_window(&mut self, max_rows: usize) -> Result<InlineScanRecord> {
        let (selected, chunks) = self.scan.next_window(max_rows).await?;

        // Rows arrive with `chunk` dense-indexed in first-reference order
        // over exactly the referenced chunks, which is the record's shape
        // already.
        let chunk_bodies: Vec<Bytes> = chunks.iter().map(|(_, body)| body.clone()).collect();
        let rows = selected
            .into_iter()
            .map(|row| {
                let (operation, _) = &chunks[row.chunk];
                let InlineOperation::Insert { schema_version, .. } = operation else {
                    return Err(Error::Corruption(format!(
                        "inline row {} references a non-insert chunk key: {operation:?}",
                        row.row_id
                    )));
                };

                Ok(InlineRowRecord {
                    row_id: row.row_id,
                    schema_version: *schema_version,
                    begin_snapshot: row.begin_snapshot,
                    end_snapshot: row.end_snapshot,
                    chunk_index: row.chunk as u64,
                    offset_in_chunk: row.offset_in_chunk,
                })
            })
            .collect::<Result<Vec<_>>>()?;

        Ok(InlineScanRecord { rows, chunk_bodies })
    }

    /// Releases the scan's read session.
    pub fn finish(self) {
        self.scan.finish();
    }
}

/// [`InlineScanCursor`]'s whole selection in one record.
///
/// # Errors
///
/// Returns an error if the underlying store scan fails or decodes
/// corrupt bytes.
#[doc(hidden)]
pub async fn scan_inline(
    catalog: &ReadOnlyCatalog,
    table_id: u64,
    kind: InlineScanKind,
    snapshot: u64,
    start: u64,
    schema_version: Option<u64>,
) -> Result<InlineScanRecord> {
    let mut cursor = InlineScanCursor::open(
        catalog,
        table_id,
        kind,
        snapshot,
        start,
        schema_version,
        InlineBodies::Fetch,
    )
    .await?;
    let record = cursor.next_window(usize::MAX).await;
    cursor.finish();

    record
}

/// Every `(schema_version, arrow_schema)` recorded for `table_id`, in
/// schema-version order — the read model behind `moraine_inline_schemas`.
///
/// # Errors
///
/// Returns an error if the underlying store scan fails or decodes
/// corrupt bytes.
#[doc(hidden)]
pub async fn inline_schemas(catalog: &ReadOnlyCatalog, table_id: u64) -> Result<Vec<(u64, Bytes)>> {
    let session = catalog.begin_read().await?;
    let schemas = store_inline::scan_inline_schemas_resolved(session.handle(), table_id).await;
    session.finish();
    schemas
}

/// Every `(table_id, schema_version)` still registered, across every
/// table — feeds the `ducklake_inlined_data_tables` projection behind
/// `moraine_inline_registered_tables`. A flushed-away version's schema
/// record is retained so its name keeps resolving, so the drop markers
/// are what deregister it here.
///
/// # Errors
///
/// Returns an error if the underlying store scan fails or decodes
/// corrupt bytes.
#[doc(hidden)]
pub async fn inline_registered_tables(catalog: &ReadOnlyCatalog) -> Result<Vec<(u64, u64)>> {
    let session = catalog.begin_read().await?;
    let scans = futures::try_join!(
        store_inline::scan_all_inline_schema_keys(session.handle()),
        store_inline::scan_all_inline_dropped_schemas(session.handle()),
    );
    session.finish();
    let (registered, dropped) = scans?;
    let dropped: HashSet<(u64, u64)> = dropped.into_iter().collect();
    Ok(registered
        .into_iter()
        .filter(|pair| !dropped.contains(pair))
        .collect())
}

/// Whether `(table_id, schema_version)` is one of the pairs
/// [`inline_registered_tables`] lists — the read model behind
/// `moraine_inline_table_registered`, which the `CREATE TABLE IF NOT
/// EXISTS ducklake_inlined_data_<t>_<v>` existence gate consults.
///
/// # Errors
///
/// Returns an error if the underlying store read fails or decodes
/// corrupt bytes.
#[doc(hidden)]
pub async fn inline_table_registered(
    catalog: &ReadOnlyCatalog,
    table_id: u64,
    schema_version: u64,
) -> Result<bool> {
    let session = catalog.begin_read().await?;
    let registered =
        store_inline::read_inline_schema_registered(session.handle(), table_id, schema_version)
            .await;
    session.finish();
    registered
}

/// Whether `table_id`'s `ducklake_inlined_delete_<table_id>` exists: it
/// does not until the first `inline/file_delete` is staged, and stays
/// existing after a flush clears the records. The record fallback covers
/// stores written before the marker.
///
/// # Errors
///
/// Returns an error if the underlying store read fails or decodes
/// corrupt bytes.
#[doc(hidden)]
pub async fn inline_file_delete_table_exists(
    catalog: &ReadOnlyCatalog,
    table_id: u64,
) -> Result<bool> {
    let session = catalog.begin_read().await?;
    let marked = store_inline::read_inline_file_delete_table(session.handle(), table_id).await;
    let exists = match marked {
        Ok(true) => Ok(true),
        Ok(false) => store_inline::scan_inline_file_deletes(session.handle(), table_id)
            .await
            .map(|records| !records.is_empty()),
        Err(error) => Err(error),
    };
    session.finish();
    exists
}

/// Every `inline/file_delete` record for `table_id` as
/// `(data_file_id, row_id, begin_snapshot)` in key order — the rows behind
/// the `ducklake_inlined_delete_<t>` projection.
///
/// # Errors
///
/// Returns an error if the underlying store scan fails or decodes
/// corrupt bytes.
#[doc(hidden)]
pub async fn inline_file_deletes(
    catalog: &ReadOnlyCatalog,
    table_id: u64,
) -> Result<Vec<(u64, u64, u64)>> {
    let session = catalog.begin_read().await?;
    let file_deletes = store_inline::scan_inline_file_deletes(session.handle(), table_id).await;
    session.finish();
    Ok(file_deletes?
        .into_iter()
        .map(|(data_file_id, row_id, value)| (data_file_id, row_id, value.begin_snapshot))
        .collect())
}

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

    use object_store::memory::InMemory;

    use super::*;
    use crate::{
        catalog::CatalogOptions,
        transaction::staged::{Cell, RowOperation, StagedTransaction, TableKind},
    };

    fn snapshot_row(id: u64) -> Vec<Cell> {
        vec![
            Cell::U64(id),
            Cell::I64(1),
            Cell::U64(0),
            Cell::U64(1),
            Cell::U64(0),
        ]
    }

    fn snapshot_changes_row(id: u64) -> Vec<Cell> {
        vec![
            Cell::U64(id),
            Cell::Str("inlined_insert:1".to_string()),
            Cell::Null,
            Cell::Null,
            Cell::Null,
        ]
    }

    async fn open() -> crate::catalog::Catalog {
        crate::catalog::Catalog::open(Arc::new(InMemory::new()), CatalogOptions::default())
            .await
            .unwrap()
    }

    /// Two chunks (rows 0-1, row 2) staged in one commit, one tombstone
    /// on row 1: `scan_inline` with `Table` at the tombstone's snapshot
    /// returns rows 0 and 2 with their chunk bodies attached, and
    /// `inline_schemas`/`inline_registered_tables` see the recorded
    /// schema.
    #[tokio::test]
    async fn scan_inline_materializes_rows_with_chunk_bodies() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);

        tx.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema".to_vec(),
        });
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 2,
            arrow_body: b"chunk-a".to_vec().into(),
        });
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 2,
            row_count: 1,
            arrow_body: b"chunk-b".to_vec().into(),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();

        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut inline_delete = StagedTransaction::begin_detached(&catalog, db_tx2);
        inline_delete.stage(RowOperation::InlineInlineDelete {
            table_id: 1,
            row_id: 1,
            end_snapshot: 2,
        });
        inline_delete.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2),
        });
        inline_delete.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2),
        });
        inline_delete.commit().await.unwrap();

        let record = scan_inline(&catalog, 1, InlineScanKind::Table, 2, 0, None)
            .await
            .unwrap();
        let mut by_id: Vec<(u64, Bytes, u64)> = record
            .rows
            .iter()
            .map(|r| {
                (
                    r.row_id,
                    record.chunk_bodies[usize::try_from(r.chunk_index).unwrap()].clone(),
                    r.offset_in_chunk,
                )
            })
            .collect();
        by_id.sort_by_key(|(id, ..)| *id);
        assert_eq!(
            by_id,
            vec![
                (0, Bytes::from_static(b"chunk-a"), 0),
                (2, Bytes::from_static(b"chunk-b"), 0)
            ]
        );
        // Each referenced chunk's body crosses once.
        assert_eq!(record.chunk_bodies.len(), 2);

        let schemas = inline_schemas(&catalog, 1).await.unwrap();
        assert_eq!(schemas, vec![(0, Bytes::from_static(b"schema"))]);

        let registered = inline_registered_tables(&catalog).await.unwrap();
        assert_eq!(registered, vec![(1, 0)]);
    }

    /// Three chunks of two rows each, committed together, then read a
    /// window at a time: the windows concatenate to what one whole scan
    /// returns, and each carries only the bodies its own rows reference.
    #[tokio::test]
    async fn inline_scan_windows_carry_only_the_bodies_their_rows_reference() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);

        tx.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema".to_vec(),
        });
        for chunk in 0..3u64 {
            tx.stage(RowOperation::InlineInsert {
                table_id: 1,
                schema_version: 0,
                begin_snapshot: 1,
                row_id_start: chunk * 2,
                row_count: 2,
                arrow_body: format!("chunk-{chunk}").into_bytes().into(),
            });
        }
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();

        let whole = scan_inline(&catalog, 1, InlineScanKind::ForFlush, 1, 0, Some(0))
            .await
            .unwrap();
        assert_eq!(whole.rows.len(), 6);

        let mut cursor = InlineScanCursor::open(
            &catalog,
            1,
            InlineScanKind::ForFlush,
            1,
            0,
            Some(0),
            InlineBodies::Fetch,
        )
        .await
        .unwrap();

        let mut streamed = Vec::new();
        let mut windows = 0;
        loop {
            let window = cursor.next_window(3).await.unwrap();
            if window.rows.is_empty() {
                // An exhausted scan keeps serving empty windows.
                assert!(window.chunk_bodies.is_empty());
                break;
            }
            assert_eq!(window.rows.len(), 3);
            // Three rows spanning two chunks carry two bodies, never the
            // table's six rows' worth.
            assert_eq!(window.chunk_bodies.len(), 2);
            for row in &window.rows {
                let body = window.chunk_bodies[usize::try_from(row.chunk_index).unwrap()].clone();
                streamed.push((row.row_id, body));
            }
            windows += 1;
        }
        assert_eq!(windows, 2);

        let expected: Vec<(u64, Bytes)> = whole
            .rows
            .iter()
            .map(|row| {
                (
                    row.row_id,
                    whole.chunk_bodies[usize::try_from(row.chunk_index).unwrap()].clone(),
                )
            })
            .collect();
        assert_eq!(streamed, expected);
        cursor.finish();

        catalog.close().await.unwrap();
    }

    /// A scan opened without bodies selects the same rows and names the
    /// same chunks, with every body empty.
    #[tokio::test]
    async fn a_body_free_inline_scan_names_its_chunks_without_reading_them() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);

        tx.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema".to_vec(),
        });
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 2,
            arrow_body: b"chunk".to_vec().into(),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();

        let mut cursor = InlineScanCursor::open(
            &catalog,
            1,
            InlineScanKind::ForFlush,
            1,
            0,
            Some(0),
            InlineBodies::Skip,
        )
        .await
        .unwrap();
        let window = cursor.next_window(usize::MAX).await.unwrap();
        cursor.finish();

        assert_eq!(
            window.rows.iter().map(|row| row.row_id).collect::<Vec<_>>(),
            [0, 1]
        );
        assert!(window.rows.iter().all(|row| row.schema_version == 0));
        assert_eq!(window.chunk_bodies, vec![Bytes::new()]);

        catalog.close().await.unwrap();
    }

    /// A dropped schema version leaves `ducklake_inlined_data_tables`
    /// but stays resolvable by name. A reader that cached the registry
    /// before the drop still binds `ducklake_inlined_data_<t>_<v>` and
    /// scans it empty, where a deregistration that also took the columns
    /// would fail its bind; a reader that loads the registry after the
    /// drop never learns the version at all.
    #[tokio::test]
    async fn a_dropped_schema_version_deregisters_but_stays_resolvable() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);

        for (schema_version, schema) in [(0u64, b"schema-v0"), (1, b"schema-v1")] {
            tx.stage(RowOperation::InlineSchema {
                table_id: 1,
                schema_version,
                arrow_schema: schema.to_vec(),
            });
        }
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 1,
            arrow_body: b"chunk".to_vec().into(),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();

        assert_eq!(
            inline_registered_tables(&catalog).await.unwrap(),
            vec![(1, 0), (1, 1)]
        );

        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut flush = StagedTransaction::begin_detached(&catalog, db_tx2);
        flush.stage(RowOperation::InlineFlushDelete {
            table_id: 1,
            schema_version: 0,
            flush_snapshot: 2,
        });
        flush.stage(RowOperation::InlineSchemaDrop {
            table_id: 1,
            schema_version: 0,
        });
        flush.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2),
        });
        flush.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2),
        });
        flush.commit().await.unwrap();

        assert_eq!(
            inline_registered_tables(&catalog).await.unwrap(),
            vec![(1, 1)],
            "the flushed version leaves the registry"
        );
        assert_eq!(
            inline_schemas(&catalog, 1).await.unwrap(),
            vec![
                (0, Bytes::from_static(b"schema-v0")),
                (1, Bytes::from_static(b"schema-v1"))
            ],
            "the flushed version still resolves its columns"
        );

        let record = scan_inline(&catalog, 1, InlineScanKind::Table, 2, 0, Some(0))
            .await
            .unwrap();
        assert!(
            record.rows.is_empty(),
            "the retained version scans empty; its rows are in the flushed file"
        );
    }

    /// A deregistered schema version registers again when its schema is
    /// staged anew, and the retained record alone never counts as
    /// registration.
    #[tokio::test]
    async fn restaging_a_dropped_schema_version_registers_it_again() {
        let catalog = open().await;

        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);
        tx.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema-v0".to_vec(),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();
        assert!(inline_table_registered(&catalog, 1, 0).await.unwrap());

        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut drop_tx = StagedTransaction::begin_detached(&catalog, db_tx2);
        drop_tx.stage(RowOperation::InlineSchemaDrop {
            table_id: 1,
            schema_version: 0,
        });
        drop_tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2),
        });
        drop_tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2),
        });
        drop_tx.commit().await.unwrap();
        assert!(
            !inline_table_registered(&catalog, 1, 0).await.unwrap(),
            "the retained schema record does not keep the version registered"
        );

        let db_tx3 = catalog.begin_write_tx().await.unwrap();
        let mut recreate = StagedTransaction::begin_detached(&catalog, db_tx3);
        recreate.stage(RowOperation::InlineSchema {
            table_id: 1,
            schema_version: 0,
            arrow_schema: b"schema-v0".to_vec(),
        });
        recreate.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(3),
        });
        recreate.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(3),
        });
        recreate.commit().await.unwrap();

        assert!(inline_table_registered(&catalog, 1, 0).await.unwrap());
        assert_eq!(
            inline_registered_tables(&catalog).await.unwrap(),
            vec![(1, 0)]
        );

        // A version no table ever recorded is registered nowhere.
        assert!(!inline_table_registered(&catalog, 1, 7).await.unwrap());
        assert!(!inline_table_registered(&catalog, 9, 0).await.unwrap());
    }

    /// The first `scan_inline` walk verifies the chunk directory and
    /// remembers it; the next serves from it, hauling only referenced
    /// bodies. The proof is divergence: a chunk planted without a locator
    /// is visible only to a body scan.
    #[tokio::test]
    async fn scan_inline_serves_from_the_directory_once_verified() {
        use crate::store::{
            key::{InlineKey, InlineOperation, Key},
            proto, value,
        };

        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached_on(&catalog, db_tx);
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 2,
            arrow_body: b"chunk-a".to_vec().into(),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();

        let record = scan_inline(&catalog, 1, InlineScanKind::Table, 1, 0, None)
            .await
            .unwrap();
        assert_eq!(record.rows.len(), 2);
        assert!(
            crate::catalog::projection::inline_directory_complete(catalog.projections(), 1),
            "the first walk must verify and remember a complete directory"
        );

        // The divergence: a chunk with no locator, behind the memo's back.
        let tx = catalog.begin_write_tx().await.unwrap();
        tx.put(
            Key::Inline(InlineKey::Live(InlineOperation::Insert {
                table_id: 1,
                schema_version: 0,
                begin_snapshot: 1,
                chunk_seq: 5,
            }))
            .encode(),
            value::encode_value(&proto::InlineChunkValue {
                body: b"rogue".to_vec().into(),
                row_id_start: 10,
                row_count: 2,
                data_file_id: None,
            }),
        )
        .unwrap();
        tx.commit().await.unwrap();

        let record = scan_inline(&catalog, 1, InlineScanKind::Table, 1, 0, None)
            .await
            .unwrap();
        let row_ids: Vec<u64> = record.rows.iter().map(|row| row.row_id).collect();
        assert_eq!(
            row_ids,
            [0, 1],
            "a directory-served scan must not see a chunk only a body scan finds"
        );
        assert_eq!(record.chunk_bodies, vec![Bytes::from_static(b"chunk-a")]);
    }

    /// A version-filtered scan returns one version's rows and hauls one
    /// version's bodies — the other version's chunks are not in the record
    /// at all, so a caller serving one version's projection never touches
    /// the rest of the table.
    #[tokio::test]
    async fn scan_inline_filtered_to_a_version_hauls_only_its_bodies() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);

        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 0,
            begin_snapshot: 1,
            row_id_start: 0,
            row_count: 2,
            arrow_body: b"chunk-v0".to_vec().into(),
        });
        tx.stage(RowOperation::InlineInsert {
            table_id: 1,
            schema_version: 1,
            begin_snapshot: 1,
            row_id_start: 2,
            row_count: 1,
            arrow_body: b"chunk-v1".to_vec().into(),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();

        // Once eagerly (verifying the directory), once served from it: the
        // filter must narrow both paths the same way.
        for pass in ["chunk scan", "directory"] {
            let record = scan_inline(&catalog, 1, InlineScanKind::Table, 1, 0, Some(1))
                .await
                .unwrap();
            let rows: Vec<(u64, u64, u64)> = record
                .rows
                .iter()
                .map(|row| (row.row_id, row.schema_version, row.chunk_index))
                .collect();
            assert_eq!(rows, vec![(2, 1, 0)], "one version-1 row ({pass})");
            assert_eq!(
                record.chunk_bodies,
                vec![Bytes::from_static(b"chunk-v1")],
                "the other version's body must not be hauled ({pass})"
            );
        }
    }

    /// Staged `inline/file_delete` rows read back as
    /// `(data_file_id, row_id, begin_snapshot)` in key order — the rows
    /// behind the `ducklake_inlined_delete_<t>` projection.
    #[tokio::test]
    async fn inline_file_deletes_read_back_in_key_order() {
        let catalog = open().await;
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);

        tx.stage(RowOperation::InlineFileDelete {
            table_id: 1,
            data_file_id: 7,
            row_id: 2,
            begin_snapshot: 6,
        });
        tx.stage(RowOperation::InlineFileDelete {
            table_id: 1,
            data_file_id: 7,
            row_id: 0,
            begin_snapshot: 6,
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(6),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(6),
        });
        tx.commit().await.unwrap();

        let rows = inline_file_deletes(&catalog, 1).await.unwrap();
        assert_eq!(rows, vec![(7, 0, 6), (7, 2, 6)]);

        assert!(inline_file_delete_table_exists(&catalog, 1).await.unwrap());
        assert_eq!(inline_file_deletes(&catalog, 9).await.unwrap(), vec![]);
    }

    /// A version left deregistered with rows still under it is re-listed,
    /// and one whose flush emptied it is left alone.
    #[tokio::test]
    async fn reregistering_relists_only_deregistered_versions_that_still_hold_rows() {
        let catalog = open().await;

        // Table 1 carries two versions, both holding rows; table 2 carries
        // one, which its flush will empty.
        let db_tx = catalog.begin_write_tx().await.unwrap();
        let mut tx = StagedTransaction::begin_detached(&catalog, db_tx);
        for (table_id, schema_version) in [(1u64, 0u64), (1, 1), (2, 0)] {
            tx.stage(RowOperation::InlineSchema {
                table_id,
                schema_version,
                arrow_schema: format!("schema-{table_id}-{schema_version}").into_bytes(),
            });
            tx.stage(RowOperation::InlineInsert {
                table_id,
                schema_version,
                begin_snapshot: 1,
                row_id_start: schema_version * 10,
                row_count: 2,
                arrow_body: b"chunk".to_vec().into(),
            });
        }
        tx.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(1),
        });
        tx.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(1),
        });
        tx.commit().await.unwrap();

        // Table 2's version is flushed and then deregistered, which is the
        // ordinary path. Table 1 version 0 is deregistered with its rows
        // still live, which is the stranding.
        let db_tx2 = catalog.begin_write_tx().await.unwrap();
        let mut flush = StagedTransaction::begin_detached(&catalog, db_tx2);
        flush.stage(RowOperation::InlineFlushDelete {
            table_id: 2,
            schema_version: 0,
            flush_snapshot: 2,
        });
        flush.stage(RowOperation::InlineSchemaDrop {
            table_id: 2,
            schema_version: 0,
        });
        flush.stage(RowOperation::InlineSchemaDrop {
            table_id: 1,
            schema_version: 0,
        });
        flush.stage(RowOperation::Insert {
            table: TableKind::Snapshot,
            cells: snapshot_row(2),
        });
        flush.stage(RowOperation::Insert {
            table: TableKind::SnapshotChanges,
            cells: snapshot_changes_row(2),
        });
        flush.commit().await.unwrap();
        assert_eq!(
            inline_registered_tables(&catalog).await.unwrap(),
            vec![(1, 1)],
            "both versions leave the registry, only one of them emptied"
        );

        assert_eq!(
            catalog.reregister_stranded_inline_schemas().await.unwrap(),
            vec![(1, 0)]
        );
        assert_eq!(
            inline_registered_tables(&catalog).await.unwrap(),
            vec![(1, 0), (1, 1)],
            "the stranded version is listed again; the emptied one stays out"
        );
        assert!(
            inline_table_registered(&catalog, 1, 0).await.unwrap(),
            "and its rows are enumerable again by the next flush"
        );

        // Nothing is stranded twice.
        assert_eq!(
            catalog.reregister_stranded_inline_schemas().await.unwrap(),
            vec![]
        );

        catalog.close().await.unwrap();
    }
}