coordinode-lsm-tree 7.0.0

use super::*;
use crate::cache::Cache;
use crate::descriptor_table::DescriptorTable;
use crate::fs::StdFs;
use crate::path::Path;
use crate::table::Writer;
#[cfg(feature = "columnar")]
use crate::{SeqNo, hash::hash64};
use alloc::sync::Arc;
use test_log::test;

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

/// A new file at `path` for table `id` to be relocated into.
fn output_at(path: &Path, id: TableId) -> crate::Result<ClaimedFile> {
    use crate::fs::Fs;
    Ok(ClaimedFile {
        id,
        path: path.to_path_buf(),
        file: StdFs.open(path, &FsOpenOptions::new().write(true).create_new(true))?,
        durable_entry: false,
        writing: crate::table::file_pool::Writing::alone(id),
    })
}

fn recover_at(file: &Path, checksum: Checksum, table_id: TableId) -> crate::Result<Table> {
    #[cfg(feature = "metrics")]
    let metrics = Arc::new(Metrics::default());
    let mut params = crate::table::RecoverParams::new(
        file.to_path_buf(),
        checksum,
        table_id,
        Arc::new(StdFs),
        crate::comparator::default_comparator(),
        Arc::new(Cache::with_capacity_bytes(1_000_000)),
    );
    params.descriptor_table = Some(Arc::new(DescriptorTable::new(10)));
    #[cfg(feature = "metrics")]
    {
        params.metrics = metrics;
    }
    Table::recover(params)
}

