aedb 0.3.1

//! Lazy primary-key row hydration for index-driven row sources.
//!
//! When an index lookup yields candidate primary keys but a residual filter
//! (or other downstream stage) decides which rows survive, eagerly
//! materializing every candidate wastes work: a bounded pipeline (LIMIT)
//! stops pulling long before the candidate list is exhausted. This source
//! fetches rows one primary key at a time as the pipeline pulls them.
//!
//! Iterators cannot propagate `Result`, so read failures are parked in a
//! shared slot and iteration ends; the executor checks the slot after the
//! pipeline drains and surfaces the error. A consequence (shared with the
//! window-bounded scan paths) is that a page fully satisfied before reaching
//! an unreadable row now succeeds where eager hydration would have failed —
//! every row actually returned was read successfully.

use crate::catalog::types::Row;
use crate::query::error::QueryError;
use crate::storage::encoded_key::EncodedKey;
use crate::storage::keyspace::{KeyspaceSnapshot, TableData};
use parking_lot::Mutex;
use std::sync::Arc;

/// Shared slot holding the first row-fetch error, if any.
pub(super) type LazyFetchError = Arc<Mutex<Option<QueryError>>>;

pub(super) struct LazyPkRowSource<'a, I> {
    snapshot: &'a KeyspaceSnapshot,
    table: &'a TableData,
    pks: I,
    error: LazyFetchError,
}

impl<'a, I: Iterator<Item = EncodedKey>> LazyPkRowSource<'a, I> {
    /// Build a lazy row source over `pks` and return it with the shared error
    /// slot the executor must check once the pipeline has drained.
    pub(super) fn new(
        snapshot: &'a KeyspaceSnapshot,
        table: &'a TableData,
        pks: I,
    ) -> (Self, LazyFetchError) {
        let error: LazyFetchError = Arc::new(Mutex::new(None));
        (
            Self {
                snapshot,
                table,
                pks,
                error: Arc::clone(&error),
            },
            error,
        )
    }
}

impl<I: Iterator<Item = EncodedKey>> Iterator for LazyPkRowSource<'_, I> {
    type Item = Row;

    fn next(&mut self) -> Option<Row> {
        for pk in self.pks.by_ref() {
            match self.snapshot.get_row_in_table(self.table, &pk) {
                Ok(Some(row)) => return Some(row.into_owned()),
                // A pk without a row (racing delete already excluded by MVCC
                // snapshots, but tolerated for defense) is skipped, matching
                // the eager loop's behavior.
                Ok(None) => continue,
                Err(err) => {
                    *self.error.lock() = Some(err.into());
                    return None;
                }
            }
        }
        None
    }
}