selene-db-graph 1.3.0

use roaring::RoaringBitmap;
use selene_core::{DbString, LabelSet, PropertyMap, Value, db_string};
use smallvec::{SmallVec, smallvec};

use super::*;
use crate::graph::{CompositePropertyIndexEntry, composite_property_key};
use crate::{CompositeTypedIndex, TypedIndexKind};

fn property_map(pairs: impl IntoIterator<Item = (DbString, Value)>) -> PropertyMap {
    PropertyMap::from_pairs(pairs).unwrap()
}

fn decimal(value: &str) -> rust_decimal::Decimal {
    value.parse().expect("test decimal parses")
}

fn entry(
    properties: SmallVec<[DbString; 4]>,
    kinds: SmallVec<[TypedIndexKind; 4]>,
) -> CompositePropertyIndexEntry {
    CompositePropertyIndexEntry::new(CompositeTypedIndex::new(kinds), properties, None)
}

fn insert_entry(
    indexes: &mut CompositeIndexMap,
    label: DbString,
    properties: SmallVec<[DbString; 4]>,
    kinds: SmallVec<[TypedIndexKind; 4]>,
) {
    indexes.insert(
        (label, composite_property_key(&properties)),
        CompositePropertyIndexEntry::new(CompositeTypedIndex::new(kinds), properties, None),
    );
}

fn rows(
    indexes: &CompositeIndexMap,
    label: DbString,
    properties: &[DbString],
    values: &[Value],
) -> RoaringBitmap {
    let Some(entry) = indexes.get(&(label, composite_property_key(properties))) else {
        return RoaringBitmap::new();
    };
    let refs = values.iter().collect::<Vec<_>>();
    let key = entry.index.key_from_values(&refs).unwrap();
    entry.index.lookup_key(&key).cloned().unwrap_or_default()
}

#[test]
fn apply_create_update_delete_moves_composite_rows() {
    let label = db_string("cpi.maintenance.label").unwrap();
    let ts = db_string("cpi.maintenance.ts").unwrap();
    let location = db_string("cpi.maintenance.location").unwrap();
    let mut indexes = CompositeIndexMap::default();
    let properties = smallvec![ts.clone(), location.clone()];
    insert_entry(
        &mut indexes,
        label.clone(),
        properties.clone(),
        smallvec![TypedIndexKind::I64, TypedIndexKind::String],
    );
    let old_props = property_map([
        (ts.clone(), Value::Int(1)),
        (location.clone(), Value::String(db_string("north").unwrap())),
    ]);
    let new_props = property_map([
        (ts, Value::Int(2)),
        (location, Value::String(db_string("north").unwrap())),
    ]);

    apply_node_create(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        3,
    )
    .unwrap();
    assert!(
        rows(
            &indexes,
            label.clone(),
            &properties,
            &[Value::Int(1), Value::String(db_string("north").unwrap())]
        )
        .contains(3)
    );

    apply_node_update(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        &LabelSet::single(label.clone()),
        &new_props,
        3,
    )
    .unwrap();
    assert!(
        rows(
            &indexes,
            label.clone(),
            &properties,
            &[Value::Int(1), Value::String(db_string("north").unwrap())]
        )
        .is_empty()
    );
    assert!(
        rows(
            &indexes,
            label.clone(),
            &properties,
            &[Value::Int(2), Value::String(db_string("north").unwrap())]
        )
        .contains(3)
    );

    apply_node_delete(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &new_props,
        3,
    )
    .unwrap();
    assert!(
        rows(
            &indexes,
            label,
            &properties,
            &[Value::Int(2), Value::String(db_string("north").unwrap())]
        )
        .is_empty()
    );
}

#[test]
fn apply_update_moves_bool_composite_component() {
    let label = db_string("cpi.bool.label").unwrap();
    let active = db_string("cpi.bool.active").unwrap();
    let location = db_string("cpi.bool.location").unwrap();
    let mut indexes = CompositeIndexMap::default();
    let properties = smallvec![active.clone(), location.clone()];
    insert_entry(
        &mut indexes,
        label.clone(),
        properties.clone(),
        smallvec![TypedIndexKind::Bool, TypedIndexKind::String],
    );
    let old_props = property_map([
        (active.clone(), Value::Bool(false)),
        (location.clone(), Value::String(db_string("north").unwrap())),
    ]);
    let new_props = property_map([
        (active.clone(), Value::Bool(true)),
        (location.clone(), Value::String(db_string("north").unwrap())),
    ]);

    apply_node_create(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        4,
    )
    .unwrap();
    apply_node_update(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        &LabelSet::single(label.clone()),
        &new_props,
        4,
    )
    .unwrap();

    assert!(
        rows(
            &indexes,
            label.clone(),
            &properties,
            &[
                Value::Bool(false),
                Value::String(db_string("north").unwrap())
            ]
        )
        .is_empty()
    );
    assert!(
        rows(
            &indexes,
            label,
            &properties,
            &[
                Value::Bool(true),
                Value::String(db_string("north").unwrap())
            ]
        )
        .contains(4)
    );
}

