hyphae 3.0.0

//! Select-cell plan node implementing [`MapQuery`].
//!
//! `select_cell` is the reactive variant of `select`: each row's inclusion is
//! gated by a [`Watchable`] producing `bool`. Returns an uncompiled plan node; call
//! [`MapQuery::materialize`] to compile a plan into a subscribable
//! [`CellMap`](crate::CellMap).

use std::{hash::Hash, marker::PhantomData, sync::Arc};

use super::ProjectCellExt;
use crate::{
    map_query::{
        BuildQueryRuntime, MapQuery,
        properties::{ByMapKey, PlanProperties, ZeroOrOne},
    },
    pipeline::{Materialize, Pipeline},
    subscription::SubscriptionGuard,
    traits::{CellValue, Gettable, MapExt},
};

/// Plan node for [`SelectCellExt::select_cell`].
///
/// Reactive filter: each row's inclusion is decided by a per-row
/// [`Watchable`] producing `bool` from `predicate(&key, &value)`. The row is
/// included while the gate is `true` and excluded when `false`.
///
/// Not [`Clone`]: cloning a plan would silently duplicate per-row
/// subscription work; share by materializing once.
#[allow(private_bounds)]
pub struct SelectCellPlan<S, K, V, W, F>
where
    S: MapQuery<Key = K, Value = V>,
    K: Hash + Eq + CellValue,
    V: CellValue,
    W: Pipeline<bool>
        + crate::pipeline::PipelineSeed<bool>
        + Gettable<bool>
        + Clone
        + Send
        + Sync
        + 'static,
    F: Fn(&K, &V) -> W + Send + Sync + 'static,
{
    pub(crate) source: S,
    pub(crate) predicate: Arc<F>,
    pub(crate) _types: PhantomData<fn() -> (K, V, W)>,
}

#[allow(private_bounds)]
impl<S, K, V, W, F> PlanProperties for SelectCellPlan<S, K, V, W, F>
where
    S: MapQuery<Key = K, Value = V>,
    K: Hash + Eq + CellValue,
    V: CellValue,
    W: Pipeline<bool>
        + crate::pipeline::PipelineSeed<bool>
        + Gettable<bool>
        + Clone
        + Send
        + Sync
        + 'static,
    F: Fn(&K, &V) -> W + Send + Sync + 'static,
{
    type Cardinality = ZeroOrOne;
    type InputPartition = S::OutputPartition;
    type OutputPartition = ByMapKey<K>;
}

impl<S, K, V, W, F> BuildQueryRuntime<K, V> for SelectCellPlan<S, K, V, W, F>
where
    S: MapQuery<Key = K, Value = V>,
    K: Hash + Eq + CellValue,
    V: CellValue,
    W: Pipeline<bool>
        + crate::pipeline::PipelineSeed<bool>
        + Gettable<bool>
        + Clone
        + Send
        + Sync
        + 'static,
    F: Fn(&K, &V) -> W + Send + Sync + 'static,
{
    fn build_into(
        self,
        cx: &mut crate::map_query::compiler::CompileContext,
        sink: crate::map_query::BoxedMapDiffSink<K, V>,
    ) -> Vec<SubscriptionGuard> {
        // Implement select_cell as a project_cell whose inner Watchable maps
        // the boolean gate into Option<(K, V)>.
        let predicate = self.predicate;
        let inner_plan = self.source.project_cell(move |k: &K, v: &V| {
            let k = k.clone();
            let v = v.clone();
            predicate(&k, &v)
                .map(move |include| {
                    if *include {
                        Some((k.clone(), v.clone()))
                    } else {
                        None
                    }
                })
                .materialize()
        });
        crate::map_query::compile_runtime_into(inner_plan, cx, sink)
    }
}

#[allow(private_bounds)]
impl<S, K, V, W, F> MapQuery for SelectCellPlan<S, K, V, W, F>
where
    S: MapQuery<Key = K, Value = V>,
    K: Hash + Eq + CellValue,
    V: CellValue,
    W: Pipeline<bool>
        + crate::pipeline::PipelineSeed<bool>
        + Gettable<bool>
        + Clone
        + Send
        + Sync
        + 'static,
    F: Fn(&K, &V) -> W + Send + Sync + 'static,
{
    type Key = K;
    type Value = V;
}

/// Select-cell operator returning a [`MapQuery`] plan node.
///
/// `select_cell` consumes `self` and returns an uncompiled plan node; call
/// [`MapQuery::materialize`] on the result to obtain a subscribable
/// [`CellMap`](crate::CellMap).
pub trait SelectCellExt<K, V>: MapQuery<Key = K, Value = V>
where
    K: Hash + Eq + CellValue,
    V: CellValue,
{
    /// Reactive filter: each row has a watchable boolean gate.
    ///
    /// `predicate(&key, &value)` returns a watchable bool; the row is
    /// included when true and excluded when false.
    #[track_caller]
    #[allow(private_bounds)]
    fn select_cell<W, F>(self, predicate: F) -> impl MapQuery<Key = K, Value = V>
    where
        W: Pipeline<bool>
            + crate::pipeline::PipelineSeed<bool>
            + Gettable<bool>
            + Clone
            + Send
            + Sync
            + 'static,
        F: Fn(&K, &V) -> W + Send + Sync + 'static,
    {
        SelectCellPlan {
            source: self,
            predicate: Arc::new(predicate),
            _types: PhantomData,
        }
    }
}

impl<K, V, M> SelectCellExt<K, V> for M
where
    K: Hash + Eq + CellValue,
    V: CellValue,
    M: MapQuery<Key = K, Value = V>,
{
}

#[cfg(test)]
mod tests {
    use std::sync::mpsc;

    use super::*;
    use crate::{Cell, CellMap, MapExt, Materialize, cell_map::MapDiff};

    #[test]
    fn select_cell_reacts_to_predicate_changes() {
        let values = CellMap::<String, i32>::new();
        let gates = CellMap::<String, bool>::new();

        values.insert("a".to_string(), 10);
        values.insert("b".to_string(), 20);
        gates.insert("a".to_string(), false);
        gates.insert("b".to_string(), true);

        let filtered = values
            .select_cell({
                let gates = gates.clone();
                move |key, _value| gates.get(key).map(|v| v.unwrap_or(false)).materialize()
            })
            .materialize();

        assert_eq!(filtered.entries().materialize().get().len(), 1);
        assert!(!filtered.contains_key(&"a".to_string()));
        assert!(filtered.contains_key(&"b".to_string()));

        gates.insert("a".to_string(), true);
        assert_eq!(filtered.entries().materialize().get().len(), 2);
        gates.insert("b".to_string(), false);
        assert_eq!(filtered.entries().materialize().get().len(), 1);
    }

    #[test]
    fn select_cell_preserves_upstream_batch_without_extra_emissions() {
        let source = CellMap::<String, i32>::new();
        let out = source
            .clone()
            .select_cell(|_, _| Cell::new(true).lock())
            .materialize();

        let (tx, rx) = mpsc::channel::<MapDiff<String, i32>>();
        let _guard = out.subscribe_diffs(move |diff| {
            let _ = tx.send(diff.clone());
        });

        source.insert_many(vec![("a".to_string(), 1), ("b".to_string(), 2)]);
        let seen: Vec<_> = rx.try_iter().collect();
        assert_eq!(seen.len(), 2);
        assert!(matches!(
            seen.last(),
            Some(MapDiff::Batch { changes }) if changes.len() == 2
        ));
    }
}