use std::collections::HashMap;
use vantage_core::{Result, error};
use vantage_table::table::Table;
use vantage_types::{EmptyEntity, Entity};
use crate::mongodb::MongoDB;
use crate::select::{JoinColumn, JoinHop, join_alias};
use crate::types::AnyMongoType;
pub(crate) fn imported_columns<E: Entity<AnyMongoType>>(table: &Table<MongoDB, E>) -> Vec<&str> {
table
.columns()
.keys()
.filter(|name| table.is_imported_column(name))
.map(String::as_str)
.collect()
}
pub(crate) fn join_alias_of<E: Entity<AnyMongoType>>(
table: &Table<MongoDB, E>,
name: &str,
) -> Option<String> {
imported_columns(table)
.iter()
.position(|n| *n == name)
.map(join_alias)
}
pub(crate) fn table_joins<E: Entity<AnyMongoType>>(
table: &Table<MongoDB, E>,
) -> Result<Vec<JoinColumn>> {
let names = imported_columns(table);
if names.is_empty() {
return Ok(Vec::new());
}
let mut resolver = Resolver {
root: table.clone().into_entity::<EmptyEntity>(),
hops: HashMap::new(),
};
names.iter().map(|name| resolver.resolve(name)).collect()
}
struct Resolver {
root: Table<MongoDB, EmptyEntity>,
hops: HashMap<String, (JoinHop, Table<MongoDB, EmptyEntity>)>,
}
impl Resolver {
fn resolve(&mut self, name: &str) -> Result<JoinColumn> {
let parts: Vec<&str> = name.split('.').collect();
let Some((field, relations)) = parts.split_last().filter(|(_, r)| !r.is_empty()) else {
return Err(error!("joined column name must be dotted", column = name));
};
let mut hops = Vec::with_capacity(relations.len());
for depth in 1..=relations.len() {
let prefix = relations[..depth].join(".");
if !self.hops.contains_key(&prefix) {
let parent = match depth {
1 => &self.root,
_ => &self.hops[&relations[..depth - 1].join(".")].1,
};
let resolved = hop(parent, relations[depth - 1], name)?;
self.hops.insert(prefix.clone(), resolved);
}
hops.push(self.hops[&prefix].0.clone());
}
let target = &self.hops[&relations.join(".")].1;
Ok(JoinColumn {
name: name.to_string(),
hops,
field: bson_path(target, field),
})
}
}
fn hop(
table: &Table<MongoDB, EmptyEntity>,
relation: &str,
name: &str,
) -> Result<(JoinHop, Table<MongoDB, EmptyEntity>)> {
let fk = table.ref_foreign_key(relation)?;
let target = table.get_ref_target_erased(relation)?;
if target.data_source().view().is_some() {
return Err(error!(
"joined column cannot traverse into a pipeline view",
column = name,
relation = relation
));
}
let foreign_field = match target.id_field() {
Some(_) => bson_path(&target, &target.id_field_name()),
None => "_id".to_string(),
};
let hop = JoinHop {
from: target.table_name().to_string(),
local_field: bson_path(table, &fk),
foreign_field,
};
Ok((hop, target))
}
fn bson_path(table: &Table<MongoDB, EmptyEntity>, column: &str) -> String {
table
.columns()
.get(column)
.and_then(|c| c.alias())
.unwrap_or(column)
.to_string()
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use vantage_table::column::core::Column;
use super::*;
async fn offline() -> MongoDB {
let client = mongodb::Client::with_uri_str("mongodb://localhost:27017")
.await
.unwrap();
MongoDB::new(client, "unit")
}
fn country(db: MongoDB) -> Table<MongoDB, EmptyEntity> {
let mut t = Table::new("countries", db)
.with_column(Column::<AnyMongoType>::new("code").with_alias("iso.alpha2"))
.with_column(Column::<AnyMongoType>::new("name").with_alias("label.en"));
t.set_id_field("code");
t
}
fn client(db: MongoDB) -> Table<MongoDB, EmptyEntity> {
Table::new("clients", db)
.with_column(Column::<AnyMongoType>::new("_id"))
.with_column(Column::<AnyMongoType>::new("name"))
.with_column(Column::<AnyMongoType>::new("country_code").with_alias("geo.country"))
.with_one("country", "country_code", country)
}
#[tokio::test]
async fn resolves_nested_fk_custom_target_id_and_nested_field() {
let db = offline().await;
let builds = Arc::new(AtomicUsize::new(0));
let counter = builds.clone();
let mut orders = Table::<MongoDB, EmptyEntity>::new("orders", db)
.with_column(Column::<AnyMongoType>::new("_id"))
.with_column(Column::<AnyMongoType>::new("client_id").with_alias("refs.client"))
.with_one("client", "client_id", move |db| {
counter.fetch_add(1, Ordering::SeqCst);
client(db)
});
orders.add_imported_column("client.name").unwrap();
orders.add_imported_column("client.country.name").unwrap();
builds.store(0, Ordering::SeqCst);
let joins = table_joins(&orders).unwrap();
assert_eq!(
builds.load(Ordering::SeqCst),
1,
"client built once per read"
);
let to_client = JoinHop {
from: "clients".into(),
local_field: "refs.client".into(),
foreign_field: "_id".into(),
};
let to_country = JoinHop {
from: "countries".into(),
local_field: "geo.country".into(),
foreign_field: "iso.alpha2".into(),
};
assert_eq!(
joins,
vec![
JoinColumn {
name: "client.name".into(),
hops: vec![to_client.clone()],
field: "name".into(),
},
JoinColumn {
name: "client.country.name".into(),
hops: vec![to_client, to_country],
field: "label.en".into(),
},
]
);
assert_eq!(
join_alias_of(&orders, "client.country.name").as_deref(),
Some("__vc1")
);
assert_eq!(join_alias_of(&orders, "client_id"), None);
}
}