coordinode-lsm-tree 7.0.0

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

//! Merge-on-read segment relocation: rewrite a columnar SST into a new file that
//! reuses every data block verbatim and carries a positional delete-bitmap,
//! deferring the expensive re-transpose that copy-on-write pays.
//!
//! Because merge-on-read keeps all rows (deleted ones are masked by the bitmap,
//! not dropped), the output segment is identical to the source except for a new
//! table id and an added `delete_bitmap` section. So this copies the data /
//! index / filter / zone-map / seqno-bounds sections (and the torn-write
//! defenses) block by block with their payloads untouched, re-encodes only the
//! two `meta` blocks with the new id, and appends the bitmap. Each copied block
//! is re-stamped for the new table and the offset it lands at, since its stored
//! checksum is bound to both. The data section stays first, so the block-index /
//! zone-map / seqno-bounds absolute offsets stay valid; every other section is
//! addressed through the table of contents.

use super::Table;
use super::block::decoder::ParsedItem;
use super::block::{Block, BlockIdentity, BlockTransform, BlockType};
use super::data_block::DataBlock;
use super::delete_bitmap::DeleteBitmap;
use super::file_pool::ClaimedFile;
use crate::checksum::ChecksummedWriter;
use crate::fs::{FsFile, FsOpenOptions, SyncMode};
use crate::io::BufWriter;
use crate::path::PathBuf;
use crate::{Checksum, InternalValue, TableId, UserValue};
use alloc::vec::Vec;

/// Largest chunk read+written when copying a section verbatim, so a multi-MiB
/// data section never has to be buffered whole.
const COPY_CHUNK: usize = 256 * 1024;

/// A relocated segment's file, as written.
#[derive(Debug)]
pub struct Relocated {
    /// The new table's id.
    pub id: TableId,
    /// The new table's file.
    pub path: PathBuf,
    /// Whether the file's directory entry was durable before the table was
    /// written into it (see [`Table::mark_entry_durable`]).
    pub entry_durable: bool,
    /// The file's checksum, for [`Table::recover`].
    pub checksum: Checksum,
    /// The bytes copied from the source as they lay: every section but the
    /// two meta blocks, encoded again, and the delete bitmap.
    pub copied: u64,
}

