coordinode-lsm-tree 6.0.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026-present, Dmitry Prudnikov

//! Reading a columnar row group: its page directory, then the pages a read
//! wants.
//!
//! The layout this reads is specified in `docs/columnar-page-format.md`. A
//! full read takes the group in one request; a projection reads the
//! directory, then only the pages of the columns it asked for.

use alloc::{sync::Arc, vec::Vec};

use super::util::{LookupFd, ReadCharge, admit_read_block, build_block_transform};
#[cfg(feature = "metrics")]
use super::util::{record_block_decoded, record_block_load_cached, record_block_read};
use super::{Block, BlockHandle, BlockOffset, GlobalTableId};
use crate::coding::Decode;
use crate::fs::FsFile;
use crate::path::Path;
use crate::table::block::{BlockType, EccParams, Header};
use crate::table::column_page::{PageDirectory, PageEntry, PageStamp, PageZones};
use crate::{
    Cache, CompressionType, Slice, encryption::EncryptionProvider, file_accessor::FileAccessor,
};

#[cfg(feature = "metrics")]
use crate::metrics::Metrics;

/// A columnar row group as read: its directory and the pages that were
/// fetched, each verified and cached as the block it is.
pub struct RowGroupBlocks {
    /// The group's directory, shared with the cache that keeps it decoded.
    pub directory: Arc<PageDirectory>,
    /// The row pages the read selected.
    pub row_pages: RowPageSet,
    /// One slot per page the read wanted, in the directory's order.
    pub pages: Vec<PageSlot>,
}

impl RowGroupBlocks {
    /// Every page of the group, `pages` holding one block per directory entry
    /// in the directory's order.
    #[must_use]
    pub fn whole(directory: Arc<PageDirectory>, pages: Vec<Block>) -> Self {
        let row_pages = RowPageSet::every(&directory);
        let pages = (0..)
            .zip(pages)
            .map(|(index, block)| PageSlot {
                index,
                block: Some(block),
            })
            .collect();
        Self {
            directory,
            row_pages,
            pages,
        }
    }
}

/// One page as [`RowGroupBlocks::page_encodings`] describes it.
#[cfg(feature = "std")]
pub struct PageDescription {
    /// The page's directory entry.
    pub entry: PageEntry,
    /// The rows of its row page.
    pub rows: u32,
    /// What its values were encoded as.
    pub expression: crate::table::columnar::Expression,
    /// For a bytes column, the bytes it spends on where its values start.
    pub offsets_len: Option<usize>,
}

/// A page a read wanted: its place in the directory's entries and its block
/// once fetched.
pub struct PageSlot {
    /// The page's index into [`PageDirectory::entries`].
    pub index: usize,
    /// The page's block, `None` until it is fetched.
    pub block: Option<Block>,
}

/// The row pages of a group a read selected, in row order.
#[derive(Clone, Debug)]
pub enum RowPageSet {
    /// A run of consecutive row pages.
    Range(core::ops::Range<u16>),
    /// Row pages picked one by one, ascending.
    Listed(Vec<u16>),
}

impl RowPageSet {
    /// Every row page of `directory`.
    fn every(directory: &PageDirectory) -> Self {
        // A directory holds at most `u16::MAX` row pages.
        #[expect(
            clippy::cast_possible_truncation,
            reason = "a directory holds at most u16::MAX row pages"
        )]
        let count = directory.row_pages().len() as u16;
        Self::Range(0..count)
    }

    /// The selected row pages, ascending.
    fn iter(&self) -> impl Iterator<Item = u16> + '_ {
        let (range, listed) = match self {
            Self::Range(range) => (range.clone(), &[][..]),
            Self::Listed(listed) => (0..0, listed.as_slice()),
        };
        range.chain(listed.iter().copied())
    }
}

/// Which row pages of a group a read wants.
#[derive(Clone, Debug, Default)]
pub enum RowPageSelect<'a> {
    /// Every row page.
    #[default]
    All,
    /// The row pages of this half-open range of ordinals.
    Range(core::ops::Range<u16>),
    /// The row pages whose statistics zone for `column_id` may hold a
    /// non-null value within the inclusive `[lower, upper]`, either side
    /// unbounded when `None`. A row page the column has no zone for is kept:
    /// only a zone can prove a page holds no match.
    Zone {
        /// The column the zones are read for.
        column_id: u16,
        /// Inclusive lower bound, or `None` for none.
        lower: Option<&'a [u8]>,
        /// Inclusive upper bound, or `None` for none.
        upper: Option<&'a [u8]>,
    },
}

/// The pages a read of a row group wants: the columns it projects, times the
/// row pages that hold the rows it asks for.
#[derive(Clone, Debug, Default)]
pub struct PageWant<'a> {
    /// The column ids read, or every column when `None`.
    pub columns: Option<&'a [u16]>,
    /// The row pages read.
    pub row_pages: RowPageSelect<'a>,
    /// Whether the read expects to want every page: a group that fits the
    /// I/O buffer is then taken in one request instead of its directory
    /// first, and the pages it does not want are neither verified nor cached.
    pub whole: bool,
}

impl<'a> PageWant<'a> {
    /// Every page of the group.
    pub const ALL: Self = Self {
        columns: None,
        row_pages: RowPageSelect::All,
        whole: true,
    };

    /// The pages of `columns` on the row pages `row_pages` selects, read
    /// directory first.
    #[must_use]
    pub const fn projected(columns: &'a [u16], row_pages: RowPageSelect<'a>) -> Self {
        Self {
            columns: Some(columns),
            row_pages,
            whole: false,
        }
    }

    /// The column this want selects row pages by, when it selects by zones.
    fn zone_column(&self) -> Option<u16> {
        match self.row_pages {
            RowPageSelect::Zone { column_id, .. } => Some(column_id),
            RowPageSelect::All | RowPageSelect::Range(_) => None,
        }
    }

    /// The column whose zone block resolving this want against `directory`
    /// needs: it selects by a column whose zones the directory does not keep,
    /// and the group has a zone block for it.
    fn zone_block_column(&self, directory: &PageDirectory) -> Option<u16> {
        self.zone_column().filter(|&column_id| {
            !directory.head_zones().describes(column_id)
                && directory.zone_block(column_id).is_some()
        })
    }

    /// This want resolved against `directory` and, when it was read, the
    /// group's zone block: the row pages it selects.
    ///
    /// # Errors
    ///
    /// [`crate::Error::InvalidHeader`] for an empty range or one past the
    /// group's row pages: a caller names a range from what it read of this
    /// group, so one outside it is a reader error, not an empty selection.
    fn resolve(
        &self,
        directory: &PageDirectory,
        block_zones: Option<&PageZones>,
    ) -> crate::Result<Wanted<'a>> {
        let row_pages = match &self.row_pages {
            RowPageSelect::All => RowPageSet::every(directory),
            RowPageSelect::Range(range) => {
                if range.start >= range.end || usize::from(range.end) > directory.row_pages().len()
                {
                    return Err(crate::Error::InvalidHeader(
                        "columnar: row page range outside the group",
                    ));
                }
                RowPageSet::Range(range.clone())
            }
            RowPageSelect::Zone {
                column_id,
                lower,
                upper,
            } => RowPageSet::Listed(
                // A directory holds at most `u16::MAX` row pages, so the
                // ordinals cover every one.
                (0u16..)
                    .zip(directory.row_pages())
                    .filter(|&(ordinal, &rows)| {
                        directory
                            .head_zones()
                            .zone(ordinal, *column_id)
                            .or_else(|| block_zones.and_then(|z| z.zone(ordinal, *column_id)))
                            .is_none_or(|zone| zone_may_match(zone, rows, *lower, *upper))
                    })
                    .map(|(ordinal, _)| ordinal)
                    .collect(),
            ),
        };
        Ok(Wanted {
            columns: self.columns,
            row_pages,
        })
    }
}

