linkrs-storage 0.1.0

use std::collections::HashMap;
use std::ops::Range;
use std::sync::Arc;

use crate::cursor::{EdgeColumnBatch, EdgeCursor, ScanOptions, ShardSlice};
use crate::edge::edge_table::core::EdgeStore;
use crate::edge::Nbr;
use crate::engine::data_store::EdgeTableKey;
use crate::engine::graph_storage::context::GraphStorageContext;
use crate::engine::graph_storage::ops::endpoint_label_id;
use crate::vertex::vertex_table::sharded::routing::ShardLayout;
use crate::vertex::ShardedVertexTable;
use linkrs_core::types::{LabelId, Timestamp, VertexId};
use linkrs_core::{Edge, StorageError, StorageResult, Value};

// ---------------------------------------------------------------------------
// GraphEdgeCursor — truly lazy CSR-scanning edge cursor
// ---------------------------------------------------------------------------

#[derive(Clone)]
struct TableDef {
    key: EdgeTableKey,
    tbl_src: LabelId,
    tbl_dst: LabelId,
}

#[derive(Clone)]
struct TargetDef {
    edge_type_name: String,
    tables: Vec<TableDef>,
}

#[derive(Clone, Debug)]
enum TablePhase {
    Mutable,
    Done,
}

#[derive(Clone, Debug)]
struct TableScanState {
    phase: TablePhase,
    /// Resumable position: group to resume from plus physical entries to skip
    /// within that group. Whole groups before `resume_group` are never
    /// revisited, so multi-batch scans stay linear instead of replaying from
    /// the table start on every batch.
    resume_group: usize,
    skip_in_group: usize,
}

impl TableScanState {
    fn new() -> Self {
        Self {
            phase: TablePhase::Mutable,
            resume_group: 0,
            skip_in_group: 0,
        }
    }
}

struct WalkContext<'a> {
    store: &'a EdgeStore,
    ctx: &'a GraphStorageContext,
    td: &'a TableDef,
}

/// Owned slice filter for one source table: the cursor's slice paired
/// with that table's snapshotted ownership space. `None` walks the
/// table unfiltered (no slice configured, or no vertex table behind
/// the label to decode ownership against). Free function (not a method)
/// so walk call sites with active field borrows can pass the two pieces
/// without borrowing the whole cursor.
fn slice_filter_for(
    slice: &Option<ShardSlice>,
    spaces: &HashMap<LabelId, EdgeSliceSpace>,
    src_label: LabelId,
) -> Option<(ShardSlice, EdgeSliceSpace)> {
    let owned = (*slice)?;
    let space = spaces.get(&src_label)?.clone();
    Some((owned, space))
}

struct ScanConfig<'a> {
    ts: Timestamp,
    src_id_range: &'a Option<Range<i64>>,
    predicate: &'a [crate::cursor::ScanPredicate],
    /// Owned slice filter for this table's source space (`None` walks the
    /// whole table). Owned (not borrowed) so table switches never fight the
    /// walk's borrows; the payload is one small vector per source label.
    slice_filter: Option<(ShardSlice, EdgeSliceSpace)>,
}

/// Source-id ownership for one partitioned edge scan.
///
/// Each entry of `chunks` is the slice-owned local-id chunk of one source
/// shard under `layout`, snapshotted when the table is first visited. Global
/// source ids decode arithmetically to `(shard, local)` pairs, so ownership
/// needs no table access and whole single-segment groups can be skipped
/// without touching any edge.
#[derive(Clone, Debug)]
struct EdgeSliceSpace {
    layout: ShardLayout,
    chunks: Vec<Range<u32>>,
}

/// Whether a global source id belongs to the slice's owned chunks.
/// `None` marks an undecodable id, counted as malformed by the caller.
fn slice_owns_global(space: &EdgeSliceSpace, global: u32) -> Option<bool> {
    let seg_bits = space.layout.segment_slots_bits;
    let segment = global >> seg_bits;
    if segment >= space.layout.total_segments {
        return None;
    }
    let shards = space.layout.num_shards as u32;
    let shard = (segment % shards) as usize;
    let local = (segment / shards) << seg_bits | (global & space.layout.segment_slots_mask());
    Some(
        space
            .chunks
            .get(shard)
            .is_some_and(|chunk| chunk.contains(&local)),
    )
}

/// Whether a whole CSR group provably holds no owned source id. Only
/// single-segment groups qualify: a group spanning segments can straddle a
/// chunk boundary, and those fall back to the per-edge ownership test.
fn group_outside_slice(space: &EdgeSliceSpace, group_base: u32, group_size: u32) -> bool {
    let seg_bits = space.layout.segment_slots_bits as u64;
    let first = (group_base as u64) >> seg_bits;
    let last = ((group_base as u64) + (group_size as u64) - 1) >> seg_bits;
    if first != last {
        return false;
    }
    let shards = space.layout.num_shards as u64;
    let shard = (first % shards) as usize;
    let Some(chunk) = space.chunks.get(shard) else {
        return false;
    };
    let seg_local_base = (first / shards) << seg_bits;
    let seg_local_end = seg_local_base + (1u64 << seg_bits);
    seg_local_end <= chunk.start as u64 || seg_local_base >= chunk.end as u64
}