impl Table {
    /// Writes a new SST into the empty file `output` opens, reusing this
    /// columnar segment's data blocks verbatim, re-pointing the on-disk table
    /// id to the file's, and adding `delete_bitmap` as a positional row mask.
    /// `output` is called only once the segment is known to be relocatable,
    /// and its file is removed on a failure after that. Returns the new file
    /// and its checksum (for [`Table::recover`]) and the bytes it copied; the
    /// caller installs and recovers it, and the install syncs the folder for
    /// a file whose entry is not durable yet. `pace` is told how long each
    /// read of this segment takes.
    ///
    /// # Errors
    ///
    /// Returns [`crate::Error::FeatureUnsupported`] when the segment cannot be
    /// relocated by verbatim block reuse: an encrypted segment (the AEAD binds
    /// the table id, so a re-pointed copy fails verification), a segment carrying
    /// Page ECC (the re-encoded meta would need the parity layout), a row-major
    /// segment, or one without a zone map (the positional mask needs the
    /// per-block row counts the zone map carries). The caller falls back to a
    /// copy-on-write rewrite in those cases.
    pub(crate) fn relocate_columnar_with_deletes(
        &self,
        output: impl FnOnce() -> crate::Result<ClaimedFile>,
        out_fs: &dyn crate::fs::Fs,
        delete_bitmap: &DeleteBitmap,
        sync_mode: SyncMode,
        pace: Option<&super::util::Pacer>,
    ) -> crate::Result<Relocated> {
        // Verbatim block reuse is sound only for a non-encrypted, non-ECC
        // columnar segment that already carries a zone map (the bitmap's
        // positional mask resolves each block's start row from the zone-map row
        // counts, and the open-time invariant rejects a bitmap without one).
        if self.encryption.is_some()
            || self.metadata.ecc_params.is_some()
            || self.metadata.ecc_unrecognized
            || !self.metadata.columnar
            || self.zone_map.is_empty()
            // A tight-space RESTRICTED view reads its live suffix only, so its
            // rows are numbered from the bound while this copies the whole
            // physical section (punched prefix blocks included) and publishes it
            // WITHOUT the restriction: the mask would land on shifted rows and
            // reads could reach the copied zeros.
            || self.restrict_lower_bound().is_some()
        {
            return Err(crate::Error::FeatureUnsupported(
                "merge-on-read block reuse needs an unrestricted, non-encrypted, non-ECC columnar segment with a zone map",
            ));
        }

        // Read the source through ITS filesystem; write the output through the
        // destination level's `out_fs` (the same one that recovers and installs
        // the relocated table), so level routing stays consistent.
        let mut timed = super::util::TimedFile::new(
            self.fs.open(&self.path, &FsOpenOptions::new().read(true))?,
        );
        if let Some(pace) = pace {
            timed.set_pace(alloc::sync::Arc::clone(pace));
        }
        let reader = crate::sfa::Reader::from_reader(&mut timed)?;
        // The section reads below are positioned: each is timed as it is made.
        let src = timed.into_inner();
        let pace = pace.map(|pace| &**pace);

        // `writing` is held to the end: the copy is among the files its folder
        // is writing until it is synced.
        let ClaimedFile {
            id: new_table_id,
            path: new_path,
            file: out,
            durable_entry,
            writing,
        } = output()?;
        let writing_below = writing.below();

        // Once the output file exists, any later failure (meta re-encode,
        // section copy, bitmap write, sync) must not leave a partial,
        // uninstalled SST behind. Best-effort remove it before propagating the
        // error.
        let result = (|| -> crate::Result<(Checksum, u64)> {
            // Re-encode the meta KV block with the new id AND the new
            // delete-bitmap descriptors. The block is loaded TAIL (`meta`)
            // since MID and TAIL carry identical content; the same patched
            // payload is written to both sections below.
            let meta_payload = self.repoint_meta_block(
                &*src,
                (new_table_id, &writing_below),
                delete_bitmap,
                pace,
            )?;
            let out = BufWriter::with_capacity(u16::MAX.into(), out);
            let out = ChecksummedWriter::new(out);
            let mut writer = crate::sfa::Writer::from_writer(out);
            let mut copied = 0;

            let meta_identity = BlockIdentity {
                table_id: new_table_id,
                block_type: BlockType::Meta,
                dict_id: 0,
                window_log: 0,
            };
            for entry in reader.toc().iter() {
                let name = entry.name();
                if name == b"delete_bitmap" {
                    // The relocated table's delete bitmap is injected fresh below;
                    // copying the source section too would leave two `delete_bitmap`
                    // entries in the TOC, so section lookup could reject the SST or
                    // select the stale copy while the metadata describes the new one.
                    continue;
                }
                writer.start(name)?;
                if name == b"meta_mid" || name == b"meta" {
                    // Re-encoded copy (new id), not the source bytes. Non-encrypted
                    // segment, so the two copies are identical (no nonce) but
                    // for the place their checksums are bound to.
                    let at = super::block::ChecksumAt::block(
                        new_table_id,
                        BlockType::Meta,
                        writer.get_ref().position(),
                    );
                    Block::write_into(
                        &mut writer,
                        &meta_payload,
                        meta_identity,
                        &BlockTransform::PLAIN,
                        at,
                    )?;
                } else if RAW_SECTIONS.contains(&name) {
                    copy_section(&*src, &mut writer, (entry.pos(), entry.len()), pace)?;
                    copied += entry.len();
                } else {
                    copy_blocks(
                        &*src,
                        &mut writer,
                        (entry.pos(), entry.len()),
                        (self.metadata.id, new_table_id),
                        pace,
                    )?;
                    copied += entry.len();
                }
            }

            // Inject the positional delete-bitmap. Its position is free (addressed
            // by name through the table of contents); appended after the copied
            // sections. Same uncompressed envelope as the other meta sections.
            writer.start(b"delete_bitmap")?;
            let encoded = delete_bitmap.encode();
            let at = super::writer::next_block_at(new_table_id, &writer);
            Block::write_into(
                &mut writer,
                &encoded,
                BlockIdentity {
                    table_id: new_table_id,
                    block_type: BlockType::DeleteBitmap,
                    dict_id: 0,
                    window_log: 0,
                },
                &BlockTransform::PLAIN,
                at,
            )?;

            let mut checksummed = writer.into_inner()?;
            let checksum = checksummed.checksum();
            let file = checksummed.inner_mut().get_mut();
            FsFile::sync_all_with(&**file, sync_mode)?;
            Ok((checksum, copied))
        })();

        match result {
            Ok((checksum, copied)) => Ok(Relocated {
                id: new_table_id,
                path: new_path,
                entry_durable: durable_entry,
                checksum,
                copied,
            }),
            Err(e) => {
                if let Err(remove) = out_fs.remove_file(&new_path) {
                    log::warn!(
                        "could not remove the failed relocation output {}: {remove}",
                        new_path.display()
                    );
                }
                Err(e)
            }
        }
    }