/// Whether a row page whose column has `zone` over `rows` rows may hold a
/// non-null value within `[lower, upper]`.
fn zone_may_match(
    zone: crate::table::column_page::Zone<'_>,
    rows: u32,
    lower: Option<&[u8]>,
    upper: Option<&[u8]>,
) -> bool {
    if zone.null_count == rows {
        // Every row is null, and a null is within no range.
        return false;
    }
    if let (Some(lower), Some(max)) = (lower, zone.max)
        && max < lower
    {
        return false;
    }
    if let Some(upper) = upper
        && zone.min > upper
    {
        return false;
    }
    true
}

/// A [`PageWant`] resolved against a group's directory.
struct Wanted<'a> {
    columns: Option<&'a [u16]>,
    row_pages: RowPageSet,
}

impl Wanted<'_> {
    /// The indexes into `directory`'s entries of the pages the read needs,
    /// ascending: the selected row pages of each projected column part.
    ///
    /// The directory proved its entries a grid, each part's pages one per row
    /// page in row page order, so a page's index is its part's times the row
    /// page count plus its row page, and only the parts are looked at.
    fn slots(&self, directory: &PageDirectory) -> Vec<PageSlot> {
        let row_page_count = directory.row_pages().len();
        let mut slots = Vec::new();
        for (part, first) in directory
            .entries()
            .iter()
            .step_by(row_page_count.max(1))
            .enumerate()
        {
            if self
                .columns
                .is_some_and(|columns| !columns.contains(&first.id.column_id))
            {
                continue;
            }
            let base = part * row_page_count;
            slots.extend(self.row_pages.iter().map(|row_page| PageSlot {
                index: base + usize::from(row_page),
                block: None,
            }));
        }
        slots
    }
}

/// A row group decoded row page by row page: one batch per row page read, in
/// row order, each holding the columns the read projected.
#[derive(Debug)]
pub struct RowPages {
    /// Each row page's ordinal, one per batch.
    pub ordinals: Vec<u16>,
    /// One batch per row page read, in row order.
    pub batches: Vec<crate::table::columnar::ColumnBatch>,
}

impl RowPages {
    /// The rows the batches hold between them.
    #[must_use]
    pub fn row_count(&self) -> u32 {
        // The directory proved the row pages sum to the group's `u32` row
        // count, so a subset of them cannot overflow.
        self.batches.iter().map(|b| b.row_count).sum()
    }

    /// The batches joined into one ([`ColumnBatch::concat`]), for a consumer
    /// that takes the group whole.
    ///
    /// [`ColumnBatch::concat`]: crate::table::columnar::ColumnBatch::concat
    ///
    /// # Errors
    ///
    /// As [`ColumnBatch::concat`].
    pub fn into_batch(self) -> crate::Result<crate::table::columnar::ColumnBatch> {
        crate::table::columnar::ColumnBatch::concat(self.batches)
    }
}

/// Everything a row group read needs to fetch, verify, count and cache the
/// blocks of one group: the context [`crate::table::util::load_block`] takes
/// for one block, plus the group's index entry.
pub struct GroupRead<'a> {
    pub table_id: GlobalTableId,
    pub path: &'a Path,
    pub file_accessor: &'a FileAccessor,
    pub cache: &'a Cache,
    /// The index entry, which spans the whole group.
    pub group: &'a BlockHandle,
    pub compression: CompressionType,
    pub encryption: Option<&'a dyn EncryptionProvider>,
    pub ecc: Option<EccParams>,
    #[cfg(zstd_any)]
    pub zstd_dict: Option<&'a crate::compression::ZstdDictionary>,
    pub heal_hints: Option<&'a crate::heal_hints::HealHints>,
    #[cfg(feature = "metrics")]
    pub metrics: &'a Metrics,
    /// Whose read this is, applied as `load_block` applies it.
    pub charge: ReadCharge,
    /// How much one request may ask for and how many go in flight at once.
    pub budget: crate::config::ReadBudget,
    /// Told how long each request to the file took, when it is active then.
    pub pace: Option<&'a dyn crate::table::util::ReadPacer>,
    /// The table file, shared with the other block reads of the same lookup.
    pub fd: &'a LookupFd,
}

