Function merge_for_retention

nodedb-array/src/query/retention.rs:170–314

Merge tile-versions for one `hilbert_prefix` according to the retention policy, operating at cell granularity. `versions` contains all [`TileEntry`] records for a single `hilbert_prefix` (the caller is responsible for grouping). They may arrive in any order; this function sorts internally. `horizon_ms` is the retention boundary expressed as an absolute `system_from_ms` timestamp: versions with `

(
    versions: &[TileEntry],
    reader: &SegmentReader<'_>,
    schema: &ArraySchema,
    horizon_ms: i64,
)

/// `system_from_ms` timestamp: versions with `system_from_ms < horizon_ms`
/// are outside the retention window.
pub fn merge_for_retention(
    versions: &[TileEntry],
    reader: &SegmentReader<'_>,
    schema: &ArraySchema,
    horizon_ms: i64,
) -> ArrayResult<RetentionMergeResult> {
    if versions.is_empty() {
        return Ok(RetentionMergeResult {
            ceiling_tile: None,
            keep_inhorizon: Vec::new(),
            dropped_tile_ids: Vec::new(),
            cells_carried_forward: 0,
        });
    }

    // Partition into inside / outside by system_from_ms.
    let mut inside: Vec<&TileEntry> = Vec::new();
    let mut outside: Vec<&TileEntry> = Vec::new();
    for entry in versions {
        if entry.tile_id.system_from_ms >= horizon_ms {
            inside.push(entry);
        } else {
            outside.push(entry);
        }
    }

    // Collect TileIds for in-horizon pass-through.
    let keep_inhorizon: Vec<TileId> = inside.iter().map(|e| e.tile_id).collect();

    // If nothing is outside the horizon, there is nothing to merge.
    if outside.is_empty() {
        return Ok(RetentionMergeResult {
            ceiling_tile: None,
            keep_inhorizon,
            dropped_tile_ids: Vec::new(),
            cells_carried_forward: 0,
        });
    }

    // Build the set of coord keys already covered by inside-horizon versions.
    // Any coord present in any inside tile (regardless of RowKind) supersedes
    // what the ceiling would contribute for that coord.
    let mut inhorizon_coords: HashSet<Vec<u8>> = HashSet::new();
    for entry in &inside {
        let tile_idx = find_tile_index(reader, entry.tile_id)?;
        let payload = reader.read_tile(tile_idx)?;
        if let TilePayload::Sparse(ref tile) = payload {
            for trow in decode_sparse_rows(tile)? {
                inhorizon_coords.insert(trow.coord_key);
            }
        }
        // Dense tiles inside horizon contribute no coords to the exclusion set
        // for the ceiling (dense tiles don't participate in sparse retention
        // merge; their coords are not relevant here).
    }

    // Walk outside-horizon versions newest → oldest, accumulating the ceiling.
    // For each coord not yet seen, the first (newest) occurrence wins.