#[test]
fn apply_update_moves_u64_composite_component() {
    let label = db_string("cpi.u64.label").unwrap();
    let count = db_string("cpi.u64.count").unwrap();
    let location = db_string("cpi.u64.location").unwrap();
    let mut indexes = CompositeIndexMap::default();
    let properties = smallvec![count.clone(), location.clone()];
    insert_entry(
        &mut indexes,
        label.clone(),
        properties.clone(),
        smallvec![TypedIndexKind::U64, TypedIndexKind::String],
    );
    let old_props = property_map([
        (count.clone(), Value::Uint(7)),
        (location.clone(), Value::String(db_string("north").unwrap())),
    ]);
    let new_props = property_map([
        (count.clone(), Value::Uint(8)),
        (location.clone(), Value::String(db_string("north").unwrap())),
    ]);

    apply_node_create(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        5,
    )
    .unwrap();
    apply_node_update(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        &LabelSet::single(label.clone()),
        &new_props,
        5,
    )
    .unwrap();

    assert!(
        rows(
            &indexes,
            label.clone(),
            &properties,
            &[Value::Uint(7), Value::String(db_string("north").unwrap())]
        )
        .is_empty()
    );
    assert!(
        rows(
            &indexes,
            label,
            &properties,
            &[Value::Uint(8), Value::String(db_string("north").unwrap())]
        )
        .contains(5)
    );
}

#[test]
fn apply_update_moves_exact_numeric_composite_components() {
    let label = db_string("cpi.exact.label").unwrap();
    let signed = db_string("cpi.exact.signed").unwrap();
    let unsigned = db_string("cpi.exact.unsigned").unwrap();
    let amount = db_string("cpi.exact.amount").unwrap();
    let mut indexes = CompositeIndexMap::default();
    let properties = smallvec![signed.clone(), unsigned.clone(), amount.clone()];
    insert_entry(
        &mut indexes,
        label.clone(),
        properties.clone(),
        smallvec![
            TypedIndexKind::I128,
            TypedIndexKind::U128,
            TypedIndexKind::Decimal
        ],
    );
    let old_props = property_map([
        (signed.clone(), Value::Int128(i128::MIN + 8)),
        (unsigned.clone(), Value::Uint128(u64::MAX as u128 + 8)),
        (amount.clone(), Value::Decimal(decimal("8.25"))),
    ]);
    let new_props = property_map([
        (signed.clone(), Value::Int128(i128::MAX - 8)),
        (unsigned.clone(), Value::Uint128(u128::MAX - 8)),
        (amount.clone(), Value::Decimal(decimal("9.50"))),
    ]);

    apply_node_create(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        6,
    )
    .unwrap();
    apply_node_update(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &old_props,
        &LabelSet::single(label.clone()),
        &new_props,
        6,
    )
    .unwrap();

    assert!(
        rows(
            &indexes,
            label.clone(),
            &properties,
            &[
                Value::Int128(i128::MIN + 8),
                Value::Uint128(u64::MAX as u128 + 8),
                Value::Decimal(decimal("8.25")),
            ]
        )
        .is_empty()
    );
    assert!(
        rows(
            &indexes,
            label,
            &properties,
            &[
                Value::Int128(i128::MAX - 8),
                Value::Uint128(u128::MAX - 8),
                Value::Decimal(decimal("9.50")),
            ]
        )
        .contains(6)
    );
}