impl GroupRead<'_> {
    /// Loads the group's directory and the pages `want` selects.
    ///
    /// Nothing is read for a block already cached. Every block read is
    /// verified and admitted separately ([`admit_read_block`]), so each is
    /// cached under its own offset with its own buffer: a page cached as a
    /// view of a shared read would keep the whole read alive while the cache
    /// charged it as one page.
    ///
    /// When the directory is cached, only the wanted pages that are not are
    /// read, each run of consecutive ones as requests of up to the budget's
    /// I/O buffer: a full read after a point read has fetched the key page
    /// reads the rest of the group, not the key page again. When it is not, a
    /// read that expects every page ([`PageWant::whole`]) of a group that fits
    /// the I/O buffer takes it in ONE request, as a full scan wants, and any
    /// other read takes exactly the directory first, whose length the index
    /// entry records, and then the wanted pages the same way. Requests go out
    /// the budget's in-flight count at a time.
    ///
    /// Each request is charged when it is issued: the whole group and page
    /// runs to the data role, the directory to the index role. Each block's
    /// load and decode are then counted under its own role as it is verified.
    ///
    /// # Errors
    ///
    /// Any block's verification error, or [`crate::Error::InvalidHeader`] when
    /// the index entry names no row group, when the directory's length is not
    /// the one the index entry records, or when the directory describes a
    /// layout that does not fill the group exactly: a gap or an overrun means
    /// the index entry and the directory disagree about where the group ends,
    /// and neither can be trusted to name the right bytes.
    pub(crate) fn load(&self, want: &PageWant<'_>) -> crate::Result<RowGroupBlocks> {
        if let Some((directory, directory_len)) = self.cache.get_directory(
            self.table_id,
            self.group.offset(),
            self.charge.touches_cache(),
        ) {
            return self.load_from_directory(directory, directory_len, want, true);
        }
        // A read that leaves the cache as it found it reads alone, as a block
        // load does.
        if !self.charge.touches_cache() {
            return self.read_selective(want, None);
        }
        match self.cache.claim_directory(
            self.table_id,
            self.group.offset(),
            crate::cache::LoadIdentity::Block {
                block_type: BlockType::ColumnPageDirectory,
                size: self.group.size(),
                compression: CompressionType::None,
                encrypted: self.encryption.is_some(),
            },
        ) {
            crate::cache::DirectoryClaim::Shared(directory, directory_len) => {
                #[cfg(feature = "metrics")]
                if self.charge.is_counted() {
                    self.metrics
                        .block_load_shared
                        .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
                }
                self.load_from_directory(directory, directory_len, want, false)
            }
            crate::cache::DirectoryClaim::Lead(lead) => {
                // A read that finished between the miss above and this claim
                // left the directory in the cache.
                if let Some((directory, directory_len)) =
                    self.cache
                        .get_directory(self.table_id, self.group.offset(), true)
                {
                    lead.publish(&directory, directory_len);
                    return self.load_from_directory(directory, directory_len, want, true);
                }
                self.read_selective(want, Some(lead))
            }
            crate::cache::DirectoryClaim::Alone => self.read_selective(want, None),
        }
    }

    /// Loads the pages `want` selects in the group whose decoded `directory`
    /// is in hand, from the cache or another read's load of it (`cached`
    /// says which), reading only the pages not cached.
    fn load_from_directory(
        &self,
        directory: Arc<PageDirectory>,
        directory_len: u32,
        want: &PageWant<'_>,
        cached: bool,
    ) -> crate::Result<RowGroupBlocks> {
        // Decoded and checked when it was cached or loaded; its length,
        // extent and tag are checked again against this read's index
        // entry, which is what names the group here.
        check_directory_len(indexed_directory_len(self.group)?, directory_len)?;
        check_group_extent(self.group, directory_len, &directory)?;
        check_head_zones_len(self.group, &directory)?;
        self.check_group_tag(&directory)?;
        // A directory cached by a read that did not select by its head
        // zone block's column lacks those zones: the first read that does
        // reads the block and caches the directory with them.
        let directory = if want
            .zone_column()
            .is_some_and(|column_id| directory.lacks_head_zones_for(column_id))
        {
            let directory = self.with_head_zones(&directory, directory_len, &Slice::empty())?;
            if self.charge.touches_cache() {
                self.cache.insert_directory(
                    self.table_id,
                    self.group.offset(),
                    Arc::clone(&directory),
                    directory_len,
                );
            }
            directory
        } else {
            directory
        };
        let zones = match want.zone_block_column(&directory) {
            Some(column_id) => Some(self.zone_block(
                &directory,
                directory_len,
                column_id,
                &Slice::empty(),
                true,
            )?),
            None => None,
        };
        let wanted = want.resolve(&directory, zones.as_ref())?;
        let mut pages = wanted.slots(&directory);
        let found = self.fill_cached(&directory, directory_len, &mut pages)?;
        self.count_cached(cached, found);
        if pages.iter().all(|slot| slot.block.is_some()) {
            return Ok(RowGroupBlocks {
                directory,
                row_pages: wanted.row_pages,
                pages,
            });
        }
        // A directory taken from another read's load: that load was
        // coordinated, so the pages are too.
        let pages = self.fetch_missing(
            self.open()?,
            &directory,
            directory_len,
            &Slice::empty(),
            pages,
            !cached,
        )?;
        Ok(RowGroupBlocks {
            directory,
            row_pages: wanted.row_pages,
            pages,
        })
    }

    /// Loads the pages `want` selects in the group `earlier` read, [`load`]
    /// having returned it for this same group: its directory, already read
    /// and checked, is used as it is, and the pages it holds are taken from it
    /// rather than looked up or read again. A read in two steps, such as a
    /// point read that finds its row pages by the key column before reading
    /// the rest of them, pays for the directory once.
    ///
    /// [`load`]: Self::load
    ///
    /// # Errors
    ///
    /// As [`Self::load`] for the pages and zone block it reads.
    pub(crate) fn load_after(
        &self,
        earlier: &RowGroupBlocks,
        want: &PageWant<'_>,
    ) -> crate::Result<RowGroupBlocks> {
        let directory = Arc::clone(&earlier.directory);
        let directory_len = indexed_directory_len(self.group)?;
        let zones = match want.zone_block_column(&directory) {
            Some(column_id) => Some(self.zone_block(
                &directory,
                directory_len,
                column_id,
                &Slice::empty(),
                true,
            )?),
            None => None,
        };
        let wanted = want.resolve(&directory, zones.as_ref())?;
        let mut pages = wanted.slots(&directory);
        // Both lists are in directory order, so one pass matches them.
        let mut held = earlier.pages.iter().peekable();
        for slot in &mut pages {
            while held.next_if(|page| page.index < slot.index).is_some() {}
            if let Some(page) = held.next_if(|page| page.index == slot.index) {
                slot.block.clone_from(&page.block);
            }
        }
        let found = self.fill_cached(&directory, directory_len, &mut pages)?;
        self.count_cached(false, found);
        let pages = if pages.iter().all(|slot| slot.block.is_some()) {
            pages
        } else {
            self.fetch_missing(
                self.open()?,
                &directory,
                directory_len,
                &Slice::empty(),
                pages,
                false,
            )?
        };
        Ok(RowGroupBlocks {
            directory,
            row_pages: wanted.row_pages,
            pages,
        })
    }

    /// `directory` with its head zone block's zones kept on it, the block
    /// read, verified and admitted like any zone block. `front` holds the
    /// group's first bytes when an earlier read brought them in.
    fn with_head_zones(
        &self,
        directory: &Arc<PageDirectory>,
        directory_len: u32,
        front: &Slice,
    ) -> crate::Result<Arc<PageDirectory>> {
        let Some(head) = directory.head_zone_block() else {
            return Ok(Arc::clone(directory));
        };
        // Kept decoded on the directory, so the block itself is not cached.
        let zones = self.zone_block(directory, directory_len, head.column_id, front, false)?;
        Ok(Arc::new(
            PageDirectory::clone(directory).with_head_zones(zones)?,
        ))
    }

    /// The group's zone block, from the cache or read, verified and admitted
    /// like a page, decoded and checked against `directory`, and cached when
    /// `cache` says so. `front` holds the group's first bytes when an earlier
    /// read brought them in.
    fn zone_block(
        &self,
        directory: &PageDirectory,
        directory_len: u32,
        column_id: u16,
        front: &Slice,
        cache: bool,
    ) -> crate::Result<PageZones> {
        let Some((after_directory, length)) = directory.zone_block(column_id) else {
            return Err(crate::Error::InvalidHeader(
                "columnar: the column has no zone block",
            ));
        };
        // The extent check proved the directory, its zone blocks and its
        // pages fill the group, and the group's length is a `u32`.
        let start = directory_len as usize + after_directory as usize;
        let end = start + length as usize;
        let handle = BlockHandle::new(BlockOffset(*self.group.offset() + start as u64), length);
        if let Some(block) = self.lookup(handle.offset(), BlockType::ColumnZones)? {
            #[cfg(feature = "metrics")]
            if self.charge.is_counted() {
                record_block_load_cached(self.metrics, BlockType::ColumnZones);
            }
            return directory.decode_zone_block(column_id, &block.data);
        }
        let bytes = if end <= front.len() {
            front.slice(start..end)
        } else {
            self.check_block_len(length)?;
            self.read(self.open()?, start, end - start, BlockType::ColumnZones)?
        };
        let block = self.verify_and_admit(
            &bytes,
            &handle,
            BlockType::ColumnZones,
            CompressionType::None,
            false,
        )?;
        let zones = directory.decode_zone_block(column_id, &block.data)?;
        if cache {
            self.cache_block(&handle, block);
        }
        Ok(zones)
    }

    /// [`check_group_tag`] against this read's index entry. Checked before the
    /// directory's zones are used: they can prune every page, and then no
    /// page stamp is ever checked against the directory.
    fn check_group_tag(&self, directory: &PageDirectory) -> crate::Result<()> {
        check_group_tag(self.group.group_tag(), directory)
    }

    /// Refuses a block length the directory declares but no block of this
    /// table can have, before a read allocates a buffer for it.
    fn check_block_len(&self, length: u32) -> crate::Result<()> {
        crate::table::block::check_on_disk_size(u64::from(length), self.encryption, self.ecc)
    }

    /// Caches a verified page or zone block under its own offset, unless this
    /// read must leave the cache as it found it.
    fn cache_block(&self, handle: &BlockHandle, block: Block) {
        if self.charge.touches_cache() {
            self.cache
                .insert_block(self.table_id, handle.offset(), block);
        }
    }

    /// The directory, then the wanted pages. A read that expects every page
    /// and fits the I/O buffer takes the whole group in one request, so the
    /// pages are served from it; any other read takes exactly the directory,
    /// with the head zone block after it when it selects by the key column,
    /// whose zones a writer puts there.
    fn read_selective(
        &self,
        want: &PageWant<'_>,
        lead: Option<crate::cache::DirectoryLead>,
    ) -> crate::Result<RowGroupBlocks> {
        let timer = crate::cache::Cache::time_load();
        let fd = self.open()?;
        let group_len = self.group.size() as usize;
        let io_buffer = self.budget.io_buffer() as usize;
        let directory_len = indexed_directory_len(self.group)?;
        self.check_block_len(directory_len)?;
        let head_len = if want.zone_column() == Some(crate::table::columnar::COL_USER_KEY) {
            self.group
                .row_group()
                .map_or(0, |group| group.head_zones_len)
        } else {
            0
        };
        if head_len > 0 {
            self.check_block_len(head_len)?;
        }
        // Both are u32s, and the index entry refused a sum past the group.
        let both = directory_len as usize + head_len as usize;
        if both > group_len {
            return Err(crate::Error::InvalidHeader(
                "columnar: page directory overruns its row group",
            ));
        }
        // The key zones ride along only while the two fit one request; else
        // the directory is read alone and the zones as a request of their own,
        // each a single block the buffer may be smaller than.
        let head_end = if both <= io_buffer {
            both
        } else {
            directory_len as usize
        };
        let prefix = if want.whole && group_len <= io_buffer {
            self.read(fd, 0, group_len, BlockType::ColumnPage)?
        } else {
            self.read(fd, 0, head_end, BlockType::ColumnPageDirectory)?
        };
        let directory = self.admit_directory(&prefix, directory_len)?;
        // The directory's load is its own request; the pages it names are
        // timed as they are fetched, in their own shards. A read that must
        // leave no trace feeds no estimate.
        if self.charge.touches_cache() {
            self.cache
                .record_directory_load(self.table_id, self.group.offset(), timer);
        }
        let directory = if want
            .zone_column()
            .is_some_and(|column_id| directory.lacks_head_zones_for(column_id))
        {
            self.with_head_zones(&directory, directory_len, &prefix)?
        } else {
            directory
        };
        let zones = match want.zone_block_column(&directory) {
            Some(column_id) => {
                Some(self.zone_block(&directory, directory_len, column_id, &prefix, true)?)
            }
            None => None,
        };
        let wanted = want.resolve(&directory, zones.as_ref())?;
        let mut pages = wanted.slots(&directory);
        // With every page in hand already, a cache lookup per page would cost
        // more than verifying it from the bytes just read.
        if prefix.len() != group_len {
            let found = self.fill_cached(&directory, directory_len, &mut pages)?;
            self.count_cached(false, found);
        }
        // The reads waiting on the directory take it only once this read's
        // pages are cached, so they find those pages there. Handing it on as
        // soon as the pages are claimed spares a read of other pages the wait
        // for these, but measured, it lost more than it saved: groups read in
        // one request had their pages read again by the waiters (a same-key
        // columnar scan on a single-queue device ran a third slower), and
        // cache-hit scans lost a few percent.
        let pages = self.fetch_missing(
            fd,
            &directory,
            directory_len,
            &prefix,
            pages,
            lead.is_some(),
        )?;
        // Cached only now that the blocks it named carried the stamps it
        // implies: another group's directory read in this one's place fails
        // the read without staying behind to refuse every healthy page later.
        if self.charge.touches_cache() {
            self.cache.insert_directory(
                self.table_id,
                self.group.offset(),
                Arc::clone(&directory),
                directory_len,
            );
        }
        if let Some(lead) = lead {
            lead.publish(&directory, directory_len);
        }
        Ok(RowGroupBlocks {
            directory,
            row_pages: wanted.row_pages,
            pages,
        })
    }

    /// Verifies and admits the directory of `directory_len` bytes, the length
    /// the index entry records, at the front of `front`, decodes it and proves
    /// it fills the group. The caller caches it once the blocks it names have
    /// checked out.
    fn admit_directory(
        &self,
        front: &Slice,
        directory_len: u32,
    ) -> crate::Result<Arc<PageDirectory>> {
        check_directory_len(
            directory_len,
            Header::decode_from(&mut &front[..])?.on_disk_size_with(self.ecc),
        )?;
        let handle = BlockHandle::new(self.group.offset(), directory_len);
        let bytes = front.get(..directory_len as usize).ok_or_else(|| {
            crate::Error::InvalidHeader("columnar: page directory overruns its row group")
        })?;
        let block = self.verify_and_admit(
            bytes,
            &handle,
            BlockType::ColumnPageDirectory,
            CompressionType::None,
            false,
        )?;
        let directory = Arc::new(PageDirectory::decode(&block.data)?);
        check_group_extent(self.group, directory_len, &directory)?;
        check_head_zones_len(self.group, &directory)?;
        self.check_group_tag(&directory)?;
        Ok(directory)
    }

    /// Fetches every page `want` selects that `pages` does not hold yet.
    ///
    /// Pages are contiguous in directory order (the extent check proved it),
    /// so a run of consecutive missing pages is one byte range, read in
    /// requests of up to the budget's I/O buffer: a run within it is one
    /// request, a longer one is cut between pages, and a page larger than it
    /// is a request of its own. Requests go out the budget's in-flight count
    /// at a time. `front` holds the group's first bytes when an earlier read
    /// brought them in, and a range it covers is served from it rather than
    /// asked for again. `with_directory` says the load of the directory was
    /// coordinated.
    fn fetch_missing(
        &self,
        fd: &dyn FsFile,
        directory: &PageDirectory,
        directory_len: u32,
        front: &Slice,
        mut pages: Vec<PageSlot>,
        with_directory: bool,
    ) -> crate::Result<Vec<PageSlot>> {
        let entries = directory.entries();
        let io_buffer = self.budget.io_buffer() as usize;
        let leads = self.claim_pages(directory, directory_len, front, &mut pages, with_directory);
        let mut leads = leads.into_iter().peekable();
        // Runs of slots whose pages are missing and lie back to back in the
        // group: consecutive slots of consecutive directory entries.
        let requests = {
            let missing = |i: usize| pages.get(i).is_some_and(|slot| slot.block.is_none());
            let follows = |i: usize| {
                i.checked_sub(1)
                    .and_then(|before| pages.get(before))
                    .zip(pages.get(i))
                    .is_some_and(|(before, slot)| before.index + 1 == slot.index)
            };
            let mut requests = Vec::new();
            let mut i = 0;
            while i < pages.len() {
                if !missing(i) {
                    i += 1;
                    continue;
                }
                let first = i;
                let mut bytes = 0usize;
                while i < pages.len() && missing(i) && (i == first || follows(i)) {
                    let length = pages
                        .get(i)
                        .and_then(|slot| entries.get(slot.index))
                        .map_or(0, |e| e.length);
                    // A page longer than any block can be is refused before a
                    // request allocates a buffer for it.
                    self.check_block_len(length)?;
                    let length = length as usize;
                    if i > first && bytes + length > io_buffer {
                        break;
                    }
                    bytes += length;
                    i += 1;
                }
                requests.push(first..i);
            }
            requests
        };
        // Every offset below is within the group: the extent check proved
        // that the directory, its zone blocks and its pages fill it exactly,
        // and the group's length is a `u32`.
        let pages_at = directory_len as usize + directory.pages_start() as usize;
        let page_at = |entry: &PageEntry| pages_at + entry.offset as usize;
        let spans = {
            let entry_of = |slot: usize| pages.get(slot).and_then(|s| entries.get(s.index));
            requests
                .iter()
                .map(|run| {
                    let first = entry_of(run.start)?;
                    let last = entry_of(run.end.checked_sub(1)?)?;
                    Some((page_at(first), page_at(last) + last.length as usize))
                })
                .collect::<Option<Vec<_>>>()
                .ok_or_else(|| {
                    crate::Error::InvalidHeader("columnar: page outside its row group")
                })?
        };
        let in_flight = usize::from(self.budget.in_flight());
        for (batch, spans) in requests.chunks(in_flight).zip(spans.chunks(in_flight)) {
            // The requests the bytes already in hand do not reach at all go
            // out together; one they reach, in whole or in part, is served
            // from them.
            let beyond: Vec<(usize, usize)> = spans
                .iter()
                .filter(|&&(start, _)| start >= front.len())
                .map(|&(start, end)| (start, end - start))
                .collect();
            let timer = crate::cache::Cache::time_load();
            let mut read = self.read_many(fd, &beyond)?.into_iter();
            for (run, &(start, end)) in batch.iter().zip(spans) {
                let span = if start >= front.len() {
                    read.next().ok_or_else(|| {
                        crate::Error::InvalidHeader("columnar: a request came back missing")
                    })?
                } else {
                    self.span(fd, front, start, end)?
                };
                self.admit_pages(
                    directory,
                    directory_len,
                    run.clone(),
                    start,
                    &span,
                    &mut pages,
                    &mut leads,
                )?;
            }
            // The pages read from disk here, read and decoded, timed into
            // the shards their own loads coordinate in, unless this read must
            // leave no trace.
            if self.charge.touches_cache() {
                let pages_from = *self.group.offset()
                    + u64::from(directory_len)
                    + u64::from(directory.pages_start());
                self.cache.record_block_loads(
                    self.table_id,
                    batch
                        .iter()
                        .zip(spans)
                        .filter(|&(_, &(start, _))| start >= front.len())
                        .flat_map(|(run, _)| run.clone())
                        .filter_map(|slot| pages.get(slot).and_then(|s| entries.get(s.index)))
                        .map(|entry| BlockOffset(pages_from + u64::from(entry.offset))),
                    timer,
                );
            }
        }
        Ok(pages)
    }

    /// Claims the load of every page `pages` lacks before any is read: one
    /// no read is loading is led here, so a read missing it meanwhile waits
    /// for this one, and one another read is loading fills its slot from
    /// that load while this read leads none yet. Once it leads one, others
    /// may wait on it, so it waits no more and reads such a page alone.
    /// Returns the leads by slot, in slot order. A read that must leave no
    /// trace in the cache claims nothing, and neither is a page claimed whose
    /// bytes `front` already holds: its read is done, and waiting on another
    /// read's load of it could only delay it. `with_directory` says the load
    /// of the directory was coordinated (see [`crate::Cache::claim_page`]).
    fn claim_pages(
        &self,
        directory: &PageDirectory,
        directory_len: u32,
        front: &Slice,
        pages: &mut [PageSlot],
        with_directory: bool,
    ) -> Vec<(usize, crate::cache::BlockLead<'_>)> {
        let mut leads = Vec::new();
        if !self.charge.touches_cache() {
            return leads;
        }
        // Within the group: the extent check proved the pages fill it, and
        // its length is a `u32`.
        let in_group = directory_len as usize + directory.pages_start() as usize;
        let pages_at = *self.group.offset() + in_group as u64;
        let entries = directory.entries();
        for (at, slot) in pages.iter_mut().enumerate() {
            let Some(entry) = entries.get(slot.index).filter(|entry| {
                slot.block.is_none()
                    && in_group + entry.offset as usize + entry.length as usize > front.len()
            }) else {
                continue;
            };
            let offset = BlockOffset(pages_at + u64::from(entry.offset));
            let claim = self.cache.claim_page(
                self.table_id,
                offset,
                crate::cache::LoadIdentity::Block {
                    block_type: BlockType::ColumnPage,
                    size: entry.length,
                    compression: self.compression,
                    encrypted: self.encryption.is_some(),
                },
                with_directory,
            );
            match claim {
                crate::cache::BlockClaim::Shared(page) => {
                    #[cfg(feature = "metrics")]
                    if self.charge.is_counted() {
                        self.metrics
                            .block_load_shared
                            .fetch_add(1, core::sync::atomic::Ordering::Relaxed);
                    }
                    slot.block = Some(page);
                }
                // A load that finished between the cache lookup before these
                // claims and this one left the page cached: hand that to the
                // reads waiting on this lead rather than reading it again. A
                // cached block of another type is left to the read, which
                // refuses it.
                crate::cache::BlockClaim::Lead(lead) => {
                    match self.cache.get_block(self.table_id, offset) {
                        Some(page) if page.header.block_type == BlockType::ColumnPage => {
                            lead.publish(&page);
                            slot.block = Some(page);
                        }
                        _ => leads.push((at, lead)),
                    }
                }
                crate::cache::BlockClaim::Alone => {}
            }
        }
        leads
    }

    /// Verifies, admits and caches the pages `run` names out of `span`, the
    /// group's bytes from `start`, into their slots of `pages`.
    ///
    /// A page is cached only once its stamp is the one `directory` implies
    /// for its slot: a page read in another page's place (a misdirected or
    /// transient read) fails the read without staying behind in the cache to
    /// fail every later read of the group. A page this read leads the load
    /// of (`leads`, by slot in slot order) is handed to the reads waiting on
    /// it once cached.
    #[expect(
        clippy::too_many_arguments,
        reason = "the run, its bytes, the slots it fills and the leads it publishes"
    )]
    fn admit_pages(
        &self,
        directory: &PageDirectory,
        directory_len: u32,
        run: core::ops::Range<usize>,
        start: usize,
        span: &Slice,
        pages: &mut [PageSlot],
        leads: &mut core::iter::Peekable<alloc::vec::IntoIter<(usize, crate::cache::BlockLead)>>,
    ) -> crate::Result<()> {
        let entries = directory.entries();
        let pages_at = directory_len as usize + directory.pages_start() as usize;
        let page_at = |entry: &PageEntry| pages_at + entry.offset as usize;
        let first = run.start;
        for (position, slot) in (first..).zip(pages.get_mut(run).into_iter().flatten()) {
            let Some(entry) = entries.get(slot.index) else {
                continue;
            };
            let at = page_at(entry) - start;
            let Some(bytes) = span.get(at..at + entry.length as usize) else {
                return Err(crate::Error::InvalidHeader(
                    "columnar: page outside its row group",
                ));
            };
            let handle = BlockHandle::new(
                BlockOffset(*self.group.offset() + page_at(entry) as u64),
                entry.length,
            );
            let page = self.verify_and_admit(
                bytes,
                &handle,
                BlockType::ColumnPage,
                self.compression,
                true,
            )?;
            let stamped = page
                .data
                .first_chunk::<{ PageStamp::LEN }>()
                .is_some_and(|stamp| PageStamp::decode(*stamp) == directory.stamp_for(entry));
            if !stamped {
                return Err(crate::Error::InvalidHeader(
                    "columnar: page belongs to another row group, column part or row page",
                ));
            }
            self.cache_block(&handle, page.clone());
            // Leads are in slot order, and runs are admitted in slot order.
            while leads.next_if(|(led, _)| *led < position).is_some() {}
            if let Some((_, lead)) = leads.next_if(|(led, _)| *led == position) {
                lead.publish(&page);
            }
            slot.block = Some(page);
        }
        Ok(())
    }

    /// Reads the `(at, len)` ranges of the group in one batched request, each
    /// charged to the data role as the request is issued.
    fn read_many(&self, fd: &dyn FsFile, ranges: &[(usize, usize)]) -> crate::Result<Vec<Slice>> {
        if ranges.is_empty() {
            return Ok(Vec::new());
        }
        #[cfg(feature = "metrics")]
        if self.charge.is_counted() {
            for &(_, len) in ranges {
                record_block_read(self.metrics, BlockType::ColumnPage, len as u64);
            }
        }
        let regions: Vec<(u64, usize)> = ranges
            .iter()
            .map(|&(at, len)| (*self.group.offset() + at as u64, len))
            .collect();
        // A backend may serve a batch one region after another, and one sample
        // of the whole batch would sum their times against a ceiling set for
        // one read: while the pacer listens, each region is read and timed on
        // its own.
        #[cfg(feature = "std")]
        if self.pace.is_some_and(crate::table::util::ReadPacer::active) {
            return regions
                .iter()
                .map(|&(offset, len)| {
                    let bytes = self.request(|| crate::file::read_exact(fd, offset, len))?;
                    self.took(len);
                    Ok(bytes)
                })
                .collect();
        }
        let bytes = crate::file::read_exact_many(fd, &regions)?;
        self.took(ranges.iter().map(|&(_, len)| len).sum());
        Ok(bytes)
    }

    /// Makes one request to the file, telling the pacer, when it is active
    /// now, how long the request took, an error included.
    fn request<T>(&self, op: impl FnOnce() -> T) -> T {
        // no-std: reads are timed only with std (see `ReadPacer::read_took`)
        #[cfg(feature = "std")]
        if let Some(pace) = self.pace.filter(|pace| pace.active()) {
            let started = std::time::Instant::now();
            let out = op();
            pace.read_took(started.elapsed());
            return out;
        }
        op()
    }

    /// The group's bytes `[start, end)`, served from `front` as far as it
    /// reaches and read for the rest.
    fn span(
        &self,
        fd: &dyn FsFile,
        front: &Slice,
        start: usize,
        end: usize,
    ) -> crate::Result<Slice> {
        if end <= front.len() {
            return Ok(front.slice(start..end));
        }
        if start >= front.len() {
            return self.read(fd, start, end - start, BlockType::ColumnPage);
        }
        let tail = self.read(fd, front.len(), end - front.len(), BlockType::ColumnPage)?;
        Ok(Slice::fused(&front.slice(start..), &tail))
    }

    /// Reads `len` bytes at `at` within the group, charged to `role` as the
    /// request is issued.
    fn read(
        &self,
        fd: &dyn FsFile,
        at: usize,
        len: usize,
        role: BlockType,
    ) -> crate::Result<Slice> {
        #[cfg(feature = "metrics")]
        if self.charge.is_counted() {
            record_block_read(self.metrics, role, len as u64);
        }
        #[cfg(not(feature = "metrics"))]
        let _ = role;
        let bytes =
            self.request(|| crate::file::read_exact(fd, *self.group.offset() + at as u64, len))?;
        self.took(len);
        Ok(bytes)
    }

    /// Tells the pacer the bytes a read of the group returned.
    fn took(&self, len: usize) {
        if let Some(pace) = self.pace {
            pace.read_bytes(len as u64);
        }
    }

    /// Fills the empty slots of `pages` with the blocks the cache already
    /// holds, returning how many it found.
    fn fill_cached(
        &self,
        directory: &PageDirectory,
        directory_len: u32,
        pages: &mut [PageSlot],
    ) -> crate::Result<usize> {
        let pages_at =
            *self.group.offset() + u64::from(directory_len) + u64::from(directory.pages_start());
        let entries = directory.entries();
        let mut found = 0;
        for slot in pages.iter_mut().filter(|slot| slot.block.is_none()) {
            if let Some(entry) = entries.get(slot.index) {
                let offset = pages_at + u64::from(entry.offset);
                slot.block = self.lookup(BlockOffset(offset), BlockType::ColumnPage)?;
                found += usize::from(slot.block.is_some());
            }
        }
        Ok(found)
    }

    /// A cached block at `offset`, looked up without promoting it when this
    /// read must not touch the cache, and refused when it is not `expected`.
    fn lookup(&self, offset: BlockOffset, expected: BlockType) -> crate::Result<Option<Block>> {
        let block = if self.charge.touches_cache() {
            self.cache.get_block(self.table_id, offset)
        } else {
            self.cache.peek_block(self.table_id, offset)
        };
        match block {
            Some(block) if block.header.block_type != expected => Err(crate::Error::InvalidTag((
                "BlockType",
                block.header.block_type.into(),
            ))),
            other => Ok(other),
        }
    }

    /// Counts the cached blocks a read served: the directory when it came
    /// from the cache, and the `pages` it found there.
    #[cfg_attr(
        not(feature = "metrics"),
        expect(clippy::unused_self, reason = "the counters are behind `metrics`")
    )]
    fn count_cached(&self, directory: bool, pages: usize) {
        #[cfg(feature = "metrics")]
        if self.charge.is_counted() {
            if directory {
                record_block_load_cached(self.metrics, BlockType::ColumnPageDirectory);
            }
            for _ in 0..pages {
                record_block_load_cached(self.metrics, BlockType::ColumnPage);
            }
        }
        #[cfg(not(feature = "metrics"))]
        let _ = (directory, pages);
    }

    /// The table's file, taken through the lookup's [`LookupFd`] as
    /// `load_block` takes it.
    fn open(&self) -> crate::Result<&dyn FsFile> {
        Ok(self.fd.resolve(
            self.file_accessor,
            &self.table_id,
            self.path,
            self.charge,
            #[cfg(feature = "metrics")]
            self.metrics,
        )?)
    }

    /// Verifies one block's on-disk `bytes` and admits it. `with_dict` says
    /// whether the table's zstd dictionary applies: pages are compressed
    /// under it, the directory is never compressed.
    fn verify_and_admit(
        &self,
        bytes: &[u8],
        handle: &BlockHandle,
        block_type: BlockType,
        compression: CompressionType,
        with_dict: bool,
    ) -> crate::Result<Block> {
        #[cfg(zstd_any)]
        let zstd_dict = if with_dict { self.zstd_dict } else { None };
        #[cfg(not(zstd_any))]
        let _ = with_dict;
        let transform = build_block_transform(
            compression,
            self.encryption,
            self.ecc,
            #[cfg(zstd_any)]
            zstd_dict,
        )?;
        let identity = crate::table::block::BlockIdentity {
            table_id: self.table_id.table_id(),
            block_type,
            dict_id: compression.dict_id(),
            window_log: 0,
        };
        // An owned copy per block, so each cached block holds exactly its own
        // bytes rather than a view that pins the read it came in with.
        let (header, payload, ecc_status, recovery) =
            Block::verify_frame(bytes.to_vec(), *handle, identity, &transform)?;
        let mut produced = 0;
        let data = Block::decompress_payload(&header, payload, &transform, &mut produced);
        #[cfg(feature = "metrics")]
        record_block_decoded(self.metrics, self.charge, produced);
        let data = data?;
        admit_read_block(
            self.table_id,
            self.path,
            self.file_accessor,
            handle,
            block_type,
            compression,
            self.encryption,
            self.ecc,
            #[cfg(zstd_any)]
            zstd_dict,
            self.heal_hints,
            #[cfg(feature = "metrics")]
            self.metrics,
            self.charge,
            Block { header, data },
            ecc_status,
            recovery,
        )
    }
}