pub(crate) struct GraphEdgeCursor {
    ctx: Arc<GraphStorageContext>,
    limit: Option<usize>,
    offset_remaining: usize,
    emitted: usize,
    src_id_range: Option<Range<i64>>,
    slice: Option<ShardSlice>,
    /// Per-source-label slice ownership, snapshotted when each source table
    /// is first visited. Labels without a vertex table stay absent and walk
    /// unfiltered, preserving the old per-edge resolution behavior there.
    slice_spaces: HashMap<LabelId, EdgeSliceSpace>,
    projection: Option<Vec<Arc<str>>>,
    /// Conjunctive predicates evaluated on decoded properties before
    /// offset/limit accounting; a pure pre-filter.
    predicate: Vec<crate::cursor::ScanPredicate>,
    /// Property names referenced by `predicate`; they are decoded even when
    /// absent from the projection so predicates can be evaluated.
    predicate_columns: Vec<Arc<str>>,
    /// Topology-only scan: resolve edge headers with zero property decode.
    /// Set from `ScanOptions`; only valid with an empty projection and no
    /// pushed predicates (rejected in `new`).
    topology_only: bool,
    exhausted: bool,
    /// Malformed/unparseable entries skipped so far, exposed through
    /// `EdgeCursor::malformed_skipped` for diagnostics.
    malformed_skipped: u64,
    ts: Timestamp,
    targets: Vec<TargetDef>,
    target_idx: usize,
    table_idx: usize,
    table_state: TableScanState,
}

impl std::fmt::Debug for GraphEdgeCursor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GraphEdgeCursor")
            .field("target_idx", &self.target_idx)
            .field("table_idx", &self.table_idx)
            .field("phase", &self.table_state.phase)
            .field("limit", &self.limit)
            .field("offset_remaining", &self.offset_remaining)
            .field("emitted", &self.emitted)
            .field("malformed_skipped", &self.malformed_skipped)
            .field("exhausted", &self.exhausted)
            .finish()
    }
}

impl GraphEdgeCursor {
    pub fn new(
        ctx: Arc<GraphStorageContext>,
        space: &str,
        options: &ScanOptions,
    ) -> StorageResult<Self> {
        let ts = options
            .read_timestamp
            .unwrap_or_else(|| ctx.get_read_timestamp());
        let allowlist = options.edge_table_allowlist.as_deref();
        let targets = if let Some(ref et) = options.edge_type {
            vec![build_target(&ctx, space, et, allowlist)?]
        } else {
            let edge_types = ctx.schema_manager().list_edge_types(space)?;
            edge_types
                .into_iter()
                .map(|et| build_target(&ctx, space, &et.edge_type_name, allowlist))
                .collect::<StorageResult<Vec<_>>>()?
        };

        let predicate = options.predicate.clone().unwrap_or_default();
        let mut predicate_columns: Vec<Arc<str>> = Vec::new();
        for pred in &predicate {
            collect_predicate_columns(pred, &mut predicate_columns);
        }

        if options.topology_only {
            let projection_non_empty = options.projection.as_ref().is_some_and(|p| !p.is_empty());
            if projection_non_empty || !predicate.is_empty() {
                return Err(StorageError::invalid_operation(
                    "topology_only requires an empty projection and no predicates",
                ));
            }
        }

        // A slice selects one of a non-empty division; anything else
        // addresses no well-defined owned set.
        if options.edge_src_slice.is_some_and(|slice| {
            slice.partition_count == 0 || slice.partition_id >= slice.partition_count
        }) {
            return Err(StorageError::invalid_operation(
                "edge_src_slice requires partition_id < partition_count with a non-zero count",
            ));
        }

        let mut slice_spaces = HashMap::new();
        if let Some(slice) = options.edge_src_slice {
            let mut labels: Vec<LabelId> = targets
                .iter()
                .flat_map(|target| target.tables.iter().map(|table| table.tbl_src))
                .collect();
            labels.sort_unstable();
            labels.dedup();
            ctx.data_store().with_vertex_tables(|tables| {
                for label in labels {
                    let Some(table) = tables.get(&label) else {
                        continue;
                    };
                    let universes = table.shard_universes();
                    slice_spaces.insert(
                        label,
                        EdgeSliceSpace {
                            layout: table.layout(),
                            chunks: universes
                                .iter()
                                .map(|universe| ShardedVertexTable::slice_chunk(*universe, &slice))
                                .collect(),
                        },
                    );
                }
            });
        }

        Ok(Self {
            ctx,
            limit: options.limit,
            offset_remaining: options.offset,
            emitted: 0,
            src_id_range: options.edge_src_id_range.clone(),
            slice: options.edge_src_slice,
            slice_spaces,
            projection: options
                .projection
                .as_ref()
                .map(|p| p.iter().map(|rp| rp.name.clone()).collect()),
            predicate,
            predicate_columns,
            topology_only: options.topology_only,
            exhausted: targets.is_empty(),
            malformed_skipped: 0,
            ts,
            targets,
            target_idx: 0,
            table_idx: 0,
            table_state: TableScanState::new(),
        })
    }