    /// Loads this segment's meta KV block, replaces the `table_id` entry's value
    /// with `new_table_id`, the two `descriptor#delete_bitmap_*` entries with
    /// values describing `delete_bitmap` (the NEW positional bitmap this
    /// relocation appends), and the `writing_below` entry with the copy's own
    /// `writing_below`, then re-encodes the payload (uncompressed, ready for a
    /// [`BlockType::Meta`] write). Every other meta field is preserved byte-exact.
    ///
    /// Without patching the delete-bitmap descriptors, the copied values would
    /// still describe the SOURCE's (absent / different) bitmap, so
    /// `verify_metadata_bounds` would flag the healthy relocated table during
    /// `repair_with_salvage`, refuse to re-emit it, and drop it.
    fn repoint_meta_block(
        &self,
        src: &dyn FsFile,
        (new_table_id, writing_below): (TableId, &[TableId]),
        delete_bitmap: &DeleteBitmap,
        pace: Option<&dyn super::util::ReadPacer>,
    ) -> crate::Result<Vec<u8>> {
        // Non-encrypted precondition (checked by the caller): PLAIN transform.
        let (block, _, _) = Block::from_file_issuing_paced(
            src,
            self.regions.metadata,
            BlockIdentity {
                table_id: self.metadata.id,
                block_type: BlockType::Meta,
                dict_id: 0,
                window_log: 0,
            },
            &BlockTransform::PLAIN,
            || {},
            pace,
            &mut 0,
        )?;
        let block = DataBlock::new(block);
        // Meta keys are lexicographic, so the default comparator orders them.
        let cmp = crate::comparator::default_comparator();
        let mut entries: Vec<InternalValue> = block
            .iter(cmp.clone())
            .map(|item| item.materialize(block.as_slice()))
            .collect();

        // New delete-bitmap descriptor values (same key, same fixed width, so the
        // in-place value swap preserves the meta block's sorted key order).
        let db_len_bytes = delete_bitmap.len().to_le_bytes();
        let db_hash_bytes = crate::hash::hash128(&delete_bitmap.encode()).to_le_bytes();
        let (mut id_patched, mut len_patched, mut hash_patched) = (false, false, false);
        for entry in &mut entries {
            match entry.key.user_key.as_ref() {
                b"table_id" => {
                    entry.value = UserValue::from(&new_table_id.to_le_bytes()[..]);
                    id_patched = true;
                }
                b"descriptor#delete_bitmap_len" => {
                    entry.value = UserValue::from(&db_len_bytes[..]);
                    len_patched = true;
                }
                b"descriptor#delete_bitmap_hash" => {
                    entry.value = UserValue::from(&db_hash_bytes[..]);
                    hash_patched = true;
                }
                _ => {}
            }
        }
        if !id_patched {
            return Err(crate::Error::InvalidHeader(
                "relocate: meta block missing table_id",
            ));
        }
        // A LEGACY columnar source predates `descriptor#delete_bitmap_len` /
        // `descriptor#delete_bitmap_hash` (the current writer always records both,
        // as `0` when it holds no bitmap). Failing here would permanently block a
        // single-input merge-on-read compaction that materializes range-tombstone
        // deletes on such a table (the planner has already chosen the relocation
        // fast path, with no copy-on-write fallback), so INSTEAD insert the
        // descriptors describing the appended bitmap. The meta block requires
        // sorted keys, so re-sort after inserting.
        if !len_patched {
            entries.push(InternalValue::from_components(
                b"descriptor#delete_bitmap_len".to_vec(),
                &db_len_bytes[..],
                0,
                crate::ValueType::Value,
            ));
        }
        if !hash_patched {
            entries.push(InternalValue::from_components(
                b"descriptor#delete_bitmap_hash".to_vec(),
                &db_hash_bytes[..],
                0,
                crate::ValueType::Value,
            ));
        }

        // A merge-on-read relocation IS a compaction output derived from
        // exactly the source, so stamp the copy's own complete single-output
        // run — lineage naming the source, no previous output (the head),
        // and the run's closing marker — in place of whatever lineage the
        // source carried. Copying the source's own (or absent) lineage
        // verbatim would leave a manifest-loss repair unable to associate
        // the copy with a still-present source (cleanup can lag the
        // install): both would be retained, and under a merge operator every
        // operand in the byte-identical data blocks would apply twice. No
        // transformed marker: the bitmap masks rows the source still holds
        // as live tombstones, so either dedup direction re-derives the same
        // materialized content.
        // The source's `writing_below` named the files its folder was writing
        // below the source; the copy records its own.
        entries.retain(|e| {
            !matches!(
                e.key.user_key.as_ref(),
                b"lineage"
                    | b"lineage_prev"
                    | b"lineage_transformed"
                    | b"lineage_last"
                    | b"writing_below"
            )
        });
        if !writing_below.is_empty() {
            let bytes: Vec<u8> = writing_below
                .iter()
                .flat_map(|id| id.to_le_bytes())
                .collect();
            entries.push(InternalValue::from_components(
                b"writing_below".to_vec(),
                &bytes[..],
                0,
                crate::ValueType::Value,
            ));
        }
        entries.push(InternalValue::from_components(
            b"lineage".to_vec(),
            &self.metadata.id.to_le_bytes()[..],
            0,
            crate::ValueType::Value,
        ));
        entries.push(InternalValue::from_components(
            b"lineage_last".to_vec(),
            &[][..],
            0,
            crate::ValueType::Value,
        ));
        // The copy's id is allocated at relocation time, but it keeps the
        // source's `recency` entry, which every table carries: its content
        // position is the source's, not its own newer id.
        entries.sort_by(|a, b| cmp.compare(a.key.user_key.as_ref(), b.key.user_key.as_ref()));

        // Same encode parameters the writer uses for the meta block
        // (restart interval 1, no hashing). The reader point-reads by key, so the
        // restart interval need not match the source; keeping it identical avoids
        // surprises.
        let mut payload = Vec::new();
        DataBlock::encode_into(&mut payload, &entries, 1, 0.0)?;
        Ok(payload)
    }
}