impl RowGroupBlocks {
    /// Decodes the pages the read selected into one batch per row page, in
    /// row order, each page a view of its own block.
    ///
    /// # Errors
    ///
    /// [`crate::Error::InvalidHeader`] for a zero-row group, a row page
    /// outside the group, a group holding an encoding part this build does
    /// not decode, a wanted page the read did not fetch, a page whose stamp
    /// names another group, part or row page, or a page whose own column id
    /// disagrees with the one the directory filed it under. They fail the
    /// group rather than the page: a reader that skipped what it could not
    /// interpret, or trusted a page to be what the directory said without
    /// checking, would return a batch that is quietly missing or substituting
    /// a column.
    ///
    /// Adds to `copied` what decoding the pages copied out of them, as each
    /// copy is made.
    pub(crate) fn to_row_pages(&self, copied: &mut usize) -> crate::Result<RowPages> {
        use crate::table::columnar::{Column, ColumnBatch, DecodeBudget};

        let outside = || crate::Error::InvalidHeader("columnar: row page outside the group");
        let mut budget = DecodeBudget::default();
        let (ordinals, columns) = self.by_row_page(
            |_| true,
            |page, rows, stamp| Column::decode_page(page, rows, stamp, copied, &mut budget),
            |column| column.column_id,
        )?;
        let mut batches = Vec::with_capacity(ordinals.len());
        for (columns, &ordinal) in columns.into_iter().zip(&ordinals) {
            let row_count = self.directory.row_page_rows(ordinal).ok_or_else(outside)?;
            batches.push(ColumnBatch { row_count, columns });
        }
        Ok(RowPages { ordinals, batches })
    }