    /// Native columnar batch collection: topology plus typed property
    /// columns decoded straight from the stores.
    ///
    /// Reuses the row-path visibility gate and segment prune, skips decoded
    /// groups without live edges, and degrades per column to `General` on
    /// type mismatch instead of falling back to row transpose.
    fn collect_column_batch(
        &mut self,
        prop_names: &[std::sync::Arc<str>],
        batch_size: usize,
    ) -> Result<EdgeColumnBatch, StorageError> {
        use crate::cursor::ColumnValues;

        let fetch_names: Vec<std::sync::Arc<str>> = if prop_names.is_empty() {
            Vec::new()
        } else {
            let mut run = prop_names.to_vec();
            for extra in self.predicate_columns.iter() {
                if !run.iter().any(|c| **c == **extra) {
                    run.push(extra.clone());
                }
            }
            run
        };
        let fetch_opt: Option<Vec<std::sync::Arc<str>>> = if prop_names.is_empty() {
            None
        } else {
            Some(fetch_names.clone())
        };

        let ctx = Arc::clone(&self.ctx);
        let ts = self.ts;
        let predicate = self.predicate.clone();
        let src_id_range = self.src_id_range.clone();
        let targets = self.targets.clone();

        let mut out_srcs: Vec<VertexId> = Vec::new();
        let mut out_dsts: Vec<VertexId> = Vec::new();
        let mut out_types: Vec<String> = Vec::new();
        let mut out_ranks: Vec<i64> = Vec::new();
        let mut union_names: Vec<std::sync::Arc<str>> = Vec::new();
        let mut union_columns: Vec<ColumnValues> = Vec::new();

        let data_store = ctx.data_store().clone();
        data_store.with_edge_tables(|edge_tables| {
            while out_srcs.len() < batch_size && !self.exhausted {
                if self.target_idx >= targets.len() {
                    self.exhausted = true;
                    break;
                }
                let target = &targets[self.target_idx];
                if self.table_idx >= target.tables.len() {
                    self.target_idx += 1;
                    self.table_idx = 0;
                    self.table_state = TableScanState::new();
                    continue;
                }
                let td = &target.tables[self.table_idx];
                let arc = match edge_tables.get(&td.key) {
                    Some(a) => a.clone(),
                    None => {
                        self.table_idx += 1;
                        self.table_state = TableScanState::new();
                        continue;
                    }
                };
                let guard = arc.read();
                let store: &EdgeStore = &guard;

                // Missing out leg reads as empty by contract, but distinguish
                // it from a genuinely empty stored leg via the capability note.
                if !store.is_direction_available(true) {
                    if let Some(note) = store.direction_note(true) {
                        log::debug!("{}", note);
                    }
                    self.table_idx += 1;
                    self.table_state = TableScanState::new();
                    continue;
                }

                if !matches!(self.table_state.phase, TablePhase::Mutable) {
                    self.table_idx += 1;
                    self.table_state = TableScanState::new();
                    continue;
                }

                let mut raw_src: Vec<u32> = Vec::new();
                let mut raw_dst: Vec<u32> = Vec::new();
                let mut raw_rank: Vec<i64> = Vec::new();
                let mut raw_edge: Vec<linkrs_core::types::EdgeId> = Vec::new();
                let mut raw_row: Vec<u32> = Vec::new();
                let mut no_stop = || false;
                let slice_filter = slice_filter_for(&self.slice, &self.slice_spaces, td.tbl_src);
                let table_done = walk_mutable_entries(
                    &WalkContext {
                        store,
                        ctx: &ctx,
                        td,
                    },
                    &ScanConfig {
                        ts,
                        src_id_range: &src_id_range,
                        predicate: &predicate,
                        slice_filter,
                    },
                    &mut self.table_state,
                    &mut self.malformed_skipped,
                    &mut no_stop,
                    |global, nbr, _| {
                        raw_src.push(global);
                        raw_dst.push(nbr.endpoint);
                        raw_rank.push(nbr.rank);
                        raw_edge.push(nbr.edge_id);
                        raw_row.push(global);
                        if out_srcs.len() + raw_src.len() >= batch_size {
                            EntryVerdict::BatchFull
                        } else {
                            EntryVerdict::Take
                        }
                    },
                );

                if raw_src.is_empty() {
                    if table_done {
                        self.table_state.phase = TablePhase::Done;
                    }
                    if matches!(self.table_state.phase, TablePhase::Done) {
                        self.table_idx += 1;
                        self.table_state = TableScanState::new();
                        continue;
                    }
                    continue;
                }

                if self.topology_only {
                    if self.offset_remaining > 0 {
                        let skip = self.offset_remaining.min(raw_src.len());
                        raw_src.drain(..skip);
                        raw_dst.drain(..skip);
                        raw_rank.drain(..skip);
                        raw_edge.drain(..skip);
                        raw_row.drain(..skip);
                        self.offset_remaining -= skip;
                    }
                    if raw_src.is_empty() {
                        if table_done {
                            self.table_state.phase = TablePhase::Done;
                        }
                        continue;
                    }
                    for (src, dst, rank) in raw_src
                        .iter()
                        .zip(raw_dst.iter())
                        .zip(raw_rank.iter())
                        .map(|((s, d), r)| (*s, *d, *r))
                    {
                        let src_ext = resolve_vertex_id(&ctx, src, td.tbl_src, ts)
                            .unwrap_or(VertexId::from_u32(src));
                        let dst_ext = resolve_vertex_id(&ctx, dst, td.tbl_dst, ts)
                            .unwrap_or(VertexId::from_u32(dst));
                        out_srcs.push(src_ext);
                        out_dsts.push(dst_ext);
                        out_types.push(target.edge_type_name.clone());
                        out_ranks.push(rank);
                    }
                    if table_done {
                        self.table_state.phase = TablePhase::Done;
                    }
                    if matches!(self.table_state.phase, TablePhase::Done) {
                        self.table_idx += 1;
                        self.table_state = TableScanState::new();
                    }
                    if let Some(limit) = self.limit {
                        if self.emitted + out_srcs.len() >= limit {
                            self.exhausted = true;
                            break;
                        }
                    }
                    continue;
                }

                let decoded = store.typed_columns_for_edges_assume_visible(
                    &raw_edge,
                    &raw_row,
                    ts,
                    fetch_opt.as_deref(),
                );
                let mut col_map: std::collections::HashMap<std::sync::Arc<str>, ColumnValues> =
                    std::collections::HashMap::new();
                for (name, values) in decoded {
                    col_map.insert(name, values);
                }
                let run_names: Vec<std::sync::Arc<str>> = if prop_names.is_empty() {
                    let mut names: Vec<std::sync::Arc<str>> =
                        col_map.keys().cloned().collect::<Vec<_>>();
                    names.sort();
                    names
                } else {
                    fetch_names.clone()
                };
                let mut run_columns: Vec<ColumnValues> = run_names
                    .iter()
                    .map(|name| {
                        col_map
                            .remove(name)
                            .unwrap_or_else(|| ColumnValues::General(vec![None; raw_src.len()]))
                    })
                    .collect();

                if !predicate.is_empty() {
                    let mut selection: Vec<usize> = Vec::new();
                    for row in 0..raw_src.len() {
                        let mut ok = true;
                        for pred in predicate.iter() {
                            let col_name = pred.column();
                            let idx_opt = run_names.iter().position(|n| *n.as_ref() == *col_name);
                            match idx_opt {
                                Some(idx) => {
                                    if !pred.matches_column(&run_columns[idx], row) {
                                        ok = false;
                                        break;
                                    }
                                }
                                None => {
                                    ok = false;
                                    break;
                                }
                            }
                        }
                        if ok {
                            selection.push(row);
                        }
                    }
                    if selection.len() != raw_src.len() {
                        let mut next_src = Vec::with_capacity(selection.len());
                        let mut next_dst = Vec::with_capacity(selection.len());
                        let mut next_rank = Vec::with_capacity(selection.len());
                        let mut next_edge = Vec::with_capacity(selection.len());
                        let mut next_row = Vec::with_capacity(selection.len());
                        for &i in &selection {
                            next_src.push(raw_src[i]);
                            next_dst.push(raw_dst[i]);
                            next_rank.push(raw_rank[i]);
                            next_edge.push(raw_edge[i]);
                            next_row.push(raw_row[i]);
                        }
                        raw_src = next_src;
                        raw_dst = next_dst;
                        raw_rank = next_rank;
                        raw_edge = next_edge;
                        raw_row = next_row;
                        for col in run_columns.iter_mut() {
                            col.select(&selection);
                        }
                    }
                }

                if !raw_src.is_empty() && self.offset_remaining > 0 {
                    let skip = self.offset_remaining.min(raw_src.len());
                    let suffix: Vec<usize> = (skip..raw_src.len()).collect();
                    for col in run_columns.iter_mut() {
                        col.select(&suffix);
                    }
                    raw_src.drain(..skip);
                    raw_dst.drain(..skip);
                    raw_rank.drain(..skip);
                    raw_edge.drain(..skip);
                    raw_row.drain(..skip);
                    self.offset_remaining -= skip;
                }

                if raw_src.is_empty() {
                    if table_done {
                        self.table_state.phase = TablePhase::Done;
                    }
                    continue;
                }

                for (name, _) in run_names.iter().zip(run_columns.iter()) {
                    if !union_names.contains(name) {
                        union_names.push(name.clone());
                        let mut new_column = ColumnValues::General(Vec::new());
                        new_column.append_nulls(out_srcs.len());
                        union_columns.push(new_column);
                    }
                }
                for (idx, uname) in union_names.iter().enumerate() {
                    match run_names.iter().position(|n| n == uname) {
                        Some(run_idx) => {
                            let run_column = run_columns[run_idx].clone();
                            union_columns[idx].append(run_column);
                        }
                        None => {
                            union_columns[idx].append_nulls(raw_src.len());
                        }
                    }
                }
                for (src, dst, rank) in raw_src
                    .iter()
                    .zip(raw_dst.iter())
                    .zip(raw_rank.iter())
                    .map(|((s, d), r)| (*s, *d, *r))
                {
                    let src_internal = src;
                    let dst_internal = dst;
                    let src_ext = resolve_vertex_id(&ctx, src_internal, td.tbl_src, ts)
                        .unwrap_or(VertexId::from_u32(src_internal));
                    let dst_ext = resolve_vertex_id(&ctx, dst_internal, td.tbl_dst, ts)
                        .unwrap_or(VertexId::from_u32(dst_internal));
                    out_srcs.push(src_ext);
                    out_dsts.push(dst_ext);
                    out_types.push(target.edge_type_name.clone());
                    out_ranks.push(rank);
                }

                if table_done {
                    self.table_state.phase = TablePhase::Done;
                }
                if matches!(self.table_state.phase, TablePhase::Done) {
                    self.table_idx += 1;
                    self.table_state = TableScanState::new();
                }
                if let Some(limit) = self.limit {
                    if self.emitted + out_srcs.len() >= limit {
                        self.exhausted = true;
                        break;
                    }
                }
            }
        });

        let mut batch = Self::assemble_edge_column_batch(
            prop_names,
            union_names,
            union_columns,
            out_srcs,
            out_dsts,
            out_types,
            out_ranks,
        );
        if let Some(limit) = self.limit {
            let remaining = limit.saturating_sub(self.emitted);
            if batch.len() > remaining {
                batch.srcs.truncate(remaining);
                batch.dsts.truncate(remaining);
                batch.edge_types.truncate(remaining);
                batch.rankings.truncate(remaining);
                for column in batch.columns.iter_mut() {
                    column.values.truncate(remaining);
                }
                self.emitted += remaining;
                self.exhausted = true;
            } else {
                self.emitted += batch.len();
            }
        }
        Ok(batch)
    }

