kglite 0.16.1

//! [`NodeView`] — the authoritative read route for a node's properties.
//!
//! # Why this exists
//!
//! A columnar node's properties live in a per-type
//! [`ColumnStore`](crate::graph::storage::column_store::ColumnStore) that the
//! **storage backend owns**. Before D1 three `Arc`s pointed at the same store —
//! the node's own handle, a `DirGraph`-level map, and `DiskGraph`'s — so a read
//! resolved *one particular replica* rather than the owner, which is how two
//! shipped defects arose (an empty columnar `property_iter`, and a spill that
//! reclaimed nothing). There is one owner now, and this type is how a caller
//! reaches it.
//!
//! `NodeView` is the single place a node's property read resolves its store.
//! Callers ask the storage backend for a view
//! ([`GraphRead::node_view`](crate::graph::storage::GraphRead::node_view)) and
//! then read through it. When the backend becomes the sole owner of the stores,
//! only [`NodeView::from_node_data`] and the backend accessors change; every
//! caller keeps compiling and keeps its meaning.
//!
//! # The columnar completeness contract
//!
//! `NodeData::property_iter` yields **nothing** for
//! `PropertyStorage::Columnar` — it cannot, because columnar values are
//! constructed on read and there is no `&'a Value` to hand out. Every
//! enumeration method on `NodeView` ([`NodeView::property_pairs`],
//! [`NodeView::property_keys`], [`NodeView::properties_cloned`],
//! [`NodeView::property_pairs_named`]) is **complete for every storage
//! variant**, columnar included. That is the contract: if you enumerate through
//! a `NodeView` you see the node's real properties.

use std::borrow::Cow;
use std::collections::HashMap;

use crate::datatypes::Value;
use crate::graph::schema::{InternedKey, NodeData, NodeInfo, PropertyStorage, StringInterner};
use crate::graph::storage::column_store::ColumnStore;
use crate::graph::storage::StrField;

/// A borrowed, backend-resolved read handle for one node.
///
/// Cheap to copy (two words + a row id). Obtained from
/// [`GraphRead::node_view`](crate::graph::storage::GraphRead::node_view).
///
/// # Lifetime discipline
///
/// A `NodeView` borrows the storage backend. On the disk backend the borrowed
/// `NodeData` lives in the per-query arena, so a view must not outlive the
/// `begin_query()` guard, and must never be held across a `Python::attach`
/// boundary or a GIL release — resolve to owned [`Value`]s at such a boundary
/// (that is what [`NodeView::properties_cloned`] and
/// [`NodeView::to_node_info`] are for).
#[derive(Clone, Copy)]
pub struct NodeView<'a> {
    data: &'a NodeData,
    /// The resolved column store for this node's row, when the node's
    /// properties are columnar. Resolving once per view rather than once per
    /// read is the point of the type.
    store: Option<(&'a ColumnStore, u32)>,
}

