use std::{hash::Hash, marker::PhantomData};
use crate::{
map_query::{MapDiffSink, MapQuery, MapQueryInstall},
subscription::SubscriptionGuard,
traits::{
CellValue, HasForeignKey, IdFor,
collections::internal::join_runtime::install_join_runtime_via_query,
},
};
pub struct LeftJoinPlan<L, R, LK, LV, RK, RV, JK, FL, FR>
where
L: MapQuery<LK, LV>,
R: MapQuery<RK, RV>,
LK: Hash + Eq + CellValue,
LV: CellValue,
RK: Hash + Eq + CellValue,
RV: CellValue,
JK: Hash + Eq + CellValue,
FL: Fn(&LK, &LV) -> JK + Send + Sync + 'static,
FR: Fn(&RK, &RV) -> JK + Send + Sync + 'static,
{
pub(crate) left: L,
pub(crate) right: R,
pub(crate) left_key: FL,
pub(crate) right_key: FR,
#[allow(clippy::type_complexity)]
pub(crate) _types: PhantomData<fn() -> (LK, LV, RK, RV, JK)>,
}
impl<L, R, LK, LV, RK, RV, JK, FL, FR> MapQueryInstall<LK, (LV, Vec<RV>)>
for LeftJoinPlan<L, R, LK, LV, RK, RV, JK, FL, FR>
where
L: MapQuery<LK, LV>,
R: MapQuery<RK, RV>,
LK: Hash + Eq + CellValue,
LV: CellValue,
RK: Hash + Eq + CellValue,
RV: CellValue,
JK: Hash + Eq + CellValue,
FL: Fn(&LK, &LV) -> JK + Send + Sync + 'static,
FR: Fn(&RK, &RV) -> JK + Send + Sync + 'static,
{
fn install(self, sink: MapDiffSink<LK, (LV, Vec<RV>)>) -> Vec<SubscriptionGuard> {
install_join_runtime_via_query::<LK, LV, RK, RV, JK, LK, (LV, Vec<RV>), _, _, _, _, _>(
self.left,
self.right,
self.left_key,
self.right_key,
|left_k: &LK, left_v: &LV, rights: &[(RK, RV)]| {
let right_values: Vec<RV> = rights.iter().map(|(_, rv)| rv.clone()).collect();
vec![(left_k.clone(), (left_v.clone(), right_values))]
},
sink,
)
}
}
#[allow(private_bounds)]
impl<L, R, LK, LV, RK, RV, JK, FL, FR> MapQuery<LK, (LV, Vec<RV>)>
for LeftJoinPlan<L, R, LK, LV, RK, RV, JK, FL, FR>
where
L: MapQuery<LK, LV>,
R: MapQuery<RK, RV>,
LK: Hash + Eq + CellValue,
LV: CellValue,
RK: Hash + Eq + CellValue,
RV: CellValue,
JK: Hash + Eq + CellValue,
FL: Fn(&LK, &LV) -> JK + Send + Sync + 'static,
FR: Fn(&RK, &RV) -> JK + Send + Sync + 'static,
{
}
pub trait LeftJoinExt<K, V>: MapQuery<K, V>
where
K: Hash + Eq + CellValue,
V: CellValue,
{
#[allow(clippy::type_complexity)]
fn left_join<R, RV>(
self,
right: R,
) -> LeftJoinPlan<
Self,
R,
K,
V,
K,
RV,
K,
impl Fn(&K, &V) -> K + Send + Sync + 'static,
impl Fn(&K, &RV) -> K + Send + Sync + 'static,
>
where
R: MapQuery<K, RV>,
RV: CellValue,
{
LeftJoinPlan {
left: self,
right,
left_key: |k: &K, _: &V| k.clone(),
right_key: |k: &K, _: &RV| k.clone(),
_types: PhantomData,
}
}
#[allow(clippy::type_complexity)]
fn left_join_fk<R, RK, RV>(
self,
right: R,
) -> LeftJoinPlan<
Self,
R,
K,
V,
RK,
RV,
K,
impl Fn(&K, &V) -> K + Send + Sync + 'static,
impl Fn(&RK, &RV) -> K + Send + Sync + 'static,
>
where
R: MapQuery<RK, RV>,
RK: Hash + Eq + CellValue,
RV: CellValue + HasForeignKey<V>,
<<RV as HasForeignKey<V>>::ForeignKey as IdFor<V>>::MapKey: Into<K>,
{
LeftJoinPlan {
left: self,
right,
left_key: |k: &K, _: &V| k.clone(),
right_key: |_: &RK, rv: &RV| rv.fk().map_key().into(),
_types: PhantomData,
}
}
fn left_join_by<R, RK, RV, JK, FL, FR>(
self,
right: R,
left_key: FL,
right_key: FR,
) -> LeftJoinPlan<Self, R, K, V, RK, RV, JK, FL, FR>
where
R: MapQuery<RK, RV>,
RK: Hash + Eq + CellValue,
RV: CellValue,
JK: Hash + Eq + CellValue,
FL: Fn(&K, &V) -> JK + Send + Sync + 'static,
FR: Fn(&RK, &RV) -> JK + Send + Sync + 'static,
{
LeftJoinPlan {
left: self,
right,
left_key,
right_key,
_types: PhantomData,
}
}
}
impl<K, V, M> LeftJoinExt<K, V> for M
where
K: Hash + Eq + CellValue,
V: CellValue,
M: MapQuery<K, V>,
{
}
#[cfg(test)]
mod tests {
use std::sync::mpsc;
use super::*;
use crate::{
CellMap, MapDiff,
traits::{Gettable, HasForeignKey, IdFor, IdType},
};
#[test]
fn left_join_keeps_unmatched_left_rows() {
let left = CellMap::<String, i32>::new();
let right = CellMap::<String, i32>::new();
let joined = left.clone().left_join(right.clone()).materialize();
left.insert("a".to_string(), 1);
assert_eq!(joined.get_value(&"a".to_string()), Some((1, vec![])));
}
#[test]
fn left_join_pairs_matched_rows() {
let left = CellMap::<String, i32>::new();
let right = CellMap::<String, i32>::new();
let joined = left.clone().left_join(right.clone()).materialize();
left.insert("a".to_string(), 1);
right.insert("a".to_string(), 10);
assert_eq!(joined.get_value(&"a".to_string()), Some((1, vec![10])));
}
#[test]
fn left_join_reacts_to_right_addition() {
let left = CellMap::<String, i32>::new();
let right = CellMap::<String, i32>::new();
let joined = left.clone().left_join(right.clone()).materialize();
left.insert("a".to_string(), 1);
assert_eq!(joined.get_value(&"a".to_string()), Some((1, vec![])));
right.insert("a".to_string(), 10);
assert_eq!(joined.get_value(&"a".to_string()), Some((1, vec![10])));
}
#[test]
fn left_join_reacts_to_right_removal() {
let left = CellMap::<String, i32>::new();
let right = CellMap::<String, i32>::new();
let joined = left.clone().left_join(right.clone()).materialize();
left.insert("a".to_string(), 1);
right.insert("a".to_string(), 10);
assert_eq!(joined.get_value(&"a".to_string()), Some((1, vec![10])));
right.remove(&"a".to_string());
assert_eq!(joined.get_value(&"a".to_string()), Some((1, vec![])));
}
#[test]
fn left_join_reacts_to_left_removal() {
let left = CellMap::<String, i32>::new();
let right = CellMap::<String, i32>::new();
let joined = left.clone().left_join(right.clone()).materialize();
left.insert("a".to_string(), 1);
right.insert("a".to_string(), 10);
assert_eq!(joined.entries().get().len(), 1);
left.remove(&"a".to_string());
assert_eq!(joined.entries().get().len(), 0);
}
#[test]
fn left_join_by_collects_multiple_right_matches() {
let left = CellMap::<String, (String, i32)>::new();
let right = CellMap::<String, (String, i32)>::new();
let joined = left
.clone()
.left_join_by(right.clone(), |_, lv| lv.0.clone(), |_, rv| rv.0.clone())
.materialize();
left.insert("l1".to_string(), ("g1".to_string(), 10));
right.insert("r1".to_string(), ("g1".to_string(), 5));
right.insert("r2".to_string(), ("g1".to_string(), 7));
let val = joined.get_value(&"l1".to_string());
assert!(val.is_some());
let (left_val, right_vals) = val.unwrap();
assert_eq!(left_val, ("g1".to_string(), 10));
assert_eq!(right_vals.len(), 2);
}
#[test]
fn left_join_by_keeps_unmatched_with_empty_vec() {
let left = CellMap::<String, (String, i32)>::new();
let right = CellMap::<String, (String, i32)>::new();
let joined = left
.clone()
.left_join_by(right.clone(), |_, lv| lv.0.clone(), |_, rv| rv.0.clone())
.materialize();
left.insert("l1".to_string(), ("g1".to_string(), 10));
let val = joined.get_value(&"l1".to_string());
assert!(val.is_some());
let (left_val, right_vals) = val.unwrap();
assert_eq!(left_val, ("g1".to_string(), 10));
assert_eq!(right_vals.len(), 0);
}
#[test]
fn left_join_by_preserves_right_batch() {
let left = CellMap::<String, (String, i32)>::new();
left.insert("l1".to_string(), ("g1".to_string(), 10));
let right = CellMap::<String, (String, i32)>::new();
let joined = left
.clone()
.left_join_by(right.clone(), |_, lv| lv.0.clone(), |_, rv| rv.0.clone())
.materialize();
let (tx, rx) = mpsc::channel::<MapDiff<String, ((String, i32), Vec<(String, i32)>)>>();
let _guard = joined.subscribe_diffs(move |diff| {
let _ = tx.send(diff.clone());
});
right.insert_many(vec![
("r1".to_string(), ("g1".to_string(), 5)),
("r2".to_string(), ("g1".to_string(), 7)),
]);
let seen: Vec<_> = rx.try_iter().collect();
assert_eq!(seen.len(), 2);
match seen.last().expect("last diff") {
MapDiff::Batch { changes } => assert!(!changes.is_empty()),
_ => panic!("expected batch diff from left_join_by"),
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct UserId(String);
#[derive(Debug, Clone, PartialEq)]
struct User {
name: String,
}
impl IdFor<User> for UserId {
type MapKey = String;
fn map_key(&self) -> String {
self.0.clone()
}
}
impl IdType for UserId {
type Parent = User;
}
#[derive(Debug, Clone, PartialEq)]
struct Post {
user_id: UserId,
title: String,
}
impl HasForeignKey<User> for Post {
type ForeignKey = UserId;
fn fk(&self) -> UserId {
self.user_id.clone()
}
}
#[test]
fn left_join_fk_keeps_unmatched_with_empty_vec() {
let users = CellMap::<String, User>::new();
let posts = CellMap::<String, Post>::new();
let joined = users.clone().left_join_fk(posts.clone()).materialize();
users.insert(
"u1".to_string(),
User {
name: "Alice".to_string(),
},
);
let val = joined.get_value(&"u1".to_string());
assert!(val.is_some());
let (user, posts) = val.unwrap();
assert_eq!(user.name, "Alice");
assert_eq!(posts.len(), 0);
}
#[test]
fn left_join_fk_collects_matching_posts() {
let users = CellMap::<String, User>::new();
let posts = CellMap::<String, Post>::new();
let joined = users.clone().left_join_fk(posts.clone()).materialize();
users.insert(
"u1".to_string(),
User {
name: "Alice".to_string(),
},
);
posts.insert(
"p1".to_string(),
Post {
user_id: UserId("u1".to_string()),
title: "Hello".to_string(),
},
);
posts.insert(
"p2".to_string(),
Post {
user_id: UserId("u1".to_string()),
title: "World".to_string(),
},
);
let val = joined.get_value(&"u1".to_string());
assert!(val.is_some());
let (_, matched_posts) = val.unwrap();
assert_eq!(matched_posts.len(), 2);
}
}