    /// Assemble the final [`EdgeColumnBatch`] from decoded union columns.
    fn assemble_edge_column_batch(
        prop_names: &[std::sync::Arc<str>],
        union_names: Vec<std::sync::Arc<str>>,
        columns: Vec<crate::cursor::ColumnValues>,
        srcs: Vec<VertexId>,
        dsts: Vec<VertexId>,
        edge_types: Vec<String>,
        rankings: Vec<i64>,
    ) -> EdgeColumnBatch {
        use crate::cursor::PropertyColumn;
        use linkrs_core::types::DataType;
        let output_columns: Vec<PropertyColumn> = if prop_names.is_empty() {
            union_names
                .into_iter()
                .zip(columns)
                .map(|(name, values)| PropertyColumn {
                    name,
                    data_type: DataType::Empty,
                    values,
                })
                .collect()
        } else {
            let row_count = srcs.len();
            prop_names
                .iter()
                .map(|name| {
                    let values = union_names
                        .iter()
                        .position(|n| n.as_ref() == name.as_ref())
                        .and_then(|index| columns.get(index).cloned())
                        .unwrap_or_else(|| {
                            crate::cursor::ColumnValues::General(vec![None; row_count])
                        });
                    PropertyColumn {
                        name: name.clone(),
                        data_type: DataType::Empty,
                        values,
                    }
                })
                .collect()
        };
        EdgeColumnBatch {
            srcs,
            dsts,
            edge_types,
            rankings,
            columns: output_columns,
        }
    }
}