impl<'a> NodeView<'a> {
    /// Pair a node with the store its backend owns.
    ///
    /// The only constructor. `store` comes from
    /// [`GraphRead::column_store`](crate::graph::storage::GraphRead::column_store),
    /// resolved by the node's type — a node itself knows only its `row_id`
    /// (D1 Phase 3). A `None` store on a columnar node means the backend has
    /// no store for that type, which reads as an empty property set.
    #[inline]
    pub(crate) fn new(data: &'a NodeData, store: Option<(&'a ColumnStore, u32)>) -> Self {
        NodeView { data, store }
    }

    /// Escape hatch to the underlying `NodeData`.
    ///
    /// Only for callers that need identity/whole-struct semantics (clone,
    /// equality, serialization) rather than a property read. **Not** for
    /// property reads — those must go through this type's methods.
    #[inline]
    pub fn data(&self) -> &'a NodeData {
        self.data
    }

    /// The node's primary type key.
    #[inline]
    pub fn node_type(&self) -> InternedKey {
        self.data.node_type
    }

    /// The node's primary type, resolved to a string.
    #[inline]
    pub fn node_type_str<'i>(&self, interner: &'i StringInterner) -> &'i str {
        interner.resolve(self.data.node_type)
    }

    /// The node's primary type, resolved to a string. Alias of
    /// [`NodeView::node_type_str`], matching `NodeData`'s two spellings so
    /// migrated call sites read unchanged.
    #[inline]
    pub fn get_node_type_ref<'i>(&self, interner: &'i StringInterner) -> &'i str {
        interner.resolve(self.data.node_type)
    }

    /// The node's id (resolving the mapped-mode `Null` sentinel through the
    /// column store).
    #[inline]
    pub fn id(&self) -> Cow<'a, Value> {
        if matches!(self.data.id, Value::Null) {
            if let Some((store, row_id)) = self.store {
                if let Some(v) = store.get_id(row_id) {
                    return Cow::Owned(v);
                }
            }
        }
        Cow::Borrowed(&self.data.id)
    }

    /// The node's title (resolving the mapped-mode `Null` sentinel through the
    /// column store).
    #[inline]
    pub fn title(&self) -> Cow<'a, Value> {
        if matches!(self.data.title, Value::Null) {
            if let Some((store, row_id)) = self.store {
                if let Some(v) = store.get_title(row_id) {
                    return Cow::Owned(v);
                }
            }
        }
        Cow::Borrowed(&self.data.title)
    }

    /// Read a property by interned key. `None` when absent or `Value::Null`.
    ///
    /// **Borrows where the storage allows it**, which for a columnar node means
    /// an in-memory `Mixed` column — the shape a list property takes. That is
    /// not a detail: the executor's list subscript reads the container through
    /// this method once per element, so an owning read makes `n.vec[i]` cost
    /// the length of `n.vec`. The 0.15.11 fix that established the borrow was
    /// undone by 0.16.0's always-columnar construction, which moved every list
    /// out of the `Map` arm and into the columnar one; it is pinned now by
    /// `tests::an_in_memory_list_property_is_borrowed_not_cloned`.
    #[inline]
    pub fn get(&self, key: InternedKey) -> Option<Cow<'a, Value>> {
        match self.store {
            Some((store, row_id)) => store.get_cow(row_id, key),
            None => self.data.properties.get(key),
        }
    }

    /// Read a property by interned key, owned. Cheaper than [`NodeView::get`]
    /// for callers that always need ownership.
    #[inline]
    pub fn get_value(&self, key: InternedKey) -> Option<Value> {
        match self.store {
            Some((store, row_id)) => store.get(row_id, key),
            None => self.data.properties.get_value(key),
        }
    }

    /// Read a property by name (excludes `id` / `title`).
    #[inline]
    pub fn get_property(&self, key: &str) -> Option<Cow<'a, Value>> {
        self.get(InternedKey::from_str(key))
    }

    /// Read a property by name, owned (excludes `id` / `title`).
    #[inline]
    pub fn get_property_value(&self, key: &str) -> Option<Value> {
        self.get_value(InternedKey::from_str(key))
    }

    /// Read a *field* by name — `id` and `title` resolve to the node's
    /// identity columns, anything else to a property.
    #[inline]
    pub fn get_field_ref(&self, field: &str) -> Option<Cow<'a, Value>> {
        match field {
            "id" => Some(self.id()),
            "title" => Some(self.title()),
            _ => self.get(InternedKey::from_str(field)),
        }
    }

    /// Read an **alias-resolved matcher field** — the value a property filter
    /// on this node actually compares against.
    ///
    /// `field`/`key` are the output of
    /// [`DirGraph::resolve_alias`](crate::graph::schema::DirGraph::resolve_alias)
    /// (a type's `unique_id_field` / `node_title_field` map onto `id` /
    /// `title`), and the resolution order is the one the pattern matcher
    /// applies: identity fields first, then a stored property (a user's own
    /// `label`/`name`/… wins — KG-1), then the structural soft-alias fallback.
    ///
    /// Every consumer of "what would a filter on `field` see?" must come
    /// through here. The planner's NDV statistic did not, read the property map
    /// alone, and so found *nothing* for a type's title field — scoring an
    /// equality filter on it as completely non-selective (Track H2).
    #[inline]
    pub fn resolved_field(
        &self,
        type_str: &str,
        field: &str,
        key: InternedKey,
    ) -> Option<Cow<'a, Value>> {
        if field == "id" {
            return Some(self.id());
        }
        if field == "title" {
            return Some(self.title());
        }
        if let Some(value) = self.get(key) {
            return Some(value);
        }
        crate::graph::schema::soft_alias_fallback(field).map(|fallback| match fallback {
            crate::graph::schema::SoftAliasFallback::Title => self.title(),
            crate::graph::schema::SoftAliasFallback::TypeString => {
                Cow::Owned(Value::String(type_str.to_string()))
            }
        })
    }

    /// Borrowed string read of an **alias-resolved matcher field** — the
    /// allocation-free companion to [`Self::resolved_field`], for matchers that
    /// only ever test the string form.
    ///
    /// The resolution order is `resolved_field`'s, step for step: identity
    /// fields, then a stored property (which *resolves* even when it holds a
    /// non-string, hence [`StrField::NotString`]), then the structural soft
    /// alias. Any divergence here would make a `WHERE` clause see a different
    /// value than the planner's statistics do.
    #[inline]
    pub fn resolved_field_str(
        &self,
        type_str: &'a str,
        field: &str,
        key: InternedKey,
    ) -> StrField<'a> {
        if field == "id" {
            return self.id_field();
        }
        if field == "title" {
            return self.title_field();
        }
        match self.str_field(key) {
            StrField::Absent => {}
            resolved => return resolved,
        }
        match crate::graph::schema::soft_alias_fallback(field) {
            Some(crate::graph::schema::SoftAliasFallback::Title) => self.title_field(),
            Some(crate::graph::schema::SoftAliasFallback::TypeString) => {
                StrField::Str(Cow::Borrowed(type_str))
            }
            None => StrField::Absent,
        }
    }

    /// Borrowed string read of a property. Mirrors [`Self::get`].
    #[inline]
    pub fn str_field(&self, key: InternedKey) -> StrField<'a> {
        match self.store {
            Some((store, row_id)) => store.str_field(row_id, key),
            None => self.data.properties.str_field(key),
        }
    }

    /// Borrowed string read of the node's title. Mirrors [`Self::title`].
    #[inline]
    pub fn title_field(&self) -> StrField<'a> {
        Self::identity_str(
            &self.data.title,
            |store, row_id| store.title_field(row_id),
            self.store,
        )
    }

    /// Borrowed string read of the node's id. Mirrors [`Self::id`].
    #[inline]
    pub fn id_field(&self) -> StrField<'a> {
        Self::identity_str(
            &self.data.id,
            |store, row_id| store.id_field(row_id),
            self.store,
        )
    }

    /// Shared shape of the two identity reads: the inline field wins unless it
    /// carries the columnar `Null` sentinel, in which case the store answers.
    #[inline]
    fn identity_str(
        inline: &'a Value,
        from_store: impl FnOnce(&'a ColumnStore, u32) -> StrField<'a>,
        store: Option<(&'a ColumnStore, u32)>,
    ) -> StrField<'a> {
        match inline {
            Value::String(s) => StrField::Str(Cow::Borrowed(s.as_str())),
            Value::Null => match store {
                Some((store, row_id)) => from_store(store, row_id),
                None => StrField::Absent,
            },
            _ => StrField::NotString,
        }
    }

    /// `true` when the property is present and non-`Null`.
    #[inline]
    pub fn contains(&self, key: InternedKey) -> bool {
        match self.store {
            Some((store, row_id)) => store.contains_value(row_id, key),
            None => self.data.properties.contains(key),
        }
    }

    /// `true` when the named property is present and non-`Null`.
    #[inline]
    pub fn has_property(&self, key: &str) -> bool {
        self.contains(InternedKey::from_str(key))
    }

    /// `true` when this node's properties live in a per-type column store.
    ///
    /// Crate-internal, and not a shape a caller can act on: every enumeration
    /// method on `NodeView` is already complete for columnar storage. It
    /// survives for the one site that branches on storage shape for *schema*
    /// reasons — completing a projection from type metadata, which only has a
    /// declared-but-unstored column to recover when the row came from a store.
    #[inline]
    pub(crate) fn properties_are_columnar(&self) -> bool {
        self.store.is_some()
    }

    /// Number of present (non-`Null`) properties.
    ///
    /// Counted without materialising the row: the columnar arm used to build
    /// the whole `Vec<(InternedKey, Value)>` to take its `len()`, and both
    /// callers (`calculate()`/`statistics()`'s capacity hint, the GraphML
    /// export's "has any property?" test) build the row again immediately
    /// afterwards.
    #[inline]
    pub fn property_count(&self) -> usize {
        match self.store {
            Some((store, row_id)) => store.row_property_count(row_id),
            None => self.data.properties.len(),
        }
    }

    /// Allocation-free string equality against a property.
    ///
    /// `None` — absent/null; `Some(true)` — equal; `Some(false)` — present but
    /// different (including non-string values).
    #[inline]
    pub fn str_prop_eq(&self, key: InternedKey, target: &str) -> Option<bool> {
        match self.store {
            Some((store, row_id)) => store.str_prop_eq(row_id, key, target),
            None => self.data.properties.str_prop_eq(key, target),
        }
    }

    /// Case-insensitive substring test on a string-typed field. `false` when
    /// the field is missing or non-string; `needle_lower` must already be
    /// lowercased.
    pub fn field_contains_ci(&self, field: &str, needle_lower: &str) -> bool {
        self.get_field_ref(field)
            .and_then(|v| match &*v {
                Value::String(s) => Some(s.to_lowercase().contains(needle_lower)),
                _ => None,
            })
            .unwrap_or(false)
    }

    /// Case-insensitive prefix test on a string-typed field. `false` when the
    /// field is missing or non-string; `prefix_lower` must already be
    /// lowercased.
    pub fn field_starts_with_ci(&self, field: &str, prefix_lower: &str) -> bool {
        self.get_field_ref(field)
            .and_then(|v| match &*v {
                Value::String(s) => Some(s.to_lowercase().starts_with(prefix_lower)),
                _ => None,
            })
            .unwrap_or(false)
    }

    /// Every present property as `(interned key, owned value)`.
    ///
    /// **Complete for columnar storage** — the removed
    /// `NodeData::property_iter` yielded nothing there.
    pub fn property_pairs(&self) -> Vec<(InternedKey, Value)> {
        match self.store {
            Some((store, row_id)) => store.row_properties(row_id),
            None => match &self.data.properties {
                // `Map` deliberately keeps `Value::Null` entries visible —
                // `NodeData::clear_property` stages a REMOVE that way for the
                // disk flush. Filtering them here would change what every
                // existing enumeration caller sees.
                PropertyStorage::Map(map) => map.iter().map(|(k, v)| (*k, v.clone())).collect(),
                PropertyStorage::Columnar(_) => unreachable!("store resolved above"),
            },
        }
    }

    /// Every present property key, resolved to a string.
    ///
    /// **Complete for columnar storage.** Keys that the interner cannot resolve
    /// are skipped, matching the pre-existing `PropertyStorage::keys` contract.
    pub fn property_keys(&self, interner: &'a StringInterner) -> Vec<&'a str> {
        match self.store {
            Some((store, row_id)) => store
                .row_property_keys(row_id)
                .into_iter()
                .filter_map(|ik| interner.try_resolve(ik))
                .collect(),
            None => self.data.properties.keys(interner).collect(),
        }
    }

    /// Every present property key, interned — the allocation-free companion to
    /// [`NodeView::property_pairs`] for callers that only need names.
    ///
    /// **Complete for columnar storage**, and key-for-key identical to what
    /// `property_pairs` yields (pinned by
    /// `column_store::tests::row_property_keys_matches_row_properties`). The
    /// `Map` arm keeps `Value::Null` entries visible for the same reason
    /// `property_pairs` does — a staged REMOVE is a key the enumeration must
    /// still report.
    pub fn property_key_set(&self) -> Vec<InternedKey> {
        match self.store {
            Some((store, row_id)) => store.row_property_keys(row_id),
            None => match &self.data.properties {
                PropertyStorage::Map(map) => map.keys().copied().collect(),
                PropertyStorage::Columnar(_) => unreachable!("store resolved above"),
            },
        }
    }

    /// Every present property as `(name, owned value)`.
    ///
    /// **Complete for columnar storage** — the replacement for
    /// `property_iter().map(|(k, v)| (k.to_string(), v.clone()))`, which
    /// silently produced an empty vector for saved graphs.
    pub fn property_pairs_named(&self, interner: &StringInterner) -> Vec<(String, Value)> {
        self.property_pairs()
            .into_iter()
            .filter_map(|(ik, v)| interner.try_resolve(ik).map(|s| (s.to_string(), v)))
            .collect()
    }

    /// Every present property as a `HashMap<String, Value>` (export / interop).
    ///
    /// **Complete for columnar storage.**
    #[inline]
    pub fn properties_cloned(&self, interner: &StringInterner) -> HashMap<String, Value> {
        self.property_pairs_named(interner).into_iter().collect()
    }

    /// Owned snapshot of the whole node (Python API / export).
    #[inline]
    pub fn to_node_info(&self, interner: &StringInterner) -> NodeInfo {
        NodeInfo {
            id: self.id().into_owned(),
            title: self.title().into_owned(),
            node_type: self.node_type_str(interner).to_string(),
            properties: self.properties_cloned(interner),
        }
    }
}

impl std::fmt::Debug for NodeView<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NodeView")
            .field("id", &self.data.id)
            .field("title", &self.data.title)
            .field("node_type", &self.data.node_type)
            .field("columnar_row", &self.store.map(|(_, r)| r))
            .finish()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::graph::schema::{ColumnarRow, TypeSchema};
    use std::sync::Arc;

    /// The regression pin the 0.15.11 borrow fix never had.
    ///
    /// That fix made a list property's container borrowable so the executor's
    /// subscript (`n.vec[i]`) stops cloning the whole list per element access.
    /// Nothing asserted the *borrow*, only the results — so when 0.16.0 made
    /// construction always-columnar, every CREATEd list moved from the `Map`
    /// arm (which borrows) to the columnar arm (which cloned), and the cost
    /// came back silently: measured at 0.19 µs/access for a 16-element list
    /// against 3.95 µs/access for a 1024-element one, release build.
    ///
    /// Asserting `Cow::Borrowed` is what makes it a pin: it goes red on the
    /// clone itself, whatever storage refactor reintroduces it.
    #[test]
    fn an_in_memory_list_property_is_borrowed_not_cloned() {
        let mut interner = StringInterner::new();
        let key = interner.get_or_intern("vec");
        let type_key = interner.get_or_intern("T");
        let mut store = ColumnStore::new(Arc::new(TypeSchema::new()), &HashMap::new(), &interner);
        let list = Value::List(vec![Value::Int64(1), Value::Int64(2), Value::Int64(3)]);
        let row_id = store.push_row(&[(key, list.clone())]);

        let data = NodeData {
            id: Value::Null,
            title: Value::Null,
            node_type: type_key,
            properties: crate::graph::schema::PropertyStorage::Columnar(ColumnarRow::new(row_id)),
        };
        let view = NodeView::new(&data, Some((&store, row_id)));

        let read = view.get(key).expect("the list property must resolve");
        assert!(
            matches!(read, Cow::Borrowed(_)),
            "a columnar in-memory list must be borrowed, not cloned per read"
        );
        assert_eq!(*read, list);
    }

    /// The borrow must not change *what* a read resolves to, on any arm.
    #[test]
    fn borrowed_and_owned_reads_agree_across_column_shapes() {
        let mut interner = StringInterner::new();
        let list_key = interner.get_or_intern("vec");
        let int_key = interner.get_or_intern("age");
        let str_key = interner.get_or_intern("name");
        let absent_key = interner.get_or_intern("nope");
        let mut store = ColumnStore::new(Arc::new(TypeSchema::new()), &HashMap::new(), &interner);
        let row_id = store.push_row(&[
            (list_key, Value::List(vec![Value::Int64(7)])),
            (int_key, Value::Int64(41)),
            (str_key, Value::String("ada".into())),
        ]);

        for key in [list_key, int_key, str_key, absent_key] {
            assert_eq!(
                store.get_cow(row_id, key).map(|v| v.into_owned()),
                store.get(row_id, key),
                "the borrowing read diverged from the owning read"
            );
        }
        // The fixed-width and string columns build their value on read and so
        // cannot lend one — this is the shape of the borrow, asserted so a
        // future "optimization" that pretends otherwise is caught here.
        assert!(matches!(
            store.get_cow(row_id, list_key),
            Some(std::borrow::Cow::Borrowed(_))
        ));
        assert!(matches!(
            store.get_cow(row_id, int_key),
            Some(std::borrow::Cow::Owned(_))
        ));
        assert!(store.get_cow(row_id, absent_key).is_none());
    }
}