    /// The pages of the columns `want` names among those the read selected,
    /// parsed but not decoded, one list of columns per row page in row order:
    /// what a lookup of a few rows reads them through. Each column borrows its
    /// page; a page of a column not wanted is not read. What the parses hold
    /// decoded, kept as long as the pages are, is charged to `budget`, the
    /// group's.
    ///
    /// # Errors
    ///
    /// As [`Self::to_row_pages`], for everything but the values themselves,
    /// and [`crate::Error::DecompressedSizeTooLarge`] past `budget`.
    pub(crate) fn page_columns(
        &self,
        want: impl Fn(u16) -> bool,
        budget: &mut crate::table::columnar::DecodeBudget,
    ) -> crate::Result<Vec<crate::table::columnar::RowPageColumns<'_>>> {
        use crate::table::columnar::{Column, RowPageColumns};

        let (ordinals, columns) = self.by_row_page(
            want,
            |page, rows, stamp| budget.parsed(Column::parse_page(page, rows, stamp)?),
            |column| column.column_id,
        )?;
        let outside = || crate::Error::InvalidHeader("columnar: row page outside the group");
        columns
            .into_iter()
            .zip(ordinals)
            .map(|(columns, ordinal)| {
                Ok(RowPageColumns {
                    ordinal,
                    start: self.directory.row_page_start(ordinal).ok_or_else(outside)?,
                    rows: self.directory.row_page_rows(ordinal).ok_or_else(outside)?,
                    columns,
                })
            })
            .collect()
    }

    /// Refuses a group no reading of it can trust: one of no rows, or holding
    /// an encoding part this build does not decode.
    fn check_readable(&self) -> crate::Result<()> {
        if self.directory.row_count() == 0 {
            return Err(crate::Error::InvalidHeader("columnar: zero-row data block"));
        }
        // Every column's encoding is one part, 0. A part this build does not
        // read is refused rather than skipped: skipping it would hand back a
        // column assembled from some of its parts. The directory proved each
        // part's pages one run, so a run's first page names the part of all of
        // them.
        let row_page_count = self.directory.row_pages().len();
        if self
            .directory
            .entries()
            .iter()
            .step_by(row_page_count.max(1))
            .any(|entry| entry.id.part != 0)
        {
            return Err(crate::Error::InvalidHeader(
                "columnar: page holds an encoding part this build does not decode",
            ));
        }
        Ok(())
    }

    /// Reads every fetched page of a column `want` names with `read`, filing
    /// what it returns under its row page: the selected row pages' ordinals
    /// and, per row page in row order, its columns in write order.
    /// `column_of` names the column a result holds, which must be the one the
    /// directory filed its page under.
    ///
    /// The checks every reading of a group makes before trusting its pages
    /// live here, so the decoded and the parsed readings cannot drift apart.
    fn by_row_page<'s, T>(
        &'s self,
        want: impl Fn(u16) -> bool,
        mut read: impl FnMut(&'s crate::Slice, u32, PageStamp) -> crate::Result<T>,
        column_of: impl Fn(&T) -> u16,
    ) -> crate::Result<(Vec<u16>, Vec<Vec<T>>)> {
        self.check_readable()?;
        let entries = self.directory.entries();
        let ordinals: Vec<u16> = self.row_pages.iter().collect();
        // One column list per row page read, sized for every column the read
        // fetched. The slots are in directory order, which is column-major,
        // so each row page's columns arrive in write order.
        let per_page = self.pages.len() / ordinals.len().max(1);
        let mut columns: Vec<Vec<T>> = ordinals
            .iter()
            .map(|_| Vec::with_capacity(per_page))
            .collect();
        let unknown_row_page = || {
            crate::Error::InvalidHeader("columnar: page names a row page the group does not have")
        };
        for slot in &self.pages {
            let entry = entries.get(slot.index).ok_or_else(unknown_row_page)?;
            if !want(entry.id.column_id) {
                continue;
            }
            let Some(page) = &slot.block else {
                return Err(crate::Error::InvalidHeader(
                    "columnar: a wanted page was not read",
                ));
            };
            let rows = self
                .directory
                .row_page_rows(entry.row_page)
                .ok_or_else(unknown_row_page)?;
            let column = read(&page.data, rows, self.directory.stamp_for(entry))?;
            if column_of(&column) != entry.id.column_id {
                return Err(crate::Error::InvalidHeader(
                    "columnar: page column disagrees with its directory entry",
                ));
            }
            let place = match &self.row_pages {
                RowPageSet::Range(range) => entry
                    .row_page
                    .checked_sub(range.start)
                    .map(usize::from)
                    .filter(|&place| place < range.len()),
                RowPageSet::Listed(listed) => listed.binary_search(&entry.row_page).ok(),
            };
            place
                .and_then(|place| columns.get_mut(place))
                .ok_or_else(unknown_row_page)?
                .push(column);
        }
        Ok((ordinals, columns))
    }

    /// Each fetched page with the rows it holds, what its values were
    /// encoded as and, for a bytes column, what it spends on where its values
    /// start, in the directory's order, without decoding the values.
    ///
    /// # Errors
    ///
    /// As [`Self::to_row_pages`], for everything but the values themselves.
    #[cfg(feature = "std")]
    pub(crate) fn page_encodings(&self) -> crate::Result<Vec<PageDescription>> {
        self.check_readable()?;
        let entries = self.directory.entries();
        let unknown = || {
            crate::Error::InvalidHeader("columnar: page names a row page the group does not have")
        };
        self.pages
            .iter()
            .map(|slot| {
                let entry = *entries.get(slot.index).ok_or_else(unknown)?;
                let Some(page) = &slot.block else {
                    return Err(crate::Error::InvalidHeader(
                        "columnar: a wanted page was not read",
                    ));
                };
                let rows = self
                    .directory
                    .row_page_rows(entry.row_page)
                    .ok_or_else(unknown)?;
                let column = crate::table::columnar::Column::parse_page(
                    &page.data,
                    rows,
                    self.directory.stamp_for(&entry),
                )?;
                // As every read of the group does: a page is described as the
                // column it holds only if that is the column it is filed under.
                if column.column_id != entry.id.column_id {
                    return Err(crate::Error::InvalidHeader(
                        "columnar: page column disagrees with its directory entry",
                    ));
                }
                Ok(PageDescription {
                    entry,
                    rows,
                    expression: column.values.describe()?,
                    offsets_len: column.values.offsets_len(column.type_tag, rows),
                })
            })
            .collect()
    }
}

