diff --git a/crates/graphforge-api/tests/e2e_baseline.rs b/crates/graphforge-api/tests/e2e_baseline.rs index fc66cb320..543ec1c5f 100644 --- a/crates/graphforge-api/tests/e2e_baseline.rs +++ b/crates/graphforge-api/tests/e2e_baseline.rs @@ -2404,6 +2404,266 @@ fn fixed_hop_nodes_preserve_endpoint_labels_and_properties() { assert_eq!(titles.value(1), "Acme"); } +/// Labels list at path-node index `node_idx` in `nodes(p)` column `col_name`. +fn path_node_labels_at( + batch: &arrow::record_batch::RecordBatch, + col_name: &str, + row: usize, + node_idx: usize, +) -> Vec { + let list = batch + .column_by_name(col_name) + .expect("column") + .as_any() + .downcast_ref::() + .expect("List column"); + let items_arr = list.value(row); + let items = items_arr + .as_any() + .downcast_ref::() + .expect("node structs"); + let labels = items + .column_by_name("labels") + .expect("labels") + .as_any() + .downcast_ref::() + .expect("labels List"); + utf8_list_cell(labels, node_idx) +} + +#[test] +fn variable_length_path_nodes_preserve_full_multi_labels() { + // #705: Person:Employee through nodes(p) keeps both labels for bounded and + // unbounded VL patterns, matching direct labels() / label predicates and + // fixed-hop path values. Empty *0.. paths and reopen stay deterministic. + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().to_str().unwrap(); + let gf = GraphForge::new(Some(path)).expect("persistent instance"); + gf.execute( + "CREATE \ + (a:Person:Employee {name:'Alice'}), \ + (b:Person {name:'Bob'}), \ + (c:Company {name:'Acme'}), \ + (a)-[:KNOWS]->(b), \ + (b)-[:KNOWS]->(a), \ + (a)-[:WORKS_AT]->(c)", + ) + .expect("create multi-label VL fixture"); + + let direct = rows( + &gf, + "MATCH (n:Person:Employee {name:'Alice'}) \ + RETURN labels(n) AS labels, n:Person AS is_person, n:Employee AS is_employee", + ); + assert_eq!(direct.stats.rows_produced, 1); + let direct_labels = utf8_list_cell( + direct.batches[0] + .column_by_name("labels") + .expect("labels") + .as_any() + .downcast_ref::() + .expect("labels List"), + 0, + ); + assert_eq!(direct_labels, vec!["Person", "Employee"]); + assert_eq!(bool_cell(&direct, "is_person", 0), Some(true)); + assert_eq!(bool_cell(&direct, "is_employee", 0), Some(true)); + + for pattern in [ + "MATCH p = (a:Person:Employee {name:'Alice'})-[:KNOWS*1..2]->(b:Person) \ + RETURN nodes(p) AS ns, length(p) AS l", + "MATCH p = (a:Person:Employee {name:'Alice'})-[:KNOWS*]->(b:Person) \ + RETURN nodes(p) AS ns, length(p) AS l", + ] { + let r = rows(&gf, pattern); + assert!(r.stats.rows_produced >= 1, "pattern must match: {pattern}"); + for batch in &r.batches { + for row in 0..batch.num_rows() { + let labels = path_node_labels_at(batch, "ns", row, 0); + assert_eq!( + labels, direct_labels, + "VL path start node labels must match direct labels()" + ); + } + } + } + + // Fixed-hop parity with the same multi-label endpoint. + let fixed = rows( + &gf, + "MATCH p = (a:Person:Employee {name:'Alice'})-[:WORKS_AT]->(c:Company) \ + RETURN nodes(p) AS ns", + ); + assert_eq!(fixed.stats.rows_produced, 1); + assert_eq!( + path_node_labels_at(&fixed.batches[0], "ns", 0, 0), + direct_labels + ); + assert_eq!( + path_node_labels_at(&fixed.batches[0], "ns", 0, 1), + vec!["Company"] + ); + + // Empty / zero-hop path still hydrates the seed node's full label set. + let zero = rows( + &gf, + "MATCH p = (a:Person:Employee {name:'Alice'})-[:KNOWS*0..0]->(b) \ + RETURN nodes(p) AS ns, length(p) AS l", + ); + assert_eq!(zero.stats.rows_produced, 1); + assert_eq!( + path_node_labels_at(&zero.batches[0], "ns", 0, 0), + direct_labels + ); + + // Repeated node on a 2-hop walk Alice→Bob→Alice keeps full labels both times. + let repeated = rows( + &gf, + "MATCH p = (a:Person:Employee {name:'Alice'})-[:KNOWS*2..2]->(b:Person:Employee) \ + RETURN nodes(p) AS ns", + ); + assert_eq!(repeated.stats.rows_produced, 1); + assert_eq!( + path_node_labels_at(&repeated.batches[0], "ns", 0, 0), + direct_labels + ); + assert_eq!( + path_node_labels_at(&repeated.batches[0], "ns", 0, 2), + direct_labels + ); + assert_eq!( + path_node_labels_at(&repeated.batches[0], "ns", 0, 1), + vec!["Person"] + ); + + // labels(x) over path nodes agrees with the hydrated list; membership + // predicates use labels(x) because `x:Label` requires a bound node var. + let via_list = rows( + &gf, + "MATCH p = (a:Person:Employee {name:'Alice'})-[:KNOWS*1..1]->(b:Person) \ + RETURN [x IN nodes(p) | labels(x)] AS path_labels, \ + [x IN nodes(p) | 'Employee' IN labels(x)] AS is_employee", + ); + assert_eq!(via_list.stats.rows_produced, 1); + let path_labels = via_list.batches[0] + .column_by_name("path_labels") + .expect("path_labels") + .as_any() + .downcast_ref::() + .expect("path_labels List"); + let first = path_labels.value(0); + let first = first + .as_any() + .downcast_ref::() + .expect("nested label lists"); + assert_eq!(utf8_list_cell(first, 0), direct_labels); + assert_eq!(utf8_list_cell(first, 1), vec!["Person"]); + let preds = via_list.batches[0] + .column_by_name("is_employee") + .expect("is_employee") + .as_any() + .downcast_ref::() + .expect("predicate list"); + let pred_vals = preds.value(0); + let pred_vals = pred_vals + .as_any() + .downcast_ref::() + .expect("bool list"); + assert_eq!(pred_vals.value(0), true); + assert_eq!(pred_vals.value(1), false); + + // Schema + values survive reopen. + let schema_before = r_schema_labels_field(&gf); + drop(gf); + let gf2 = GraphForge::new(Some(path)).expect("reopen"); + let after = rows( + &gf2, + "MATCH p = (a:Person:Employee {name:'Alice'})-[:KNOWS*1..2]->(b:Person) \ + RETURN nodes(p) AS ns", + ); + assert!(after.stats.rows_produced >= 1); + assert_eq!( + path_node_labels_at(&after.batches[0], "ns", 0, 0), + direct_labels + ); + assert_eq!(r_schema_labels_field(&gf2), schema_before); +} + +fn r_schema_labels_field(gf: &GraphForge) -> DataType { + let r = rows( + gf, + "MATCH p = (a:Person:Employee {name:'Alice'})-[:KNOWS*1..1]->(b:Person) \ + RETURN nodes(p) AS ns", + ); + let field = r.schema.field_with_name("ns").expect("ns"); + let DataType::List(item) = field.data_type() else { + panic!("ns must be List, got {:?}", field.data_type()); + }; + let DataType::Struct(fields) = item.data_type() else { + panic!("ns item must be Struct, got {:?}", item.data_type()); + }; + fields + .iter() + .find(|f| f.name() == "labels") + .expect("labels field") + .data_type() + .clone() +} + +#[test] +fn variable_length_path_nodes_single_label_and_advisory_runtime() { + // #705: single-label nodes stay single-element lists; advisory/runtime + // labels resolve through the runtime catalog domain (not ontology IDs). + let gf = GraphForge::new(None).expect("in-memory instance"); + gf.execute( + "CREATE (a:Person {name:'Alice'})-[:KNOWS]->(b:Person {name:'Bob'}), \ + (a)-[:KNOWS]->(c:Contractor {name:'Chris'})", + ) + .expect("create single-label / runtime-label fixture"); + + let single = rows( + &gf, + "MATCH p = (:Person {name:'Alice'})-[:KNOWS*1..1]->(:Person {name:'Bob'}) \ + RETURN nodes(p) AS ns", + ); + assert_eq!(single.stats.rows_produced, 1); + assert_eq!( + path_node_labels_at(&single.batches[0], "ns", 0, 0), + vec!["Person"] + ); + assert_eq!( + path_node_labels_at(&single.batches[0], "ns", 0, 1), + vec!["Person"] + ); + + let runtime = rows( + &gf, + "MATCH p = (:Person {name:'Alice'})-[:KNOWS*1..1]->(:Contractor) \ + RETURN nodes(p) AS ns", + ); + assert_eq!(runtime.stats.rows_produced, 1); + assert_eq!( + path_node_labels_at(&runtime.batches[0], "ns", 0, 1), + vec!["Contractor"] + ); + let direct_runtime = rows( + &gf, + "MATCH (n:Contractor {name:'Chris'}) RETURN labels(n) AS labels", + ); + assert_eq!( + utf8_list_cell( + direct_runtime.batches[0] + .column_by_name("labels") + .expect("labels") + .as_any() + .downcast_ref::() + .expect("labels List"), + 0, + ), + path_node_labels_at(&runtime.batches[0], "ns", 0, 1) + ); +} + #[test] fn fixed_hop_return_p_carries_relationship_properties() { // #889: fixed-hop `RETURN p` uses the same relationship value shape as diff --git a/crates/graphforge-rel/src/expr.rs b/crates/graphforge-rel/src/expr.rs index 9ea2ea22e..2c86c81a0 100644 --- a/crates/graphforge-rel/src/expr.rs +++ b/crates/graphforge-rel/src/expr.rs @@ -12041,53 +12041,64 @@ fn hydration_fsb16( }) } -/// The `labels` child (#1024): one-element `List` per flattened node, -/// resolved through `topology/nodes.parquet`'s `type_id` and the baked -/// (sorted) map — the element is NULL when the type is unknown (mirroring -/// `node_value_struct`'s no-map case). +/// The `labels` child (#1024 / #705): full `List` per flattened node from +/// authoritative `topology/nodes.parquet` `type_ids`, resolved through the +/// baked (id-sorted) catalog map — the same complete set `node_labels_list` +/// projects for direct node values. Unknown catalog ids are skipped; a missing +/// topology row keeps a single-null list element. fn path_node_labels_child( h: &PathNodeHydration, flat: &[[u8; 16]], ) -> datafusion::error::Result { - use datafusion::arrow::array::{Array, ListBuilder, StringBuilder}; + use datafusion::arrow::array::{Array, ListArray, ListBuilder, StringBuilder, UInt32Array}; use datafusion::error::DataFusionError; use std::collections::HashMap; let exec_err = |m: String| DataFusionError::Execution(m); let node_batches = graphforge_storage::read_nodes(&h.dir).map_err(|e| exec_err(e.to_string()))?; - let mut label_of: HashMap<[u8; 16], usize> = HashMap::new(); + let mut label_ids_of: HashMap<[u8; 16], Vec> = HashMap::new(); for b in &node_batches { let uuids = hydration_fsb16(b, "node_uuid")?; let type_ids = b - .column_by_name("type_id") - .and_then(|c| { - c.as_any() - .downcast_ref::() - .cloned() - }) - .ok_or_else(|| exec_err("cypher_path_nodes: no UInt32 type_id column".into()))?; + .column_by_name("type_ids") + .and_then(|c| c.as_any().downcast_ref::()) + .ok_or_else(|| exec_err("cypher_path_nodes: no List type_ids column".into()))?; for r in 0..b.num_rows() { if uuids.is_null(r) || type_ids.is_null(r) { continue; } let mut u = [0u8; 16]; u.copy_from_slice(uuids.value(r)); - if let Ok(i) = h - .labels_by_type - .binary_search_by_key(&type_ids.value(r), |(id, _)| *id) - { - label_of.insert(u, i); + let values = type_ids.value(r); + let values = values + .as_any() + .downcast_ref::() + .ok_or_else(|| { + exec_err("cypher_path_nodes: type_ids values are not UInt32".into()) + })?; + let mut ids = Vec::with_capacity(values.len()); + for i in 0..values.len() { + if !values.is_null(i) { + ids.push(values.value(i)); + } } + label_ids_of.insert(u, ids); } } let mut labels_b = ListBuilder::new(StringBuilder::new()); for u in flat { - match label_of.get(u) { - Some(&i) => labels_b.values().append_value(&h.labels_by_type[i].1), - None => labels_b.values().append_null(), + if let Some(ids) = label_ids_of.get(u) { + for id in ids { + if let Ok(i) = h.labels_by_type.binary_search_by_key(id, |(tid, _)| *tid) { + labels_b.values().append_value(&h.labels_by_type[i].1); + } + } + labels_b.append(true); + } else { + labels_b.values().append_null(); + labels_b.append(true); } - labels_b.append(true); } Ok(std::sync::Arc::new(labels_b.finish())) } @@ -14815,6 +14826,201 @@ mod tests { ); } + /// Hydrated `cypher_path_nodes` invoke over a real topology directory (#705). + fn invoke_hydrated_path_nodes( + hydrate: PathNodeHydration, + seed: datafusion::arrow::array::ArrayRef, + rels: datafusion::arrow::array::ArrayRef, + ) -> datafusion::error::Result { + use std::sync::Arc; + + use datafusion::arrow::datatypes::Field; + use datafusion::config::ConfigOptions; + + let udf = CypherPathNodes::with_hydration(hydrate); + let n = seed.len(); + let args = ScalarFunctionArgs { + args: vec![ + ColumnarValue::Array(Arc::clone(&seed)), + ColumnarValue::Array(Arc::clone(&rels)), + ], + arg_fields: vec![ + Arc::new(Field::new("seed", seed.data_type().clone(), true)), + Arc::new(Field::new("rels", rels.data_type().clone(), true)), + ], + number_rows: n, + return_field: Arc::new(Field::new("nodes", udf.return_type(&[])?, true)), + config_options: Arc::new(ConfigOptions::default()), + }; + udf.invoke_with_args(args).map(|v| match v { + ColumnarValue::Array(a) => a, + ColumnarValue::Scalar(s) => s.to_array_of_size(n).unwrap(), + }) + } + + fn path_node_label_lists( + out: &datafusion::arrow::array::ArrayRef, + row: usize, + ) -> Option>>> { + use datafusion::arrow::array::{Array, ListArray, StringArray, StructArray}; + let list = out.as_any().downcast_ref::().unwrap(); + if list.is_null(row) { + return None; + } + let items = list.value(row); + let items = items.as_any().downcast_ref::().unwrap(); + let labels = items + .column_by_name("labels") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + Some( + (0..labels.len()) + .map(|i| { + let values = labels.value(i); + let strings = values.as_any().downcast_ref::().unwrap(); + (0..strings.len()) + .map(|j| (!strings.is_null(j)).then(|| strings.value(j).to_owned())) + .collect() + }) + .collect(), + ) + } + + #[test] + fn hydrated_path_nodes_preserve_full_type_ids_labels() { + // #705: multi-label nodes keep every catalog-resolved label from + // authoritative `type_ids` (not the legacy primary `type_id` alone). + use datafusion::arrow::array::FixedSizeBinaryArray; + use datafusion::arrow::datatypes::Field; + use graphforge_core::uuid::{new_v7, to_bytes}; + use graphforge_core::{OntologyMode, TypeId}; + use graphforge_storage::GraphWriter; + + let dir = tempfile::TempDir::new().unwrap(); + let mut w = GraphWriter::open_at(dir.path(), OntologyMode::Exploratory, 0).unwrap(); + let multi = new_v7(); + let single = new_v7(); + let unknown_only = new_v7(); + w.create_node_with_labels(multi, &[TypeId(1), TypeId(3)]) + .unwrap(); + w.create_node_with_labels(single, &[TypeId(2)]).unwrap(); + // type_ids present but absent from the baked catalog → empty label list. + w.create_node_with_labels(unknown_only, &[TypeId(99)]) + .unwrap(); + w.flush().unwrap(); + + let multi_bytes = to_bytes(&multi); + let single_bytes = to_bytes(&single); + let unknown_bytes = to_bytes(&unknown_only); + let missing_bytes = [0xABu8; 16]; + + let hydrate = PathNodeHydration { + dir: dir.path().to_path_buf(), + labels_by_type: vec![ + (1, "Person".to_owned()), + (2, "Company".to_owned()), + (3, "Employee".to_owned()), + ], + prop_stems: vec![], + fields: vec![ + Field::new("node_uuid", DataType::FixedSizeBinary(16), false), + Field::new("labels", DataType::new_list(DataType::Utf8, true), true), + ] + .into(), + }; + + // Zero-hop walk over four seeds: multi-label, single-label, missing + // catalog id, and uuid absent from topology. + let seed = std::sync::Arc::new( + FixedSizeBinaryArray::try_from_iter( + [multi_bytes, single_bytes, unknown_bytes, missing_bytes] + .iter() + .copied(), + ) + .unwrap(), + ) as datafusion::arrow::array::ArrayRef; + let rels = edge_list(&[Some(&[]), Some(&[]), Some(&[]), Some(&[])]); + let out = invoke_hydrated_path_nodes(hydrate, seed, rels).unwrap(); + let labels = path_node_label_lists(&out, 0).unwrap(); + assert_eq!( + labels[0], + vec![Some("Person".into()), Some("Employee".into())], + "multi-label node keeps full type_ids set in catalog order" + ); + // Remaining seeds are separate rows (one zero-hop path each). + let labels1 = path_node_label_lists(&out, 1).unwrap(); + assert_eq!(labels1[0], vec![Some("Company".into())]); + let labels2 = path_node_label_lists(&out, 2).unwrap(); + assert_eq!( + labels2[0], + Vec::>::new(), + "unknown catalog ids skipped" + ); + let labels3 = path_node_label_lists(&out, 3).unwrap(); + assert_eq!( + labels3[0], + vec![None], + "missing topology row keeps a single-null labels element" + ); + + // Repeated node on a self-loop walk must repeat the full label set. + let seed_loop = std::sync::Arc::new( + FixedSizeBinaryArray::try_from_iter([multi_bytes].iter().copied()).unwrap(), + ) as datafusion::arrow::array::ArrayRef; + // Build a one-edge list with real uuids (not the byte-tag helper). + use datafusion::arrow::array::{FixedSizeBinaryBuilder, ListBuilder, StructBuilder}; + let fields: datafusion::arrow::datatypes::Fields = vec![ + Field::new("src_uuid", DataType::FixedSizeBinary(16), false), + Field::new("dst_uuid", DataType::FixedSizeBinary(16), false), + ] + .into(); + let mut b = ListBuilder::new(StructBuilder::new( + fields, + vec![ + Box::new(FixedSizeBinaryBuilder::new(16)), + Box::new(FixedSizeBinaryBuilder::new(16)), + ], + )); + b.values() + .field_builder::(0) + .unwrap() + .append_value(multi_bytes) + .unwrap(); + b.values() + .field_builder::(1) + .unwrap() + .append_value(multi_bytes) + .unwrap(); + b.values().append(true); + b.append(true); + let loop_rels = std::sync::Arc::new(b.finish()) as datafusion::arrow::array::ArrayRef; + + let hydrate2 = PathNodeHydration { + dir: dir.path().to_path_buf(), + labels_by_type: vec![ + (1, "Person".to_owned()), + (2, "Company".to_owned()), + (3, "Employee".to_owned()), + ], + prop_stems: vec![], + fields: vec![ + Field::new("node_uuid", DataType::FixedSizeBinary(16), false), + Field::new("labels", DataType::new_list(DataType::Utf8, true), true), + ] + .into(), + }; + let looped = invoke_hydrated_path_nodes(hydrate2, seed_loop, loop_rels).unwrap(); + let loop_labels = path_node_label_lists(&looped, 0).unwrap(); + assert_eq!(loop_labels.len(), 2); + assert_eq!(loop_labels[0], loop_labels[1]); + assert_eq!( + loop_labels[0], + vec![Some("Person".into()), Some("Employee".into())] + ); + } + // ----------------------------------------------------------------------- // Parameter test // -----------------------------------------------------------------------