impl EdgeCursor for GraphEdgeCursor {
    fn malformed_skipped(&self) -> u64 {
        self.malformed_skipped
    }

    fn next_column_batch(
        &mut self,
        prop_names: &[std::sync::Arc<str>],
        batch_size: usize,
    ) -> Result<EdgeColumnBatch, StorageError> {
        if self.exhausted {
            return Ok(EdgeColumnBatch::empty());
        }
        let batch_size = batch_size.max(1);
        loop {
            let batch = self.collect_column_batch(prop_names, batch_size)?;
            if !batch.is_empty() || self.exhausted {
                return Ok(batch);
            }
        }
    }

    fn next_batch(&mut self, batch_size: usize) -> Result<Vec<Edge>, StorageError> {
        if self.exhausted {
            return Ok(Vec::new());
        }

        let batch_size = batch_size.max(1);
        let mut candidates = Vec::new();

        let ctx = &*self.ctx;
        let ts = self.ts;
        let limit = self.limit;
        let src_id_range = &self.src_id_range;
        let projection = &self.projection;
        let predicate = &self.predicate;
        let predicate_columns = &self.predicate_columns;
        let targets = &self.targets;
        let target_idx = &mut self.target_idx;
        let table_idx = &mut self.table_idx;
        let table_state = &mut self.table_state;
        let emitted = &mut self.emitted;
        let offset_remaining = &mut self.offset_remaining;
        let exhausted = &mut self.exhausted;
        let malformed = &mut self.malformed_skipped;
        let slice = &self.slice;
        let slice_spaces = &self.slice_spaces;

        let data_store = ctx.data_store().clone();
        data_store.with_edge_tables(|edge_tables| {
            'outer: while candidates.len() < batch_size {
                if *target_idx >= targets.len() {
                    *exhausted = true;
                    break 'outer;
                }

                let target = &targets[*target_idx];

                if *table_idx >= target.tables.len() {
                    *target_idx += 1;
                    *table_idx = 0;
                    *table_state = TableScanState::new();
                    continue 'outer;
                }

                let td = &target.tables[*table_idx];
                let arc = match edge_tables.get(&td.key) {
                    Some(a) => a.clone(),
                    None => {
                        *table_idx += 1;
                        *table_state = TableScanState::new();
                        continue;
                    }
                };
                let guard = arc.read();
                let store: &EdgeStore = &guard;

                if !store.is_direction_available(true) {
                    if let Some(note) = store.direction_note(true) {
                        log::debug!("{}", note);
                    }
                    *table_idx += 1;
                    *table_state = TableScanState::new();
                    continue 'outer;
                }

                match table_state.phase {
                    TablePhase::Mutable => {
                        let slice_filter = slice_filter_for(slice, slice_spaces, td.tbl_src);
                        scan_mutable(ScanArgs {
                            ctx,
                            store,
                            target,
                            td,
                            ts,
                            src_id_range,
                            projection,
                            predicate,
                            predicate_columns,
                            limit,
                            emitted,
                            offset_remaining,
                            slice_filter,
                            state: table_state,
                            batch: &mut candidates,
                            batch_size,
                            malformed,
                        });
                    }
                    TablePhase::Done => {
                        *table_idx += 1;
                        *table_state = TableScanState::new();
                        continue;
                    }
                }

                if limit.is_some_and(|l| *emitted >= l) {
                    *exhausted = true;
                }
            }
        });

        Ok(candidates
            .into_iter()
            .filter_map(|candidate| materialize_edge(ctx, candidate, ts))
            .collect())
    }
}

struct ScanArgs<'a> {
    ctx: &'a GraphStorageContext,
    store: &'a EdgeStore,
    target: &'a TargetDef,
    td: &'a TableDef,
    ts: Timestamp,
    src_id_range: &'a Option<Range<i64>>,
    projection: &'a Option<Vec<Arc<str>>>,
    predicate: &'a [crate::cursor::ScanPredicate],
    predicate_columns: &'a [Arc<str>],
    limit: Option<usize>,
    emitted: &'a mut usize,
    offset_remaining: &'a mut usize,
    slice_filter: Option<(ShardSlice, EdgeSliceSpace)>,
    state: &'a mut TableScanState,
    batch: &'a mut Vec<EdgeCandidate>,
    batch_size: usize,
    malformed: &'a mut u64,
}