/// Refuses a directory whose tag is not `expected`, the one the group's index
/// entry names it by, and a group whose entry names none.
///
/// A directory verifies as a block wherever it is read, so another group's,
/// read in this one's place, would otherwise be taken for this group's: every
/// read of a group's blocks (an indexed read, a scan, verification, salvage)
/// checks it before trusting anything the directory says.
pub fn check_group_tag(
    expected: Option<core::num::NonZeroU64>,
    directory: &PageDirectory,
) -> crate::Result<()> {
    match expected {
        Some(tag) if tag.get() == directory.group_tag() => Ok(()),
        Some(_) => Err(crate::Error::InvalidHeader(
            "columnar: page directory belongs to another row group",
        )),
        None => Err(crate::Error::InvalidHeader(
            "columnar: the index entry names no row group",
        )),
    }
}

/// The on-disk length of `group`'s directory, as its index entry records it.
///
/// # Errors
///
/// [`crate::Error::InvalidHeader`] when the entry names no row group, which a
/// columnar table's entries always do.
pub fn indexed_directory_len(group: &BlockHandle) -> crate::Result<u32> {
    group
        .row_group()
        .map(|group| group.directory_len.get())
        .ok_or_else(|| crate::Error::InvalidHeader("columnar: the index entry names no row group"))
}