#[test]
fn apply_create_skips_partial_composite_values() {
    let label = db_string("cpi.partial.label").unwrap();
    let ts = db_string("cpi.partial.ts").unwrap();
    let location = db_string("cpi.partial.location").unwrap();
    let mut indexes = CompositeIndexMap::default();
    insert_entry(
        &mut indexes,
        label.clone(),
        smallvec![ts.clone(), location],
        smallvec![TypedIndexKind::I64, TypedIndexKind::String],
    );
    let props = property_map([(ts, Value::Int(1))]);

    apply_node_create(&mut indexes, &LabelSet::single(label), &props, 0).unwrap();

    let entry = indexes
        .values()
        .next()
        .expect("composite registration remains");
    assert_eq!(entry.index.cardinality(), 0);
}

#[test]
fn apply_update_label_remove_deletes_composite_row() {
    let label = db_string("cpi.label-remove.label").unwrap();
    let ts = db_string("cpi.label-remove.ts").unwrap();
    let location = db_string("cpi.label-remove.location").unwrap();
    let mut indexes = CompositeIndexMap::default();
    let properties = smallvec![ts.clone(), location.clone()];
    insert_entry(
        &mut indexes,
        label.clone(),
        properties.clone(),
        smallvec![TypedIndexKind::I64, TypedIndexKind::String],
    );
    let props = property_map([
        (ts, Value::Int(1)),
        (location, Value::String(db_string("north").unwrap())),
    ]);
    apply_node_create(&mut indexes, &LabelSet::single(label.clone()), &props, 8).unwrap();

    apply_node_update(
        &mut indexes,
        &LabelSet::single(label.clone()),
        &props,
        &LabelSet::new(),
        &props,
        8,
    )
    .unwrap();

    assert!(
        rows(
            &indexes,
            label,
            &properties,
            &[Value::Int(1), Value::String(db_string("north").unwrap())]
        )
        .is_empty()
    );
}

#[test]
fn rebuild_composite_property_indexes_is_lenient_on_kind_drift() {
    let label = db_string("cpi.rebuild.label").unwrap();
    let ts = db_string("cpi.rebuild.ts").unwrap();
    let location = db_string("cpi.rebuild.location").unwrap();
    let mut graph = crate::SeleneGraph::new(selene_core::GraphId::new(1));
    graph
        .node_store
        .labels
        .push(LabelSet::single(label.clone()));
    graph.node_store.properties.push(property_map([
        (ts.clone(), Value::Int(1)),
        (location.clone(), Value::String(db_string("north").unwrap())),
    ]));
    graph.node_store.alive_mut().insert(0);
    graph
        .node_store
        .labels
        .push(LabelSet::single(label.clone()));
    graph.node_store.properties.push(property_map([
        (ts.clone(), Value::String(db_string("wrong").unwrap())),
        (location.clone(), Value::String(db_string("south").unwrap())),
    ]));
    graph.node_store.alive_mut().insert(1);
    graph.composite_property_index.insert(
        (
            label.clone(),
            composite_property_key(&[ts.clone(), location.clone()]),
        ),
        entry(
            smallvec![ts.clone(), location.clone()],
            smallvec![TypedIndexKind::I64, TypedIndexKind::String],
        ),
    );

    rebuild_composite_property_indexes(&mut graph).unwrap();

    let rows = rows(
        &graph.composite_property_index,
        label,
        &[ts, location],
        &[Value::Int(1), Value::String(db_string("north").unwrap())],
    );
    assert_eq!(rows.iter().collect::<Vec<_>>(), vec![0]);
}

#[test]
fn apply_create_admits_string_string_component() {
    // Composite (I64, STRING) admits a row whose STRING component is a
    // `Value::String`, and the row is findable by the same key.
    let label = db_string("cpi.string.create.label").unwrap();
    let ts = db_string("cpi.string.create.ts").unwrap();
    let location = db_string("cpi.string.create.location").unwrap();
    let mut indexes = CompositeIndexMap::default();
    let properties = smallvec![ts.clone(), location.clone()];
    insert_entry(
        &mut indexes,
        label.clone(),
        properties.clone(),
        smallvec![TypedIndexKind::I64, TypedIndexKind::String],
    );
    let probe = db_string("cpi.string.create.unique-1").unwrap();
    let props = property_map([
        (ts, Value::Int(42)),
        (location, Value::String(probe.clone())),
    ]);

    apply_node_create(&mut indexes, &LabelSet::single(label.clone()), &props, 4).unwrap();

    assert!(
        rows(
            &indexes,
            label,
            &properties,
            &[Value::Int(42), Value::String(probe)]
        )
        .contains(4)
    );
}