Method scan_tiles_at

nodedb/src/engine/array/store/catalog.rs:255–364

Bitemporal scan: resolve the ceiling for every cell coordinate at the given `system_as_of` and optional `valid_at_ms` point. Returns one `(hilbert_prefix, SparseTile)` pair per prefix that has at least one `Live` cell after ceiling resolution. Tombstoned and erased coords are omitted. Also returns `truncated_before_horizon`: `true` when the store contains at least one tile version but the `syste

(
        &self,
        system_as_of: i64,
        valid_at_ms: Option<i64>,
    )

    /// at least one tile version but the `system_as_of` cutoff is below every
    /// version's `system_from_ms` (i.e., the cutoff predates all data).
    pub fn scan_tiles_at(
        &self,
        system_as_of: i64,
        valid_at_ms: Option<i64>,
    ) -> Result<(Vec<(u64, SparseTile)>, bool), nodedb_array::ArrayError> {
        let params = CeilingParams {
            system_as_of,
            valid_at_ms,
        };

        // Collect all distinct hilbert_prefix values present in any version.
        let mut all_prefixes: HashSet<u64> = HashSet::new();
        for h in self.segments.values() {
            let reader = h.reader();
            for entry in reader.tiles() {
                all_prefixes.insert(entry.tile_id.hilbert_prefix);
            }
        }
        for (tile_id, _) in self.memtable.iter() {
            all_prefixes.insert(tile_id.hilbert_prefix);
        }

        // Did any version exist at all in the store?
        let any_versions = !all_prefixes.is_empty();

        let mut out: Vec<(u64, SparseTile)> = Vec::new();
        let mut any_qualifying = false;

        for prefix in all_prefixes {
            // Collect all distinct coords across every version for this prefix.
            let mut coords: Vec<Vec<CoordValue>> = Vec::new();

            // From memtable versions.
            for (_, buf) in self.memtable.iter_tile_versions(prefix, i64::MAX) {
                for coord_key in buf.all_coord_keys() {
                    let coord = Memtable::decode_coord_key(coord_key)?;
                    if !coords.contains(&coord) {
                        coords.push(coord);
                    }
                }
            }

            // From segment versions (newest-first per segment, but we only need
            // coords here so order within a segment doesn't matter).
            for h in self.segments.values() {
                let reader = h.reader();
                for item in reader.iter_tile_versions(prefix, i64::MAX)? {
                    let (_, tile_payload) = item?;
                    if let TilePayload::Sparse(sparse) = &tile_payload {
                        let n = sparse.nnz() as usize;
                        for row in 0..n {
                            let coord: Vec<CoordValue> = sparse
                                .dim_dicts
                                .iter()
                                .map(|d| d.values[d.indices[row] as usize].clone())
                                .collect();
                            if !coords.contains(&coord) {
                                coords.push(coord);