/// Refuses a directory whose on-disk length, `found`, is not the `indexed`
/// one its group's index entry records.
///
/// A reader reads exactly the indexed length, so a directory of another
/// length in the group's place would either be cut short or run into the
/// first page; the two are independent statements a writer makes agree.
///
/// # Errors
///
/// [`crate::Error::InvalidHeader`] when they differ.
pub fn check_directory_len(indexed: u32, found: u32) -> crate::Result<()> {
    if indexed == found {
        Ok(())
    } else {
        Err(crate::Error::InvalidHeader(
            "columnar: page directory length disagrees with its index entry",
        ))
    }
}

/// Refuses a directory whose head zone block is not the length `group`'s index
/// entry records: a read that selects by the key column takes that many bytes
/// after the directory as the block, so a disagreement would cut it short or
/// run it into the first page.
///
/// # Errors
///
/// [`crate::Error::InvalidHeader`] when they differ, or when the entry names
/// no row group.
pub fn check_head_zones_len(group: &BlockHandle, directory: &PageDirectory) -> crate::Result<()> {
    let Some(indexed) = group.row_group().map(|group| group.head_zones_len) else {
        return Err(crate::Error::InvalidHeader(
            "columnar: the index entry names no row group",
        ));
    };
    if indexed == directory.pages_start() {
        Ok(())
    } else {
        Err(crate::Error::InvalidHeader(
            "columnar: head zone block length disagrees with its index entry",
        ))
    }
}

/// Proves the directory describes a layout that fills the group exactly: the
/// directory, its pages back to back (which a [`PageDirectory`] holds by
/// construction), then its zone blocks.
///
/// The index entry and the directory are two independent statements of where
/// the group ends. A writer makes them agree by construction, so a
/// disagreement is corruption, and it is refused before any page offset
/// derived from the directory is used to slice the group.
pub fn check_group_extent(
    group: &BlockHandle,
    directory_len: u32,
    directory: &PageDirectory,
) -> crate::Result<()> {
    if directory.group_len(directory_len) == Some(group.size()) {
        Ok(())
    } else {
        Err(crate::Error::InvalidHeader(
            "columnar: page directory does not fill its row group",
        ))
    }
}