Skip to main content

mongreldb_core/
sorted_run.rs

1//! Sorted Run — the immutable columnar unit (`.sr`).
2//!
3//! On-disk layout: a 256-byte header, a columnar page region (PAX), a column
4//! directory, an index trailer, and a checksummed footer. [`RunWriter`] flushes
5//! drained memtable rows into encoded columns
6//! (system columns `_row_id` / `_epoch` / `_deleted` plus user columns), and
7//! [`RunReader`] decodes them back, answering MVCC point lookups and scans.
8//!
9//! # HLC stamps (P0.5-T3)
10//!
11//! Sorted runs may carry an optional [`SYS_COMMIT_TS`] system column (16-byte
12//! little-endian HLC encoding). Writers emit it whenever any flushed row has a
13//! `commit_ts`; readers restore stamps when the column is present and treat a
14//! missing column as legacy (`commit_ts: None`). Epoch remains the always-on
15//! system column so pre-stamp runs stay readable without a format bump.
16//! Full PITR-at-HLC still depends on archive targets (P0.5-X10).
17
18use crate::columnar;
19use crate::encryption::{setup_run_encryption, Cipher, Kek, RunEncryption};
20use crate::epoch::Epoch;
21use crate::error::{MongrelError, Result};
22use crate::index::pgm::PgmIndex;
23use crate::memtable::{Row, Value};
24use crate::page::{Encoding, PageStat};
25use crate::row_id_set::RowIdSet;
26use crate::rowid::RowId;
27use crate::schema::{Schema, TypeId};
28use mongreldb_types::hlc::HlcTimestamp;
29use serde::{Deserialize, Serialize};
30use sha2::{Digest, Sha256};
31use std::collections::HashMap;
32use std::fs::{File, OpenOptions};
33use std::io::{Read, Seek, SeekFrom, Write};
34use std::path::{Path, PathBuf};
35use std::sync::Arc;
36
37pub const RUN_MAGIC: [u8; 8] = *b"MONGRRUN";
38pub const RUN_FORMAT_VERSION: u16 = 1;
39/// v2 appends `encrypted_stats_offset`/`encrypted_stats_len` to [`RunHeader`].
40/// The fields are trailing and the header region is zero-padded, so a run
41/// written before v2 deserializes with both fields = 0 ("no encrypted stats
42/// section"). Pre-v2 *encrypted* runs are not readable (their MAC covers the
43/// shorter v1 header serialization) — recreate them; no released data exists.
44pub const RUN_HEADER_VERSION: u16 = 2;
45pub const RUN_HEADER_PAD: usize = 256;
46const MAX_RUN_PAGE_BYTES: u64 = 64 * 1024 * 1024 + 32;
47
48/// Reserved `(column_id, page_seq)` nonce coordinates for the encrypted
49/// page-stats envelope (see [`RunHeader::encrypted_stats_offset`]). Pages per
50/// column are capped strictly below `u16::MAX` at encode, so no page nonce can
51/// ever collide with this pair under the run's DEK.
52const ENC_STATS_NONCE_COLUMN: u16 = u16::MAX;
53const ENC_STATS_NONCE_SEQ: u32 = u16::MAX as u32;
54
55/// One page's `(min, max)` bounds as stored in the encrypted stats envelope.
56type PageMinMax = (Option<Vec<u8>>, Option<Vec<u8>>);
57
58/// Per-page `(min, max)` bounds of one encrypted column, keyed by column id —
59/// the plaintext shape of the encrypted stats envelope. The cleartext column
60/// directory must not carry these (they would leak values to anyone holding
61/// the file without the key), so they travel AES-256-GCM-encrypted under the
62/// run DEK and are overlaid onto the in-memory [`PageStat`]s at open, which
63/// restores zone-map page pruning for encrypted columns.
64type EncryptedColumnStats = Vec<(u16, Vec<PageMinMax>)>;
65
66/// AES-256-GCM authentication-tag length appended to every ciphertext. Used to
67/// precompute on-disk page sizes so the direct-to-mmap writer can lay the whole
68/// run out before encrypting (Phase 14.6). Validated against the cipher's real
69/// output in [`write_run_mmap`].
70const GCM_TAG_LEN: usize = 16;
71
72/// On-disk length of a page payload: ciphertext = plaintext + GCM tag when
73/// encrypted, else the plaintext length itself.
74fn on_disk_len(page_len: usize, encrypted: bool) -> usize {
75    if encrypted {
76        page_len + GCM_TAG_LEN
77    } else {
78        page_len
79    }
80}
81
82pub const RUN_FLAG_ENCRYPTED: u8 = 1 << 0;
83pub const RUN_FLAG_TOMBSTONE_ONLY: u8 = 1 << 1;
84/// Run is "clean": exactly one version per `RowId`, no tombstones, and row_ids
85/// are strictly ascending. Bulk-loaded and compacted runs are clean by
86/// construction. For a clean run, MVCC visibility is trivially *every* position
87/// (the newest visible version of a rid is the only version, and it is not
88/// deleted), so the visibility pass can skip decoding the epoch/deleted system
89/// columns and the group-collapse loop — only the row_id column is needed (for
90/// survivor↔position mapping). Old runs lack this bit and default to not-clean
91/// (safe fallback to the full pass).
92pub const RUN_FLAG_CLEAN: u8 = 1 << 2;
93/// Run is "uniform-epoch": every row's commit epoch is the run's commit epoch,
94/// which is **not** baked into the file (it is assigned at commit/link time and
95/// recorded in the manifest `RunRef.epoch_created`). Spill runs from large
96/// transactions are written before the commit epoch is known, so their stored
97/// `_epoch` column is a placeholder; the reader must overlay the real epoch from
98/// the `RunRef`. The engine calls [`RunReader::set_uniform_epoch`] with that
99/// value after opening such a run.
100pub const RUN_FLAG_UNIFORM_EPOCH: u8 = 1 << 3;
101pub const SORT_KEY_ROW_ID: u16 = 0xFFFF;
102
103/// Reserved column ids for the MVCC system columns, stored in every run.
104pub const SYS_ROW_ID: u16 = 0xFFFE;
105pub const SYS_EPOCH: u16 = 0xFFFD;
106pub const SYS_DELETED: u16 = 0xFFFC;
107/// Optional HLC commit stamp (P0.5-T3). Absent on legacy runs → `commit_ts: None`.
108///
109/// Encoding: 16 little-endian bytes
110/// `(physical_micros:u64, logical:u32, node_tiebreaker:u32)`, or `Null` when
111/// a particular row version was not stamped.
112pub const SYS_COMMIT_TS: u16 = 0xFFFB;
113
114/// Bytes length of the on-disk HLC encoding for [`SYS_COMMIT_TS`].
115const COMMIT_TS_BYTES: usize = 16;
116
117/// Encode an optional HLC stamp for the optional [`SYS_COMMIT_TS`] column.
118fn encode_commit_ts_value(ts: Option<HlcTimestamp>) -> Value {
119    match ts {
120        None => Value::Null,
121        Some(ts) => {
122            let mut buf = [0u8; COMMIT_TS_BYTES];
123            buf[0..8].copy_from_slice(&ts.physical_micros.to_le_bytes());
124            buf[8..12].copy_from_slice(&ts.logical.to_le_bytes());
125            buf[12..16].copy_from_slice(&ts.node_tiebreaker.to_le_bytes());
126            Value::Bytes(buf.to_vec())
127        }
128    }
129}
130
131/// Decode a [`SYS_COMMIT_TS`] cell. Malformed/null/absent → `None` (legacy).
132fn decode_commit_ts_value(value: Option<&Value>) -> Option<HlcTimestamp> {
133    match value {
134        Some(Value::Bytes(bytes)) if bytes.len() == COMMIT_TS_BYTES => {
135            let physical_micros = u64::from_le_bytes(bytes[0..8].try_into().ok()?);
136            let logical = u32::from_le_bytes(bytes[8..12].try_into().ok()?);
137            let node_tiebreaker = u32::from_le_bytes(bytes[12..16].try_into().ok()?);
138            Some(HlcTimestamp {
139                physical_micros,
140                logical,
141                node_tiebreaker,
142            })
143        }
144        _ => None,
145    }
146}
147
148/// True when `column_id` is a reserved system column (including optional HLC).
149fn is_system_column_id(column_id: u16) -> bool {
150    matches!(
151        column_id,
152        SYS_ROW_ID | SYS_EPOCH | SYS_DELETED | SYS_COMMIT_TS
153    )
154}
155
156/// Guaranteed positional error of the stored PGM model (the predicted offset is
157/// within `± LEARNED_EPSILON` of the true position; a tiny final scan corrects).
158const LEARNED_EPSILON: usize = 64;
159
160/// Build the learned-index trailer as a compressed PGM-index over
161/// `(row_id, array_index)`. Near-linear row-id sequences collapse to a single
162/// segment, so the trailer is typically a few dozen bytes regardless of run size.
163fn build_learned_trailer(row_ids: &[Value]) -> Vec<u8> {
164    let points: Vec<(u64, usize)> = row_ids
165        .iter()
166        .enumerate()
167        .filter_map(|(i, v)| match v {
168            Value::Int64(r) => Some((*r as u64, i)),
169            _ => None,
170        })
171        .collect();
172    let pgm = PgmIndex::build(&points, LEARNED_EPSILON);
173    bincode::serialize(&pgm).expect("pgm serialize")
174}
175
176#[derive(Debug, Clone, Serialize, Deserialize)]
177pub struct RunHeader {
178    pub magic: [u8; 8],
179    pub format_version: u16,
180    pub header_layout_version: u16,
181    pub run_id: u128,
182    pub content_hash: [u8; 32],
183    pub schema_id: u64,
184    pub epoch_created: u64,
185    pub level: u8,
186    pub flags: u8,
187    pub sort_key_column_id: u16,
188    pub row_count: u64,
189    pub min_row_id: u64,
190    pub max_row_id: u64,
191    pub column_count: u64,
192    pub column_dir_offset: u64,
193    pub index_trailer_offset: u64,
194    pub encryption_descriptor_offset: u64,
195    pub footer_offset: u64,
196    /// Offset/length of the AES-256-GCM-encrypted per-page min/max envelope
197    /// for encrypted columns (0/0 = absent; always 0 for plaintext runs and
198    /// pre-v2 files, whose zero header padding deserializes to 0 here).
199    pub encrypted_stats_offset: u64,
200    pub encrypted_stats_len: u64,
201}
202
203impl RunHeader {
204    pub fn is_encrypted(&self) -> bool {
205        self.flags & RUN_FLAG_ENCRYPTED != 0
206    }
207    pub fn is_clean(&self) -> bool {
208        self.flags & RUN_FLAG_CLEAN != 0
209    }
210    pub fn is_uniform_epoch(&self) -> bool {
211        self.flags & RUN_FLAG_UNIFORM_EPOCH != 0
212    }
213}
214
215#[derive(Debug, Clone, Serialize, Deserialize)]
216pub struct ColumnPageHeader {
217    pub column_id: u16,
218    pub type_id_tag: u16,
219    pub encoding: u8,
220    pub flags: u8,
221    pub page_count: u32,
222    pub page_region_offset: u64,
223    pub page_region_len: u64,
224    pub page_stats: Vec<PageStat>,
225}
226
227impl ColumnPageHeader {
228    const PAGE_ENCRYPTED: u8 = 1 << 0;
229}
230
231/// Length of the run-metadata HMAC tag appended after the footer of an encrypted
232/// run (HMAC-SHA256, see [`crate::encryption::run_metadata_mac`]).
233const RUN_MAC_LEN: usize = 32;
234
235/// Compute the run-metadata MAC tag for an encrypted run, or `None` for a
236/// plaintext run (or when the `encryption` feature is off, where `enc` is always
237/// `None`). MACs `header ‖ dir ‖ descriptor` under the run's KEK-derived key.
238fn compute_run_mac(
239    enc: Option<&RunEncryption>,
240    header_bytes: &[u8],
241    dir_bytes: &[u8],
242) -> Option<[u8; RUN_MAC_LEN]> {
243    {
244        if let Some(e) = enc {
245            if let Some(mac_key) = &e.mac_key {
246                return Some(crate::encryption::run_metadata_mac(
247                    mac_key,
248                    header_bytes,
249                    dir_bytes,
250                    &e.descriptor_bytes,
251                ));
252            }
253        }
254    }
255    None
256}
257
258/// A column's pages handed to the low-level writer.
259pub struct ColumnPayload {
260    pub column_id: u16,
261    pub type_id_tag: u16,
262    pub encoding: Encoding,
263    pub pages: Vec<Vec<u8>>,
264    /// Optional value-derived stats per page (parallel to [`Self::pages`]). When
265    /// present, [`write_run_with`] fills only the offset/length slots; when a
266    /// slot is missing it falls back to an empty stat.
267    pub page_stats: Vec<PageStat>,
268}
269
270/// Specification handed to [`write_run`] / [`write_run_with`].
271pub struct RunSpec<'a> {
272    pub run_id: u128,
273    pub schema_id: u64,
274    pub epoch_created: u64,
275    pub level: u8,
276    pub flags: u8,
277    pub sort_key_column_id: u16,
278    pub row_count: u64,
279    pub min_row_id: u64,
280    pub max_row_id: u64,
281    pub columns: &'a [ColumnPayload],
282}
283
284/// Write a run with no encryption and no index trailer (back-compat entry point).
285pub fn write_run(path: impl AsRef<Path>, spec: &RunSpec) -> Result<RunHeader> {
286    write_run_with(path, spec, None, &[], None)
287}
288
289/// Write a run, optionally encrypting page payloads with a per-file DEK
290/// (wrapped by the table `kek`, see §7) and appending an `index_trailer` blob
291/// (used by [`RunWriter`] for the learned index).
292///
293/// Tries the direct-to-mmap writer (Phase 14.6 — pages are encrypted + placed
294/// straight into a memory mapping of the output file, in parallel, with no
295/// intermediate whole-file `Vec<u8>`), and falls back to the in-buffer writer
296/// only when the mapping itself can't be created (some filesystems/environments
297/// reject `mmap`). The two writers produce a byte-identical run.
298pub fn write_run_with(
299    path: impl AsRef<Path>,
300    spec: &RunSpec,
301    kek: Option<&Kek>,
302    indexable_columns: &[(u16, u8)],
303    index_trailer: Option<&[u8]>,
304) -> Result<RunHeader> {
305    // Assemble per-run encryption material (fresh DEK + nonce prefix + wrapped
306    // descriptor) when a KEK is supplied. The page cipher lives only for this
307    // write; the wrapped DEK is embedded in the run below.
308    let enc: Option<RunEncryption> = match kek {
309        Some(k) => Some(setup_run_encryption(k, indexable_columns)?),
310        None => None,
311    };
312    match write_run_mmap(path.as_ref(), spec, enc.as_ref(), index_trailer) {
313        Ok(h) => Ok(h),
314        Err(e) if is_mmap_unavailable(&e) => write_run_vec(path, spec, enc, index_trailer),
315        Err(e) => Err(e),
316    }
317}
318
319fn write_run_with_file(
320    mut file: File,
321    spec: &RunSpec,
322    kek: Option<&Kek>,
323    indexable_columns: &[(u16, u8)],
324    index_trailer: Option<&[u8]>,
325) -> Result<RunHeader> {
326    let enc = match kek {
327        Some(key) => Some(setup_run_encryption(key, indexable_columns)?),
328        None => None,
329    };
330    match write_run_mmap_file(&file, spec, enc.as_ref(), index_trailer) {
331        Ok(header) => Ok(header),
332        Err(error) if is_mmap_unavailable(&error) => {
333            file.set_len(0)?;
334            file.seek(SeekFrom::Start(0))?;
335            let (bytes, header) = encode_run_vec(spec, enc, index_trailer)?;
336            file.write_all(&bytes)?;
337            file.sync_all()?;
338            Ok(header)
339        }
340        Err(error) => Err(error),
341    }
342}
343
344/// `true` when `write_run_mmap` could not create the mapping (the only case
345/// where we fall back to the in-buffer writer). Any other error — encryption,
346/// serialization, a genuine write failure — must surface.
347fn is_mmap_unavailable(e: &MongrelError) -> bool {
348    matches!(e, MongrelError::InvalidArgument(m) if m.starts_with("__mmap_unavailable__:"))
349}
350
351/// Direct-to-mmap run writer (Phases 14.5 + 14.6).
352///
353/// Lays the whole run out up front (computing each page's on-disk offset/length
354/// without encrypting), sizes the file, memory-maps it, then encrypts + copies
355/// every page **straight into the mapping** on the rayon pool. Because pages
356/// land in the kernel page cache (the mapping) instead of a heap `Vec<u8>`,
357/// the kernel can begin writeback of earlier pages while later ones are still
358/// being encrypted — CPU encryption overlaps with disk I/O (the double-buffer
359/// intent of 14.5). There is no whole-run buffer; peak extra memory is a few
360/// transient ciphertext `Vec`s (one per worker thread), not the full run.
361/// Direct-to-mmap run writer (Phases 14.5 + 14.6).
362///
363/// Lays the whole run out ([`plan_run`]), sizes the file, memory-maps it, then
364/// delegates to [`place_run`] which encrypts + copies every page **straight into
365/// the mapping** on the rayon pool. Because pages land in the kernel page cache
366/// (the mapping) instead of a heap `Vec<u8>`, the kernel can begin writeback of
367/// earlier pages while later ones are still being encrypted — CPU encryption
368/// overlaps with disk I/O (the double-buffer intent of 14.5). There is no
369/// whole-run buffer; peak extra memory is a few transient ciphertext `Vec`s
370/// (one per worker thread), not the full run.
371fn write_run_mmap(
372    path: &Path,
373    spec: &RunSpec,
374    enc: Option<&RunEncryption>,
375    index_trailer: Option<&[u8]>,
376) -> Result<RunHeader> {
377    let file = OpenOptions::new().create_new(true).write(true).open(path)?;
378    let result = write_run_mmap_file(&file, spec, enc, index_trailer);
379    if result.as_ref().is_err_and(is_mmap_unavailable) {
380        drop(file);
381        let _ = std::fs::remove_file(path);
382    }
383    result
384}
385
386fn write_run_mmap_file(
387    file: &File,
388    spec: &RunSpec,
389    enc: Option<&RunEncryption>,
390    index_trailer: Option<&[u8]>,
391) -> Result<RunHeader> {
392    let plan = plan_run(spec, enc, index_trailer)?;
393    file.set_len(plan.total as u64)?;
394    let mut mmap = match unsafe { memmap2::MmapMut::map_mut(file) } {
395        Ok(m) => m,
396        Err(e) => {
397            return Err(MongrelError::InvalidArgument(format!(
398                "__mmap_unavailable__: {e}"
399            )));
400        }
401    };
402    let header = place_run(spec, enc, index_trailer, &plan, &mut mmap[..])?;
403    mmap.flush()?;
404    file.sync_all()?;
405    Ok(header)
406}
407
408/// A precomputed run layout: the page-placement jobs, the serialized column
409/// directory (stats already filled), and every on-disk offset/length. The
410/// layout is independent of the backing buffer, so [`place_run`] can target a
411/// mmap'd file or a plain `Vec` interchangeably — which is what lets the
412/// placement path be unit-tested even where file mmap is unavailable.
413struct RunPlan {
414    jobs: Vec<(usize, usize, u64, usize)>, // (col_idx, page_seq, offset, on_disk_len)
415    dir_bytes: Vec<u8>,
416    encrypted: bool,
417    column_dir_offset: u64,
418    index_trailer_offset: u64,
419    encryption_descriptor_offset: u64,
420    footer_offset: u64,
421    total: usize,
422    /// Serialized (plaintext) [`EncryptedColumnStats`]; encrypted into the
423    /// stats section by the writer. `None` when the run is plaintext or no
424    /// encrypted column carries min/max.
425    encrypted_stats_plain: Option<Vec<u8>>,
426    encrypted_stats_offset: u64,
427    encrypted_stats_len: u64,
428}
429
430/// Compute the run layout: per-page offset + on-disk length, the filled column
431/// directory (serialized), and the dir / trailer / descriptor / footer offsets.
432/// Validates the encrypted-page-seq nonce-exhaustion bound.
433fn plan_run(
434    spec: &RunSpec,
435    enc: Option<&RunEncryption>,
436    index_trailer: Option<&[u8]>,
437) -> Result<RunPlan> {
438    let encrypted = enc.is_some();
439    let columns = spec.columns;
440    let mut jobs: Vec<(usize, usize, u64, usize)> = Vec::new();
441    let mut dir: Vec<ColumnPageHeader> = Vec::with_capacity(columns.len());
442    let mut enc_stats: EncryptedColumnStats = Vec::new();
443    let mut cursor: u64 = RUN_HEADER_PAD as u64;
444    for (ci, col) in columns.iter().enumerate() {
445        let region_offset = cursor;
446        let mut region_len = 0u64;
447        let mut stats = Vec::with_capacity(col.pages.len());
448        let mut col_minmax: Vec<PageMinMax> = Vec::new();
449        for (ps, page) in col.pages.iter().enumerate() {
450            // The per-page GCM nonce encodes page_seq in 2 bytes; refuse to
451            // silently truncate at 65 535 pages/column (4.29e9 rows), which
452            // would otherwise reuse a nonce under the run's DEK. Sequence
453            // 0xFFFF itself is reserved for the encrypted-stats envelope.
454            if encrypted && ps >= u16::MAX as usize {
455                return Err(MongrelError::Full(format!(
456                    "column {:#x} exceeds 65534 pages; encrypted-run page-seq nonce space exhausted",
457                    col.column_id
458                )));
459            }
460            let odl = on_disk_len(page.len(), encrypted);
461            jobs.push((ci, ps, cursor, odl));
462            let mut stat = col.page_stats.get(ps).cloned().unwrap_or(PageStat {
463                first_row_id: 0,
464                last_row_id: 0,
465                null_count: 0,
466                row_count: 0,
467                min: None,
468                max: None,
469                offset: 0,
470                compressed_len: 0,
471                uncompressed_len: 0,
472            });
473            stat.offset = cursor;
474            stat.compressed_len = odl as u32;
475            stat.uncompressed_len = page.len() as u32;
476            // The column directory is serialized in cleartext, so per-page
477            // min/max would leak raw plaintext values (literal bytes for `Bytes`
478            // columns) of every encrypted page to anyone reading the file
479            // without the key. Move them out of the directory and into the
480            // DEK-encrypted stats envelope, which the reader overlays back at
481            // open — zone-map page pruning works identically to plaintext runs.
482            if encrypted {
483                col_minmax.push((stat.min.take(), stat.max.take()));
484            }
485            stats.push(stat);
486            cursor += odl as u64;
487            region_len += odl as u64;
488        }
489        if col_minmax
490            .iter()
491            .any(|(mn, mx)| mn.is_some() || mx.is_some())
492        {
493            enc_stats.push((col.column_id, col_minmax));
494        }
495        let page_flags = if encrypted {
496            ColumnPageHeader::PAGE_ENCRYPTED
497        } else {
498            0
499        };
500        dir.push(ColumnPageHeader {
501            column_id: col.column_id,
502            type_id_tag: col.type_id_tag,
503            encoding: col.encoding as u8,
504            flags: page_flags,
505            page_count: col.pages.len() as u32,
506            page_region_offset: region_offset,
507            page_region_len: region_len,
508            page_stats: stats,
509        });
510    }
511    let dir_bytes = bincode::serialize(&dir)?;
512    let column_dir_offset = cursor;
513    cursor += dir_bytes.len() as u64;
514    let index_trailer_offset = match index_trailer {
515        Some(t) => {
516            let off = cursor;
517            cursor += t.len() as u64;
518            off
519        }
520        None => 0,
521    };
522    let encryption_descriptor_offset = match enc {
523        Some(e) => {
524            let off = cursor;
525            cursor += 4 + e.descriptor_bytes.len() as u64;
526            off
527        }
528        None => 0,
529    };
530    let (encrypted_stats_plain, encrypted_stats_offset, encrypted_stats_len) =
531        if encrypted && !enc_stats.is_empty() {
532            let plain = bincode::serialize(&enc_stats)?;
533            let ct_len = on_disk_len(plain.len(), true) as u64;
534            let off = cursor;
535            cursor += ct_len;
536            (Some(plain), off, ct_len)
537        } else {
538            (None, 0, 0)
539        };
540    let footer_offset = cursor;
541    // footer = MAGIC(8) + footer_offset(8) + checksum(32); encrypted runs append
542    // a 32-byte HMAC tag (RUN_MAC_LEN) authenticating header+dir+descriptor.
543    let total = footer_offset as usize + 8 + 8 + 32 + if encrypted { RUN_MAC_LEN } else { 0 };
544    Ok(RunPlan {
545        jobs,
546        dir_bytes,
547        encrypted,
548        column_dir_offset,
549        index_trailer_offset,
550        encryption_descriptor_offset,
551        footer_offset,
552        total,
553        encrypted_stats_plain,
554        encrypted_stats_offset,
555        encrypted_stats_len,
556    })
557}
558
559/// Place a run into `buf` (which must be exactly [`RunPlan::total`] bytes):
560/// encrypt + copy every page into its precomputed slot on the rayon pool, then
561/// write the column directory, index trailer, encryption descriptor, header,
562/// and checksummed footer. Backed by a mmap'd file in production and by a
563/// `Vec<u8>` in tests — the placement logic is identical either way, which is
564/// what makes the Phase 14.6 path unit-testable.
565fn place_run(
566    spec: &RunSpec,
567    enc: Option<&RunEncryption>,
568    index_trailer: Option<&[u8]>,
569    plan: &RunPlan,
570    buf: &mut [u8],
571) -> Result<RunHeader> {
572    use rayon::prelude::*;
573    use std::borrow::Cow;
574    debug_assert_eq!(
575        buf.len(),
576        plan.total,
577        "place_run: buffer must be exactly plan.total bytes"
578    );
579
580    let cipher: Option<&dyn Cipher> = enc.map(|e| e.cipher.as_ref());
581    let nonce_prefix = enc.map(|e| e.nonce_prefix);
582    let columns = spec.columns;
583
584    // ---- parallel placement: encrypt + copy each page into its slot ----
585    // Disjointness is by construction: every job targets [offset, offset+len)
586    // and `plan_run` made those ranges non-overlapping. The `SyncPtr` newtype
587    // lets the base pointer cross thread boundaries for that purpose.
588    struct SyncPtr(*mut u8);
589    unsafe impl Send for SyncPtr {}
590    unsafe impl Sync for SyncPtr {}
591    impl SyncPtr {
592        fn get(&self) -> *mut u8 {
593            self.0
594        }
595    }
596    let base = SyncPtr(buf.as_mut_ptr());
597    plan.jobs
598        .par_iter()
599        .map(
600            |&(ci, ps, offset, odl): &(usize, usize, u64, usize)| -> Result<()> {
601                let page = &columns[ci].pages[ps];
602                let dst =
603                    unsafe { std::slice::from_raw_parts_mut(base.get().add(offset as usize), odl) };
604                let bytes: Cow<[u8]> = match cipher {
605                    Some(c) => {
606                        let nonce =
607                            page_nonce(nonce_prefix.unwrap(), columns[ci].column_id, ps as u32);
608                        let ct = c.encrypt_page(&nonce, page)?;
609                        // Guards the GCM_TAG_LEN assumption: a future cipher with a
610                        // different tag size must change `on_disk_len` too.
611                        assert_eq!(
612                        ct.len(), odl,
613                        "ciphertext length {} != predicted {}; GCM tag size assumption is stale",
614                        ct.len(), odl
615                    );
616                        Cow::Owned(ct)
617                    }
618                    None => {
619                        debug_assert_eq!(page.len(), odl);
620                        Cow::Borrowed(page.as_slice())
621                    }
622                };
623                dst.copy_from_slice(&bytes);
624                Ok(())
625            },
626        )
627        .collect::<Result<()>>()?;
628
629    // ---- content hash over the on-disk page region (column-major, page order) ----
630    let content_hash = {
631        let mut h = Sha256::new();
632        h.update(&buf[RUN_HEADER_PAD..plan.column_dir_offset as usize]);
633        h.finalize()
634    };
635
636    // ---- dir / trailer / descriptor ----
637    let doff = plan.column_dir_offset as usize;
638    buf[doff..doff + plan.dir_bytes.len()].copy_from_slice(&plan.dir_bytes);
639    if let Some(t) = index_trailer {
640        let off = plan.index_trailer_offset as usize;
641        buf[off..off + t.len()].copy_from_slice(t);
642    }
643    if let Some(e) = enc {
644        let off = plan.encryption_descriptor_offset as usize;
645        buf[off..off + 4].copy_from_slice(&(e.descriptor_bytes.len() as u32).to_le_bytes());
646        buf[off + 4..off + 4 + e.descriptor_bytes.len()].copy_from_slice(&e.descriptor_bytes);
647    }
648    if let (Some(e), Some(plain)) = (enc, plan.encrypted_stats_plain.as_ref()) {
649        let nonce = page_nonce(e.nonce_prefix, ENC_STATS_NONCE_COLUMN, ENC_STATS_NONCE_SEQ);
650        let ct = e.cipher.encrypt_page(&nonce, plain)?;
651        debug_assert_eq!(ct.len() as u64, plan.encrypted_stats_len);
652        let off = plan.encrypted_stats_offset as usize;
653        buf[off..off + ct.len()].copy_from_slice(&ct);
654    }
655
656    // ---- header + footer ----
657    let header_flags = if plan.encrypted {
658        spec.flags | RUN_FLAG_ENCRYPTED
659    } else {
660        spec.flags
661    };
662    let header = RunHeader {
663        magic: RUN_MAGIC,
664        format_version: RUN_FORMAT_VERSION,
665        header_layout_version: RUN_HEADER_VERSION,
666        run_id: spec.run_id,
667        content_hash: content_hash.into(),
668        schema_id: spec.schema_id,
669        epoch_created: spec.epoch_created,
670        level: spec.level,
671        flags: header_flags,
672        sort_key_column_id: spec.sort_key_column_id,
673        row_count: spec.row_count,
674        min_row_id: spec.min_row_id,
675        max_row_id: spec.max_row_id,
676        column_count: columns.len() as u64,
677        column_dir_offset: plan.column_dir_offset,
678        index_trailer_offset: plan.index_trailer_offset,
679        encryption_descriptor_offset: plan.encryption_descriptor_offset,
680        footer_offset: plan.footer_offset,
681        encrypted_stats_offset: plan.encrypted_stats_offset,
682        encrypted_stats_len: plan.encrypted_stats_len,
683    };
684    let header_bytes = bincode::serialize(&header)?;
685    if header_bytes.len() > RUN_HEADER_PAD {
686        return Err(MongrelError::InvalidArgument(format!(
687            "run header too large: {} > {RUN_HEADER_PAD}",
688            header_bytes.len()
689        )));
690    }
691    buf[..header_bytes.len()].copy_from_slice(&header_bytes);
692
693    let checksum = Sha256::digest(&buf[..plan.footer_offset as usize]);
694    let foot = plan.footer_offset as usize;
695    buf[foot..foot + 8].copy_from_slice(&RUN_MAGIC);
696    buf[foot + 8..foot + 16].copy_from_slice(&plan.footer_offset.to_le_bytes());
697    buf[foot + 16..foot + 48].copy_from_slice(&checksum);
698    // Encrypted runs: append the keyed metadata MAC over header‖dir‖descriptor.
699    if let Some(tag) = compute_run_mac(enc, &header_bytes, &plan.dir_bytes) {
700        buf[foot + 48..foot + 48 + RUN_MAC_LEN].copy_from_slice(&tag);
701    }
702    Ok(header)
703}
704
705/// Fallback in-buffer writer (used when the output file can't be mmap'd).
706/// Produces a byte-identical run to [`write_run_mmap`].
707fn write_run_vec(
708    path: impl AsRef<Path>,
709    spec: &RunSpec,
710    enc: Option<RunEncryption>,
711    index_trailer: Option<&[u8]>,
712) -> Result<RunHeader> {
713    let (buf, header) = encode_run_vec(spec, enc, index_trailer)?;
714    let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
715    file.write_all(&buf)?;
716    file.sync_all()?;
717    Ok(header)
718}
719
720fn encode_run_vec(
721    spec: &RunSpec,
722    enc: Option<RunEncryption>,
723    index_trailer: Option<&[u8]>,
724) -> Result<(Vec<u8>, RunHeader)> {
725    let mut buf: Vec<u8> = vec![0; RUN_HEADER_PAD]; // reserve header region
726    let mut content_hasher = Sha256::new();
727    let mut dir: Vec<ColumnPageHeader> = Vec::with_capacity(spec.columns.len());
728    let mut enc_stats: EncryptedColumnStats = Vec::new();
729
730    for col in spec.columns {
731        let region_offset = buf.len() as u64;
732        let mut region_len = 0u64;
733        let mut stats = Vec::with_capacity(col.pages.len());
734        let mut col_minmax: Vec<PageMinMax> = Vec::new();
735        for (page_seq, page) in col.pages.iter().enumerate() {
736            if enc.is_some() && page_seq >= u16::MAX as usize {
737                return Err(MongrelError::Full(format!(
738                    "column {:#x} exceeds 65534 pages; encrypted-run page-seq nonce space exhausted",
739                    col.column_id
740                )));
741            }
742            let on_disk: Vec<u8> = match &enc {
743                Some(e) => e.cipher.encrypt_page(
744                    &page_nonce(e.nonce_prefix, col.column_id, page_seq as u32),
745                    page,
746                )?,
747                None => page.clone(),
748            };
749            let offset = buf.len() as u64;
750            buf.write_all(&on_disk)?;
751            content_hasher.update(&on_disk);
752            region_len += on_disk.len() as u64;
753            let mut stat = if let Some(s) = col.page_stats.get(page_seq) {
754                s.clone()
755            } else {
756                PageStat {
757                    first_row_id: 0,
758                    last_row_id: 0,
759                    null_count: 0,
760                    row_count: 0,
761                    min: None,
762                    max: None,
763                    offset: 0,
764                    compressed_len: 0,
765                    uncompressed_len: 0,
766                }
767            };
768            stat.offset = offset;
769            stat.compressed_len = on_disk.len() as u32;
770            stat.uncompressed_len = page.len() as u32;
771            // See plan_run: the cleartext directory must not carry plaintext
772            // min/max for encrypted columns — they move into the encrypted
773            // stats envelope. Keep this byte-identical to plan_run so both
774            // writers emit the same run.
775            if enc.is_some() {
776                col_minmax.push((stat.min.take(), stat.max.take()));
777            }
778            stats.push(stat);
779        }
780        if col_minmax
781            .iter()
782            .any(|(mn, mx)| mn.is_some() || mx.is_some())
783        {
784            enc_stats.push((col.column_id, col_minmax));
785        }
786        let page_flags = if enc.is_some() {
787            ColumnPageHeader::PAGE_ENCRYPTED
788        } else {
789            0
790        };
791        dir.push(ColumnPageHeader {
792            column_id: col.column_id,
793            type_id_tag: col.type_id_tag,
794            encoding: col.encoding as u8,
795            flags: page_flags,
796            page_count: col.pages.len() as u32,
797            page_region_offset: region_offset,
798            page_region_len: region_len,
799            page_stats: stats,
800        });
801    }
802
803    let column_dir_offset = buf.len() as u64;
804    let dir_bytes = bincode::serialize(&dir)?;
805    buf.write_all(&dir_bytes)?;
806
807    let index_trailer_offset = match index_trailer {
808        Some(trailer) => {
809            let off = buf.len() as u64;
810            buf.write_all(trailer)?;
811            off
812        }
813        None => 0,
814    };
815    let encryption_descriptor_offset = match &enc {
816        Some(e) => {
817            let off = buf.len() as u64;
818            buf.write_all(&(e.descriptor_bytes.len() as u32).to_le_bytes())?;
819            buf.write_all(&e.descriptor_bytes)?;
820            off
821        }
822        None => 0,
823    };
824    let (encrypted_stats_offset, encrypted_stats_len) = match &enc {
825        Some(e) if !enc_stats.is_empty() => {
826            let plain = bincode::serialize(&enc_stats)?;
827            let nonce = page_nonce(e.nonce_prefix, ENC_STATS_NONCE_COLUMN, ENC_STATS_NONCE_SEQ);
828            let ct = e.cipher.encrypt_page(&nonce, &plain)?;
829            let off = buf.len() as u64;
830            let len = ct.len() as u64;
831            buf.write_all(&ct)?;
832            (off, len)
833        }
834        _ => (0, 0),
835    };
836    let footer_offset = buf.len() as u64;
837
838    let header_flags = if enc.is_some() {
839        spec.flags | RUN_FLAG_ENCRYPTED
840    } else {
841        spec.flags
842    };
843    let header = RunHeader {
844        magic: RUN_MAGIC,
845        format_version: RUN_FORMAT_VERSION,
846        header_layout_version: RUN_HEADER_VERSION,
847        run_id: spec.run_id,
848        content_hash: content_hasher.finalize().into(),
849        schema_id: spec.schema_id,
850        epoch_created: spec.epoch_created,
851        level: spec.level,
852        flags: header_flags,
853        sort_key_column_id: spec.sort_key_column_id,
854        row_count: spec.row_count,
855        min_row_id: spec.min_row_id,
856        max_row_id: spec.max_row_id,
857        column_count: spec.columns.len() as u64,
858        column_dir_offset,
859        index_trailer_offset,
860        encryption_descriptor_offset,
861        footer_offset,
862        encrypted_stats_offset,
863        encrypted_stats_len,
864    };
865    let header_bytes = bincode::serialize(&header)?;
866    if header_bytes.len() > RUN_HEADER_PAD {
867        return Err(MongrelError::InvalidArgument(format!(
868            "run header too large: {} > {RUN_HEADER_PAD}",
869            header_bytes.len()
870        )));
871    }
872    buf[..header_bytes.len()].copy_from_slice(&header_bytes);
873
874    let checksum = Sha256::digest(&buf[..footer_offset as usize]);
875    buf.write_all(&RUN_MAGIC)?;
876    buf.write_all(&footer_offset.to_le_bytes())?;
877    buf.write_all(&checksum)?;
878    // Encrypted runs: append the keyed metadata MAC (byte-identical to the mmap
879    // writer). `dir_bytes`/`header_bytes` here are the exact serialized forms.
880    if let Some(tag) = compute_run_mac(enc.as_ref(), &header_bytes, &dir_bytes) {
881        buf.write_all(&tag)?;
882    }
883
884    Ok((buf, header))
885}
886
887fn page_nonce(nonce_prefix: [u8; 12], column_id: u16, page_seq: u32) -> [u8; 12] {
888    let mut n = nonce_prefix;
889    n[8..10].copy_from_slice(&column_id.to_le_bytes());
890    n[10..12].copy_from_slice(&(page_seq as u16).to_le_bytes());
891    n
892}
893
894/// Decrypt the run's per-page min/max envelope ([`EncryptedColumnStats`]) and
895/// overlay the bounds onto the in-memory column directory, restoring zone-map
896/// page pruning for encrypted columns. Caller must have verified the run
897/// metadata MAC first (the envelope's offset/len live in the header); the
898/// envelope itself is AES-256-GCM-authenticated, so tampering fails loudly —
899/// the same posture as a tampered page payload.
900fn overlay_encrypted_stats(
901    file: &mut File,
902    header: &RunHeader,
903    cipher: &dyn Cipher,
904    nonce_prefix: [u8; 12],
905    dir: &mut [ColumnPageHeader],
906) -> Result<()> {
907    file.seek(SeekFrom::Start(header.encrypted_stats_offset))?;
908    const MAX_ENCRYPTED_STATS_BYTES: u64 = 64 * 1024 * 1024;
909    if header.encrypted_stats_len > MAX_ENCRYPTED_STATS_BYTES {
910        return Err(MongrelError::InvalidArgument(format!(
911            "encrypted run stats length {} exceeds {MAX_ENCRYPTED_STATS_BYTES}",
912            header.encrypted_stats_len
913        )));
914    }
915    let mut ct = vec![0u8; header.encrypted_stats_len as usize];
916    file.read_exact(&mut ct)?;
917    let nonce = page_nonce(nonce_prefix, ENC_STATS_NONCE_COLUMN, ENC_STATS_NONCE_SEQ);
918    let plain = cipher.decrypt_page(&nonce, &ct)?;
919    let stats: EncryptedColumnStats = bincode::deserialize(&plain)
920        .map_err(|e| MongrelError::Encryption(format!("bad encrypted page-stats envelope: {e}")))?;
921    for (cid, minmax) in stats {
922        let Some(col) = dir.iter_mut().find(|c| c.column_id == cid) else {
923            continue;
924        };
925        for (stat, (mn, mx)) in col.page_stats.iter_mut().zip(minmax) {
926            stat.min = mn;
927            stat.max = mx;
928        }
929    }
930    Ok(())
931}
932
933/// Stable content-address of an immutable run page (the cache key): SHA-256 of
934/// `(table_id, run_id, column_id, page_seq)`. Runs are immutable, so this
935/// identity is also the page's content address — a rewritten page lives in a
936/// different run (different id) and so gets a different key without any
937/// invalidation sweep. `table_id` namespaces the shared cache across tables in
938/// a `Database` so two tables' identically-numbered runs never collide.
939pub(crate) fn page_cache_key(
940    table_id: u64,
941    run_id: u128,
942    column_id: u16,
943    page_seq: usize,
944) -> [u8; 32] {
945    let mut h = Sha256::new();
946    h.update(table_id.to_be_bytes());
947    h.update(run_id.to_be_bytes());
948    h.update(column_id.to_be_bytes());
949    h.update((page_seq as u64).to_be_bytes());
950    let out = h.finalize();
951    let mut k = [0u8; 32];
952    k.copy_from_slice(&out);
953    k
954}
955
956/// Decrypt a raw (on-disk) page when the column is encrypted, else pass it
957/// through. The shared page cache stores the raw bytes (ciphertext when
958/// encrypted), so this runs after every cache hit or disk read.
959fn decrypt_or_passthrough(
960    cipher: Option<&dyn Cipher>,
961    nonce_prefix: [u8; 12],
962    column_id: u16,
963    page_seq: usize,
964    encrypted: bool,
965    buf: &[u8],
966) -> Result<Vec<u8>> {
967    if encrypted {
968        match cipher {
969            Some(c) => {
970                Ok(c.decrypt_page(&page_nonce(nonce_prefix, column_id, page_seq as u32), buf)?)
971            }
972            None => Err(MongrelError::Decryption(
973                "encrypted page but no cipher".into(),
974            )),
975        }
976    } else {
977        Ok(buf.to_vec())
978    }
979}
980
981/// Read just the header and confirm the footer magic, without hashing the
982/// body. Cheap: two small fixed-size reads, no allocation proportional to
983/// run size. Only safe as a substitute for [`read_header`] when the caller
984/// has independent evidence this exact `run_id`'s checksum already verified
985/// in this process (see `open_with_cache`'s `verified_runs` cache) — `.sr`
986/// runs are immutable once written, so a checksum verified once cannot
987/// regress later in the same process lifetime. Still catches gross
988/// corruption (truncation, garbled header) via the magic checks and the
989/// bincode deserialize.
990fn read_header_fast_from_file(file: &mut File) -> Result<RunHeader> {
991    file.seek(SeekFrom::Start(0))?;
992    let mut header_buf = [0u8; RUN_HEADER_PAD];
993    file.read_exact(&mut header_buf)?;
994    let header: RunHeader = bincode::deserialize(&header_buf)
995        .map_err(|e| MongrelError::InvalidArgument(format!("bad run header: {e}")))?;
996    validate_run_header_bytes(&header, &header_buf)?;
997    validate_run_layout(&header, file.metadata()?.len())?;
998    file.seek(SeekFrom::Start(header.footer_offset))?;
999    let mut footer_magic = [0u8; 8];
1000    file.read_exact(&mut footer_magic)?;
1001    if footer_magic != RUN_MAGIC {
1002        return Err(MongrelError::MagicMismatch {
1003            what: "sorted run footer",
1004            expected: RUN_MAGIC,
1005            got: footer_magic,
1006        });
1007    }
1008    Ok(header)
1009}
1010
1011/// Read and validate a run header (magic + footer checksum).
1012pub fn read_header(path: impl AsRef<Path>) -> Result<RunHeader> {
1013    let mut file = crate::durable_file::open_regular_nofollow(path.as_ref())?;
1014    read_header_from_file(&mut file)
1015}
1016
1017fn validate_run_layout(header: &RunHeader, file_len: u64) -> Result<()> {
1018    let footer_len = 8u64
1019        + 8
1020        + 32
1021        + if header.is_encrypted() {
1022            RUN_MAC_LEN as u64
1023        } else {
1024            0
1025        };
1026    let expected_len = header
1027        .footer_offset
1028        .checked_add(footer_len)
1029        .ok_or_else(|| MongrelError::InvalidArgument("sorted run length overflow".into()))?;
1030    if header.footer_offset < RUN_HEADER_PAD as u64 || expected_len != file_len {
1031        return Err(MongrelError::InvalidArgument(format!(
1032            "invalid sorted run layout: footer={} file={file_len}",
1033            header.footer_offset
1034        )));
1035    }
1036    let in_body = |offset: u64| {
1037        offset == 0 || (offset >= RUN_HEADER_PAD as u64 && offset <= header.footer_offset)
1038    };
1039    if header.column_dir_offset < RUN_HEADER_PAD as u64
1040        || header.column_dir_offset > header.footer_offset
1041        || !in_body(header.index_trailer_offset)
1042        || !in_body(header.encryption_descriptor_offset)
1043        || !in_body(header.encrypted_stats_offset)
1044        || header
1045            .encrypted_stats_offset
1046            .checked_add(header.encrypted_stats_len)
1047            .is_none_or(|end| end > header.footer_offset)
1048    {
1049        return Err(MongrelError::InvalidArgument(
1050            "sorted run metadata offsets are outside the file".into(),
1051        ));
1052    }
1053    Ok(())
1054}
1055
1056fn validate_run_header_bytes(header: &RunHeader, bytes: &[u8; RUN_HEADER_PAD]) -> Result<()> {
1057    if header.magic != RUN_MAGIC {
1058        return Err(MongrelError::MagicMismatch {
1059            what: "sorted run",
1060            expected: RUN_MAGIC,
1061            got: header.magic,
1062        });
1063    }
1064    const KNOWN_FLAGS: u8 =
1065        RUN_FLAG_ENCRYPTED | RUN_FLAG_TOMBSTONE_ONLY | RUN_FLAG_CLEAN | RUN_FLAG_UNIFORM_EPOCH;
1066    if header.format_version != RUN_FORMAT_VERSION
1067        || header.header_layout_version != RUN_HEADER_VERSION
1068        || header.flags & !KNOWN_FLAGS != 0
1069        || (header.row_count == 0 && (header.min_row_id != 0 || header.max_row_id != 0))
1070        || (header.row_count != 0 && header.min_row_id > header.max_row_id)
1071    {
1072        return Err(MongrelError::InvalidArgument(
1073            "unsupported or invalid sorted run header".into(),
1074        ));
1075    }
1076    let canonical = bincode::serialize(header)?;
1077    if canonical.len() > RUN_HEADER_PAD
1078        || bytes[..canonical.len()] != canonical
1079        || bytes[canonical.len()..].iter().any(|byte| *byte != 0)
1080    {
1081        return Err(MongrelError::InvalidArgument(
1082            "sorted run header has noncanonical bytes or nonzero padding".into(),
1083        ));
1084    }
1085    Ok(())
1086}
1087
1088fn read_header_from_file(file: &mut File) -> Result<RunHeader> {
1089    file.seek(SeekFrom::Start(0))?;
1090    let mut header_buf = [0u8; RUN_HEADER_PAD];
1091    file.read_exact(&mut header_buf)?;
1092    let header: RunHeader = bincode::deserialize(&header_buf)
1093        .map_err(|e| MongrelError::InvalidArgument(format!("bad run header: {e}")))?;
1094    validate_run_header_bytes(&header, &header_buf)?;
1095
1096    validate_run_layout(&header, file.metadata()?.len())?;
1097    file.seek(SeekFrom::Start(header.footer_offset))?;
1098    let mut footer = [0u8; 8 + 8 + 32];
1099    file.read_exact(&mut footer)?;
1100    if footer[..8] != RUN_MAGIC {
1101        return Err(MongrelError::MagicMismatch {
1102            what: "sorted run footer",
1103            expected: RUN_MAGIC,
1104            got: footer[..8].try_into().unwrap(),
1105        });
1106    }
1107    let mut hasher = Sha256::new();
1108    file.seek(SeekFrom::Start(0))?;
1109    let mut remaining = header.footer_offset;
1110    let mut buffer = [0u8; 64 * 1024];
1111    while remaining != 0 {
1112        let length = usize::try_from(remaining.min(buffer.len() as u64)).unwrap();
1113        file.read_exact(&mut buffer[..length])?;
1114        hasher.update(&buffer[..length]);
1115        remaining -= length as u64;
1116    }
1117    let computed: [u8; 32] = hasher.finalize().into();
1118    let stored: [u8; 32] = footer[16..].try_into().unwrap();
1119    if computed != stored {
1120        return Err(MongrelError::ChecksumMismatch {
1121            expected: u64::from_be_bytes(stored[..8].try_into().unwrap()),
1122            actual: u64::from_be_bytes(computed[..8].try_into().unwrap()),
1123            context: "sorted run footer".into(),
1124        });
1125    }
1126    file.seek(SeekFrom::Start(RUN_HEADER_PAD as u64))?;
1127    let mut remaining = header.column_dir_offset - RUN_HEADER_PAD as u64;
1128    let mut content = Sha256::new();
1129    while remaining != 0 {
1130        let length = usize::try_from(remaining.min(buffer.len() as u64)).unwrap();
1131        file.read_exact(&mut buffer[..length])?;
1132        content.update(&buffer[..length]);
1133        remaining -= length as u64;
1134    }
1135    let content_hash: [u8; 32] = content.finalize().into();
1136    if content_hash != header.content_hash {
1137        return Err(MongrelError::ChecksumMismatch {
1138            expected: u64::from_be_bytes(header.content_hash[..8].try_into().unwrap()),
1139            actual: u64::from_be_bytes(content_hash[..8].try_into().unwrap()),
1140            context: "sorted run content hash".into(),
1141        });
1142    }
1143    Ok(header)
1144}
1145
1146/// Read the column directory.
1147pub fn read_column_dir(
1148    path: impl AsRef<Path>,
1149    header: &RunHeader,
1150) -> Result<Vec<ColumnPageHeader>> {
1151    let mut file = crate::durable_file::open_regular_nofollow(path.as_ref())?;
1152    read_column_dir_from_file(&mut file, header)
1153}
1154
1155fn read_column_dir_from_file(file: &mut File, header: &RunHeader) -> Result<Vec<ColumnPageHeader>> {
1156    file.seek(SeekFrom::Start(header.column_dir_offset))?;
1157    let end = [
1158        header.index_trailer_offset,
1159        header.encryption_descriptor_offset,
1160        header.encrypted_stats_offset,
1161        header.footer_offset,
1162    ]
1163    .into_iter()
1164    .filter(|offset| *offset > header.column_dir_offset)
1165    .min()
1166    .ok_or_else(|| {
1167        MongrelError::InvalidArgument("sorted run column directory has no end".into())
1168    })?;
1169    let len = end.checked_sub(header.column_dir_offset).ok_or_else(|| {
1170        MongrelError::InvalidArgument("sorted run column directory offsets are reversed".into())
1171    })?;
1172    const MAX_COLUMN_DIR_BYTES: u64 = 64 * 1024 * 1024;
1173    if len > MAX_COLUMN_DIR_BYTES {
1174        return Err(MongrelError::InvalidArgument(format!(
1175            "sorted run column directory length {len} exceeds {MAX_COLUMN_DIR_BYTES}"
1176        )));
1177    }
1178    let mut buf = vec![0u8; len as usize];
1179    file.read_exact(&mut buf)?;
1180    let dir: Vec<ColumnPageHeader> = bincode::deserialize(&buf)
1181        .map_err(|e| MongrelError::InvalidArgument(format!("bad column dir: {e}")))?;
1182    Ok(dir)
1183}
1184
1185fn validate_column_directory_layout(header: &RunHeader, dir: &[ColumnPageHeader]) -> Result<()> {
1186    if header.column_count != dir.len() as u64 || dir.len() > u16::MAX as usize {
1187        return Err(MongrelError::InvalidArgument(
1188            "sorted run column count is invalid".into(),
1189        ));
1190    }
1191    let mut regions = Vec::with_capacity(dir.len());
1192    let mut ids = std::collections::HashSet::new();
1193    for column in dir {
1194        if !ids.insert(column.column_id)
1195            || column.flags & !ColumnPageHeader::PAGE_ENCRYPTED != 0
1196            || column.page_count as usize != column.page_stats.len()
1197        {
1198            return Err(MongrelError::InvalidArgument(
1199                "sorted run column directory identity is invalid".into(),
1200            ));
1201        }
1202        let region_end = column
1203            .page_region_offset
1204            .checked_add(column.page_region_len)
1205            .ok_or_else(|| MongrelError::InvalidArgument("run page region overflows".into()))?;
1206        if column.page_region_offset < RUN_HEADER_PAD as u64
1207            || region_end > header.column_dir_offset
1208        {
1209            return Err(MongrelError::InvalidArgument(
1210                "sorted run page region is outside its body".into(),
1211            ));
1212        }
1213        let mut cursor = column.page_region_offset;
1214        let mut rows = 0_u64;
1215        for stat in &column.page_stats {
1216            let length = stat.compressed_len as u64;
1217            let end = stat
1218                .offset
1219                .checked_add(length)
1220                .ok_or_else(|| MongrelError::InvalidArgument("run page offset overflows".into()))?;
1221            if stat.offset != cursor
1222                || length == 0
1223                || length > MAX_RUN_PAGE_BYTES
1224                || stat.uncompressed_len as u64 > MAX_RUN_PAGE_BYTES
1225                || stat.row_count as usize > PAGE_ROWS
1226                || end > region_end
1227            {
1228                return Err(MongrelError::InvalidArgument(
1229                    "sorted run page metadata is outside its region".into(),
1230                ));
1231            }
1232            rows = rows.checked_add(stat.row_count as u64).ok_or_else(|| {
1233                MongrelError::InvalidArgument("sorted run row count overflows".into())
1234            })?;
1235            cursor = end;
1236        }
1237        if cursor != region_end || rows != header.row_count {
1238            return Err(MongrelError::InvalidArgument(
1239                "sorted run page region length or row count is inconsistent".into(),
1240            ));
1241        }
1242        regions.push((column.page_region_offset, region_end));
1243    }
1244    regions.sort_unstable();
1245    if regions.windows(2).any(|pair| pair[0].1 > pair[1].0) {
1246        return Err(MongrelError::InvalidArgument(
1247            "sorted run page regions overlap".into(),
1248        ));
1249    }
1250    Ok(())
1251}
1252
1253fn read_encryption_descriptor_bytes_from_file(
1254    file: &mut File,
1255    header: &RunHeader,
1256) -> Result<Vec<u8>> {
1257    file.seek(SeekFrom::Start(header.encryption_descriptor_offset))?;
1258    let mut len_buf = [0u8; 4];
1259    file.read_exact(&mut len_buf)?;
1260    let len = u32::from_le_bytes(len_buf) as usize;
1261    // The descriptor is tiny (~60 bytes + a few column entries); clamp to a
1262    // sane ceiling so a corrupt/malicious header can't trigger a multi-GiB
1263    // allocation. (The footer checksum already guards integrity; this is a
1264    // defense-in-depth bound on the pre-checksum read.)
1265    const MAX_DESCRIPTOR_BYTES: usize = 65_536;
1266    if len > MAX_DESCRIPTOR_BYTES {
1267        return Err(MongrelError::InvalidArgument(format!(
1268            "encryption descriptor length {len} exceeds {MAX_DESCRIPTOR_BYTES}"
1269        )));
1270    }
1271    let mut buf = vec![0u8; len];
1272    file.read_exact(&mut buf)?;
1273    Ok(buf)
1274}
1275
1276/// Authenticate an encrypted run's cleartext metadata (`header ‖ dir ‖
1277/// descriptor`) against the keyed MAC tag stored after the footer. Run BEFORE
1278/// any offset/stat from the directory is trusted to drive a read. Errors if the
1279/// tag is missing (a run written before run-metadata MACs existed) or does not
1280/// match (tampering, or the wrong key). The on-disk page payloads are AEAD-
1281/// authenticated separately, so they are not covered here.
1282fn verify_run_mac(
1283    file: &mut File,
1284    header: &RunHeader,
1285    dir: &[ColumnPageHeader],
1286    kek: &Kek,
1287    desc_bytes: &[u8],
1288) -> Result<()> {
1289    let header_bytes = bincode::serialize(header)?;
1290    let dir_bytes = bincode::serialize(dir)?;
1291    let mac_key = kek.derive_run_mac_key();
1292    let expected =
1293        crate::encryption::run_metadata_mac(&mac_key, &header_bytes, &dir_bytes, desc_bytes);
1294    file.seek(SeekFrom::Start(header.footer_offset + 8 + 8 + 32))?;
1295    let mut tag = [0u8; RUN_MAC_LEN];
1296    file.read_exact(&mut tag).map_err(|_| {
1297        MongrelError::Decryption(
1298            "encrypted run is missing or truncated its metadata MAC; cannot \
1299             authenticate metadata"
1300                .into(),
1301        )
1302    })?;
1303    // Constant-time comparison (no early-exit timing oracle on the tag).
1304    let mut diff = 0u8;
1305    for (x, y) in tag.iter().zip(expected.iter()) {
1306        diff |= x ^ y;
1307    }
1308    if diff != 0 {
1309        return Err(MongrelError::Decryption(
1310            "run metadata authentication failed — tampered run or wrong key".into(),
1311        ));
1312    }
1313    Ok(())
1314}
1315
1316// ============================ high-level writer ============================
1317
1318/// Builds and writes a sorted run from drained memtable rows.
1319///
1320/// `rows` must be sorted ascending by `(row_id, epoch)` (the memtable's natural
1321/// drain order). System columns `_row_id`, `_epoch`, `_deleted` are always
1322/// emitted; each user column in `schema` is emitted, with `Null` for rows that
1323/// don't set it.
1324pub struct RunWriter<'a> {
1325    schema: &'a Schema,
1326    run_id: u128,
1327    epoch_created: Epoch,
1328    level: u8,
1329    kek: Option<&'a Kek>,
1330    /// `(column_id, scheme)` for each ENCRYPTED_INDEXABLE column — wrapped into
1331    /// the run's Encryption Descriptor (Phase 10.2).
1332    indexable_columns: Vec<(u16, u8)>,
1333    /// Per-page compression policy (Phase 14.4 / 15.3). Default `Zstd(3)`
1334    /// (compaction); the bulk path uses `Zstd(1)` or `Lz4` (hot, scan-heavy
1335    /// runs), and `Plain` skips compression entirely (`bulk_load_fast`).
1336    compress: columnar::Compress,
1337    /// Whether this run is "clean" (one version per RowId, no tombstones,
1338    /// ascending row_ids) — written into [`RUN_FLAG_CLEAN`] so readers can skip
1339    /// the MVCC visibility pass. Set true only by paths that construct clean
1340    /// system columns by construction (typed bulk load, compaction output).
1341    clean: bool,
1342    /// Whether this run's stored `_epoch` column is a placeholder and its real
1343    /// commit epoch lives in the manifest `RunRef.epoch_created` (set only by the
1344    /// large-transaction spill path, which writes before the epoch is assigned).
1345    /// Stamps [`RUN_FLAG_UNIFORM_EPOCH`] so the reader overlays the real epoch.
1346    uniform_epoch: bool,
1347    /// Write fixed-width page payloads in little-endian (Phase 15.7) so the
1348    /// decode path is a memcpy on real (x86/ARM) hardware instead of a
1349    /// per-element `swap_bytes`. Default **true** for the typed write paths
1350    /// (`write_native`); the legacy `write` (Value) path keeps big-endian for
1351    /// back-compat with pre-15.7 runs. The flag is stored per-page (bit 3 of the
1352    /// algo byte), so a run may freely mix LE and BE pages.
1353    le: bool,
1354}
1355
1356impl<'a> RunWriter<'a> {
1357    pub fn new(schema: &'a Schema, run_id: u128, epoch_created: Epoch, level: u8) -> Self {
1358        Self {
1359            schema,
1360            run_id,
1361            epoch_created,
1362            level,
1363            kek: None,
1364            indexable_columns: Vec::new(),
1365            compress: columnar::Compress::Zstd(3),
1366            clean: false,
1367            uniform_epoch: false,
1368            le: false,
1369        }
1370    }
1371
1372    /// Mark this run as uniform-epoch: its stored `_epoch` column is a
1373    /// placeholder and the real commit epoch is supplied at read time from the
1374    /// manifest `RunRef.epoch_created` (see [`RUN_FLAG_UNIFORM_EPOCH`]). Used by
1375    /// the large-transaction spill path, which writes the run before the commit
1376    /// epoch is assigned.
1377    pub fn uniform_epoch(mut self, uniform: bool) -> Self {
1378        self.uniform_epoch = uniform;
1379        self
1380    }
1381
1382    /// Encrypt this run's pages with a fresh per-file DEK wrapped by `kek`.
1383    /// `indexable_columns` are the ENCRYPTED_INDEXABLE `(column_id, scheme)`
1384    /// pairs whose column keys are derived+wrapped into the descriptor.
1385    pub fn with_encryption(mut self, kek: &'a Kek, indexable_columns: Vec<(u16, u8)>) -> Self {
1386        self.kek = Some(kek);
1387        self.indexable_columns = indexable_columns;
1388        self
1389    }
1390
1391    /// Override the zstd level for this run's pages (Phase 14.4). The bulk
1392    /// ingest path uses level 1; background compaction upgrades cold runs to 3.
1393    pub fn with_zstd_level(mut self, level: i32) -> Self {
1394        self.compress = columnar::Compress::Zstd(level);
1395        self
1396    }
1397
1398    /// Compress hot/mutable-run pages with LZ4 (Phase 15.3): 3–5× faster decode
1399    /// than zstd with ~10% worse ratio. The right default for runs that get
1400    /// scanned (bulk-loaded analytical runs).
1401    pub fn with_lz4(mut self) -> Self {
1402        self.compress = columnar::Compress::Lz4;
1403        self
1404    }
1405
1406    /// Emit raw `ALGO_PLAIN` pages with no compression (Phase 14.4
1407    /// `bulk_load_fast`): maximal encode throughput at the cost of ~3–4× size.
1408    pub fn with_plain(mut self) -> Self {
1409        self.compress = columnar::Compress::Plain;
1410        self
1411    }
1412
1413    /// Mark this run as "clean" (one version per RowId, no tombstones, ascending
1414    /// row_ids). Only set when the caller constructs the system columns so by
1415    /// construction (typed bulk load of fresh, contiguous row_ids; compaction
1416    /// output that has collapsed versions and dropped tombstones). Stamps
1417    /// [`RUN_FLAG_CLEAN`] into the header so readers skip the MVCC pass.
1418    pub fn clean(mut self, clean: bool) -> Self {
1419        self.clean = clean;
1420        self
1421    }
1422
1423    /// Write fixed-width page payloads in little-endian (Phase 15.7): the decode
1424    /// path becomes a memcpy on little-endian targets. Only effective on the
1425    /// typed `write_native` path; no-op on big-endian writers (they keep the
1426    /// portable BE layout, since "native" would not be LE there).
1427    pub fn with_native_endian(mut self) -> Self {
1428        if cfg!(target_endian = "little") {
1429            self.le = true;
1430        }
1431        self
1432    }
1433
1434    /// Write a run straight from typed columns (no `Value`). `user_columns` are
1435    /// the schema's user columns as [`NativeColumn`]s; the system columns
1436    /// (`_row_id`/`_epoch`/`_deleted`) are built from `first_row_id..+n` /
1437    /// `epoch_created` / all-false. Sorted Int64 columns (always the system
1438    /// `_row_id`, plus any sorted user Int64) use delta encoding unless
1439    /// [`RunWriter::with_plain`] forces raw `ALGO_PLAIN` everywhere.
1440    pub fn write_native(
1441        self,
1442        path: impl AsRef<Path>,
1443        user_columns: &[(u16, columnar::NativeColumn)],
1444        n: usize,
1445        first_row_id: u64,
1446    ) -> Result<RunHeader> {
1447        self.write_native_target(Some(path.as_ref()), None, user_columns, n, first_row_id)
1448    }
1449
1450    pub(crate) fn write_native_file(
1451        self,
1452        file: File,
1453        user_columns: &[(u16, columnar::NativeColumn)],
1454        n: usize,
1455        first_row_id: u64,
1456    ) -> Result<RunHeader> {
1457        self.write_native_target(None, Some(file), user_columns, n, first_row_id)
1458    }
1459
1460    fn write_native_target(
1461        self,
1462        path: Option<&Path>,
1463        file: Option<File>,
1464        user_columns: &[(u16, columnar::NativeColumn)],
1465        n: usize,
1466        first_row_id: u64,
1467    ) -> Result<RunHeader> {
1468        use columnar::NativeColumn;
1469        let row_id_col = NativeColumn::int64_sequence(first_row_id as i64, n);
1470        let epoch_col = NativeColumn::int64_constant(self.epoch_created.0 as i64, n);
1471        let deleted_col = NativeColumn::bool_constant(false, n);
1472
1473        let learned_trailer = build_learned_trailer_native(&row_id_col);
1474
1475        // All columns split on the same row-position boundaries so a reader can
1476        // skip whole pages of any column via its [min,max] stat.
1477        let row_ids = match &row_id_col {
1478            NativeColumn::Int64 { data, .. } => data.as_slice(),
1479            _ => &[],
1480        };
1481        let bounds = page_bounds(row_ids);
1482        let compress = self.compress;
1483        let le = self.le;
1484        let plain = matches!(compress, columnar::Compress::Plain);
1485        // In plain mode every column is `Encoding::Plain` (raw); otherwise keep
1486        // the chosen encoding (Delta for the sorted _row_id, Zstd for the rest).
1487        let row_id_enc = if plain {
1488            Encoding::Plain
1489        } else {
1490            Encoding::Delta
1491        };
1492        let sys_enc = if plain {
1493            Encoding::Plain
1494        } else {
1495            Encoding::Zstd
1496        };
1497
1498        let mut columns: Vec<ColumnPayload> = Vec::with_capacity(3 + self.schema.columns.len());
1499        let (pages, stats) = native_column_pages(
1500            TypeId::Int64,
1501            &row_id_col,
1502            row_id_enc,
1503            compress,
1504            le,
1505            &bounds,
1506        )?;
1507        columns.push(ColumnPayload {
1508            column_id: SYS_ROW_ID,
1509            type_id_tag: type_tag(&TypeId::Int64),
1510            encoding: row_id_enc,
1511            pages,
1512            page_stats: stats,
1513        });
1514        let (pages, stats) =
1515            native_column_pages(TypeId::Int64, &epoch_col, sys_enc, compress, le, &bounds)?;
1516        columns.push(ColumnPayload {
1517            column_id: SYS_EPOCH,
1518            type_id_tag: type_tag(&TypeId::Int64),
1519            encoding: sys_enc,
1520            pages,
1521            page_stats: stats,
1522        });
1523        let (pages, stats) =
1524            native_column_pages(TypeId::Bool, &deleted_col, sys_enc, compress, le, &bounds)?;
1525        columns.push(ColumnPayload {
1526            column_id: SYS_DELETED,
1527            type_id_tag: type_tag(&TypeId::Bool),
1528            encoding: sys_enc,
1529            pages,
1530            page_stats: stats,
1531        });
1532        // Encode all user columns in parallel (Phase 14.3): each column's pages
1533        // are independent, and the page-level work is itself parallel, so a
1534        // wide table saturates the pool without oversubscribing (rayon
1535        // work-steals across the nested parallelism). Order is preserved so the
1536        // column directory matches `schema.columns` order.
1537        use rayon::prelude::*;
1538        let user_cols: Vec<ColumnPayload> = self
1539            .schema
1540            .columns
1541            .par_iter()
1542            .map(|cdef| -> Result<ColumnPayload> {
1543                let col = user_columns
1544                    .iter()
1545                    .find(|(id, _)| *id == cdef.id)
1546                    .map(|(_, c)| c)
1547                    .ok_or_else(|| {
1548                        MongrelError::ColumnNotFound(format!("user column {}", cdef.id))
1549                    })?;
1550                let encoding = if plain {
1551                    Encoding::Plain
1552                } else {
1553                    choose_encoding_native(&cdef.ty, col)
1554                };
1555                let (pages, stats) =
1556                    native_column_pages(cdef.ty.clone(), col, encoding, compress, le, &bounds)?;
1557                Ok(ColumnPayload {
1558                    column_id: cdef.id,
1559                    type_id_tag: type_tag(&cdef.ty),
1560                    encoding,
1561                    pages,
1562                    page_stats: stats,
1563                })
1564            })
1565            .collect::<Result<Vec<_>>>()?;
1566        columns.extend(user_cols);
1567
1568        // A uniform-epoch run's `_epoch` column is a placeholder (the real commit
1569        // epoch is overlaid from the RunRef at read time), so it must NOT be
1570        // marked clean — the clean fast path skips snapshot gating — and must
1571        // carry the overlay flag. write_native is not the spill path today, but
1572        // keep it sound if it ever is.
1573        let flags = if self.uniform_epoch {
1574            RUN_FLAG_UNIFORM_EPOCH
1575        } else {
1576            RUN_FLAG_CLEAN
1577        };
1578        let spec = RunSpec {
1579            run_id: self.run_id,
1580            schema_id: self.schema.schema_id,
1581            epoch_created: self.epoch_created.0,
1582            level: self.level,
1583            flags,
1584            sort_key_column_id: SYS_ROW_ID,
1585            row_count: n as u64,
1586            min_row_id: first_row_id,
1587            max_row_id: first_row_id + n as u64 - 1,
1588            columns: &columns,
1589        };
1590        match file {
1591            Some(file) => write_run_with_file(
1592                file,
1593                &spec,
1594                self.kek,
1595                &self.indexable_columns,
1596                Some(&learned_trailer),
1597            ),
1598            None => write_run_with(
1599                path.ok_or_else(|| {
1600                    MongrelError::InvalidArgument("sorted run output is missing".into())
1601                })?,
1602                &spec,
1603                self.kek,
1604                &self.indexable_columns,
1605                Some(&learned_trailer),
1606            ),
1607        }
1608    }
1609
1610    pub fn write(self, path: impl AsRef<Path>, rows: &[Row]) -> Result<RunHeader> {
1611        self.write_target(Some(path.as_ref()), None, rows)
1612    }
1613
1614    pub(crate) fn write_file(self, file: File, rows: &[Row]) -> Result<RunHeader> {
1615        self.write_target(None, Some(file), rows)
1616    }
1617
1618    fn write_target(
1619        self,
1620        path: Option<&Path>,
1621        file: Option<File>,
1622        rows: &[Row],
1623    ) -> Result<RunHeader> {
1624        let n = rows.len();
1625        // System columns.
1626        let mut row_ids = Vec::with_capacity(n);
1627        let mut epochs = Vec::with_capacity(n);
1628        let mut deleted = Vec::with_capacity(n);
1629        let mut commit_ts_vals = Vec::with_capacity(n);
1630        // Emit SYS_COMMIT_TS when any version is HLC-stamped (P0.5-T3).
1631        let has_commit_ts = rows.iter().any(|r| r.commit_ts.is_some());
1632        for r in rows {
1633            row_ids.push(Value::Int64(r.row_id.0 as i64));
1634            epochs.push(Value::Int64(r.committed_epoch.0 as i64));
1635            deleted.push(Value::Bool(r.deleted));
1636            if has_commit_ts {
1637                commit_ts_vals.push(encode_commit_ts_value(r.commit_ts));
1638            }
1639        }
1640        let learned_trailer = build_learned_trailer(&row_ids);
1641        let (min_rid, max_rid) = row_id_bounds(rows);
1642        let row_id_i64: Vec<i64> = rows.iter().map(|r| r.row_id.0 as i64).collect();
1643        let bounds = page_bounds(&row_id_i64);
1644
1645        let mut columns: Vec<ColumnPayload> =
1646            Vec::with_capacity(3 + usize::from(has_commit_ts) + self.schema.columns.len());
1647        let (pages, stats, enc) = value_column_pages(TypeId::Int64, &row_ids, &bounds)?;
1648        columns.push(ColumnPayload {
1649            column_id: SYS_ROW_ID,
1650            type_id_tag: type_tag(&TypeId::Int64),
1651            encoding: enc,
1652            pages,
1653            page_stats: stats,
1654        });
1655        let (pages, stats, enc) = value_column_pages(TypeId::Int64, &epochs, &bounds)?;
1656        columns.push(ColumnPayload {
1657            column_id: SYS_EPOCH,
1658            type_id_tag: type_tag(&TypeId::Int64),
1659            encoding: enc,
1660            pages,
1661            page_stats: stats,
1662        });
1663        let (pages, stats, enc) = value_column_pages(TypeId::Bool, &deleted, &bounds)?;
1664        columns.push(ColumnPayload {
1665            column_id: SYS_DELETED,
1666            type_id_tag: type_tag(&TypeId::Bool),
1667            encoding: enc,
1668            pages,
1669            page_stats: stats,
1670        });
1671        if has_commit_ts {
1672            let (pages, stats, enc) = value_column_pages(TypeId::Bytes, &commit_ts_vals, &bounds)?;
1673            columns.push(ColumnPayload {
1674                column_id: SYS_COMMIT_TS,
1675                type_id_tag: type_tag(&TypeId::Bytes),
1676                encoding: enc,
1677                pages,
1678                page_stats: stats,
1679            });
1680        }
1681        // User columns — choose an encoding per column from run-time stats.
1682        for cdef in &self.schema.columns {
1683            let vals: Vec<Value> = rows
1684                .iter()
1685                .map(|r| r.columns.get(&cdef.id).cloned().unwrap_or(Value::Null))
1686                .collect();
1687            let (pages, stats, encoding) = value_column_pages(cdef.ty.clone(), &vals, &bounds)?;
1688            columns.push(ColumnPayload {
1689                column_id: cdef.id,
1690                type_id_tag: type_tag(&cdef.ty),
1691                encoding,
1692                pages,
1693                page_stats: stats,
1694            });
1695        }
1696
1697        let spec = RunSpec {
1698            run_id: self.run_id,
1699            schema_id: self.schema.schema_id,
1700            epoch_created: self.epoch_created.0,
1701            level: self.level,
1702            flags: {
1703                let mut f = if self.clean { RUN_FLAG_CLEAN } else { 0 };
1704                if self.uniform_epoch {
1705                    f |= RUN_FLAG_UNIFORM_EPOCH;
1706                }
1707                f
1708            },
1709            sort_key_column_id: SYS_ROW_ID,
1710            row_count: n as u64,
1711            min_row_id: min_rid,
1712            max_row_id: max_rid,
1713            columns: &columns,
1714        };
1715        match file {
1716            Some(file) => write_run_with_file(
1717                file,
1718                &spec,
1719                self.kek,
1720                &self.indexable_columns,
1721                Some(&learned_trailer),
1722            ),
1723            None => write_run_with(
1724                path.ok_or_else(|| {
1725                    MongrelError::InvalidArgument("sorted run output is missing".into())
1726                })?,
1727                &spec,
1728                self.kek,
1729                &self.indexable_columns,
1730                Some(&learned_trailer),
1731            ),
1732        }
1733    }
1734}
1735
1736fn type_tag(ty: &TypeId) -> u16 {
1737    // Informational only; the reader resolves the full type from the schema.
1738    match ty {
1739        TypeId::Bool => 1,
1740        TypeId::Int64 => 8,
1741        TypeId::Float64 => 9,
1742        TypeId::Bytes => 12,
1743        TypeId::Embedding { .. } => 13,
1744        _ => 0,
1745    }
1746}
1747
1748/// Decode a big-endian i64 from a `PageStat` min/max slot (None if absent or
1749/// truncated — treated as "all-null page" by the caller).
1750pub(crate) fn be_i64(b: Option<&[u8]>) -> Option<i64> {
1751    let b = b?;
1752    (b.len() == 8).then(|| i64::from_be_bytes(b.try_into().unwrap()))
1753}
1754
1755/// Decode a big-endian f64 (stored as `to_bits`) from a stat slot.
1756pub(crate) fn be_f64(b: Option<&[u8]>) -> Option<f64> {
1757    let b = b?;
1758    (b.len() == 8).then(|| f64::from_bits(u64::from_be_bytes(b.try_into().unwrap())))
1759}
1760
1761/// Rows per columnar page. Small enough to prune effectively, large enough to
1762/// keep per-page overhead negligible. ~16 pages per 1M-row column.
1763const PAGE_ROWS: usize = 65_536;
1764
1765/// `(start, end, first_row_id, last_row_id)` for each page, derived from the
1766/// actual row-id column so flush paths with non-contiguous ids stay correct.
1767fn page_bounds(row_ids: &[i64]) -> Vec<(usize, usize, u64, u64)> {
1768    let n = row_ids.len();
1769    if n == 0 {
1770        return vec![(0, 0, 0, 0)];
1771    }
1772    let mut out = Vec::new();
1773    let mut start = 0;
1774    while start < n {
1775        let end = (start + PAGE_ROWS).min(n);
1776        out.push((start, end, row_ids[start] as u64, row_ids[end - 1] as u64));
1777        start = end;
1778    }
1779    out
1780}
1781
1782/// Split a typed column into per-page encoded bytes + value-derived stats.
1783/// Pages are encoded in parallel across the rayon pool when a column spans more
1784/// than one page (Phase 14.3) — a 1M-row column has 16 independent encode tasks.
1785/// `compress` selects the page algorithm (Phase 14.4 / 15.3). Order is preserved
1786/// so the column directory stays sequential by page_seq.
1787fn native_column_pages(
1788    ty: TypeId,
1789    col: &columnar::NativeColumn,
1790    encoding: Encoding,
1791    compress: columnar::Compress,
1792    le: bool,
1793    bounds: &[(usize, usize, u64, u64)],
1794) -> Result<(Vec<Vec<u8>>, Vec<PageStat>)> {
1795    use rayon::prelude::*;
1796    let encode_one =
1797        |&(s, e, frid, lrid): &(usize, usize, u64, u64)| -> Result<(Vec<u8>, PageStat)> {
1798            let chunk = col.slice_range(s, e);
1799            let stat = columnar::page_stat_for(ty.clone(), &chunk, frid, lrid);
1800            let page = columnar::encode_page_native(ty.clone(), &chunk, encoding, compress, le)?;
1801            Ok((page, stat))
1802        };
1803    // Single-page columns skip the thread-pool handshake.
1804    let pairs: Vec<(Vec<u8>, PageStat)> = if bounds.len() > 1 {
1805        bounds
1806            .par_iter()
1807            .map(encode_one)
1808            .collect::<Result<Vec<_>>>()?
1809    } else {
1810        bounds.iter().map(encode_one).collect::<Result<Vec<_>>>()?
1811    };
1812    let (pages, stats) = pairs.into_iter().unzip();
1813    Ok((pages, stats))
1814}
1815
1816/// Split a `Value` column into per-page encoded bytes + stats (flush path).
1817fn value_column_pages(
1818    ty: TypeId,
1819    vals: &[Value],
1820    bounds: &[(usize, usize, u64, u64)],
1821) -> Result<(Vec<Vec<u8>>, Vec<PageStat>, Encoding)> {
1822    let encoding = choose_encoding(&ty, vals);
1823    let mut pages = Vec::with_capacity(bounds.len());
1824    let mut stats = Vec::with_capacity(bounds.len());
1825    for &(s, e, frid, lrid) in bounds {
1826        let chunk = &vals[s..e];
1827        pages.push(columnar::encode_page(ty.clone(), chunk, encoding)?);
1828        let native = columnar::values_to_native(ty.clone(), chunk);
1829        stats.push(columnar::page_stat_for(ty.clone(), &native, frid, lrid));
1830    }
1831    Ok((pages, stats, encoding))
1832}
1833
1834/// Pick a page encoding from run-time stats: dictionary for low-cardinality
1835/// strings, zstd-plain otherwise.
1836fn choose_encoding(ty: &TypeId, values: &[Value]) -> Encoding {
1837    use std::collections::HashSet;
1838    if matches!(ty, TypeId::Bytes) {
1839        let n = values.len();
1840        if n > 0 {
1841            let distinct = values
1842                .iter()
1843                .filter(|v| !matches!(v, Value::Null))
1844                .map(|v| v.encode_key())
1845                .collect::<HashSet<_>>()
1846                .len();
1847            if (distinct as f64 / n as f64) < 0.5 {
1848                return Encoding::Dictionary;
1849            }
1850        }
1851    }
1852    Encoding::Zstd
1853}
1854
1855/// Encoding choice for the typed (native) path: delta for sorted Int64,
1856/// dictionary for low-cardinality Bytes, zstd otherwise.
1857fn choose_encoding_native(ty: &TypeId, col: &columnar::NativeColumn) -> Encoding {
1858    use std::collections::HashSet;
1859    match (ty, col) {
1860        (TypeId::Int64 | TypeId::TimestampNanos, columnar::NativeColumn::Int64 { data, .. }) => {
1861            if data.windows(2).all(|w| w[0] <= w[1]) {
1862                Encoding::Delta
1863            } else {
1864                Encoding::Zstd
1865            }
1866        }
1867        (
1868            TypeId::Bytes,
1869            columnar::NativeColumn::Bytes {
1870                offsets, values, ..
1871            },
1872        ) => {
1873            let n = offsets.len().saturating_sub(1);
1874            if n > 0 {
1875                let distinct: HashSet<&[u8]> = (0..n)
1876                    .map(|i| &values[offsets[i] as usize..offsets[i + 1] as usize])
1877                    .collect();
1878                if (distinct.len() as f64 / n as f64) < 0.5 {
1879                    return Encoding::Dictionary;
1880                }
1881            }
1882            Encoding::Zstd
1883        }
1884        _ => Encoding::Zstd,
1885    }
1886}
1887
1888/// PGM-index trailer built straight from the typed row-id column.
1889fn build_learned_trailer_native(col: &columnar::NativeColumn) -> Vec<u8> {
1890    let points: Vec<(u64, usize)> = match col {
1891        columnar::NativeColumn::Int64 { data, .. } => data
1892            .iter()
1893            .enumerate()
1894            .map(|(i, v)| (*v as u64, i))
1895            .collect(),
1896        _ => Vec::new(),
1897    };
1898    let pgm = PgmIndex::build(&points, LEARNED_EPSILON);
1899    bincode::serialize(&pgm).expect("pgm serialize")
1900}
1901
1902fn row_id_bounds(rows: &[Row]) -> (u64, u64) {
1903    match (rows.first(), rows.last()) {
1904        (Some(f), Some(l)) => (f.row_id.0, l.row_id.0),
1905        _ => (0, 0),
1906    }
1907}
1908
1909// ============================ high-level reader ============================
1910
1911/// Reads a sorted run: decodes columns lazily (cached), answers MVCC point
1912/// lookups via page-pruned `SYS_ROW_ID` bounds, and materializes visible rows
1913/// for scans.
1914pub struct RunReader {
1915    file: File,
1916    mmap: Option<memmap2::Mmap>,
1917    header: RunHeader,
1918    dir: Vec<ColumnPageHeader>,
1919    schema: Schema,
1920    /// Owning table id — namespaces the shared page cache across tables.
1921    table_id: u64,
1922    /// Per-run page cipher, built from the unwrapped DEK (None when plaintext).
1923    cipher: Option<Box<dyn Cipher>>,
1924    /// Per-run nonce prefix (overlaid per page with column_id + page_seq).
1925    nonce_prefix: [u8; 12],
1926    col_cache: HashMap<u16, Vec<Value>>,
1927    /// Shared, MVCC content-addressed page cache (Phase 9.2). Caches raw page
1928    /// bytes (ciphertext when encrypted) so all readers share decoded/decrypted
1929    /// pages. `None` only in standalone tests.
1930    page_cache: Option<Arc<crate::cache::Sharded<crate::cache::PageCache>>>,
1931    /// Shared decoded-page cache (Phase 15.4): the post-decompress/decrypt typed
1932    /// page, so a repeat scan skips decode. Keyed by `(run_id, column_id,
1933    /// page_seq)` identity; `None` in standalone tests.
1934    decoded_cache: Option<Arc<crate::cache::Sharded<crate::cache::DecodedPageCache>>>,
1935    /// Uniform-epoch overlay (see [`RUN_FLAG_UNIFORM_EPOCH`]). When `Some`, every
1936    /// row's commit epoch is taken to be this value instead of the placeholder
1937    /// stored in the `_epoch` column. Set by [`Self::set_uniform_epoch`].
1938    epoch_override: Option<Epoch>,
1939}
1940
1941#[derive(Debug, Clone, Copy)]
1942pub(crate) struct RunVisibleVersion {
1943    pub(crate) row_id: RowId,
1944    pub(crate) committed_epoch: Epoch,
1945    pub(crate) deleted: bool,
1946    /// Optional HLC from SYS_COMMIT_TS when the run carries that column.
1947    /// Used by controlled merge for HLC-authoritative cross-tier winners
1948    /// without forcing full materialization first.
1949    pub(crate) commit_ts: Option<HlcTimestamp>,
1950    page_seq: usize,
1951    within_page: usize,
1952}
1953
1954/// Page-bounded cursor over one run's newest snapshot-visible version per row.
1955/// Only the three compact system columns and one user-data page are decoded at
1956/// a time. Full `Vec<Row>` materialization is deliberately avoided.
1957pub(crate) struct RunVisibleVersionCursor {
1958    reader: RunReader,
1959    snapshot: Epoch,
1960    page_row_counts: Vec<usize>,
1961    page_seq: usize,
1962    within_page: usize,
1963    row_ids: Vec<i64>,
1964    epochs: Vec<i64>,
1965    deleted: Vec<u8>,
1966    /// Per-row optional HLC stamps for the loaded system page (parallel to
1967    /// `row_ids`). Empty when the run has no SYS_COMMIT_TS column.
1968    commit_ts: Vec<Option<HlcTimestamp>>,
1969    has_commit_ts_col: bool,
1970    lookahead: Option<RunVisibleVersion>,
1971    materialized_page: Option<usize>,
1972    materialized_columns: Vec<(u16, columnar::NativeColumn)>,
1973}
1974
1975impl RunVisibleVersionCursor {
1976    fn new(reader: RunReader, snapshot: Epoch) -> Result<Self> {
1977        let page_row_counts = reader.page_row_counts(SYS_ROW_ID)?;
1978        let has_commit_ts_col = reader.has_column(SYS_COMMIT_TS);
1979        Ok(Self {
1980            reader,
1981            snapshot,
1982            page_row_counts,
1983            page_seq: 0,
1984            within_page: 0,
1985            row_ids: Vec::new(),
1986            epochs: Vec::new(),
1987            deleted: Vec::new(),
1988            commit_ts: Vec::new(),
1989            has_commit_ts_col,
1990            lookahead: None,
1991            materialized_page: None,
1992            materialized_columns: Vec::new(),
1993        })
1994    }
1995
1996    fn load_system_page(&mut self, control: &crate::ExecutionControl) -> Result<bool> {
1997        while self.page_seq < self.page_row_counts.len() {
1998            control.checkpoint()?;
1999            let rows = self.page_row_counts[self.page_seq];
2000            if rows == 0 {
2001                self.page_seq += 1;
2002                continue;
2003            }
2004            self.row_ids = match columnar::decode_page_native(
2005                TypeId::Int64,
2006                &self.reader.read_page(SYS_ROW_ID, self.page_seq)?,
2007                rows,
2008            )? {
2009                columnar::NativeColumn::Int64 { data, .. } => data,
2010                _ => return Err(MongrelError::InvalidArgument("sys row_id not int64".into())),
2011            };
2012            self.epochs = if let Some(epoch) = self.reader.epoch_override {
2013                vec![epoch.0 as i64; rows]
2014            } else {
2015                match columnar::decode_page_native(
2016                    TypeId::Int64,
2017                    &self.reader.read_page(SYS_EPOCH, self.page_seq)?,
2018                    rows,
2019                )? {
2020                    columnar::NativeColumn::Int64 { data, .. } => data,
2021                    _ => return Err(MongrelError::InvalidArgument("sys epoch not int64".into())),
2022                }
2023            };
2024            self.deleted = match columnar::decode_page_native(
2025                TypeId::Bool,
2026                &self.reader.read_page(SYS_DELETED, self.page_seq)?,
2027                rows,
2028            )? {
2029                columnar::NativeColumn::Bool { data, .. } => data,
2030                _ => return Err(MongrelError::InvalidArgument("sys deleted not bool".into())),
2031            };
2032            self.commit_ts.clear();
2033            if self.has_commit_ts_col {
2034                let page = self.reader.read_page(SYS_COMMIT_TS, self.page_seq)?;
2035                let native = columnar::decode_page_native(TypeId::Bytes, &page, rows)?;
2036                self.commit_ts.reserve(rows);
2037                for i in 0..rows {
2038                    self.commit_ts
2039                        .push(decode_commit_ts_value(native.value_at(i).as_ref()));
2040                }
2041            }
2042            self.within_page = 0;
2043            return Ok(true);
2044        }
2045        Ok(false)
2046    }
2047
2048    fn next_raw(&mut self, control: &crate::ExecutionControl) -> Result<Option<RunVisibleVersion>> {
2049        if self.within_page >= self.row_ids.len() {
2050            if !self.row_ids.is_empty() {
2051                self.page_seq += 1;
2052                self.row_ids.clear();
2053                self.epochs.clear();
2054                self.deleted.clear();
2055                self.commit_ts.clear();
2056            }
2057            if !self.load_system_page(control)? {
2058                return Ok(None);
2059            }
2060        }
2061        if self.within_page.is_multiple_of(256) {
2062            control.checkpoint()?;
2063        }
2064        let position = self.within_page;
2065        self.within_page += 1;
2066        let commit_ts = self.commit_ts.get(position).copied().flatten();
2067        Ok(Some(RunVisibleVersion {
2068            row_id: RowId(self.row_ids[position] as u64),
2069            committed_epoch: Epoch(self.epochs[position] as u64),
2070            deleted: self.deleted[position] != 0,
2071            commit_ts,
2072            page_seq: self.page_seq,
2073            within_page: position,
2074        }))
2075    }
2076
2077    pub(crate) fn next_visible_version(
2078        &mut self,
2079        control: &crate::ExecutionControl,
2080    ) -> Result<Option<RunVisibleVersion>> {
2081        loop {
2082            let first = match self.lookahead.take() {
2083                Some(version) => version,
2084                None => match self.next_raw(control)? {
2085                    Some(version) => version,
2086                    None => return Ok(None),
2087                },
2088            };
2089            let row_id = first.row_id;
2090            let mut best = (first.committed_epoch <= self.snapshot).then_some(first);
2091            while let Some(candidate) = self.next_raw(control)? {
2092                if candidate.row_id != row_id {
2093                    self.lookahead = Some(candidate);
2094                    break;
2095                }
2096                if candidate.committed_epoch <= self.snapshot
2097                    && best.is_none_or(|current| {
2098                        crate::epoch::Snapshot::version_is_newer(
2099                            candidate.committed_epoch,
2100                            candidate.commit_ts,
2101                            current.committed_epoch,
2102                            current.commit_ts,
2103                        )
2104                    })
2105                {
2106                    best = Some(candidate);
2107                }
2108            }
2109            if best.is_some() {
2110                return Ok(best);
2111            }
2112        }
2113    }
2114
2115    pub(crate) fn materialize(
2116        &mut self,
2117        version: RunVisibleVersion,
2118        control: &crate::ExecutionControl,
2119    ) -> Result<Row> {
2120        if self.materialized_page != Some(version.page_seq) {
2121            let rows = self.page_row_counts[version.page_seq];
2122            let columns = self.reader.schema.columns.clone();
2123            let mut materialized = Vec::with_capacity(columns.len());
2124            for (index, column) in columns.into_iter().enumerate() {
2125                if index % 16 == 0 {
2126                    control.checkpoint()?;
2127                }
2128                let native = if self.reader.has_column(column.id) {
2129                    columnar::decode_page_native(
2130                        column.ty,
2131                        &self.reader.read_page(column.id, version.page_seq)?,
2132                        rows,
2133                    )?
2134                } else {
2135                    columnar::null_native(column.ty, rows)
2136                };
2137                materialized.push((column.id, native));
2138            }
2139            self.materialized_columns = materialized;
2140            self.materialized_page = Some(version.page_seq);
2141        }
2142        let columns = self
2143            .materialized_columns
2144            .iter()
2145            .map(|(column_id, column)| {
2146                (
2147                    *column_id,
2148                    column.value_at(version.within_page).unwrap_or(Value::Null),
2149                )
2150            })
2151            .collect();
2152        // Prefer the stamp already loaded on the candidate; fall back to a
2153        // page read for legacy cursors that predate per-candidate stamps.
2154        let commit_ts = version.commit_ts.or_else(|| {
2155            if self.reader.has_column(SYS_COMMIT_TS) {
2156                let page = self
2157                    .reader
2158                    .read_page(SYS_COMMIT_TS, version.page_seq)
2159                    .ok()?;
2160                let native = columnar::decode_page_native(
2161                    TypeId::Bytes,
2162                    &page,
2163                    self.page_row_counts[version.page_seq],
2164                )
2165                .ok()?;
2166                decode_commit_ts_value(native.value_at(version.within_page).as_ref())
2167            } else {
2168                None
2169            }
2170        });
2171        Ok(Row {
2172            row_id: version.row_id,
2173            committed_epoch: version.committed_epoch,
2174            columns,
2175            deleted: version.deleted,
2176            commit_ts,
2177        })
2178    }
2179}
2180
2181impl RunReader {
2182    pub fn open(path: impl AsRef<Path>, schema: Schema, kek: Option<Arc<Kek>>) -> Result<Self> {
2183        Self::open_with_cache(path, schema, kek, None, None, 0, None)
2184    }
2185
2186    pub(crate) fn open_file(file: File, schema: Schema, kek: Option<Arc<Kek>>) -> Result<Self> {
2187        Self::open_file_with_cache(file, schema, kek, None, None, 0, None, None)
2188    }
2189
2190    #[allow(clippy::too_many_arguments)]
2191    pub(crate) fn open_with_cache(
2192        path: impl AsRef<Path>,
2193        schema: Schema,
2194        kek: Option<Arc<Kek>>,
2195        page_cache: Option<Arc<crate::cache::Sharded<crate::cache::PageCache>>>,
2196        decoded_cache: Option<Arc<crate::cache::Sharded<crate::cache::DecodedPageCache>>>,
2197        table_id: u64,
2198        verified_runs: Option<&parking_lot::Mutex<std::collections::HashSet<u128>>>,
2199    ) -> Result<Self> {
2200        let path = path.as_ref().to_path_buf();
2201        let file = crate::durable_file::open_regular_nofollow(&path)?;
2202        Self::open_file_with_cache(
2203            file,
2204            schema,
2205            kek,
2206            page_cache,
2207            decoded_cache,
2208            table_id,
2209            verified_runs,
2210            Some(path),
2211        )
2212    }
2213
2214    #[allow(clippy::too_many_arguments)]
2215    pub(crate) fn open_file_with_cache(
2216        mut file: File,
2217        schema: Schema,
2218        kek: Option<Arc<Kek>>,
2219        page_cache: Option<Arc<crate::cache::Sharded<crate::cache::PageCache>>>,
2220        decoded_cache: Option<Arc<crate::cache::Sharded<crate::cache::DecodedPageCache>>>,
2221        table_id: u64,
2222        verified_runs: Option<&parking_lot::Mutex<std::collections::HashSet<u128>>>,
2223        path: Option<PathBuf>,
2224    ) -> Result<Self> {
2225        let header = match verified_runs {
2226            Some(cache) => {
2227                let header = read_header_fast_from_file(&mut file)?;
2228                if cache.lock().contains(&header.run_id) {
2229                    header
2230                } else {
2231                    let verified = read_header_from_file(&mut file)?;
2232                    cache.lock().insert(verified.run_id);
2233                    verified
2234                }
2235            }
2236            None => read_header_from_file(&mut file)?,
2237        };
2238        let mut dir = read_column_dir_from_file(&mut file, &header)?;
2239        validate_column_directory_layout(&header, &dir)?;
2240        if header.is_encrypted() != kek.is_some() {
2241            return Err(MongrelError::Encryption(
2242                "sorted-run encryption mode differs from the database".into(),
2243            ));
2244        }
2245        // Unwrap this run's per-file DEK (stored wrapped in its Encryption
2246        // Descriptor) using the table KEK, then build the page cipher.
2247        let (cipher, nonce_prefix): (Option<Box<dyn Cipher>>, [u8; 12]) = if header.is_encrypted() {
2248            let kek = kek.as_ref().ok_or_else(|| {
2249                MongrelError::Encryption(
2250                    "run is encrypted but no key-encryption key was provided".into(),
2251                )
2252            })?;
2253            if header.encryption_descriptor_offset == 0 {
2254                return Err(MongrelError::Encryption(
2255                    "encrypted run has no encryption descriptor".into(),
2256                ));
2257            }
2258            let desc_bytes = read_encryption_descriptor_bytes_from_file(&mut file, &header)?;
2259            // Authenticate the cleartext metadata (header‖dir‖descriptor) under
2260            // the KEK-derived MAC key BEFORE trusting any offset/stat to drive a
2261            // read. Required for every encrypted run (no downgrade path: an
2262            // attacker can neither strip the encryption — pages stay ciphertext —
2263            // nor forge the tag without the key).
2264            verify_run_mac(&mut file, &header, &dir, kek, &desc_bytes)?;
2265            let enc = crate::encryption::build_run_cipher(kek, &desc_bytes)?;
2266            // With the metadata authenticated, decrypt the per-page min/max
2267            // envelope (v2 runs) and overlay it so zone-map pruning works on
2268            // encrypted columns exactly as it does on plaintext ones.
2269            if header.encrypted_stats_offset != 0 {
2270                overlay_encrypted_stats(
2271                    &mut file,
2272                    &header,
2273                    enc.cipher.as_ref(),
2274                    enc.nonce_prefix,
2275                    &mut dir,
2276                )?;
2277            }
2278            (Some(enc.cipher), enc.nonce_prefix)
2279        } else {
2280            (None, [0u8; 12])
2281        };
2282        // Best-effort memory map: lets the OS page cache manage I/O and removes
2283        // per-page seek+read syscalls. Falls back to read() on empty/unmappable
2284        // files. The `file` handle is kept for the lifetime of the mapping.
2285        // (Per-column `MADV_WILLNEED` read-ahead is issued from
2286        // `column_native_shared` — see Phase 15.2 — so a global advice policy is
2287        // not set here; that would degrade concurrent point lookups.)
2288        let mmap = unsafe { memmap2::MmapOptions::new().map(&file) }.ok();
2289        let _ = path;
2290        Ok(Self {
2291            file,
2292            mmap,
2293            header,
2294            dir,
2295            schema,
2296            table_id,
2297            cipher,
2298            nonce_prefix,
2299            col_cache: HashMap::new(),
2300            epoch_override: None,
2301            page_cache,
2302            decoded_cache,
2303        })
2304    }
2305
2306    pub fn header(&self) -> &RunHeader {
2307        &self.header
2308    }
2309
2310    pub(crate) fn validate_all_pages(&mut self) -> Result<()> {
2311        let mut columns = std::collections::HashSet::new();
2312        let row_header = self.find_header(SYS_ROW_ID)?.clone();
2313        let expected_rows = row_header
2314            .page_stats
2315            .iter()
2316            .map(|stat| stat.row_count)
2317            .collect::<Vec<_>>();
2318        let mut row_bounds = Vec::with_capacity(row_header.page_stats.len());
2319        let mut previous = None::<u64>;
2320        let mut first = None::<u64>;
2321        let mut last = None::<u64>;
2322        let mut counted = 0_u64;
2323        for (page, stat) in row_header.page_stats.iter().enumerate() {
2324            let bytes = self.read_page(SYS_ROW_ID, page)?;
2325            let native =
2326                columnar::decode_page_native(TypeId::Int64, &bytes, stat.row_count as usize)?;
2327            if columnar::page_stat_for(TypeId::Int64, &native, 0, 0).null_count != 0 {
2328                return Err(MongrelError::InvalidArgument(
2329                    "sorted run row-id page contains nulls".into(),
2330                ));
2331            }
2332            let columnar::NativeColumn::Int64 { data, validity } = native else {
2333                return Err(MongrelError::InvalidArgument(
2334                    "sorted run row-id page has the wrong type".into(),
2335                ));
2336            };
2337            let _ = validity;
2338            if data.is_empty() {
2339                if self.header.row_count != 0 || stat.row_count != 0 {
2340                    return Err(MongrelError::InvalidArgument(
2341                        "sorted run row-id page is unexpectedly empty".into(),
2342                    ));
2343                }
2344                row_bounds.push((0, 0));
2345                continue;
2346            }
2347            if data.iter().any(|value| *value < 0) {
2348                return Err(MongrelError::InvalidArgument(
2349                    "sorted run contains a negative row id".into(),
2350                ));
2351            }
2352            let page_first = data[0] as u64;
2353            let page_last = data[data.len() - 1] as u64;
2354            for value in data {
2355                let value = value as u64;
2356                if previous.is_some_and(|prior| {
2357                    value < prior || (self.header.is_clean() && value == prior)
2358                }) {
2359                    return Err(MongrelError::InvalidArgument(
2360                        "sorted run row ids are not ordered".into(),
2361                    ));
2362                }
2363                first.get_or_insert(value);
2364                previous = Some(value);
2365                last = Some(value);
2366                counted += 1;
2367            }
2368            if stat.first_row_id != page_first || stat.last_row_id != page_last {
2369                return Err(MongrelError::InvalidArgument(
2370                    "sorted run row-id page bounds are stale".into(),
2371                ));
2372            }
2373            row_bounds.push((page_first, page_last));
2374        }
2375        if counted != self.header.row_count
2376            || first.unwrap_or(0) != self.header.min_row_id
2377            || last.unwrap_or(0) != self.header.max_row_id
2378        {
2379            return Err(MongrelError::InvalidArgument(
2380                "sorted run header row bounds differ from its pages".into(),
2381            ));
2382        }
2383        for column in self.dir.clone() {
2384            if !columns.insert(column.column_id) {
2385                return Err(MongrelError::InvalidArgument(format!(
2386                    "sorted run contains duplicate column {}",
2387                    column.column_id
2388                )));
2389            }
2390            let rows = column
2391                .page_stats
2392                .iter()
2393                .map(|stat| stat.row_count)
2394                .collect::<Vec<_>>();
2395            if rows != expected_rows {
2396                return Err(MongrelError::InvalidArgument(
2397                    "sorted run columns have inconsistent page row counts".into(),
2398                ));
2399            }
2400            let ty = self.resolve_type(column.column_id);
2401            if column.type_id_tag != type_tag(&ty)
2402                || column.encoding > Encoding::Zstd as u8
2403                || (!is_system_column_id(column.column_id)
2404                    && self
2405                        .schema
2406                        .columns
2407                        .iter()
2408                        .all(|item| item.id != column.column_id))
2409            {
2410                return Err(MongrelError::InvalidArgument(
2411                    "sorted run column type or encoding is invalid".into(),
2412                ));
2413            }
2414            for (page, stat) in column.page_stats.iter().enumerate() {
2415                let bytes = self.read_page(column.column_id, page)?;
2416                if bytes.len() != stat.uncompressed_len as usize {
2417                    return Err(MongrelError::InvalidArgument(
2418                        "sorted run page length differs from its metadata".into(),
2419                    ));
2420                }
2421                let native =
2422                    match columnar::decode_page_native(ty.clone(), &bytes, stat.row_count as usize)
2423                    {
2424                        Ok(native) => native,
2425                        Err(MongrelError::InvalidArgument(message))
2426                            if message.starts_with("decode_page_native: unsupported ty") =>
2427                        {
2428                            let values =
2429                                columnar::decode_page(ty.clone(), &bytes, stat.row_count as usize)?;
2430                            columnar::values_to_native(ty.clone(), &values)
2431                        }
2432                        Err(error) => return Err(error),
2433                    };
2434                let (first_row_id, last_row_id) =
2435                    row_bounds.get(page).copied().ok_or_else(|| {
2436                        MongrelError::InvalidArgument(
2437                            "sorted run column has more pages than its row-id column".into(),
2438                        )
2439                    })?;
2440                let expected =
2441                    columnar::page_stat_for(ty.clone(), &native, first_row_id, last_row_id);
2442                if stat.first_row_id != expected.first_row_id
2443                    || stat.last_row_id != expected.last_row_id
2444                    || stat.null_count != expected.null_count
2445                    || stat.min != expected.min
2446                    || stat.max != expected.max
2447                {
2448                    return Err(MongrelError::InvalidArgument(
2449                        "sorted run page statistics differ from decoded values".into(),
2450                    ));
2451                }
2452            }
2453        }
2454        if !columns.contains(&SYS_ROW_ID)
2455            || !columns.contains(&SYS_EPOCH)
2456            || !columns.contains(&SYS_DELETED)
2457            || expected_rows.into_iter().map(u64::from).sum::<u64>() != self.header.row_count
2458        {
2459            return Err(MongrelError::InvalidArgument(
2460                "sorted run system columns or row count are invalid".into(),
2461            ));
2462        }
2463        if self.header.schema_id == self.schema.schema_id
2464            && self
2465                .schema
2466                .columns
2467                .iter()
2468                .any(|column| !columns.contains(&column.id))
2469        {
2470            return Err(MongrelError::InvalidArgument(
2471                "sorted run is missing a column from its declared schema".into(),
2472            ));
2473        }
2474
2475        // Page encodings and statistics are not enough to establish semantic
2476        // validity.  Reconstruct every materialized row and apply the schema's
2477        // row-level rules before the run can participate in recovery.
2478        for row_index in 0..usize::try_from(self.header.row_count).map_err(|_| {
2479            MongrelError::InvalidArgument("sorted run row count exceeds this platform".into())
2480        })? {
2481            let epoch = match self.column(SYS_EPOCH)?.get(row_index) {
2482                Some(Value::Int64(value)) if *value >= 0 => *value as u64,
2483                _ => {
2484                    return Err(MongrelError::InvalidArgument(
2485                        "sorted run contains an invalid commit epoch".into(),
2486                    ))
2487                }
2488            };
2489            if epoch > self.header.epoch_created || (self.header.is_uniform_epoch() && epoch != 0) {
2490                return Err(MongrelError::InvalidArgument(
2491                    "sorted run commit epoch exceeds its creation epoch".into(),
2492                ));
2493            }
2494            let deleted = match self.column(SYS_DELETED)?.get(row_index) {
2495                Some(Value::Bool(value)) => *value,
2496                _ => {
2497                    return Err(MongrelError::InvalidArgument(
2498                        "sorted run contains an invalid tombstone marker".into(),
2499                    ))
2500                }
2501            };
2502            let mut values = Vec::with_capacity(self.schema.columns.len());
2503            for column in self.schema.columns.clone() {
2504                let value = if columns.contains(&column.id) {
2505                    self.column(column.id)?
2506                        .get(row_index)
2507                        .cloned()
2508                        .unwrap_or(Value::Null)
2509                } else {
2510                    Value::Null
2511                };
2512                values.push((column.id, value));
2513            }
2514            if !deleted {
2515                self.schema
2516                    .validate_persisted_values(&values)
2517                    .map_err(|error| {
2518                        MongrelError::InvalidArgument(format!(
2519                            "sorted run row violates its schema: {error}"
2520                        ))
2521                    })?;
2522            }
2523        }
2524        Ok(())
2525    }
2526
2527    /// Overlay the real commit epoch for a uniform-epoch run (see
2528    /// [`RUN_FLAG_UNIFORM_EPOCH`]). No-op unless the run carries that flag, so it
2529    /// is always safe for the engine to call with the `RunRef.epoch_created`.
2530    pub(crate) fn set_uniform_epoch(&mut self, epoch: Epoch) {
2531        if self.header.is_uniform_epoch() {
2532            self.epoch_override = Some(epoch);
2533            // Drop any cached placeholder epoch column so the overlay takes hold.
2534            self.col_cache.remove(&SYS_EPOCH);
2535        }
2536    }
2537
2538    pub(crate) fn into_visible_version_cursor(
2539        self,
2540        snapshot: Epoch,
2541    ) -> Result<RunVisibleVersionCursor> {
2542        RunVisibleVersionCursor::new(self, snapshot)
2543    }
2544
2545    /// Whether this run is "clean" (one version per RowId, no tombstones,
2546    /// ascending row_ids) — stamped at write time via [`RUN_FLAG_CLEAN`].
2547    pub fn is_clean(&self) -> bool {
2548        self.header.is_clean()
2549    }
2550
2551    pub fn row_count(&self) -> usize {
2552        self.header.row_count as usize
2553    }
2554
2555    pub(crate) fn clean_contiguous_row_ids(&self) -> bool {
2556        let n = self.row_count();
2557        n > 0
2558            && self.is_clean()
2559            && self.epoch_override.is_none()
2560            && self.header.max_row_id >= self.header.min_row_id
2561            && self
2562                .header
2563                .max_row_id
2564                .checked_sub(self.header.min_row_id)
2565                .and_then(|span| span.checked_add(1))
2566                == Some(n as u64)
2567    }
2568
2569    pub(crate) fn position_for_row_id_fast(&self, row_id: u64) -> Option<usize> {
2570        if !self.clean_contiguous_row_ids()
2571            || row_id < self.header.min_row_id
2572            || row_id > self.header.max_row_id
2573        {
2574            return None;
2575        }
2576        Some((row_id - self.header.min_row_id) as usize)
2577    }
2578
2579    pub(crate) fn positions_for_row_ids_fast(&self, row_ids: &[u64]) -> Option<Vec<usize>> {
2580        if !self.clean_contiguous_row_ids() {
2581            return None;
2582        }
2583        let mut positions = Vec::with_capacity(row_ids.len());
2584        for &row_id in row_ids {
2585            if let Some(pos) = self.position_for_row_id_fast(row_id) {
2586                positions.push(pos);
2587            }
2588        }
2589        positions.sort_unstable();
2590        Some(positions)
2591    }
2592
2593    fn resolve_type(&self, column_id: u16) -> TypeId {
2594        match column_id {
2595            SYS_ROW_ID | SYS_EPOCH => TypeId::Int64,
2596            SYS_DELETED => TypeId::Bool,
2597            SYS_COMMIT_TS => TypeId::Bytes,
2598            _ => self
2599                .schema
2600                .columns
2601                .iter()
2602                .find(|c| c.id == column_id)
2603                .map(|c| c.ty.clone())
2604                .unwrap_or(TypeId::Bytes),
2605        }
2606    }
2607
2608    /// Optional HLC stamp at row `index` (P0.5-T3). Missing column → `None`.
2609    fn commit_ts_at(&mut self, index: usize) -> Result<Option<HlcTimestamp>> {
2610        if !self.has_column(SYS_COMMIT_TS) {
2611            return Ok(None);
2612        }
2613        Ok(decode_commit_ts_value(
2614            self.column(SYS_COMMIT_TS)?.get(index),
2615        ))
2616    }
2617
2618    fn find_header(&self, column_id: u16) -> Result<&ColumnPageHeader> {
2619        self.dir
2620            .iter()
2621            .find(|h| h.column_id == column_id)
2622            .ok_or_else(|| MongrelError::ColumnNotFound(format!("column {column_id}")))
2623    }
2624
2625    /// Whether `column_id`'s pages are encrypted. Encrypted runs carry no
2626    /// cleartext per-page min/max (they would leak plaintext values), so the
2627    /// range resolvers must NOT prune by stats for such columns — a missing
2628    /// stat there means "hidden", not "all-null". They fall back to decrypting
2629    /// and scanning every page.
2630    fn col_encrypted(&self, column_id: u16) -> bool {
2631        self.dir
2632            .iter()
2633            .find(|h| h.column_id == column_id)
2634            .map(|h| h.flags & ColumnPageHeader::PAGE_ENCRYPTED != 0)
2635            .unwrap_or(false)
2636    }
2637
2638    pub(crate) fn read_page(&mut self, column_id: u16, page_seq: usize) -> Result<Vec<u8>> {
2639        let (offset, compressed_len, encrypted) = {
2640            let ch = self.find_header(column_id)?;
2641            let stat = ch
2642                .page_stats
2643                .get(page_seq)
2644                .ok_or_else(|| MongrelError::InvalidArgument("page seq out of range".into()))?;
2645            (
2646                stat.offset,
2647                stat.compressed_len,
2648                ch.flags & ColumnPageHeader::PAGE_ENCRYPTED != 0,
2649            )
2650        };
2651        let end = offset
2652            .checked_add(compressed_len as u64)
2653            .ok_or_else(|| MongrelError::InvalidArgument("run page offset overflows".into()))?;
2654        let start = usize::try_from(offset)
2655            .map_err(|_| MongrelError::InvalidArgument("run page offset is too large".into()))?;
2656        let end = usize::try_from(end)
2657            .map_err(|_| MongrelError::InvalidArgument("run page end is too large".into()))?;
2658        // Shared cache: serve the raw (on-disk / ciphertext) page bytes if
2659        // present, so concurrent readers never re-read or re-decrypt a page.
2660        let key = page_cache_key(self.table_id, self.header.run_id, column_id, page_seq);
2661        if let Some(cache) = &self.page_cache {
2662            if let Some(bytes) = cache.lock(&key).get(
2663                &key,
2664                crate::epoch::Snapshot::at(crate::epoch::Epoch(u64::MAX)),
2665            ) {
2666                return decrypt_or_passthrough(
2667                    self.cipher.as_deref(),
2668                    self.nonce_prefix,
2669                    column_id,
2670                    page_seq,
2671                    encrypted,
2672                    &bytes,
2673                );
2674            }
2675        }
2676        let buf = match &self.mmap {
2677            // Slice the mapping — no seek/read syscalls; the OS page cache fills
2678            // the pages on first touch.
2679            Some(m) => m
2680                .get(start..end)
2681                .ok_or_else(|| {
2682                    MongrelError::InvalidArgument("run page is outside the mapped file".into())
2683                })?
2684                .to_vec(),
2685            None => {
2686                self.file.seek(SeekFrom::Start(offset))?;
2687                let mut buf = vec![0u8; compressed_len as usize];
2688                self.file.read_exact(&mut buf)?;
2689                buf
2690            }
2691        };
2692        // Spill the raw bytes into the shared cache (post-read, pre-decrypt).
2693        if let Some(cache) = &self.page_cache {
2694            cache.lock(&key).insert(crate::page::CachedPage {
2695                committed_epoch: crate::epoch::Epoch(self.header.epoch_created),
2696                content_hash: key,
2697                bytes: bytes::Bytes::copy_from_slice(&buf),
2698            });
2699        }
2700        decrypt_or_passthrough(
2701            self.cipher.as_deref(),
2702            self.nonce_prefix,
2703            column_id,
2704            page_seq,
2705            encrypted,
2706            &buf,
2707        )
2708    }
2709
2710    /// `&self` version of [`Self::read_page`] restricted to the mmap-backed
2711    /// path, so page bytes can be read concurrently (rayon) without the
2712    /// `&mut self` file handle. Decryption (when enabled) uses the `Sync`
2713    /// cipher and a deterministic per-page nonce, so it is also parallel-safe.
2714    fn read_page_shared(&self, column_id: u16, page_seq: usize) -> Result<Vec<u8>> {
2715        let ch = self.find_header(column_id)?;
2716        let stat = ch
2717            .page_stats
2718            .get(page_seq)
2719            .ok_or_else(|| MongrelError::InvalidArgument("page seq out of range".into()))?;
2720        let encrypted = ch.flags & ColumnPageHeader::PAGE_ENCRYPTED != 0;
2721        // Non-blocking probe of the shared cache: never block the rayon pool on
2722        // a contended lock. On a hit, avoid the mmap slice + decrypt entirely.
2723        let key = page_cache_key(self.table_id, self.header.run_id, column_id, page_seq);
2724        if let Some(cache) = &self.page_cache {
2725            if let Some(guard) = cache.try_lock(&key) {
2726                if let Some(bytes) = guard.try_get(
2727                    &key,
2728                    crate::epoch::Snapshot::at(crate::epoch::Epoch(u64::MAX)),
2729                ) {
2730                    return decrypt_or_passthrough(
2731                        self.cipher.as_deref(),
2732                        self.nonce_prefix,
2733                        column_id,
2734                        page_seq,
2735                        encrypted,
2736                        &bytes,
2737                    );
2738                }
2739            }
2740        }
2741        let mmap = self.mmap.as_ref().ok_or_else(|| {
2742            MongrelError::InvalidArgument("parallel page decode requires an mmap-backed run".into())
2743        })?;
2744        let end = stat
2745            .offset
2746            .checked_add(stat.compressed_len as u64)
2747            .ok_or_else(|| MongrelError::InvalidArgument("run page offset overflows".into()))?;
2748        let start = usize::try_from(stat.offset)
2749            .map_err(|_| MongrelError::InvalidArgument("run page offset is too large".into()))?;
2750        let end = usize::try_from(end)
2751            .map_err(|_| MongrelError::InvalidArgument("run page end is too large".into()))?;
2752        let buf = mmap
2753            .get(start..end)
2754            .ok_or_else(|| {
2755                MongrelError::InvalidArgument("run page is outside the mapped file".into())
2756            })?
2757            .to_vec();
2758        // Opportunistic, non-blocking insert: populate the shared cache so later
2759        // readers (and encrypted re-reads) skip the mmap slice + decrypt. Never
2760        // block the rayon pool — if the lock is contended, just skip the insert.
2761        if let Some(cache) = &self.page_cache {
2762            if let Some(mut guard) = cache.try_lock(&key) {
2763                guard.insert(crate::page::CachedPage {
2764                    committed_epoch: crate::epoch::Epoch(self.header.epoch_created),
2765                    content_hash: key,
2766                    bytes: bytes::Bytes::copy_from_slice(&buf),
2767                });
2768            }
2769        }
2770        decrypt_or_passthrough(
2771            self.cipher.as_deref(),
2772            self.nonce_prefix,
2773            column_id,
2774            page_seq,
2775            encrypted,
2776            &buf,
2777        )
2778    }
2779
2780    /// Decode (and cache) a full column, concatenating all pages.
2781    fn column(&mut self, column_id: u16) -> Result<&[Value]> {
2782        // Uniform-epoch overlay: serve the `_epoch` column as a constant of the
2783        // real commit epoch instead of the placeholder stored on disk.
2784        if column_id == SYS_EPOCH {
2785            if let Some(ov) = self.epoch_override {
2786                if !self.col_cache.contains_key(&column_id) {
2787                    let n = self.row_count();
2788                    self.col_cache
2789                        .insert(column_id, vec![Value::Int64(ov.0 as i64); n]);
2790                }
2791                return Ok(self.col_cache.get(&column_id).unwrap().as_slice());
2792            }
2793        }
2794        if !self.col_cache.contains_key(&column_id) {
2795            let ty = self.resolve_type(column_id);
2796            let page_rows: Vec<usize> = {
2797                let ch = self.find_header(column_id)?;
2798                ch.page_stats.iter().map(|s| s.row_count as usize).collect()
2799            };
2800            let mut decoded: Vec<Value> = Vec::with_capacity(self.row_count());
2801            for (seq, &pr) in page_rows.iter().enumerate() {
2802                let page = self.read_page(column_id, seq)?;
2803                decoded.extend(columnar::decode_page(ty.clone(), &page, pr)?);
2804            }
2805            self.col_cache.insert(column_id, decoded);
2806        }
2807        Ok(self.col_cache.get(&column_id).unwrap().as_slice())
2808    }
2809
2810    /// Newest version of `row_id` with `epoch <= snapshot`, including tombstones
2811    /// (returned as a `Row` with `deleted=true`). `None` if no such version.
2812    ///
2813    /// Page-pruned: `SYS_ROW_ID` pages carry exact `first_row_id`/`last_row_id`
2814    /// bounds (rows are written in ascending `(RowId, Epoch)` order), so this
2815    /// decodes only the page(s) that can contain `row_id` instead of the whole
2816    /// column — the old implementation decoded every row's `SYS_ROW_ID` (and
2817    /// `SYS_EPOCH`) up front, making every single-row point lookup (the common
2818    /// case for a PK/unique check feeding insert/update/delete) pay a
2819    /// full-column decode. A row's version group can span at most two adjacent
2820    /// pages (split at a page boundary mid-group); `candidate_pages` below
2821    /// collects every page whose bounds include `row_id`, which is normally
2822    /// one page and two only in that split case.
2823    pub fn get_version(&mut self, row_id: RowId, snapshot: Epoch) -> Result<Option<(Epoch, Row)>> {
2824        match self.find_version_page(row_id, snapshot)? {
2825            None => Ok(None),
2826            Some((epoch, seq, local_index)) => Ok(Some((
2827                Epoch(epoch),
2828                self.materialize_in_page(seq, local_index)?,
2829            ))),
2830        }
2831    }
2832
2833    /// Page-pruned search for the newest version of `row_id` with `epoch <=
2834    /// snapshot`: `(epoch, page_seq, local_index)`, or `None` if no such
2835    /// version exists in this run. Factored out of [`Self::get_version`] so
2836    /// [`Self::get_version_column`] can reuse the exact same page-finding
2837    /// logic without re-deriving it.
2838    fn find_version_page(
2839        &mut self,
2840        row_id: RowId,
2841        snapshot: Epoch,
2842    ) -> Result<Option<(u64, usize, usize)>> {
2843        let n = self.row_count();
2844        if n == 0 {
2845            return Ok(None);
2846        }
2847        let target = row_id.0 as i64;
2848        let ch = self.find_header(SYS_ROW_ID)?;
2849        let mut page_start = 0u64;
2850        let candidate_pages: Vec<(usize, u64)> = ch // (page_seq, row offset of page start)
2851            .page_stats
2852            .iter()
2853            .enumerate()
2854            .filter_map(|(seq, s)| {
2855                let start = page_start;
2856                page_start += s.row_count as u64;
2857                (s.first_row_id <= row_id.0 && row_id.0 <= s.last_row_id).then_some((seq, start))
2858            })
2859            .collect();
2860        if candidate_pages.is_empty() {
2861            return Ok(None);
2862        }
2863        let ty = self.resolve_type(SYS_ROW_ID);
2864        let mut best: Option<(u64, usize, usize)> = None; // (epoch, page_seq, local index)
2865        for (seq, _page_row_start) in candidate_pages {
2866            let page_rows = self.find_header(SYS_ROW_ID)?.page_stats[seq].row_count as usize;
2867            let row_ids =
2868                match self.decode_page_native_cached(ty.clone(), SYS_ROW_ID, seq, page_rows)? {
2869                    columnar::NativeColumn::Int64 { data, .. } => data,
2870                    _ => return Err(MongrelError::InvalidArgument("sys row_id not int64".into())),
2871                };
2872            let local = match row_ids.binary_search(&target) {
2873                Ok(i) => i,
2874                Err(_) => continue,
2875            };
2876            let epoch_ty = self.resolve_type(SYS_EPOCH);
2877            let epochs =
2878                match self.decode_page_native_cached(epoch_ty, SYS_EPOCH, seq, page_rows)? {
2879                    columnar::NativeColumn::Int64 { data, .. } => data,
2880                    _ => return Err(MongrelError::InvalidArgument("sys epoch not int64".into())),
2881                };
2882            // `local` is one match; the row-id's full version group is the
2883            // contiguous run of equal row-ids around it within this page.
2884            let mut lo = local;
2885            while lo > 0 && row_ids[lo - 1] == target {
2886                lo -= 1;
2887            }
2888            let mut hi = local;
2889            while hi + 1 < row_ids.len() && row_ids[hi + 1] == target {
2890                hi += 1;
2891            }
2892            for (i, &epoch) in epochs[lo..=hi].iter().enumerate() {
2893                let epoch = epoch as u64;
2894                if epoch <= snapshot.0 && best.map(|(be, ..)| epoch > be).unwrap_or(true) {
2895                    best = Some((epoch, seq, lo + i));
2896                }
2897            }
2898        }
2899        Ok(best)
2900    }
2901
2902    /// Like [`Self::get_version`], but decodes only `column_id` (plus the
2903    /// `SYS_DELETED` flag) instead of materializing every schema column via
2904    /// [`Self::materialize_in_page`]. For a wide schema this avoids paying to
2905    /// decode every other column's page just to read one value and throw the
2906    /// rest away — e.g. `Table::remove_hot_for_row`'s PK-only lookup, which
2907    /// used to pull the whole row (every column, every page) just for the
2908    /// primary key.
2909    pub fn get_version_column(
2910        &mut self,
2911        row_id: RowId,
2912        snapshot: Epoch,
2913        column_id: u16,
2914    ) -> Result<Option<(Epoch, bool, Option<Value>)>> {
2915        let Some((epoch, seq, local_index)) = self.find_version_page(row_id, snapshot)? else {
2916            return Ok(None);
2917        };
2918        let page_rows = self.find_header(SYS_ROW_ID)?.page_stats[seq].row_count as usize;
2919        let page_start: usize = self.find_header(SYS_ROW_ID)?.page_stats[..seq]
2920            .iter()
2921            .map(|s| s.row_count as usize)
2922            .sum();
2923        let global_index = page_start + local_index;
2924        let native_at = |slf: &mut Self, cid: u16| -> Result<Option<Value>> {
2925            if !slf.dir.iter().any(|h| h.column_id == cid) {
2926                return Ok(None);
2927            }
2928            let ty = slf.resolve_type(cid);
2929            if !matches!(
2930                ty,
2931                TypeId::Bool
2932                    | TypeId::Int8
2933                    | TypeId::Int16
2934                    | TypeId::Int32
2935                    | TypeId::Int64
2936                    | TypeId::UInt8
2937                    | TypeId::UInt16
2938                    | TypeId::UInt32
2939                    | TypeId::UInt64
2940                    | TypeId::Float32
2941                    | TypeId::Float64
2942                    | TypeId::TimestampNanos
2943                    | TypeId::Date32
2944                    | TypeId::Bytes
2945            ) {
2946                return Ok(slf.column(cid)?.get(global_index).cloned());
2947            }
2948            Ok(slf
2949                .decode_page_native_cached(ty, cid, seq, page_rows)?
2950                .value_at(local_index))
2951        };
2952        let deleted = matches!(native_at(self, SYS_DELETED)?, Some(Value::Bool(true)));
2953        let value = native_at(self, column_id)?;
2954        Ok(Some((Epoch(epoch), deleted, value)))
2955    }
2956
2957    /// Newest version epoch and tombstone flag without decoding a user column.
2958    pub fn get_version_visibility(
2959        &mut self,
2960        row_id: RowId,
2961        snapshot: Epoch,
2962    ) -> Result<Option<(Epoch, bool)>> {
2963        let Some((epoch, seq, local_index)) = self.find_version_page(row_id, snapshot)? else {
2964            return Ok(None);
2965        };
2966        let page_rows = self.find_header(SYS_ROW_ID)?.page_stats[seq].row_count as usize;
2967        let deleted = match self.decode_page_native_cached(
2968            self.resolve_type(SYS_DELETED),
2969            SYS_DELETED,
2970            seq,
2971            page_rows,
2972        )? {
2973            columnar::NativeColumn::Bool { data, .. } => data[local_index] != 0,
2974            _ => return Err(MongrelError::InvalidArgument("sys deleted not bool".into())),
2975        };
2976        Ok(Some((Epoch(epoch), deleted)))
2977    }
2978
2979    /// Build a `Row` from page `seq`'s data at `local_index`, decoding only
2980    /// that one page per column instead of [`Self::materialize`]'s whole-column
2981    /// `Vec<Value>` decode — used by [`Self::get_version`], which already knows
2982    /// the exact page from its page-pruned search. Only for scalar types with a
2983    /// [`columnar::NativeColumn`] representation; any other column (e.g. a
2984    /// fixed-size `Embedding`, which `NativeColumn` has no variant for) falls
2985    /// back to the whole-column path for that column specifically, so this is
2986    /// never wrong, only sometimes not faster.
2987    fn materialize_in_page(&mut self, seq: usize, local_index: usize) -> Result<Row> {
2988        let page_rows = self.find_header(SYS_ROW_ID)?.page_stats[seq].row_count as usize;
2989        let page_start: usize = self.find_header(SYS_ROW_ID)?.page_stats[..seq]
2990            .iter()
2991            .map(|s| s.row_count as usize)
2992            .sum();
2993        let global_index = page_start + local_index;
2994        let native_at = |slf: &mut Self, column_id: u16| -> Result<Option<Value>> {
2995            // Absent column (schema evolution: added after this run was
2996            // written) reads as null, matching `materialize`'s own guard.
2997            if !slf.dir.iter().any(|h| h.column_id == column_id) {
2998                return Ok(None);
2999            }
3000            let ty = slf.resolve_type(column_id);
3001            if !matches!(
3002                ty,
3003                TypeId::Bool
3004                    | TypeId::Int8
3005                    | TypeId::Int16
3006                    | TypeId::Int32
3007                    | TypeId::Int64
3008                    | TypeId::UInt8
3009                    | TypeId::UInt16
3010                    | TypeId::UInt32
3011                    | TypeId::UInt64
3012                    | TypeId::Float32
3013                    | TypeId::Float64
3014                    | TypeId::TimestampNanos
3015                    | TypeId::Date32
3016                    | TypeId::Bytes
3017            ) {
3018                // Not a NativeColumn-representable scalar (e.g. Embedding) —
3019                // fall back to the always-correct whole-column decode.
3020                return Ok(slf.column(column_id)?.get(global_index).cloned());
3021            }
3022            Ok(slf
3023                .decode_page_native_cached(ty, column_id, seq, page_rows)?
3024                .value_at(local_index))
3025        };
3026        let row_id = RowId(match native_at(self, SYS_ROW_ID)? {
3027            Some(Value::Int64(x)) => x as u64,
3028            _ => 0,
3029        });
3030        let committed_epoch = Epoch(match native_at(self, SYS_EPOCH)? {
3031            Some(Value::Int64(x)) => x as u64,
3032            _ => 0,
3033        });
3034        let deleted = matches!(native_at(self, SYS_DELETED)?, Some(Value::Bool(true)));
3035        let commit_ts = if self.has_column(SYS_COMMIT_TS) {
3036            decode_commit_ts_value(native_at(self, SYS_COMMIT_TS)?.as_ref())
3037        } else {
3038            None
3039        };
3040        let col_ids: Vec<u16> = self.schema.columns.iter().map(|c| c.id).collect();
3041        let mut columns = HashMap::new();
3042        for id in col_ids {
3043            columns.insert(id, native_at(self, id)?.unwrap_or(Value::Null));
3044        }
3045        Ok(Row {
3046            row_id,
3047            committed_epoch,
3048            columns,
3049            deleted,
3050            commit_ts,
3051        })
3052    }
3053
3054    /// Every row in the run (all versions), in `(RowId, Epoch)` order. Used by
3055    /// compaction, which must see every version to apply snapshot retention.
3056    pub fn all_rows(&mut self) -> Result<Vec<Row>> {
3057        let n = self.row_count();
3058        let mut out = Vec::with_capacity(n);
3059        for i in 0..n {
3060            out.push(self.materialize(i)?);
3061        }
3062        Ok(out)
3063    }
3064
3065    /// Every version with cooperative cancellation and a hard row bound.
3066    pub fn all_rows_controlled(
3067        &mut self,
3068        control: &crate::ExecutionControl,
3069        max_rows: usize,
3070    ) -> Result<Vec<Row>> {
3071        let n = self.row_count();
3072        if n > max_rows {
3073            return Err(MongrelError::ResourceLimitExceeded {
3074                resource: "controlled run row materialization",
3075                requested: n,
3076                limit: max_rows,
3077            });
3078        }
3079        let mut out = Vec::with_capacity(n);
3080        for i in 0..n {
3081            if i % 256 == 0 {
3082                control.checkpoint()?;
3083            }
3084            out.push(self.materialize(i)?);
3085        }
3086        control.checkpoint()?;
3087        Ok(out)
3088    }
3089
3090    /// Indices of the newest non-deleted version per `RowId` visible at
3091    /// `snapshot`, ascending. This is the columnar scan primitive: compute the
3092    /// visible set once (one pass over the row-id/epoch/deleted columns), then
3093    /// [`Self::gather_column`] each user column at these indices — no per-row
3094    /// `HashMap`/`Row` materialization.
3095    pub fn visible_indices(&mut self, snapshot: Epoch) -> Result<Vec<usize>> {
3096        let n = self.row_count();
3097        if n == 0 {
3098            return Ok(Vec::new());
3099        }
3100        let row_ids = self.column(SYS_ROW_ID)?.to_vec();
3101        let epochs = self.column(SYS_EPOCH)?.to_vec();
3102        let deleted = self.column(SYS_DELETED)?.to_vec();
3103        let mut best: HashMap<u64, (u64, usize)> = HashMap::new();
3104        for i in 0..n {
3105            let rid = int_at(&row_ids, i);
3106            let e = int_at(&epochs, i);
3107            if e > snapshot.0 {
3108                continue;
3109            }
3110            best.entry(rid)
3111                .and_modify(|(be, bi)| {
3112                    if e > *be {
3113                        *be = e;
3114                        *bi = i;
3115                    }
3116                })
3117                .or_insert((e, i));
3118        }
3119        let mut idxs: Vec<usize> = best.into_values().map(|(_, i)| i).collect();
3120        idxs.retain(|&i| !bool_at(&deleted, i));
3121        idxs.sort_unstable();
3122        Ok(idxs)
3123    }
3124
3125    /// Gather `column_id`'s values at the given indices (column cached). Used
3126    /// with [`Self::visible_indices`] for vectorized scans. A column absent from
3127    /// this run (e.g. added via schema evolution after the run was written)
3128    /// yields all-nulls.
3129    pub fn gather_column(&mut self, column_id: u16, indices: &[usize]) -> Result<Vec<Value>> {
3130        if !self.dir.iter().any(|h| h.column_id == column_id) {
3131            return Ok(vec![Value::Null; indices.len()]);
3132        }
3133        let col = self.column(column_id)?;
3134        Ok(indices
3135            .iter()
3136            .map(|&i| col.get(i).cloned().unwrap_or(Value::Null))
3137            .collect())
3138    }
3139
3140    /// Decode a column straight to a typed [`NativeColumn`] (no `Value`),
3141    /// concatenating all pages. A column absent from this run (schema evolution)
3142    /// yields an all-null column. Pages are decoded in parallel (rayon) when the
3143    /// run is mmap-backed and has more than one page; otherwise sequentially.
3144    pub fn column_native(&mut self, column_id: u16) -> Result<columnar::NativeColumn> {
3145        use rayon::prelude::*;
3146        if column_id == SYS_EPOCH {
3147            if let Some(ov) = self.epoch_override {
3148                return Ok(columnar::NativeColumn::int64_constant(
3149                    ov.0 as i64,
3150                    self.row_count(),
3151                ));
3152            }
3153        }
3154        let ty = self.resolve_type(column_id);
3155        let n = self.row_count();
3156        let Some(ch) = self.dir.iter().find(|h| h.column_id == column_id) else {
3157            return Ok(columnar::null_native(ty, n));
3158        };
3159        let page_count = ch.page_count as usize;
3160        let page_rows: Vec<usize> = ch.page_stats.iter().map(|s| s.row_count as usize).collect();
3161        if page_count == 0 {
3162            return Ok(columnar::null_native(ty, n));
3163        }
3164        let parts: Vec<columnar::NativeColumn> = if self.mmap.is_some() && page_count > 1 {
3165            // Parallel decode: each page is an independent region of the shared
3166            // mmap; the cipher (if any) is Sync. Spread the (CPU-bound) decode
3167            // across the rayon thread pool.
3168            let reader: &RunReader = self;
3169            (0..page_count)
3170                .into_par_iter()
3171                .map(|seq| {
3172                    let raw = reader.read_page_shared(column_id, seq)?;
3173                    columnar::decode_page_native(ty.clone(), &raw, page_rows[seq])
3174                })
3175                .collect::<Result<Vec<_>>>()?
3176        } else {
3177            let mut out = Vec::with_capacity(page_count);
3178            for (seq, &pr) in page_rows.iter().enumerate() {
3179                let page = self.read_page(column_id, seq)?;
3180                out.push(columnar::decode_page_native(ty.clone(), &page, pr)?);
3181            }
3182            out
3183        };
3184        Ok(columnar::NativeColumn::concat(&parts))
3185    }
3186
3187    /// Whether this reader is backed by a memory map (the prerequisite for the
3188    /// `&self` parallel decode paths). Readers on filesystems that reject mmap
3189    /// fall back to per-page `read()` and `has_mmap()` is false.
3190    pub fn has_mmap(&self) -> bool {
3191        self.mmap.is_some()
3192    }
3193
3194    /// `&self` variant of [`column_native`] for cross-column parallel scans
3195    /// (Phase 15.1). Requires the mmap backing (uses [`read_page_shared`], which
3196    /// is rayon-safe); callers without mmap use the `&mut` [`column_native`].
3197    /// Pages within the column decode in parallel when there is more than one,
3198    /// and `MADV_WILLNEED` is hinted up front so the kernel pre-faults the whole
3199    /// column's byte range (Phase 15.2) before the decode workers touch it.
3200    pub fn column_native_shared(&self, column_id: u16) -> Result<columnar::NativeColumn> {
3201        use rayon::prelude::*;
3202        if column_id == SYS_EPOCH {
3203            if let Some(ov) = self.epoch_override {
3204                return Ok(columnar::NativeColumn::int64_constant(
3205                    ov.0 as i64,
3206                    self.row_count(),
3207                ));
3208            }
3209        }
3210        let ty = self.resolve_type(column_id);
3211        let n = self.row_count();
3212        let Some(ch) = self.dir.iter().find(|h| h.column_id == column_id) else {
3213            return Ok(columnar::null_native(ty, n));
3214        };
3215        let page_count = ch.page_count as usize;
3216        let page_rows: Vec<usize> = ch.page_stats.iter().map(|s| s.row_count as usize).collect();
3217        if page_count == 0 {
3218            return Ok(columnar::null_native(ty, n));
3219        }
3220        // Phase 15.2: best-effort read-ahead. Tell the kernel to page-in the
3221        // column's full byte range before the workers fan out, overlapping the
3222        // disk I/O with the upcoming decode CPU.
3223        #[cfg(unix)]
3224        {
3225            if let (Some(m), Some(first)) = (&self.mmap, ch.page_stats.first()) {
3226                let start = first.offset as usize;
3227                let end = (ch.page_region_offset as usize) + (ch.page_region_len as usize);
3228                if end > start {
3229                    let _ = m.advise_range(memmap2::Advice::WillNeed, start, end - start);
3230                }
3231            }
3232        }
3233        let run_id = self.header.run_id;
3234        // Decode in parallel (cache probes use `try_lock` → no worker blocking).
3235        // Each item is the decoded page plus its key when it was a cache miss
3236        // (hits return `None` for the key so we don't re-insert/clone them).
3237        let mut parts_keys: Vec<(columnar::NativeColumn, Option<[u8; 32]>)> = if page_count > 1 {
3238            (0..page_count)
3239                .into_par_iter()
3240                .map(|seq| {
3241                    self.decode_page_cached(ty.clone(), column_id, seq, page_rows[seq], run_id)
3242                })
3243                .collect::<Result<Vec<_>>>()?
3244        } else {
3245            vec![self.decode_page_cached(ty, column_id, 0, page_rows[0], run_id)?]
3246        };
3247        // Sequentially cache the freshly-decoded pages — no parallel contention
3248        // on the insert, so every miss is reliably stored for the next scan.
3249        if let Some(cache) = &self.decoded_cache {
3250            for (col, key) in parts_keys.iter_mut() {
3251                if let Some(k) = key.take() {
3252                    cache.lock(&k).insert(k, Arc::new(col.clone()));
3253                }
3254            }
3255        }
3256        let parts: Vec<columnar::NativeColumn> = parts_keys.into_iter().map(|(c, _)| c).collect();
3257        Ok(columnar::NativeColumn::concat(&parts))
3258    }
3259
3260    /// Decode one page for the shared scan path, consulting the decoded-page
3261    /// cache first (Phase 15.4). Returns the decoded page plus `Some(key)` on a
3262    /// cache miss (so the caller can insert it) or `None` on a hit (already
3263    /// cached). Cache probes use `try_lock` so rayon workers never block.
3264    fn decode_page_cached(
3265        &self,
3266        ty: TypeId,
3267        column_id: u16,
3268        seq: usize,
3269        nrows: usize,
3270        run_id: u128,
3271    ) -> Result<(columnar::NativeColumn, Option<[u8; 32]>)> {
3272        let key = page_cache_key(self.table_id, run_id, column_id, seq);
3273        if let Some(cache) = &self.decoded_cache {
3274            if let Some(g) = cache.try_lock(&key) {
3275                if let Some(hit) = g.try_get(&key) {
3276                    return Ok(((*hit).clone(), None));
3277                }
3278            }
3279        }
3280        let raw = self.read_page_shared(column_id, seq)?;
3281        let col = columnar::decode_page_native(ty, &raw, nrows)?;
3282        Ok((col, Some(key)))
3283    }
3284
3285    /// [`Self::decode_page_cached`], but for the sequential point-lookup paths
3286    /// (`find_version_page`, `materialize_in_page`, `get_version_column`) via
3287    /// [`Self::read_page`] (handles the non-mmap-backed fallback) instead of
3288    /// [`Self::read_page_shared`] (mmap-only, for the parallel rayon scan
3289    /// path). No rayon pool to avoid blocking here, so a plain `.lock()` is
3290    /// fine — unlike the scan path's `try_lock`, this always consults the
3291    /// cache rather than skipping it under contention. Without this, every
3292    /// point lookup re-decompressed its page from scratch even when a prior
3293    /// lookup had just decoded the exact same page (the dominant remaining
3294    /// cost measured for `remove_hot_for_row`'s on-disk path).
3295    fn decode_page_native_cached(
3296        &mut self,
3297        ty: TypeId,
3298        column_id: u16,
3299        seq: usize,
3300        nrows: usize,
3301    ) -> Result<columnar::NativeColumn> {
3302        let key = page_cache_key(self.table_id, self.header.run_id, column_id, seq);
3303        if let Some(cache) = &self.decoded_cache {
3304            if let Some(hit) = cache.lock(&key).try_get(&key) {
3305                return Ok((*hit).clone());
3306            }
3307        }
3308        let raw = self.read_page(column_id, seq)?;
3309        let col = columnar::decode_page_native(ty, &raw, nrows)?;
3310        if let Some(cache) = &self.decoded_cache {
3311            cache
3312                .lock(&key)
3313                .insert(key, std::sync::Arc::new(col.clone()));
3314        }
3315        Ok(col)
3316    }
3317
3318    /// Row ids whose Int64 value is in `[lo, hi]`, **skipping pages whose
3319    /// `[min,max]` stat excludes the range** (Parquet-style page-index pruning).
3320    /// Nulls are excluded. Used by `Table::query_columns_native` to serve
3321    /// `Condition::Range` without decoding every page.
3322    pub fn range_row_ids_i64(
3323        &mut self,
3324        column_id: u16,
3325        lo: i64,
3326        hi: i64,
3327    ) -> Result<std::collections::HashSet<u64>> {
3328        Ok(self
3329            .range_row_id_set_i64(column_id, lo, hi)?
3330            .into_sorted_vec()
3331            .into_iter()
3332            .collect())
3333    }
3334
3335    pub(crate) fn range_row_id_set_i64(
3336        &mut self,
3337        column_id: u16,
3338        lo: i64,
3339        hi: i64,
3340    ) -> Result<RowIdSet> {
3341        let info: Vec<(Option<i64>, Option<i64>, usize)> =
3342            match self.dir.iter().find(|h| h.column_id == column_id) {
3343                Some(ch) => ch
3344                    .page_stats
3345                    .iter()
3346                    .map(|s| {
3347                        (
3348                            be_i64(s.min.as_deref()),
3349                            be_i64(s.max.as_deref()),
3350                            s.row_count as usize,
3351                        )
3352                    })
3353                    .collect(),
3354                None => return Ok(RowIdSet::empty()),
3355            };
3356        // Encrypted columns are pruneable only when this run carries the
3357        // decrypted stats envelope (overlaid at open). Without one (a run
3358        // whose writer recorded no stats at all), a missing min/max means
3359        // "unknown" — never prune — whereas with the envelope (and always for
3360        // plaintext runs) a missing min/max means an all-null page.
3361        let stats_pruneable =
3362            !self.col_encrypted(column_id) || self.header.encrypted_stats_offset != 0;
3363        let clean_contiguous = self.clean_contiguous_row_ids();
3364        let mut out = Vec::new();
3365        let mut page_start = 0usize;
3366        for (seq, (mn, mx, nrows)) in info.into_iter().enumerate() {
3367            let current_page_start = page_start;
3368            page_start += nrows;
3369            // Skip pages that cannot contain a match (or are all-null).
3370            let skip = stats_pruneable
3371                && match (mn, mx) {
3372                    (Some(mn), Some(mx)) => mx < lo || mn > hi,
3373                    _ => true,
3374                };
3375            if skip {
3376                continue;
3377            }
3378            let val_page = self.read_page(column_id, seq)?;
3379            let vals =
3380                columnar::decode_page_native(self.resolve_type(column_id), &val_page, nrows)?;
3381            if let columnar::NativeColumn::Int64 { data: v, validity } = vals {
3382                if clean_contiguous {
3383                    for (i, val) in v.iter().enumerate() {
3384                        if columnar::validity_bit(&validity, i) && *val >= lo && *val <= hi {
3385                            out.push(self.header.min_row_id + current_page_start as u64 + i as u64);
3386                        }
3387                    }
3388                } else {
3389                    let rid_page = self.read_page(SYS_ROW_ID, seq)?;
3390                    let rids = columnar::decode_page_native(TypeId::Int64, &rid_page, nrows)?;
3391                    if let columnar::NativeColumn::Int64 { data: r, .. } = rids {
3392                        for (i, val) in v.iter().enumerate() {
3393                            if columnar::validity_bit(&validity, i) && *val >= lo && *val <= hi {
3394                                out.push(r[i] as u64);
3395                            }
3396                        }
3397                    }
3398                }
3399            }
3400        }
3401        Ok(RowIdSet::from_unsorted(out))
3402    }
3403
3404    /// Float64 analogue of [`Self::range_row_ids_i64`] with per-bound
3405    /// inclusivity, for `Condition::RangeF64`.
3406    pub fn range_row_ids_f64(
3407        &mut self,
3408        column_id: u16,
3409        lo: f64,
3410        lo_inclusive: bool,
3411        hi: f64,
3412        hi_inclusive: bool,
3413    ) -> Result<std::collections::HashSet<u64>> {
3414        Ok(self
3415            .range_row_id_set_f64(column_id, lo, lo_inclusive, hi, hi_inclusive)?
3416            .into_sorted_vec()
3417            .into_iter()
3418            .collect())
3419    }
3420
3421    pub(crate) fn range_row_id_set_f64(
3422        &mut self,
3423        column_id: u16,
3424        lo: f64,
3425        lo_inclusive: bool,
3426        hi: f64,
3427        hi_inclusive: bool,
3428    ) -> Result<RowIdSet> {
3429        let info: Vec<(Option<f64>, Option<f64>, usize)> =
3430            match self.dir.iter().find(|h| h.column_id == column_id) {
3431                Some(ch) => ch
3432                    .page_stats
3433                    .iter()
3434                    .map(|s| {
3435                        (
3436                            be_f64(s.min.as_deref()),
3437                            be_f64(s.max.as_deref()),
3438                            s.row_count as usize,
3439                        )
3440                    })
3441                    .collect(),
3442                None => return Ok(RowIdSet::empty()),
3443            };
3444        // Encrypted columns are pruneable only when this run carries the
3445        // decrypted stats envelope (overlaid at open). Without one (a run
3446        // whose writer recorded no stats at all), a missing min/max means
3447        // "unknown" — never prune — whereas with the envelope (and always for
3448        // plaintext runs) a missing min/max means an all-null page.
3449        let stats_pruneable =
3450            !self.col_encrypted(column_id) || self.header.encrypted_stats_offset != 0;
3451        let clean_contiguous = self.clean_contiguous_row_ids();
3452        let mut out = Vec::new();
3453        let mut page_start = 0usize;
3454        for (seq, (mn, mx, nrows)) in info.into_iter().enumerate() {
3455            let current_page_start = page_start;
3456            page_start += nrows;
3457            // A page can be dropped iff every value fails the predicate, i.e. the
3458            // largest fails the lo-test or the smallest fails the hi-test.
3459            let skip = stats_pruneable
3460                && match (mn, mx) {
3461                    (Some(mn), Some(mx)) => {
3462                        let skip_lo = mx < lo || (!lo_inclusive && mx == lo);
3463                        let skip_hi = mn > hi || (!hi_inclusive && mn == hi);
3464                        skip_lo || skip_hi
3465                    }
3466                    _ => true,
3467                };
3468            if skip {
3469                continue;
3470            }
3471            let val_page = self.read_page(column_id, seq)?;
3472            let vals = columnar::decode_page_native(TypeId::Float64, &val_page, nrows)?;
3473            if let columnar::NativeColumn::Float64 { data: v, validity } = vals {
3474                if clean_contiguous {
3475                    for (i, val) in v.iter().enumerate() {
3476                        if !columnar::validity_bit(&validity, i) || val.is_nan() {
3477                            continue;
3478                        }
3479                        let ok_lo = if lo_inclusive { *val >= lo } else { *val > lo };
3480                        let ok_hi = if hi_inclusive { *val <= hi } else { *val < hi };
3481                        if ok_lo && ok_hi {
3482                            out.push(self.header.min_row_id + current_page_start as u64 + i as u64);
3483                        }
3484                    }
3485                } else {
3486                    let rid_page = self.read_page(SYS_ROW_ID, seq)?;
3487                    let rids = columnar::decode_page_native(TypeId::Int64, &rid_page, nrows)?;
3488                    if let columnar::NativeColumn::Int64 { data: r, .. } = rids {
3489                        for (i, val) in v.iter().enumerate() {
3490                            if !columnar::validity_bit(&validity, i) || val.is_nan() {
3491                                continue;
3492                            }
3493                            let ok_lo = if lo_inclusive { *val >= lo } else { *val > lo };
3494                            let ok_hi = if hi_inclusive { *val <= hi } else { *val < hi };
3495                            if ok_lo && ok_hi {
3496                                out.push(r[i] as u64);
3497                            }
3498                        }
3499                    }
3500                }
3501            }
3502        }
3503        Ok(RowIdSet::from_unsorted(out))
3504    }
3505
3506    /// Page-pruned row-id set for `IS NULL` / `IS NOT NULL` on `column_id`.
3507    /// Skips pages whose `null_count` makes a match impossible (no nulls for
3508    /// `IS NULL`, all-nulls for `IS NOT NULL`), then decodes the validity bitmap
3509    /// of surviving pages to pinpoint matching rows.
3510    pub(crate) fn null_row_id_set(&mut self, column_id: u16, want_nulls: bool) -> Result<RowIdSet> {
3511        let stats: Vec<(usize, usize)> = match self.dir.iter().find(|h| h.column_id == column_id) {
3512            Some(ch) => ch
3513                .page_stats
3514                .iter()
3515                .map(|s| (s.null_count as usize, s.row_count as usize))
3516                .collect(),
3517            None => return Ok(RowIdSet::empty()),
3518        };
3519        let ty = self.resolve_type(column_id);
3520        let clean_contiguous = self.clean_contiguous_row_ids();
3521        let mut out = Vec::new();
3522        let mut page_start = 0usize;
3523        for (seq, (null_count, nrows)) in stats.into_iter().enumerate() {
3524            let current_page_start = page_start;
3525            page_start += nrows;
3526            // Skip pages that cannot match.
3527            if want_nulls && null_count == 0 {
3528                continue;
3529            }
3530            if !want_nulls && null_count == nrows {
3531                continue;
3532            }
3533            let val_page = self.read_page(column_id, seq)?;
3534            let col = columnar::decode_page_native(ty.clone(), &val_page, nrows)?;
3535            let validity = col.validity();
3536            if clean_contiguous {
3537                for i in 0..nrows {
3538                    let is_null = !columnar::validity_bit(validity, i);
3539                    if is_null == want_nulls {
3540                        out.push(self.header.min_row_id + current_page_start as u64 + i as u64);
3541                    }
3542                }
3543            } else {
3544                let rid_page = self.read_page(SYS_ROW_ID, seq)?;
3545                let rids = columnar::decode_page_native(TypeId::Int64, &rid_page, nrows)?;
3546                if let columnar::NativeColumn::Int64 { data: r, .. } = rids {
3547                    for (i, &rid) in r.iter().enumerate().take(nrows) {
3548                        let is_null = !columnar::validity_bit(validity, i);
3549                        if is_null == want_nulls {
3550                            out.push(rid as u64);
3551                        }
3552                    }
3553                }
3554            }
3555        }
3556        Ok(RowIdSet::from_unsorted(out))
3557    }
3558
3559    pub fn visible_indices_native(&mut self, snapshot: Epoch) -> Result<Vec<usize>> {
3560        let n = self.row_count();
3561        if n == 0 {
3562            return Ok(Vec::new());
3563        }
3564        let (row_ids, epochs, deleted) = self.system_columns_native()?;
3565        let mut best: HashMap<u64, (u64, usize)> = HashMap::new();
3566        for i in 0..n {
3567            let rid = row_ids[i] as u64;
3568            let e = epochs[i] as u64;
3569            if e > snapshot.0 {
3570                continue;
3571            }
3572            best.entry(rid)
3573                .and_modify(|(be, bi)| {
3574                    if e > *be {
3575                        *be = e;
3576                        *bi = i;
3577                    }
3578                })
3579                .or_insert((e, i));
3580        }
3581        let mut idxs: Vec<usize> = best.into_values().map(|(_, i)| i).collect();
3582        idxs.retain(|&i| deleted[i] == 0);
3583        idxs.sort_unstable();
3584        Ok(idxs)
3585    }
3586
3587    /// Page-pruned, **MVCC-visible** Int64 range resolution (Phase 16.3).
3588    ///
3589    /// Like [`Self::range_row_ids_i64`] (skips pages whose `[min,max]` excludes
3590    /// `[lo, hi]`) but restricts the output to the newest non-deleted version per
3591    /// `RowId` visible at `snapshot`. This is the layout-independent range
3592    /// primitive: correct under any memtable / multi-run / deletion-vector state,
3593    /// so the engine no longer has to fall back to a full-column decode when the
3594    /// "single clean run" invariant doesn't hold. Nulls are excluded.
3595    pub fn range_row_ids_visible_i64(
3596        &mut self,
3597        column_id: u16,
3598        lo: i64,
3599        hi: i64,
3600        snapshot: Epoch,
3601    ) -> Result<Vec<u64>> {
3602        let stats: Vec<(Option<i64>, Option<i64>, usize)> = match self.column_page_stats(column_id)
3603        {
3604            Some(s) => s
3605                .iter()
3606                .map(|st| {
3607                    (
3608                        be_i64(st.min.as_deref()),
3609                        be_i64(st.max.as_deref()),
3610                        st.row_count as usize,
3611                    )
3612                })
3613                .collect(),
3614            None => return Ok(Vec::new()),
3615        };
3616        // Encrypted columns are pruneable only when this run carries the
3617        // decrypted stats envelope (overlaid at open). Without one (a run
3618        // whose writer recorded no stats at all), a missing min/max means
3619        // "unknown" — never prune — whereas with the envelope (and always for
3620        // plaintext runs) a missing min/max means an all-null page.
3621        let stats_pruneable =
3622            !self.col_encrypted(column_id) || self.header.encrypted_stats_offset != 0;
3623        let (positions, rids) = self.visible_positions_with_rids(snapshot)?;
3624        let mut out: Vec<u64> = Vec::new();
3625        let mut vis = 0usize;
3626        let mut page_start = 0usize;
3627        for (seq, &(mn, mx, nrows)) in stats.iter().enumerate() {
3628            let page_end = page_start + nrows;
3629            // A page can be dropped iff every value fails the predicate.
3630            let skip = stats_pruneable
3631                && match (mn, mx) {
3632                    (Some(mn), Some(mx)) => mx < lo || mn > hi,
3633                    _ => true, // all-null / no stats → nulls never match
3634                };
3635            if !skip {
3636                let val_page = self.read_page(column_id, seq)?;
3637                let vals = columnar::decode_page_native(TypeId::Int64, &val_page, nrows)?;
3638                if let columnar::NativeColumn::Int64 { data: v, validity } = vals {
3639                    while vis < positions.len() && positions[vis] < page_end {
3640                        let local = positions[vis] - page_start;
3641                        if columnar::validity_bit(&validity, local)
3642                            && v[local] >= lo
3643                            && v[local] <= hi
3644                        {
3645                            out.push(rids[vis] as u64);
3646                        }
3647                        vis += 1;
3648                    }
3649                }
3650            } else {
3651                while vis < positions.len() && positions[vis] < page_end {
3652                    vis += 1;
3653                }
3654            }
3655            page_start = page_end;
3656        }
3657        Ok(out)
3658    }
3659
3660    /// Float64 analogue of [`Self::range_row_ids_visible_i64`] with per-bound
3661    /// inclusivity (Phase 16.3).
3662    pub fn range_row_ids_visible_f64(
3663        &mut self,
3664        column_id: u16,
3665        lo: f64,
3666        lo_inclusive: bool,
3667        hi: f64,
3668        hi_inclusive: bool,
3669        snapshot: Epoch,
3670    ) -> Result<Vec<u64>> {
3671        let stats: Vec<(Option<f64>, Option<f64>, usize)> = match self.column_page_stats(column_id)
3672        {
3673            Some(s) => s
3674                .iter()
3675                .map(|st| {
3676                    (
3677                        be_f64(st.min.as_deref()),
3678                        be_f64(st.max.as_deref()),
3679                        st.row_count as usize,
3680                    )
3681                })
3682                .collect(),
3683            None => return Ok(Vec::new()),
3684        };
3685        // Encrypted columns are pruneable only when this run carries the
3686        // decrypted stats envelope (overlaid at open). Without one (a run
3687        // whose writer recorded no stats at all), a missing min/max means
3688        // "unknown" — never prune — whereas with the envelope (and always for
3689        // plaintext runs) a missing min/max means an all-null page.
3690        let stats_pruneable =
3691            !self.col_encrypted(column_id) || self.header.encrypted_stats_offset != 0;
3692        let (positions, rids) = self.visible_positions_with_rids(snapshot)?;
3693        let mut out: Vec<u64> = Vec::new();
3694        let mut vis = 0usize;
3695        let mut page_start = 0usize;
3696        for (seq, &(mn, mx, nrows)) in stats.iter().enumerate() {
3697            let page_end = page_start + nrows;
3698            let skip = stats_pruneable
3699                && match (mn, mx) {
3700                    (Some(mn), Some(mx)) => {
3701                        let skip_lo = mx < lo || (!lo_inclusive && mx == lo);
3702                        let skip_hi = mn > hi || (!hi_inclusive && mn == hi);
3703                        skip_lo || skip_hi
3704                    }
3705                    _ => true,
3706                };
3707            if !skip {
3708                let val_page = self.read_page(column_id, seq)?;
3709                let vals = columnar::decode_page_native(TypeId::Float64, &val_page, nrows)?;
3710                if let columnar::NativeColumn::Float64 { data: v, validity } = vals {
3711                    while vis < positions.len() && positions[vis] < page_end {
3712                        let local = positions[vis] - page_start;
3713                        if columnar::validity_bit(&validity, local) && !v[local].is_nan() {
3714                            let val = v[local];
3715                            let ok_lo = if lo_inclusive { val >= lo } else { val > lo };
3716                            let ok_hi = if hi_inclusive { val <= hi } else { val < hi };
3717                            if ok_lo && ok_hi {
3718                                out.push(rids[vis] as u64);
3719                            }
3720                        }
3721                        vis += 1;
3722                    }
3723                }
3724            } else {
3725                while vis < positions.len() && positions[vis] < page_end {
3726                    vis += 1;
3727                }
3728            }
3729            page_start = page_end;
3730        }
3731        Ok(out)
3732    }
3733
3734    /// MVCC-visible `IS NULL` / `IS NOT NULL` resolution. Follows the same
3735    /// page-stat-pruned + visible-positions pattern as
3736    /// [`Self::range_row_ids_visible_i64`], but checks the validity bitmap
3737    /// instead of a value range. Pages with no nulls (for IS NULL) or all-nulls
3738    /// (for IS NOT NULL) are skipped.
3739    pub fn null_row_ids_visible(
3740        &mut self,
3741        column_id: u16,
3742        want_nulls: bool,
3743        snapshot: Epoch,
3744    ) -> Result<Vec<u64>> {
3745        let stats: Vec<(usize, usize)> = match self.column_page_stats(column_id) {
3746            Some(s) => s
3747                .iter()
3748                .map(|st| (st.null_count as usize, st.row_count as usize))
3749                .collect(),
3750            None => return Ok(Vec::new()),
3751        };
3752        let ty = self.resolve_type(column_id);
3753        let (positions, rids) = self.visible_positions_with_rids(snapshot)?;
3754        let mut out: Vec<u64> = Vec::new();
3755        let mut vis = 0usize;
3756        let mut page_start = 0usize;
3757        for (seq, &(null_count, nrows)) in stats.iter().enumerate() {
3758            let page_end = page_start + nrows;
3759            let skip = (want_nulls && null_count == 0) || (!want_nulls && null_count == nrows);
3760            if !skip {
3761                let val_page = self.read_page(column_id, seq)?;
3762                let col = columnar::decode_page_native(ty.clone(), &val_page, nrows)?;
3763                let validity = col.validity();
3764                while vis < positions.len() && positions[vis] < page_end {
3765                    let local = positions[vis] - page_start;
3766                    let is_null = !columnar::validity_bit(validity, local);
3767                    if is_null == want_nulls {
3768                        out.push(rids[vis] as u64);
3769                    }
3770                    vis += 1;
3771                }
3772            } else {
3773                while vis < positions.len() && positions[vis] < page_end {
3774                    vis += 1;
3775                }
3776            }
3777            page_start = page_end;
3778        }
3779        Ok(out)
3780    }
3781
3782    /// Row ids whose newest visible version in this run at `snapshot` is a
3783    /// tombstone (`deleted=true`). These rids must be subtracted from a
3784    /// multi-run range survivor set, otherwise a tombstone landing in a
3785    /// newer run can't strip its alive preimage from an older run that the
3786    /// learned index (built only from run 0) or `range_row_ids_visible_i64`
3787    /// happily returned.
3788    pub fn tombstoned_row_ids(&mut self, snapshot: Epoch) -> Result<Vec<u64>> {
3789        let n = self.row_count();
3790        if n == 0 {
3791            return Ok(Vec::new());
3792        }
3793        // Clean runs have no tombstones.
3794        if self.is_clean()
3795            && self.epoch_override.is_none()
3796            && self.header.epoch_created <= snapshot.0
3797        {
3798            return Ok(Vec::new());
3799        }
3800        let (row_ids, epochs, deleted) = self.system_columns_native()?;
3801        let mut out = Vec::new();
3802        let mut i = 0;
3803        while i < n {
3804            let rid = row_ids[i] as u64;
3805            let mut best: Option<usize> = None;
3806            let mut j = i;
3807            while j < n && row_ids[j] as u64 == rid {
3808                if epochs[j] as u64 <= snapshot.0 {
3809                    best = Some(j);
3810                }
3811                j += 1;
3812            }
3813            if let Some(b) = best {
3814                if deleted[b] != 0 {
3815                    out.push(rid);
3816                }
3817            }
3818            i = j;
3819        }
3820        Ok(out)
3821    }
3822
3823    /// tombstones excluded) paired with each position's `RowId`, in one pass.
3824    /// Used by [`crate::cursor::NativePageCursor`] to map survivors to pages
3825    /// without re-decoding the system columns.
3826    pub fn visible_positions_with_rids(
3827        &mut self,
3828        snapshot: Epoch,
3829    ) -> Result<(Vec<usize>, Vec<i64>)> {
3830        let n = self.row_count();
3831        if n == 0 {
3832            return Ok((Vec::new(), Vec::new()));
3833        }
3834        // Clean-run fast path (Phase 16.3c): one version per RowId, no
3835        // tombstones, ascending row_ids ⟹ every position is the newest (and
3836        // only) visible version. Skip decoding epoch/deleted and the group-
3837        // collapse loop; just decode row_ids (needed for survivor↔position
3838        // mapping) and return identity positions [0..n).
3839        // A uniform-epoch overlay must still gate by snapshot, so skip the clean
3840        // fast path when an override is active (defensive: spill runs are never
3841        // written clean).
3842        if self.is_clean()
3843            && self.epoch_override.is_none()
3844            && self.header.epoch_created <= snapshot.0
3845        {
3846            let row_ids = match self.column_native_shared(SYS_ROW_ID)? {
3847                columnar::NativeColumn::Int64 { data, .. } => data,
3848                _ => return Err(MongrelError::InvalidArgument("sys row_id not int64".into())),
3849            };
3850            let positions: Vec<usize> = (0..n).collect();
3851            return Ok((positions, row_ids));
3852        }
3853        let (row_ids, epochs, deleted) = self.system_columns_native()?;
3854        // Runs are written in `(RowId, Epoch)` ascending order (Bε-tree
3855        // composite key), so same-rid positions are consecutive with the
3856        // newest (highest epoch) last. One linear pass keeps the last position
3857        // per rid with `epoch <= snapshot`, dropping tombstones — no HashMap,
3858        // no per-row hashing, sequential memory access. (Phase 16.3.)
3859        //
3860        // Invariant: every write path (memtable drain, mutable-run spill,
3861        // bulk_load's sequential alloc) produces runs in this order; if a path
3862        // ever wrote an unsorted run this would under-count (only consecutive
3863        // dup groups merge) and must be reverted to the HashMap form.
3864        let mut idxs: Vec<usize> = Vec::new();
3865        let mut i = 0;
3866        while i < n {
3867            let rid = row_ids[i] as u64;
3868            // Walk the consecutive rid group; epochs rise within it, so the
3869            // last position with epoch <= snapshot is the newest visible one.
3870            let mut best: Option<usize> = None;
3871            let mut j = i;
3872            while j < n && row_ids[j] as u64 == rid {
3873                if epochs[j] as u64 <= snapshot.0 {
3874                    best = Some(j);
3875                }
3876                j += 1;
3877            }
3878            if let Some(b) = best {
3879                if deleted[b] == 0 {
3880                    idxs.push(b);
3881                }
3882            }
3883            i = j;
3884        }
3885        // Groups are processed in rid-ascending = position-ascending order.
3886        let rids: Vec<i64> = idxs.iter().map(|&k| row_ids[k]).collect();
3887        Ok((idxs, rids))
3888    }
3889
3890    /// Row count of each PAX page of `column_id`, in page order. Every column
3891    /// in a run shares the same PAX row partition, so this yields the table's
3892    /// page layout (cumulative sums give page start offsets).
3893    pub fn page_row_counts(&self, column_id: u16) -> Result<Vec<usize>> {
3894        Ok(self
3895            .find_header(column_id)?
3896            .page_stats
3897            .iter()
3898            .map(|s| s.row_count as usize)
3899            .collect())
3900    }
3901
3902    /// Whether this run stores `column_id` (false for a column added via
3903    /// `add_column` after the run was written — those read as all-null).
3904    pub fn has_column(&self, column_id: u16) -> bool {
3905        self.dir.iter().any(|h| h.column_id == column_id)
3906    }
3907
3908    /// The per-page [`PageStat`]s for `column_id`, or `None` if the column is
3909    /// absent from this run (schema evolution). Used to compute exact column
3910    /// min/max/null_count for the analytical aggregate fast path.
3911    pub fn column_page_stats(&self, column_id: u16) -> Option<&[crate::page::PageStat]> {
3912        self.dir
3913            .iter()
3914            .find(|h| h.column_id == column_id)
3915            .map(|ch| ch.page_stats.as_slice())
3916    }
3917
3918    /// Decode the system columns once, as typed buffers. Uses the shared
3919    /// parallel + decoded-page-cached path (`column_native_shared`) so the
3920    /// row-id/epoch/deleted pages decode concurrently and stay cached across
3921    /// queries — MVCC visibility resolution is on every scan's hot path.
3922    pub(crate) fn system_columns_native(&mut self) -> Result<(Vec<i64>, Vec<i64>, Vec<u8>)> {
3923        let row_ids = match self.column_native_shared(SYS_ROW_ID)? {
3924            columnar::NativeColumn::Int64 { data, .. } => data,
3925            _ => return Err(MongrelError::InvalidArgument("sys row_id not int64".into())),
3926        };
3927        let epochs = match self.column_native_shared(SYS_EPOCH)? {
3928            columnar::NativeColumn::Int64 { data, .. } => data,
3929            _ => return Err(MongrelError::InvalidArgument("sys epoch not int64".into())),
3930        };
3931        let deleted = match self.column_native_shared(SYS_DELETED)? {
3932            columnar::NativeColumn::Bool { data, .. } => data,
3933            _ => return Err(MongrelError::InvalidArgument("sys deleted not bool".into())),
3934        };
3935        Ok((row_ids, epochs, deleted))
3936    }
3937
3938    /// Newest visible version per `RowId` at `snapshot`, **including
3939    /// tombstones** (as `Row`s with `deleted=true`). Ascending `RowId`. Used by
3940    /// the engine to merge versions across runs and the memtable.
3941    ///
3942    /// HLC stamps are restored when [`SYS_COMMIT_TS`] is present (P0.5-T3);
3943    /// legacy runs without the column materialise `commit_ts: None`. Callers
3944    /// that hold a full [`crate::epoch::Snapshot`] still apply
3945    /// [`crate::epoch::Snapshot::observes_row`].
3946    pub fn visible_versions(&mut self, snapshot: Epoch) -> Result<Vec<Row>> {
3947        let n = self.row_count();
3948        if n == 0 {
3949            return Ok(Vec::new());
3950        }
3951        let row_ids = self.column(SYS_ROW_ID)?.to_vec();
3952        let epochs = self.column(SYS_EPOCH)?.to_vec();
3953        let mut best: HashMap<u64, (u64, usize)> = HashMap::new();
3954        for i in 0..n {
3955            let rid = int_at(&row_ids, i);
3956            let epoch = int_at(&epochs, i);
3957            if epoch > snapshot.0 {
3958                continue;
3959            }
3960            best.entry(rid)
3961                .and_modify(|e| {
3962                    if epoch > e.0 {
3963                        *e = (epoch, i);
3964                    }
3965                })
3966                .or_insert((epoch, i));
3967        }
3968        let mut picks: Vec<usize> = best.into_values().map(|(_, i)| i).collect();
3969        picks.sort();
3970        let mut out = Vec::with_capacity(picks.len());
3971        for i in picks {
3972            out.push(self.materialize(i)?);
3973        }
3974        Ok(out)
3975    }
3976
3977    /// All non-deleted rows visible at `snapshot`. Ascending `RowId`.
3978    pub fn visible_rows(&mut self, snapshot: Epoch) -> Result<Vec<Row>> {
3979        Ok(self
3980            .visible_versions(snapshot)?
3981            .into_iter()
3982            .filter(|r| !r.deleted)
3983            .collect())
3984    }
3985
3986    pub(crate) fn materialize(&mut self, index: usize) -> Result<Row> {
3987        let row_id = RowId(int_at(self.column(SYS_ROW_ID)?, index));
3988        let epoch = Epoch(int_at(self.column(SYS_EPOCH)?, index));
3989        let deleted = bool_at(self.column(SYS_DELETED)?, index);
3990        let commit_ts = self.commit_ts_at(index)?;
3991        let col_ids: Vec<u16> = self.schema.columns.iter().map(|c| c.id).collect();
3992        let mut columns = HashMap::new();
3993        for id in col_ids {
3994            let val = if self.dir.iter().any(|h| h.column_id == id) {
3995                self.column(id)?.get(index).cloned().unwrap_or(Value::Null)
3996            } else {
3997                // Column added via schema evolution after this run was written.
3998                Value::Null
3999            };
4000            columns.insert(id, val);
4001        }
4002        Ok(Row {
4003            row_id,
4004            committed_epoch: epoch,
4005            columns,
4006            deleted,
4007            commit_ts,
4008        })
4009    }
4010
4011    /// Batched row materialization (Phase 16.3b finish): decode each system +
4012    /// user column **once** via the typed, page-cached `column_native` path and
4013    /// gather only the requested `indices` straight into `Row`s. This replaces
4014    /// N independent `materialize` calls, each of which rebuilt a full-column
4015    /// `Vec<Value>` (heap-allocating one `Value` per row) and then `.cloned()`
4016    /// a single slot. Exact semantics retained: schema-evolved columns absent
4017    /// from this run read `Value::Null`; deleted rows are still returned (the
4018    /// caller filters them).
4019    pub(crate) fn materialize_batch(&mut self, indices: &[usize]) -> Result<Vec<Row>> {
4020        if indices.is_empty() {
4021            return Ok(Vec::new());
4022        }
4023        use std::collections::HashMap;
4024        let rid_col = self.column_native_shared(SYS_ROW_ID)?;
4025        let epoch_col = self.column_native_shared(SYS_EPOCH)?;
4026        let del_col = self.column_native_shared(SYS_DELETED)?;
4027        let commit_ts_col = if self.has_column(SYS_COMMIT_TS) {
4028            Some(self.column_native_shared(SYS_COMMIT_TS)?)
4029        } else {
4030            None
4031        };
4032        let mut user: HashMap<u16, columnar::NativeColumn> = HashMap::new();
4033        let present: Vec<u16> = self
4034            .schema
4035            .columns
4036            .iter()
4037            .map(|c| c.id)
4038            .filter(|id| self.dir.iter().any(|h| h.column_id == *id))
4039            .collect();
4040        for id in present {
4041            user.insert(id, self.column_native(id)?);
4042        }
4043        let i64_at = |col: &columnar::NativeColumn, i: usize| -> i64 {
4044            match col {
4045                columnar::NativeColumn::Int64 { data, .. } => data.get(i).copied().unwrap_or(0),
4046                _ => 0,
4047            }
4048        };
4049        let bool_at_native = |col: &columnar::NativeColumn, i: usize| -> bool {
4050            match col {
4051                columnar::NativeColumn::Bool { data, .. } => {
4052                    data.get(i).copied().map(|b| b != 0).unwrap_or(false)
4053                }
4054                _ => false,
4055            }
4056        };
4057        let mut rows = Vec::with_capacity(indices.len());
4058        for &idx in indices {
4059            let row_id = RowId(i64_at(&rid_col, idx) as u64);
4060            let epoch = Epoch(i64_at(&epoch_col, idx) as u64);
4061            let deleted = bool_at_native(&del_col, idx);
4062            let commit_ts = commit_ts_col
4063                .as_ref()
4064                .and_then(|col| decode_commit_ts_value(col.value_at(idx).as_ref()));
4065            let mut columns = HashMap::with_capacity(self.schema.columns.len());
4066            for cdef in self.schema.columns.iter() {
4067                let val = match user.get(&cdef.id) {
4068                    Some(col) => col.value_at(idx).unwrap_or(Value::Null),
4069                    None => Value::Null,
4070                };
4071                columns.insert(cdef.id, val);
4072            }
4073            rows.push(Row {
4074                row_id,
4075                committed_epoch: epoch,
4076                columns,
4077                deleted,
4078                commit_ts,
4079            });
4080        }
4081        Ok(rows)
4082    }
4083}
4084
4085fn int_at(vals: &[Value], i: usize) -> u64 {
4086    match vals.get(i) {
4087        Some(Value::Int64(x)) => *x as u64,
4088        _ => 0,
4089    }
4090}
4091
4092fn bool_at(vals: &[Value], i: usize) -> bool {
4093    matches!(vals.get(i), Some(Value::Bool(true)))
4094}
4095
4096#[cfg(test)]
4097mod tests {
4098    use super::*;
4099    use crate::columnar::NativeColumn;
4100    use crate::memtable::Value;
4101    use crate::rowid::RowId;
4102    use crate::schema::{ColumnDef, ColumnFlags};
4103    use tempfile::tempdir;
4104
4105    fn schema() -> Schema {
4106        Schema {
4107            schema_id: 1,
4108            columns: vec![
4109                ColumnDef {
4110                    id: 1,
4111                    name: "id".into(),
4112                    ty: TypeId::Int64,
4113                    flags: ColumnFlags::empty().with(ColumnFlags::PRIMARY_KEY),
4114                    default_value: None,
4115                    embedding_source: None,
4116                },
4117                ColumnDef {
4118                    id: 2,
4119                    name: "name".into(),
4120                    ty: TypeId::Bytes,
4121                    flags: ColumnFlags::empty(),
4122                    default_value: None,
4123                    embedding_source: None,
4124                },
4125            ],
4126            indexes: Vec::new(),
4127            colocation: vec![],
4128            constraints: Default::default(),
4129            clustered: false,
4130        }
4131    }
4132
4133    fn rows() -> Vec<Row> {
4134        vec![
4135            Row::new(RowId(1), Epoch(10))
4136                .with_column(1, Value::Int64(1))
4137                .with_column(2, Value::Bytes(b"alice".to_vec())),
4138            Row::new(RowId(2), Epoch(10))
4139                .with_column(1, Value::Int64(2))
4140                .with_column(2, Value::Bytes(b"bob".to_vec())),
4141            Row::new(RowId(3), Epoch(10))
4142                .with_column(1, Value::Int64(3))
4143                .with_column(2, Value::Bytes(b"carol".to_vec())),
4144        ]
4145    }
4146
4147    #[test]
4148    fn flush_then_read_mvcc() {
4149        let dir = tempdir().unwrap();
4150        let path = dir.path().join("r-1.sr");
4151        let header = RunWriter::new(&schema(), 1, Epoch(10), 0)
4152            .write(&path, &rows())
4153            .unwrap();
4154        assert_eq!(header.row_count, 3);
4155
4156        let mut r = RunReader::open(&path, schema(), None).unwrap();
4157        // Point lookup.
4158        let (e, row) = r.get_version(RowId(2), Epoch(20)).unwrap().unwrap();
4159        assert_eq!(e, Epoch(10));
4160        assert_eq!(row.row_id, RowId(2));
4161        assert!(matches!(row.columns.get(&2), Some(Value::Bytes(_))));
4162        // Missing.
4163        assert!(r.get_version(RowId(99), Epoch(20)).unwrap().is_none());
4164        // Scan.
4165        let all = r.visible_rows(Epoch(20)).unwrap();
4166        assert_eq!(all.len(), 3);
4167        // Unstamped flush omits SYS_COMMIT_TS (legacy path).
4168        assert!(!r.has_column(SYS_COMMIT_TS));
4169        assert!(all.iter().all(|row| row.commit_ts.is_none()));
4170    }
4171
4172    /// ID: P0.5-T3 — flush preserves HLC stamps via optional SYS_COMMIT_TS;
4173    /// legacy runs without the column materialise commit_ts as None.
4174    #[test]
4175    fn p05_t3_sys_commit_ts_preserved_on_flush_and_legacy_is_none() {
4176        let dir = tempdir().unwrap();
4177        let stamped_path = dir.path().join("r-hlc.sr");
4178        let stamp = HlcTimestamp {
4179            physical_micros: 42_000_000,
4180            logical: 7,
4181            node_tiebreaker: 3,
4182        };
4183        let stamped = vec![
4184            Row::new_with_hlc(RowId(1), Epoch(10), stamp)
4185                .with_column(1, Value::Int64(1))
4186                .with_column(2, Value::Bytes(b"alice".to_vec())),
4187            Row::new(RowId(2), Epoch(11)) // mixed: unstamped sibling
4188                .with_column(1, Value::Int64(2))
4189                .with_column(2, Value::Bytes(b"bob".to_vec())),
4190        ];
4191        RunWriter::new(&schema(), 9, Epoch(11), 0)
4192            .write(&stamped_path, &stamped)
4193            .unwrap();
4194        let mut reader = RunReader::open(&stamped_path, schema(), None).unwrap();
4195        assert!(
4196            reader.has_column(SYS_COMMIT_TS),
4197            "stamped flush must emit optional SYS_COMMIT_TS"
4198        );
4199        let (_, row1) = reader
4200            .get_version(RowId(1), Epoch(20))
4201            .unwrap()
4202            .expect("row 1");
4203        assert_eq!(row1.commit_ts, Some(stamp));
4204        let (_, row2) = reader
4205            .get_version(RowId(2), Epoch(20))
4206            .unwrap()
4207            .expect("row 2");
4208        assert_eq!(row2.commit_ts, None, "Null stamp cell stays None");
4209        let versions = reader.visible_versions(Epoch(20)).unwrap();
4210        assert_eq!(versions.len(), 2);
4211        assert_eq!(
4212            versions
4213                .iter()
4214                .find(|r| r.row_id == RowId(1))
4215                .and_then(|r| r.commit_ts),
4216            Some(stamp)
4217        );
4218
4219        // Legacy run (no stamps on write) lacks the column → always None.
4220        let legacy_path = dir.path().join("r-legacy.sr");
4221        RunWriter::new(&schema(), 10, Epoch(10), 0)
4222            .write(&legacy_path, &rows())
4223            .unwrap();
4224        let mut legacy = RunReader::open(&legacy_path, schema(), None).unwrap();
4225        assert!(!legacy.has_column(SYS_COMMIT_TS));
4226        let all = legacy.visible_versions(Epoch(20)).unwrap();
4227        assert!(all.iter().all(|r| r.commit_ts.is_none()));
4228    }
4229
4230    #[test]
4231    fn visible_version_cursor_preserves_order_snapshot_and_tombstones() {
4232        let dir = tempdir().unwrap();
4233        let path = dir.path().join("r-cursor.sr");
4234        let mut tombstone = Row::new(RowId(2), Epoch(2));
4235        tombstone.deleted = true;
4236        let versions = vec![
4237            Row::new(RowId(1), Epoch(4)).with_column(1, Value::Int64(10)),
4238            Row::new(RowId(2), Epoch(1)).with_column(1, Value::Int64(2)),
4239            tombstone,
4240            Row::new(RowId(3), Epoch(1)).with_column(1, Value::Int64(20)),
4241            Row::new(RowId(3), Epoch(3)).with_column(1, Value::Int64(23)),
4242            Row::new(RowId(4), Epoch(4)).with_column(1, Value::Int64(30)),
4243        ];
4244        RunWriter::new(&schema(), 7, Epoch(4), 0)
4245            .write(&path, &versions)
4246            .unwrap();
4247
4248        let control = crate::ExecutionControl::new(None);
4249        let mut cursor = RunReader::open(&path, schema(), None)
4250            .unwrap()
4251            .into_visible_version_cursor(Epoch(2))
4252            .unwrap();
4253        let first = cursor.next_visible_version(&control).unwrap().unwrap();
4254        assert_eq!(first.row_id, RowId(2));
4255        assert!(first.deleted);
4256        assert_eq!(first.committed_epoch, Epoch(2));
4257        let second = cursor.next_visible_version(&control).unwrap().unwrap();
4258        assert_eq!(second.row_id, RowId(3));
4259        assert_eq!(second.committed_epoch, Epoch(1));
4260        let row = cursor.materialize(second, &control).unwrap();
4261        assert_eq!(row.columns.get(&1), Some(&Value::Int64(20)));
4262        assert!(cursor.next_visible_version(&control).unwrap().is_none());
4263    }
4264
4265    #[test]
4266    fn learned_index_was_stored() {
4267        let dir = tempdir().unwrap();
4268        let path = dir.path().join("r-2.sr");
4269        RunWriter::new(&schema(), 2, Epoch(1), 0)
4270            .write(&path, &rows())
4271            .unwrap();
4272        let header = read_header(&path).unwrap();
4273        assert!(
4274            header.index_trailer_offset != 0,
4275            "trailer should be present"
4276        );
4277    }
4278
4279    #[test]
4280    fn low_level_container_still_round_trips() {
4281        let dir = tempdir().unwrap();
4282        let path = dir.path().join("r-3.sr");
4283        let cols = vec![ColumnPayload {
4284            column_id: 1,
4285            type_id_tag: 8,
4286            encoding: Encoding::Plain,
4287            pages: vec![vec![1, 2, 3, 4]],
4288            page_stats: Vec::new(),
4289        }];
4290        let header = write_run(
4291            &path,
4292            &RunSpec {
4293                run_id: 1,
4294                schema_id: 1,
4295                epoch_created: 1,
4296                level: 0,
4297                flags: 0,
4298                sort_key_column_id: SORT_KEY_ROW_ID,
4299                row_count: 1,
4300                min_row_id: 0,
4301                max_row_id: 0,
4302                columns: &cols,
4303            },
4304        )
4305        .unwrap();
4306        let back = read_header(&path).unwrap();
4307        assert_eq!(back.content_hash, header.content_hash);
4308        assert_eq!(read_column_dir(&path, &back).unwrap().len(), 1);
4309    }
4310
4311    #[test]
4312    fn detects_corruption() {
4313        let dir = tempdir().unwrap();
4314        let path = dir.path().join("r-4.sr");
4315        write_run(
4316            &path,
4317            &RunSpec {
4318                run_id: 1,
4319                schema_id: 1,
4320                epoch_created: 1,
4321                level: 0,
4322                flags: 0,
4323                sort_key_column_id: SORT_KEY_ROW_ID,
4324                row_count: 1,
4325                min_row_id: 0,
4326                max_row_id: 0,
4327                columns: &[ColumnPayload {
4328                    column_id: 1,
4329                    type_id_tag: 8,
4330                    encoding: Encoding::Plain,
4331                    pages: vec![vec![1, 2, 3]],
4332                    page_stats: Vec::new(),
4333                }],
4334            },
4335        )
4336        .unwrap();
4337        let mut bytes = std::fs::read(&path).unwrap();
4338        bytes[300] ^= 0xFF;
4339        std::fs::write(&path, bytes).unwrap();
4340        let err = read_header(&path).unwrap_err();
4341        assert!(
4342            matches!(err, MongrelError::ChecksumMismatch { .. }),
4343            "got {err:?}"
4344        );
4345    }
4346
4347    /// Phase 14.6: the direct-to-mmap placement logic must produce a byte-
4348    /// identical run to the in-buffer fallback. We drive `plan_run` + `place_run`
4349    /// into a plain `Vec<u8>` (so it's testable even where file mmap is denied —
4350    /// e.g. this sandbox), compare against `write_run_vec`, and additionally
4351    /// check that `write_run_mmap` agrees wherever a mapping can be created.
4352    #[test]
4353    fn mmap_and_vec_writers_are_byte_identical() {
4354        let dir = tempdir().unwrap();
4355        let stats = vec![PageStat {
4356            first_row_id: 0,
4357            last_row_id: 1,
4358            null_count: 0,
4359            row_count: 2,
4360            min: Some(vec![0]),
4361            max: Some(vec![9]),
4362            offset: 0,
4363            compressed_len: 0,
4364            uncompressed_len: 0,
4365        }];
4366        let spec = RunSpec {
4367            run_id: 7,
4368            schema_id: 1,
4369            epoch_created: 10,
4370            level: 0,
4371            flags: 0,
4372            sort_key_column_id: SORT_KEY_ROW_ID,
4373            row_count: 2,
4374            min_row_id: 0,
4375            max_row_id: 1,
4376            columns: &[
4377                ColumnPayload {
4378                    column_id: 1,
4379                    type_id_tag: 8,
4380                    encoding: Encoding::Plain,
4381                    pages: vec![vec![1, 2, 3, 4], vec![5, 6, 7, 8]],
4382                    page_stats: stats.clone(),
4383                },
4384                ColumnPayload {
4385                    column_id: 2,
4386                    type_id_tag: 8,
4387                    encoding: Encoding::Plain,
4388                    pages: vec![vec![10, 20], vec![30, 40]],
4389                    page_stats: stats.clone(),
4390                },
4391            ],
4392        };
4393        let trailer = b"learned-trailer-bytes";
4394
4395        // (1) placement path into a Vec-backed buffer.
4396        let plan = plan_run(&spec, None, Some(trailer)).expect("plan");
4397        let mut buf = vec![0u8; plan.total];
4398        let h_place = place_run(&spec, None, Some(trailer), &plan, &mut buf)
4399            .expect("place_run into Vec succeeds");
4400
4401        // (2) in-buffer fallback writer to a real file.
4402        let path_vec = dir.path().join("vec.sr");
4403        let h_vec = write_run_vec(&path_vec, &spec, None, Some(trailer)).unwrap();
4404        let bv = std::fs::read(&path_vec).unwrap();
4405
4406        assert_eq!(h_place.content_hash, h_vec.content_hash);
4407        assert_eq!(h_place.footer_offset, h_vec.footer_offset);
4408        assert_eq!(
4409            buf, bv,
4410            "place_run and write_run_vec must be byte-identical"
4411        );
4412        assert!(read_header(&path_vec).is_ok());
4413
4414        // (3) the mmap writer, when the FS supports it, must also agree. Where
4415        // file mmap is denied (some sandboxes), it reports the fallback sentinel
4416        // and `write_run_with` transparently uses the vec path instead.
4417        let path_mmap = dir.path().join("mmap.sr");
4418        match write_run_mmap(&path_mmap, &spec, None, Some(trailer)) {
4419            Ok(h_mmap) => {
4420                let bm = std::fs::read(&path_mmap).unwrap();
4421                assert_eq!(bm, bv, "mmap run must be byte-identical to vec run");
4422                assert_eq!(h_mmap.content_hash, h_vec.content_hash);
4423            }
4424            Err(e) if is_mmap_unavailable(&e) => {
4425                eprintln!("note: file mmap unavailable here; vec path covered (1)/(2)");
4426            }
4427            Err(e) => panic!("unexpected mmap error: {e:?}"),
4428        }
4429    }
4430
4431    /// Phase 15.1: the `&self` parallel-decode path (`column_native_shared`)
4432    /// must yield the same `NativeColumn` as the `&mut` `column_native` path,
4433    /// column for column, on a multi-page run. This guards the cross-column
4434    /// parallel scan used by `visible_columns_native`.
4435    #[test]
4436    fn column_native_shared_matches_column_native() {
4437        let dir = tempdir().unwrap();
4438        let path = dir.path().join("r.sr");
4439        // Enough rows to span >1 page (PAGE_ROWS = 65 536) so the parallel page
4440        // branch runs, plus a partial tail page.
4441        let n = 65_536 * 2 + 9;
4442        let id_col = NativeColumn::int64_sequence(1, n);
4443        let mut offsets = vec![0u32];
4444        let mut values = Vec::new();
4445        for i in 0..n {
4446            values.extend_from_slice(format!("v{}", i % 17).as_bytes());
4447            offsets.push(values.len() as u32);
4448        }
4449        let name_col = NativeColumn::Bytes {
4450            offsets,
4451            values,
4452            validity: vec![0xFF; n.div_ceil(8)],
4453        };
4454        let header = RunWriter::new(&schema(), 9, Epoch(5), 0)
4455            .write_native(&path, &[(1, id_col), (2, name_col)], n, 1)
4456            .unwrap();
4457
4458        let mut reader = RunReader::open_with_cache(&path, schema(), None, None, None, 0, None)
4459            .expect("open reader");
4460        assert!(reader.has_mmap(), "test env must support read-only mmap");
4461        assert_eq!(reader.row_count(), header.row_count as usize);
4462
4463        for cid in [1u16, 2] {
4464            let a = reader.column_native(cid).expect("column_native");
4465            let b = reader
4466                .column_native_shared(cid)
4467                .expect("column_native_shared");
4468            assert_eq!(a.len(), b.len(), "len mismatch col {cid}");
4469            // Byte-level equality of the typed buffers.
4470            match (&a, &b) {
4471                (
4472                    NativeColumn::Int64 {
4473                        data: da,
4474                        validity: va,
4475                    },
4476                    NativeColumn::Int64 {
4477                        data: db,
4478                        validity: vb,
4479                    },
4480                ) => {
4481                    assert_eq!(da, db, "Int64 data col {cid}");
4482                    assert_eq!(va, vb, "Int64 validity col {cid}");
4483                }
4484                (
4485                    NativeColumn::Bytes {
4486                        offsets: oa,
4487                        values: ua,
4488                        validity: va,
4489                    },
4490                    NativeColumn::Bytes {
4491                        offsets: ob,
4492                        values: ub,
4493                        validity: vb,
4494                    },
4495                ) => {
4496                    assert_eq!(oa, ob, "Bytes offsets col {cid}");
4497                    assert_eq!(ua, ub, "Bytes values col {cid}");
4498                    assert_eq!(va, vb, "Bytes validity col {cid}");
4499                }
4500                _ => panic!("type mismatch col {cid}: {a:?} vs {b:?}"),
4501            }
4502        }
4503    }
4504
4505    #[test]
4506    fn page_cache_key_distinguishes_tables() {
4507        let a = page_cache_key(1, 5, 2, 3);
4508        let b = page_cache_key(2, 5, 2, 3);
4509        assert_ne!(a, b, "same run/col/page, different table must differ");
4510        // same table different run/col/page also differ (sanity)
4511        assert_ne!(page_cache_key(1, 5, 2, 3), page_cache_key(1, 6, 2, 3));
4512        assert_ne!(page_cache_key(1, 5, 2, 3), page_cache_key(1, 5, 2, 4));
4513    }
4514}