// ---------------------------------------------------------------------------
// Free-function scan helpers
// ---------------------------------------------------------------------------

/// Verdict returned by the per-entry callback of [`walk_mutable_entries`].
enum EntryVerdict {
    /// Entry accepted; keep walking.
    Take,
    /// Stop walking: the caller's output is full. Resume stays at the
    /// current (group, skip) so the next call continues after it.
    BatchFull,
    /// Stop walking: the scan is complete (limit reached). Resume advances
    /// past the current group.
    StopAll,
}

/// Shared "group traversal + resume + filtering" walk over the mutable
/// out-CSR of one edge table, from the current resume point in `state`.
///
/// Advances `state.resume_group` / `state.skip_in_group` exactly like a full
/// scan: every consumed entry increments `skip_in_group`, finished groups
/// advance `resume_group` and reset the skip. Per entry, visibility and
/// corruption are filtered (corrupt rows counted in `malformed`), the
/// src-id range is enforced, and accepted `(global src, nbr)` pairs go to
/// `emit`, which may also count decode failures through the passed
/// malformed counter.
///
/// `pre_stop` is polled at group start and once per consumed entry (before
/// the entry is decoded); when it fires the walk stops early with
/// `false` and resume positioned after the last consumed entry.
///
/// Returns `true` when every remaining group was consumed (the caller marks
/// the table done), `false` when the walk stopped early.
fn walk_mutable_entries(
    walk: &WalkContext<'_>,
    config: &ScanConfig<'_>,
    state: &mut TableScanState,
    malformed: &mut u64,
    pre_stop: &mut dyn FnMut() -> bool,
    mut emit: impl FnMut(u32, Nbr, &mut u64) -> EntryVerdict,
) -> bool {
    let WalkContext { store, ctx, td } = walk;
    let ScanConfig {
        ts,
        src_id_range,
        predicate,
        slice_filter,
    } = config;
    let gate = ctx.pending_gate();

    // Segment pruning before decoding: groups whose flushed statistics
    // provably exclude the predicates are skipped without touching property
    // columns. Safe mid-scan because visibility is snapshot-fixed, bounds
    // only widen, and dirty groups never prune; the resume accounting below
    // advances past skipped groups exactly like a full walk.
    let mut pruned: std::collections::HashSet<usize> = std::collections::HashSet::new();
    if !predicate.is_empty() {
        for gid in store.out_csr.existing_group_ids() {
            if !store.segment_may_contain(gid as u32, predicate) {
                pruned.insert(gid);
            }
        }
        if !pruned.is_empty() {
            log::debug!(
                "edge scan pruned {} of {} groups",
                pruned.len(),
                store.out_csr.existing_group_ids().len()
            );
        }
    }

    let existing = store.out_csr.existing_group_ids();
    let group_bits = store.out_csr.group_bits();
    let group_size = 1u32 << group_bits;
    let start_pos = existing.partition_point(|gid| *gid < state.resume_group);
    for gid in existing.into_iter().skip(start_pos) {
        if pre_stop() {
            return false;
        }
        let base = crate::edge::node_group::group_base(gid, group_bits);
        if let Some((_, space)) = slice_filter {
            if group_outside_slice(space, base, group_size) {
                // Fully consumed without decoding: no owned source id can
                // live in this group, so resume advances past the whole
                // group exactly like a pruned group below.
                state.resume_group = gid + 1;
                state.skip_in_group = 0;
                continue;
            }
        }
        if pruned.contains(&gid) {
            // Fully consumed without decoding: pruned groups yield no
            // candidates, so batches never end inside them and resume
            // advances past the whole group.
            state.resume_group = gid + 1;
            state.skip_in_group = 0;
            continue;
        }
        let Some(variant) = store.out_csr.group_variant(gid) else {
            // Listed as existing but unreadable: metadata inconsistency,
            // counted instead of silently dropped.
            *malformed += 1;
            state.resume_group = gid + 1;
            state.skip_in_group = 0;
            continue;
        };
        let mut iter = variant.iter_all();
        if gid == state.resume_group {
            let skip = state.skip_in_group;
            for _ in 0..skip {
                if iter.next().is_none() {
                    break;
                }
            }
            // Keep the absolute physical offset: entries below keep incrementing
            // it so the next batch resumes after all consumed entries.
        } else {
            state.resume_group = gid;
            state.skip_in_group = 0;
        }
        for (local_vid, nbr) in iter.by_ref() {
            state.skip_in_group += 1;
            if pre_stop() {
                return false;
            }
            if !store.is_visible_with_gate(nbr.edge_id, *ts, &gate) {
                continue;
            }
            let Some(local) = local_vid.as_internal_u32() else {
                // A stored group row that is not an internal id indicates
                // corruption; count it instead of silently dropping it.
                *malformed += 1;
                continue;
            };
            let Some(global) = local.checked_add(base) else {
                *malformed += 1;
                continue;
            };
            if let Some((_, space)) = slice_filter {
                // Pure arithmetic ownership: unowned edges skip the
                // external-id resolution below, which is the dominant
                // per-edge cost of a partitioned scan.
                match slice_owns_global(space, global) {
                    Some(true) => {}
                    Some(false) => continue,
                    None => {
                        *malformed += 1;
                        continue;
                    }
                }
            }
            if let Some(ref r) = *src_id_range {
                let src_internal = VertexId::from_u32(global)
                    .as_internal_u32()
                    .unwrap_or(u32::MAX);
                let Some(src_ext) = resolve_vertex_id(ctx, src_internal, td.tbl_src, *ts) else {
                    *malformed += 1;
                    continue;
                };
                // An external id outside the integer domain cannot satisfy a
                // numeric range; count and skip instead of mapping it to a
                // sentinel that silently mis-filters the row.
                let src_int = match src_ext.as_int64() {
                    Some(v) => v,
                    None => match src_ext.as_u64() {
                        Some(v) => match i64::try_from(v) {
                            Ok(v) => v,
                            Err(_) => {
                                *malformed += 1;
                                continue;
                            }
                        },
                        None => {
                            *malformed += 1;
                            continue;
                        }
                    },
                };
                if src_int < r.start || src_int >= r.end {
                    continue;
                }
            }
            match emit(global, nbr, malformed) {
                EntryVerdict::Take => {}
                EntryVerdict::BatchFull => return false,
                EntryVerdict::StopAll => {
                    state.resume_group = gid + 1;
                    state.skip_in_group = 0;
                    return false;
                }
            }
        }
        state.resume_group = gid + 1;
        state.skip_in_group = 0;
    }
    true
}