/// The sections a table writes as raw bytes rather than as blocks: everything
/// else is a run of blocks, each re-stamped as it is copied.
const RAW_SECTIONS: [&[u8]; 3] = [b"linked_blob_files", b"table_version", b"meta_separator"];

/// Copies the blocks of the section at `(pos, len)` of `src`, a table of id
/// `from_table`, into `writer` for table `to_table`: each block's payload is
/// written as it is, and its stored checksum is moved from where it was to
/// where it lands. `pace` is told how long each read takes.
///
/// The blocks are framed by their headers alone, which is exact for the
/// segments a relocation accepts: they carry no parity trailer.
///
/// # Errors
///
/// Returns [`crate::Error::InvalidHeader`] when the section does not tile into
/// whole blocks, and any I/O error.
fn copy_blocks<W: crate::io::Write + crate::io::Seek>(
    src: &dyn FsFile,
    writer: &mut crate::sfa::Writer<ChecksummedWriter<W>>,
    (pos, len): (u64, u64),
    (from_table, to_table): (TableId, TableId),
    pace: Option<&dyn super::util::ReadPacer>,
) -> crate::Result<()> {
    use super::block::{ChecksumAt, Header};
    use crate::coding::Decode;
    #[cfg(not(feature = "std"))]
    use crate::io::Write;
    #[cfg(feature = "std")]
    use std::io::Write;

    let end = pos
        .checked_add(len)
        .ok_or(crate::Error::InvalidHeader("relocate: section overflows"))?;
    let mut offset = pos;
    while offset < end {
        let head = crate::file::read_exact_paced(
            src,
            offset,
            usize::try_from((end - offset).min(Header::MAX_LEN as u64)).unwrap_or(Header::MAX_LEN),
            pace,
        )?;
        let header = Header::decode_from(&mut &head[..])?;
        let frame_len =
            Header::header_len(header.block_type) as u64 + u64::from(header.data_length);
        if frame_len > end - offset {
            return Err(crate::Error::InvalidHeader(
                "relocate: a block runs past its section",
            ));
        }
        // Only the header changes: re-stamp it, then stream the payload through
        // in bounded chunks, so a block near the size cap is never buffered.
        let header_len = Header::header_len(header.block_type);
        let mut header_bytes = head
            .get(..header_len)
            .ok_or(crate::Error::InvalidHeader(
                "relocate: truncated block header",
            ))?
            .to_vec();
        let to = ChecksumAt::block(to_table, header.block_type, writer.get_ref().position());
        Header::rebind_frame(
            &mut header_bytes,
            ChecksumAt::block(from_table, header.block_type, offset),
            to,
        )?;
        writer.write_all(&header_bytes)?;
        copy_section(
            src,
            writer,
            (offset + header_len as u64, u64::from(header.data_length)),
            pace,
        )?;
        offset += frame_len;
    }
    Ok(())
}

/// Copies `len` bytes from `src` at absolute offset `pos` into `writer`,
/// in bounded chunks so a large data section is never buffered whole. An
/// active `pace` has each chunk read a portion at a time and told how long
/// each portion took.
fn copy_section<W: crate::io::Write>(
    src: &dyn FsFile,
    writer: &mut W,
    (pos, len): (u64, u64),
    pace: Option<&dyn super::util::ReadPacer>,
) -> crate::Result<()> {
    let mut offset = pos;
    let end = pos + len;
    while offset < end {
        // `end - offset` is bounded by the section length (a u64 file size);
        // the `min` caps each read at COPY_CHUNK, so the cast cannot truncate.
        #[expect(clippy::cast_possible_truncation, reason = "capped at COPY_CHUNK")]
        let want = (end - offset).min(COPY_CHUNK as u64) as usize;
        let bytes = crate::file::read_exact_paced(src, offset, want, pace)?;
        writer.write_all(&bytes)?;
        offset += want as u64;
    }
    Ok(())
}

#[cfg(test)]
#[expect(
    clippy::unwrap_used,
    clippy::expect_used,
    reason = "tests intentionally unwrap setup failures to keep assertions focused"
)]
mod tests;