#[cfg(feature = "columnar")]
#[test]
fn relocate_reuses_blocks_and_masks_deleted_rows() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let src_path = dir.path().join("src");
    let out_path = dir.path().join("out");

    let n = 96u32;
    // Positions follow write (= key) order.
    let deleted = [4u32, 7, 40, 95];

    // Source: a columnar segment with a zone map and NO deletes.
    let mut writer = Writer::new(src_path.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..n {
        let key = format!("k{i:04}").into_bytes();
        writer.write(InternalValue::from_components(
            key,
            b"val",
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, src_checksum) = writer.finish()?.expect("source table written");
    let source = recover_at(&src_path, src_checksum, 0)?;

    // Relocate into a new segment (id 1) carrying the bitmap.
    let mut bitmap = DeleteBitmap::new();
    for &row in &deleted {
        bitmap.insert(row);
    }
    let out_checksum = source
        .relocate_columnar_with_deletes(
            || output_at(&out_path, 1),
            &StdFs,
            &bitmap,
            SyncMode::Normal,
            None,
        )?
        .checksum;

    let relocated = recover_at(&out_path, out_checksum, 1)?;

    // (i) format flags + id preserved; (ii) deleted rows masked, live found.
    assert_eq!(relocated.metadata.id, 1, "meta carries the new table id");
    assert!(relocated.metadata.columnar, "columnar flag preserved");
    for i in 0..n {
        let key = format!("k{i:04}").into_bytes();
        let got = relocated.get(&key, SeqNo::MAX, hash64(&key))?;
        if deleted.contains(&i) {
            assert!(
                got.is_none(),
                "deleted row {i} must read absent after relocate"
            );
        } else {
            let got = got.expect("live row must survive relocate");
            assert_eq!(&*got.value, b"val", "live value preserved verbatim");
        }
    }
    Ok(())
}

/// A relocated copy records the files its own folder was writing below it
/// when it was finished, not the ones its source recorded: a repair reads
/// them to tell an output a power loss cut short from a table that lost its
/// bytes.
#[cfg(feature = "columnar")]
#[test]
fn a_relocated_copy_records_what_its_folder_was_writing_below_it() -> crate::Result<()> {
    use crate::table::file_pool::TableFilePools;

    let dir = tempfile::tempdir()?;
    let fs: Arc<dyn crate::fs::Fs> = Arc::new(StdFs);
    let pools = TableFilePools::new(crate::SequenceNumberCounter::default(), 0, SyncMode::Normal);
    let claim = || pools.claim(dir.path(), &fs, None, true);

    let earlier = claim()?;
    let source_file = claim()?;
    let (source_id, source_path) = (source_file.id, source_file.path.clone());
    let mut writer = Writer::into_claimed(source_file, 0, Arc::clone(&fs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..96u32 {
        writer.write(InternalValue::from_components(
            format!("k{i:04}").into_bytes(),
            b"val",
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, source_checksum) = writer.finish()?.expect("source table written");
    let source = recover_at(&source_path, source_checksum, source_id)?;
    assert_eq!(
        source.metadata.writing_below,
        vec![earlier.id],
        "self-check: the source records the file written below it"
    );
    drop(earlier);

    let pending = claim()?;
    let mut bitmap = DeleteBitmap::new();
    bitmap.insert(3);
    let relocated =
        source.relocate_columnar_with_deletes(claim, &StdFs, &bitmap, SyncMode::Normal, None)?;
    let copy = recover_at(&relocated.path, relocated.checksum, relocated.id)?;
    assert_eq!(copy.metadata.writing_below, vec![pending.id]);
    Ok(())
}

/// A relocation reads its whole source, section by section, as a merge would:
/// those reads report to the compaction's pacer, or a latency backoff on the
/// limiter it shares with a verification scan has nothing to steer by while
/// the relocation runs.
#[cfg(all(feature = "columnar", feature = "std"))]
#[test]
fn a_paced_relocation_reports_the_reads_of_its_source() -> crate::Result<()> {
    use crate::table::util::{Pacer, ReadPacer};
    use core::sync::atomic::{AtomicU64, Ordering};

    /// Sums the bytes of the positioned reads it is told about and counts
    /// every timed read.
    struct Recorder {
        bytes: AtomicU64,
        timed: AtomicU64,
    }
    impl ReadPacer for Recorder {
        fn active(&self) -> bool {
            true
        }
        fn pace(&self, _offset: u64, len: u64) {
            self.bytes.fetch_add(len, Ordering::Relaxed);
        }
        fn read_took(&self, _elapsed: core::time::Duration) {
            self.timed.fetch_add(1, Ordering::Relaxed);
        }
    }

    let dir = tempfile::tempdir()?;
    let src_path = dir.path().join("src");
    let out_path = dir.path().join("out");
    let mut writer = Writer::new(src_path.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..2_000u32 {
        writer.write(InternalValue::from_components(
            format!("k{i:06}").into_bytes(),
            format!("value-{i}-payload").into_bytes(),
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, src_checksum) = writer.finish()?.expect("source table written");
    let source = recover_at(&src_path, src_checksum, 0)?;
    let mut bitmap = DeleteBitmap::new();
    bitmap.insert(3);

    let recorder = Arc::new(Recorder {
        bytes: AtomicU64::new(0),
        timed: AtomicU64::new(0),
    });
    let pace: Pacer = recorder.clone();
    source.relocate_columnar_with_deletes(
        || output_at(&out_path, 1),
        &StdFs,
        &bitmap,
        SyncMode::Normal,
        Some(&pace),
    )?;

    let file_len = std::fs::metadata(&src_path)?.len();
    let paced = recorder.bytes.load(Ordering::Relaxed);
    assert!(
        paced * 2 > file_len,
        "the section copy was not paced: {paced} of {file_len} bytes",
    );
    assert!(
        recorder.timed.load(Ordering::Relaxed) > 0,
        "no read was timed"
    );
    Ok(())
}

/// A merge-on-read relocated columnar SST appends a NEW delete bitmap, so its
/// re-encoded meta must describe THAT bitmap: `verify_metadata_bounds` (the
/// repair-time forgery cross-check) authenticates `descriptor#delete_bitmap_len`
/// and `descriptor#delete_bitmap_hash` against the on-disk section. If the
/// relocation copied the source's (absent-bitmap) descriptors verbatim, the
/// check would flag the healthy table and `repair_with_salvage` would drop
/// it. The descriptors must be repointed to the appended bitmap.
#[cfg(feature = "columnar")]
#[test]
fn relocated_mor_table_passes_metadata_bounds_cross_check() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let src_path = dir.path().join("src");
    let out_path = dir.path().join("out");

    let n = 96u32;
    let mut writer = Writer::new(src_path.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..n {
        writer.write(InternalValue::from_components(
            format!("k{i:04}").into_bytes(),
            b"val",
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, src_checksum) = writer.finish()?.expect("source table written");
    let source = recover_at(&src_path, src_checksum, 0)?;

    let mut bitmap = DeleteBitmap::new();
    for &row in &[4u32, 7, 40, 95] {
        bitmap.insert(row);
    }
    let out_checksum = source
        .relocate_columnar_with_deletes(
            || output_at(&out_path, 1),
            &StdFs,
            &bitmap,
            SyncMode::Normal,
            None,
        )?
        .checksum;
    let relocated = recover_at(&out_path, out_checksum, 1)?;

    // The re-encoded meta describes the appended bitmap, so the forgery
    // cross-checks accept the healthy relocated table.
    if let Err((gate, e)) = relocated.verify_reconcile_gates(None, false) {
        panic!("a healthy relocated table must pass every gate, {gate:?} refused it: {e}");
    }
    Ok(())
}

/// A merge-on-read relocation is a compaction output derived from exactly its
/// source, and its meta must SAY so: lineage naming the source id, no
/// `lineage_prev` (the run's head), and `lineage_last` (its closing output) —
/// a complete single-output run. Copying the source's own (or absent)
/// lineage verbatim leaves a manifest-loss repair unable to associate the
/// copy with a still-present source (cleanup can lag the install), so BOTH
/// are retained — and under a merge operator every operand in the
/// byte-identical data blocks applies twice. The copy also states its
/// content recency: its id is allocated at relocation time, so without the
/// key its own (newer) id would masquerade as its content position.
#[cfg(feature = "columnar")]
#[test]
fn a_relocated_table_carries_a_complete_single_output_lineage() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let src_path = dir.path().join("src");
    let out_path = dir.path().join("out");

    let mut writer = Writer::new(src_path.clone(), 7, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..96u32 {
        writer.write(InternalValue::from_components(
            format!("k{i:04}").into_bytes(),
            b"val",
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, src_checksum) = writer.finish()?.expect("source table written");
    let source = recover_at(&src_path, src_checksum, 7)?;

    let mut bitmap = DeleteBitmap::new();
    bitmap.insert(4);
    let out_checksum = source
        .relocate_columnar_with_deletes(
            || output_at(&out_path, 9),
            &StdFs,
            &bitmap,
            SyncMode::Normal,
            None,
        )?
        .checksum;
    let relocated = recover_at(&out_path, out_checksum, 9)?;

    assert_eq!(
        relocated.metadata.lineage.as_deref(),
        Some(&[7u64][..]),
        "the copy's lineage names exactly its source",
    );
    assert_eq!(
        relocated.metadata.lineage_prev, None,
        "a single-output run has no previous output",
    );
    assert!(
        relocated.metadata.lineage_last,
        "the single output closes its run",
    );
    assert_eq!(
        relocated.metadata.recency, 7,
        "the copy's content position is its source's, not its own newer id",
    );
    Ok(())
}

/// A tight-space RESTRICTED view reads its live suffix only: `scan()` starts at
/// the bound and numbers rows from zero, while the relocation copies the whole
/// physical data section — punched prefix blocks included — and publishes the
/// copy WITHOUT the restriction. The bitmap would then mask shifted rows and
/// reads could reach the copied zero blocks, so block reuse must refuse and let
/// the caller rewrite copy-on-write.
#[cfg(feature = "columnar")]
#[test]
fn relocate_rejects_a_restricted_view() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let src_path = dir.path().join("src");

    let mut writer = Writer::new(src_path.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    for i in 0..32u32 {
        writer.write(InternalValue::from_components(
            format!("k{i:04}").into_bytes(),
            b"val",
            1,
            crate::ValueType::Value,
        ))?;
    }
    let (_, checksum) = writer.finish()?.expect("table written");
    let source = recover_at(&src_path, checksum, 0)?;
    let restricted = source.reopen_restricted(crate::UserKey::from(&b"k0016"[..]))?;

    let out_path = dir.path().join("out");
    let mut bitmap = DeleteBitmap::new();
    bitmap.insert(0);
    let err = restricted
        .relocate_columnar_with_deletes(
            || output_at(&out_path, 1),
            &StdFs,
            &bitmap,
            SyncMode::Normal,
            None,
        )
        .unwrap_err();
    assert!(
        matches!(err, crate::Error::FeatureUnsupported(_)),
        "a restricted view must be rejected, got {err:?}",
    );
    Ok(())
}

#[test]
fn relocate_rejects_row_major_segment() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let src_path = dir.path().join("src");

    // Row-major (no columnar): block reuse must refuse and let the caller CoW.
    let mut writer = Writer::new(src_path.clone(), 0, 0, Arc::new(StdFs))?.use_zone_map(true);
    writer.write(InternalValue::from_components(
        b"a",
        b"v",
        1,
        crate::ValueType::Value,
    ))?;
    let (_, checksum) = writer.finish()?.expect("table written");
    let source = recover_at(&src_path, checksum, 0)?;

    let out_path = dir.path().join("out");
    let mut bitmap = DeleteBitmap::new();
    bitmap.insert(0);
    let err = source
        .relocate_columnar_with_deletes(
            || output_at(&out_path, 1),
            &StdFs,
            &bitmap,
            SyncMode::Normal,
            None,
        )
        .unwrap_err();
    assert!(
        matches!(err, crate::Error::FeatureUnsupported(_)),
        "row-major segment must be rejected, got {err:?}",
    );
    Ok(())
}

/// A merge-on-read relocated columnar SST keeps its PHYSICAL rows (the
/// delete bitmap only masks them) and copies the source `linked_blob_files`
/// accounting verbatim — so the blob-link cross-check must derive its
/// accounting from the UNMASKED physical rows. Deriving from the masked
/// view marks a healthy relocated table corrupt, and repair-with-salvage
/// would then drop it (its retained range tombstones make it
/// unsalvageable).
#[cfg(feature = "columnar")]
#[test]
fn relocated_mor_table_passes_the_blob_link_cross_check() -> crate::Result<()> {
    use crate::blob_tree::handle::BlobIndirection;
    use crate::coding::Encode;
    use crate::vlog::ValueHandle;

    let dir = tempfile::tempdir()?;
    let src_path = dir.path().join("src");
    let out_path = dir.path().join("out");

    // A columnar segment whose every row is a blob indirection into file 5,
    // with the blob-link accounting the writer derives from those rows.
    let n = 8u32;
    let mut writer = Writer::new(src_path.clone(), 0, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_zone_map(true);
    let mut bytes_sum = 0u64;
    let mut on_disk_sum = 0u64;
    for i in 0..n {
        let ind = BlobIndirection {
            vhandle: ValueHandle {
                blob_file_id: 5,
                offset: u64::from(i) * 100,
                on_disk_size: 40,
            },
            size: 80,
        };
        bytes_sum += u64::from(ind.size);
        on_disk_sum += u64::from(ind.vhandle.on_disk_size);
        let mut val = alloc::vec::Vec::new();
        ind.encode_into(&mut val)?;
        writer.write(InternalValue::from_components(
            format!("k{i:04}").into_bytes(),
            val,
            1,
            crate::ValueType::Indirection,
        ))?;
    }
    writer.link_blob_file(crate::table::writer::LinkedFile {
        blob_file_id: 5,
        len: n as usize,
        bytes: bytes_sum,
        on_disk_bytes: on_disk_sum,
        first_key: b"k0000".into(),
        last_key: format!("k{:04}", n - 1).into_bytes().into(),
    });
    let (_, src_checksum) = writer.finish()?.expect("source table written");
    let source = recover_at(&src_path, src_checksum, 0)?;

    // Relocate with one indirection row MASKED: the physical rows (and the
    // copied accounting) still include it.
    let mut bitmap = DeleteBitmap::new();
    bitmap.insert(3);
    let out_checksum = source
        .relocate_columnar_with_deletes(
            || output_at(&out_path, 1),
            &StdFs,
            &bitmap,
            SyncMode::Normal,
            None,
        )?
        .checksum;
    let relocated = recover_at(&out_path, out_checksum, 1)?;

    relocated
        .verify_blob_links()
        .expect("a healthy relocated MoR table passes the blob-link cross-check");
    Ok(())
}