Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions crates/lance-csr-python/src/graph.rs
Original file line number Diff line number Diff line change
Expand Up @@ -340,13 +340,19 @@ impl GraphConfig {
/// Get node labels
fn node_labels(&self) -> Vec<String> {
// Extract from the config's node mappings
self.inner.node_mappings.keys().cloned().collect()
self.inner
.iter_node_mappings()
.map(|(name, _)| name.clone())
.collect()
}

/// Get relationship types
fn relationship_types(&self) -> Vec<String> {
// Extract from the config's relationship mappings
self.inner.relationship_mappings.keys().cloned().collect()
self.inner
.iter_relationship_mappings()
.map(|(name, _)| name.clone())
.collect()
}

fn __repr__(&self) -> String {
Expand Down Expand Up @@ -1481,8 +1487,8 @@ impl CypherEngine {
let config = self.inner.config();
format!(
"CypherEngine(nodes={}, relationships={})",
config.node_mappings.len(),
config.relationship_mappings.len()
config.iter_node_mappings().len(),
config.iter_relationship_mappings().len()
)
}
}
Expand Down
7 changes: 7 additions & 0 deletions crates/lance-csr/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,13 @@ let config = GraphConfig::builder()
.build()?;
```

For Rust callers migrating away from direct mapping-field access, use `get_node_mapping(label)` and
`get_relationship_mapping(type_name)` for case-insensitive lookups. `iter_node_mappings()` and
`iter_relationship_mappings()` yield read-only `(normalized_key, mapping)` pairs; iteration order is
unspecified. Build mappings through `GraphConfigBuilder` so keys are normalized. Public fields remain
available for compatibility until a breaking release; replacing direct writes with builder calls avoids
that future migration. The serialized configuration and Python API are unchanged.

## Executing Cypher Queries

- `CypherQuery::new` parses Cypher text into the internal AST.
Expand Down
51 changes: 39 additions & 12 deletions crates/lance-csr/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -21,16 +21,8 @@ use std::collections::HashMap;
/// # Future API Changes
///
/// These fields may become private in a future major version to enforce invariants.
/// Code should migrate to using accessor methods (`get_node_mapping()`, `get_relationship_mapping()`)
/// rather than direct field access.
///
/// # TODO: API Safety
///
/// TODO: Make `node_mappings` and `relationship_mappings` private to prevent external code
/// from bypassing key normalization. This would require:
/// 1. Adding iterator methods for external access (e.g., `iter_node_mappings()`)
/// 2. Making this a breaking API change in next major version
/// 3. Ensuring all internal code uses accessor methods
/// Use the lookup and iterator accessors for reads, and [`GraphConfigBuilder`] for writes.
/// The mapping fields remain public for source compatibility until a breaking release.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraphConfig {
/// Mapping of node labels to their field configurations
Expand Down Expand Up @@ -126,6 +118,18 @@ impl GraphConfig {
self.relationship_mappings.get(&rel_type.to_lowercase())
}

/// Iterate over normalized node labels and their mappings without allowing mutation.
pub fn iter_node_mappings(&self) -> impl ExactSizeIterator<Item = (&String, &NodeMapping)> {
self.node_mappings.iter()
}

/// Iterate over normalized relationship types and their mappings without allowing mutation.
pub fn iter_relationship_mappings(
&self,
) -> impl ExactSizeIterator<Item = (&String, &RelationshipMapping)> {
self.relationship_mappings.iter()
}

/// Validate the configuration
///
/// Checks for:
Expand Down Expand Up @@ -430,6 +434,29 @@ impl RelationshipMapping {
mod tests {
use super::*;

#[test]
fn test_mapping_iterators_keep_normalized_keys_and_original_labels() {
let config = GraphConfig::builder()
.node_table("Person", "id")
.rel_table("KNOWS", "Person", "Person", "src", "dst")
.build()
.unwrap();
let (key, node) = config.iter_node_mappings().next().unwrap();
assert_eq!(key, "person");
assert_eq!(node.label, "Person");
assert!(std::ptr::eq(
node,
config.get_node_mapping("PERSON").unwrap()
));
let (key, relationship) = config.iter_relationship_mappings().next().unwrap();
assert_eq!(key, "knows");
assert_eq!(relationship.relationship_type, "KNOWS");
assert!(std::ptr::eq(
relationship,
config.get_relationship_mapping("KnOwS").unwrap()
));
}

#[test]
fn test_graph_config_builder() {
let config = GraphConfig::builder()
Expand All @@ -439,8 +466,8 @@ mod tests {
.build()
.unwrap();

assert_eq!(config.node_mappings.len(), 2);
assert_eq!(config.relationship_mappings.len(), 1);
assert_eq!(config.iter_node_mappings().len(), 2);
assert_eq!(config.iter_relationship_mappings().len(), 1);

let person_mapping = config.get_node_mapping("Person").unwrap();
assert_eq!(person_mapping.id_field, "person_id");
Expand Down
49 changes: 45 additions & 4 deletions crates/lance-csr/src/datafusion_planner/builder/join_builder.rs
Original file line number Diff line number Diff line change
Expand Up @@ -242,10 +242,8 @@ impl DataFusionPlanner {
.find(|r| r.alias == *var)
{
// Get the relationship mapping to find src/dst field names
if let Some(rel_map) = self
.config
.relationship_mappings
.get(&rel_instance.rel_type)
if let Some(rel_map) =
self.config.get_relationship_mapping(&rel_instance.rel_type)
{
// Generate composite join keys for both src_id and dst_id
// This ensures we're matching the exact same relationship instance
Expand Down Expand Up @@ -696,4 +694,47 @@ mod tests {
"Shared variables should include 'r'"
);
}

#[test]
fn test_shared_relationship_variable_joins_on_its_endpoints() {
// MATCH (a)-[r:KNOWS]->(b) and (c)-[r:KNOWS]->(d), joined on r. Mapping keys are
// stored lower-case, so the upper-case type must still find the KNOWS mapping.
use crate::datafusion_planner::test_fixtures::{person_knows_config, person_scan};
let expand = |src: &str, dst: &str| LogicalOperator::Expand {
input: Box::new(person_scan(src)),
source_variable: src.to_string(),
target_variable: dst.to_string(),
target_label: "Person".to_string(),
relationship_types: vec!["KNOWS".to_string()],
direction: crate::ast::RelationshipDirection::Outgoing,
relationship_variable: Some("r".to_string()),
properties: Default::default(),
target_properties: Default::default(),
};
let join = LogicalOperator::Join {
left: Box::new(expand("a", "b")),
right: Box::new(expand("c", "d")),
join_type: crate::logical_plan::JoinType::Inner,
};
let planner = DataFusionPlanner::with_catalog(person_knows_config(), make_catalog());
let analysis = analysis::analyze(&join).unwrap();
let ctx = crate::datafusion_planner::analysis::PlanningContext::new(&analysis);
let (left, right) = planner.infer_join_keys(&ctx, &join_left(&join), &join_right(&join));
assert_eq!(left, ["r__src_person_id", "r__dst_person_id"]);
assert_eq!(right, left);
}

fn join_left(op: &LogicalOperator) -> LogicalOperator {
match op {
LogicalOperator::Join { left, .. } => (**left).clone(),
_ => unreachable!(),
}
}

fn join_right(op: &LogicalOperator) -> LogicalOperator {
match op {
LogicalOperator::Join { right, .. } => (**right).clone(),
_ => unreachable!(),
}
}
}
12 changes: 6 additions & 6 deletions crates/lance-csr/src/engine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -363,10 +363,10 @@ fn validate_sources(config: &GraphConfig, schemas: &HashMap<String, SchemaRef>)
}
Ok(())
};
for mapping in config.node_mappings.values() {
for (_, mapping) in config.iter_node_mappings() {
check(&mapping.label, &mapping.id_field)?;
}
for mapping in config.relationship_mappings.values() {
for (_, mapping) in config.iter_relationship_mappings() {
check(&mapping.relationship_type, &mapping.source_id_field)?;
check(&mapping.relationship_type, &mapping.target_id_field)?;
}
Expand Down Expand Up @@ -464,7 +464,7 @@ async fn build_node_index(
previous: Option<&EngineState>,
) -> Result<HashMap<String, NodeIndexEntry>> {
let mut entries = HashMap::new();
for mapping in config.node_mappings.values() {
for (_, mapping) in config.iter_node_mappings() {
let label = &mapping.label;
let Some(source) = sources
.iter()
Expand Down Expand Up @@ -561,15 +561,15 @@ async fn build_key_statistics(
let mut to_count = Vec::new();
// Relationship type and endpoint columns, for the fan-out once counts are known.
let mut endpoints = None;
for mapping in config.node_mappings.values() {
for (_, mapping) in config.iter_node_mappings() {
if !mapping.label.eq_ignore_ascii_case(name) {
continue;
}
if let Some(id) = column(&mapping.id_field) {
distinct.insert(id.name().clone(), source.num_rows(name).await?);
}
}
for mapping in config.relationship_mappings.values() {
for (_, mapping) in config.iter_relationship_mappings() {
if !mapping.relationship_type.eq_ignore_ascii_case(name) {
continue;
}
Expand Down Expand Up @@ -662,7 +662,7 @@ async fn build_csr_entries(
previous: Option<&EngineState>,
) -> Result<HashMap<String, CsrEntry>> {
let mut entries = HashMap::new();
for mapping in config.relationship_mappings.values() {
for (_, mapping) in config.iter_relationship_mappings() {
let rel_type = &mapping.relationship_type;
let Some((name, source)) = sources
.iter()
Expand Down
19 changes: 11 additions & 8 deletions crates/lance-csr/src/query.rs
Original file line number Diff line number Diff line change
Expand Up @@ -480,7 +480,7 @@ impl CypherQuery {
let mut catalog = InMemoryCatalog::new();

// Register node sources
for label in config.node_mappings.keys() {
for (label, _) in config.iter_node_mappings() {
let table_provider =
ctx.table_provider(label)
.await
Expand All @@ -497,7 +497,7 @@ impl CypherQuery {
}

// Register relationship sources
for rel_type in config.relationship_mappings.keys() {
for (rel_type, _) in config.iter_relationship_mappings() {
let table_provider =
ctx.table_provider(rel_type)
.await
Expand Down Expand Up @@ -745,12 +745,15 @@ impl CypherQuery {
let mut required_tables: HashSet<String> = HashSet::new();
// Use original label/type names (not lowercase keys) for namespace resolution
// The namespace needs the original casing to find files on disk
required_tables.extend(config.node_mappings.values().map(|m| m.label.clone()));
required_tables.extend(
config
.relationship_mappings
.values()
.map(|m| m.relationship_type.clone()),
.iter_node_mappings()
.map(|(_, mapping)| mapping.label.clone()),
);
required_tables.extend(
config
.iter_relationship_mappings()
.map(|(_, mapping)| mapping.relationship_type.clone()),
);

if required_tables.is_empty() {
Expand Down Expand Up @@ -816,7 +819,7 @@ impl CypherQuery {
providers.insert(normalized_table_name.clone(), provider);
}

for label in config.node_mappings.keys() {
for (label, _) in config.iter_node_mappings() {
let provider = providers
.get(label)
.ok_or_else(|| GraphError::ConfigError {
Expand All @@ -831,7 +834,7 @@ impl CypherQuery {
catalog = catalog.with_node_source(label, table_source);
}

for rel_type in config.relationship_mappings.keys() {
for (rel_type, _) in config.iter_relationship_mappings() {
let provider = providers
.get(rel_type)
.ok_or_else(|| GraphError::ConfigError {
Expand Down
Loading