fn scan_mutable(args: ScanArgs) {
    let ScanArgs {
        ctx,
        store,
        target,
        td,
        ts,
        src_id_range,
        projection,
        predicate,
        predicate_columns,
        limit,
        emitted,
        offset_remaining,
        slice_filter,
        state,
        batch,
        batch_size,
        malformed,
    } = args;
    // Column pruning: fetch the projection plus any predicate-only columns
    // in one storage read instead of decoding every column per edge.
    // `None` still means all columns.
    let fetch_columns: Option<Vec<Arc<str>>> = match *projection {
        None => None,
        Some(ref names) => {
            let mut cols = names.clone();
            for extra in predicate_columns.iter() {
                if !cols.iter().any(|c| c == extra) {
                    cols.push(extra.clone());
                }
            }
            Some(cols)
        }
    };

    let mut no_stop = || false;
    let table_done = walk_mutable_entries(
        &WalkContext { store, ctx, td },
        &ScanConfig {
            ts,
            src_id_range,
            predicate,
            slice_filter,
        },
        state,
        malformed,
        &mut no_stop,
        |global, nbr, malformed| {
            let src_vid = VertexId::from_u32(global);

            // Decode once with predicate columns included so pushed predicates
            // can be evaluated; matching rows are then trimmed back to the
            // projection. Filtering happens before offset/limit accounting.
            // With predicates, the column-scan layer filters row numbers
            // first on predicate columns only, and hits decode the fetch
            // set afterwards; misses never materialize a record and nulls
            // use bitmap semantics (missing never matches).
            if !predicate.is_empty() {
                match try_decode_edge_properties(store, nbr.edge_id, ts, Some(predicate_columns)) {
                    None => return EntryVerdict::Take,
                    Some(Err(e)) => {
                        log::warn!(
                            "edge {:?} predicate decode failed: {}; counting as malformed",
                            nbr.edge_id,
                            e
                        );
                        *malformed += 1;
                        return EntryVerdict::Take;
                    }
                    Some(Ok(probe)) => {
                        if !predicate.iter().all(|p| p.matches(probe.as_slice())) {
                            return EntryVerdict::Take;
                        }
                    }
                }
            }
            let mut properties = match try_decode_edge_properties(
                store,
                nbr.edge_id,
                ts,
                fetch_columns.as_deref(),
            ) {
                None => Vec::new(),
                Some(Ok(props)) => props,
                Some(Err(e)) => {
                    log::warn!(
                        "edge {:?} property decode failed: {}; counting as malformed",
                        nbr.edge_id,
                        e
                    );
                    *malformed += 1;
                    return EntryVerdict::Take;
                }
            };
            if !predicate.iter().all(|p| p.matches(properties.as_slice())) {
                return EntryVerdict::Take;
            }
            trim_to_projection(&mut properties, projection);

            if *offset_remaining > 0 {
                *offset_remaining -= 1;
                return EntryVerdict::Take;
            }

            let edge = build_edge_candidate(EdgeBuildArgs {
                target,
                td,
                src_vid: &src_vid,
                nbr,
                props: properties,
            });
            batch.push(edge);
            *emitted += 1;

            if batch.len() >= batch_size {
                return EntryVerdict::BatchFull;
            }
            if limit.is_some_and(|l| *emitted >= l) {
                return EntryVerdict::StopAll;
            }
            EntryVerdict::Take
        },
    );

    if table_done {
        state.phase = TablePhase::Done;
    }
}

// ---------------------------------------------------------------------------
// Edge construction
// ---------------------------------------------------------------------------

struct EdgeBuildArgs<'a> {
    target: &'a TargetDef,
    td: &'a TableDef,
    src_vid: &'a VertexId,
    nbr: Nbr,
    props: Vec<(Arc<str>, Value)>,
}

struct EdgeCandidate {
    edge_type_name: String,
    src_label: LabelId,
    dst_label: LabelId,
    src_vid: VertexId,
    dst_vid: VertexId,
    rank: i64,
    props: HashMap<Arc<str>, Value>,
}

fn build_edge_candidate(args: EdgeBuildArgs<'_>) -> EdgeCandidate {
    let src_internal = args.src_vid.as_internal_u32().unwrap_or(u32::MAX);
    let rank = args.nbr.rank;
    let dst_vid = VertexId::from_u32(args.nbr.endpoint);

    let src_vid = VertexId::from_u32(src_internal);
    let props: HashMap<Arc<str>, Value> = args.props.into_iter().collect();
    EdgeCandidate {
        edge_type_name: args.target.edge_type_name.clone(),
        src_label: args.td.tbl_src,
        dst_label: args.td.tbl_dst,
        src_vid,
        dst_vid,
        rank,
        props,
    }
}

