Function apply_retention

nodedb/src/engine/array/compaction/merger.rs:167–256

Apply cell-level retention to the merged tile map. For each Hilbert prefix group: - Tile versions with `system_from_ms >= horizon_ms` are in-horizon and pass through unchanged. - Tile versions with `system_from_ms < horizon_ms` are out-of-horizon. Their cells are collapsed into a single synthetic ceiling tile: - For each coordinate, the newest out-of-horizon row wins. - Coordinates already covere

(
    merged: BTreeMap<TileId, MergedTile>,
    schema: &ArraySchema,
    horizon_ms: i64,
)

/// strictly before all in-horizon tiles within the same prefix in the output
/// segment (TileId order is `(prefix, system_from_ms)` ascending).
fn apply_retention(
    merged: BTreeMap<TileId, MergedTile>,
    schema: &ArraySchema,
    horizon_ms: i64,
) -> Result<BTreeMap<TileId, MergedTile>, CompactionError> {
    // Group tile versions by hilbert_prefix.
    let mut by_prefix: HashMap<u64, Vec<(TileId, MergedTile)>> = HashMap::new();
    for (tile_id, mt) in merged {
        by_prefix
            .entry(tile_id.hilbert_prefix)
            .or_default()
            .push((tile_id, mt));
    }

    let mut out: BTreeMap<TileId, MergedTile> = BTreeMap::new();

    for (prefix, versions) in by_prefix {
        // Partition into inside-horizon and outside-horizon.
        let mut inside: Vec<(TileId, MergedTile)> = Vec::new();
        let mut outside: Vec<(TileId, MergedTile)> = Vec::new();
        for (tile_id, mt) in versions {
            if tile_id.system_from_ms >= horizon_ms {
                inside.push((tile_id, mt));
            } else {
                outside.push((tile_id, mt));
            }
        }

        // Collect coord keys present in any in-horizon version so they can
        // be excluded from the ceiling (the in-horizon version supersedes).
        let mut inhorizon_coord_keys: HashSet<Vec<u8>> = HashSet::new();
        for (_tile_id, mt) in &inside {
            for row in &mt.rows {
                let key = encode_coord_key(&row.coord)?;
                inhorizon_coord_keys.insert(key);
            }
        }

        // Pass in-horizon tiles through unchanged.
        for (tile_id, mt) in inside {
            out.insert(tile_id, mt);
        }

        // Nothing to collapse.
        if outside.is_empty() {
            continue;
        }

        // Sort outside-horizon versions newest → oldest.
        outside.sort_by_key(|(tid, _)| std::cmp::Reverse(tid.system_from_ms));

        // Build ceiling: coord → (newest out-of-horizon row).
        // Maps encoded coord key → MergedRow.
        let mut ceiling_rows: HashMap<Vec<u8>, MergedRow> = HashMap::new();
        for (_tile_id, mt) in outside {
            for row in mt.rows {
                let key = encode_coord_key(&row.coord)?;
                // Skip coords already covered by in-horizon versions.