fn materialize_edge(
    ctx: &GraphStorageContext,
    candidate: EdgeCandidate,
    ts: Timestamp,
) -> Option<Edge> {
    let src_internal = candidate.src_vid.as_internal_u32().unwrap_or(u32::MAX);
    let dst_internal = candidate.dst_vid.as_internal_u32().unwrap_or(u32::MAX);
    let src_external =
        resolve_vertex_id(ctx, src_internal, candidate.src_label, ts).unwrap_or(candidate.src_vid);
    let dst_external =
        resolve_vertex_id(ctx, dst_internal, candidate.dst_label, ts).unwrap_or(candidate.dst_vid);
    Some(Edge {
        src: src_external,
        dst: dst_external,
        edge_type: candidate.edge_type_name,
        ranking: candidate.rank,
        props: candidate.props,
    })
}

/// Decode edge properties for the precomputed fetch set (projection plus
/// any extra columns required by pushed scan predicates).
///
/// MVCCManager is the single visibility authority; the property row
/// timestamps are physical replicas and must not decide visibility here.
/// Strict property decode distinguishing invisible rows from corrupt payloads.
///
/// `None` means invisible or unmapped (normal empty). `Some(Ok)` carries the
/// projected properties. `Some(Err)` means the row is visible but the payload
/// is corrupt or below the history floor: callers must count it as malformed
/// instead of serving it as null.
fn try_decode_edge_properties(
    store: &EdgeStore,
    edge_id: linkrs_core::types::EdgeId,
    ts: Timestamp,
    fetch: Option<&[Arc<str>]>,
) -> Option<StorageResult<Vec<(Arc<str>, Value)>>> {
    if !store.is_visible(edge_id, ts) {
        return None;
    }
    match store
        .properties
        .try_get_projected_physical_by_edge_id(edge_id, ts, fetch)
    {
        Ok(None) => None,
        Ok(Some(cells)) => Some(Ok(cells
            .into_iter()
            .filter_map(|(k, v)| v.map(|value| (k, value)))
            .collect())),
        Err(e) => Some(Err(e)),
    }
}

/// Drop predicate-only columns so emitted rows carry projected properties.
fn trim_to_projection(props: &mut Vec<(Arc<str>, Value)>, projection: &Option<Vec<Arc<str>>>) {
    if let Some(names) = projection {
        props.retain(|(k, _)| names.iter().any(|name| name == k));
    }
}

/// Collect the property names a scan predicate references.
fn collect_predicate_columns(pred: &crate::cursor::ScanPredicate, out: &mut Vec<Arc<str>>) {
    use crate::cursor::ScanPredicate as P;
    match pred {
        P::ColumnEqual { column, .. } => {
            if !out.iter().any(|name| name == column) {
                out.push(column.clone());
            }
        }
        P::ColumnRange { column, .. } => {
            if !out.iter().any(|name| name == column) {
                out.push(column.clone());
            }
        }
    }
}

pub(crate) fn resolve_vertex_id(
    ctx: &GraphStorageContext,
    internal: u32,
    label: LabelId,
    ts: Timestamp,
) -> Option<VertexId> {
    if let Some(vid) = ctx.get_external_id_by_internal_id(label, internal) {
        return Some(vid);
    }
    ctx.get_external_vertex_id(label, internal, ts)
}

// ---------------------------------------------------------------------------
// Target resolution
// ---------------------------------------------------------------------------

fn build_target(
    ctx: &Arc<GraphStorageContext>,
    space: &str,
    edge_type: &str,
    allowlist: Option<&[EdgeTableKey]>,
) -> StorageResult<TargetDef> {
    let edge_info = ctx
        .schema_manager()
        .get_edge_type(space, edge_type)?
        .ok_or_else(|| {
            StorageError::not_found(format!(
                "Edge type {} not found in space {}",
                edge_type, space
            ))
        })?;

    let edge_label_id = edge_info.edge_type_id;
    let src_label_id = endpoint_label_id(ctx, space, &edge_info.src_tag_name)?.unwrap_or(0);
    let dst_label_id = endpoint_label_id(ctx, space, &edge_info.dst_tag_name)?.unwrap_or(0);

    let tables = if src_label_id == 0 && dst_label_id == 0 {
        ctx.data_store().with_edge_tables(|edge_tables| {
            edge_tables
                .iter()
                .filter(|(_, arc)| arc.read().label() == edge_label_id)
                .map(|(key, arc)| {
                    let store = arc.read();
                    TableDef {
                        key: *key,
                        tbl_src: store.src_label(),
                        tbl_dst: store.dst_label(),
                    }
                })
                .collect()
        })
    } else {
        let key = EdgeTableKey::new(src_label_id, dst_label_id, edge_label_id);
        vec![TableDef {
            key,
            tbl_src: src_label_id,
            tbl_dst: dst_label_id,
        }]
    };
    // Table-sharded parallel scans drain one key subset per shard; the
    // union of shards covers every table exactly once.
    let tables = match allowlist {
        Some(allow) => tables
            .into_iter()
            .filter(|td| allow.contains(&td.key))
            .collect(),
        None => tables,
    };

    Ok(TargetDef {
        edge_type_name: edge_type.to_string(),
        tables,
    })
}

/// Open a hot single-segment edge scan cursor.
pub(crate) fn create_edge_cursor(
    ctx: Arc<GraphStorageContext>,
    space: &str,
    options: &ScanOptions,
) -> StorageResult<Box<dyn EdgeCursor>> {
    let cursor = GraphEdgeCursor::new(ctx, space, options)?;
    Ok(Box::new(cursor))
}