diff --git a/crates/graphforge-api/src/embedding_refresh.rs b/crates/graphforge-api/src/embedding_refresh.rs index 5ed0c0b96..a4c5c40c8 100644 --- a/crates/graphforge-api/src/embedding_refresh.rs +++ b/crates/graphforge-api/src/embedding_refresh.rs @@ -255,6 +255,7 @@ impl GraphForge { lifecycle_mode: self.lifecycle_mode, resolved_generation: self.resolved_generation.clone(), property_authority: Arc::clone(&self.property_authority), + ordinal_identities: Arc::clone(&self.ordinal_identities), read_only: self.read_only, current_generation_uuid: Arc::clone(&self.current_generation_uuid), uuid_membership_index: std::sync::Mutex::new(None), diff --git a/crates/graphforge-api/src/lib.rs b/crates/graphforge-api/src/lib.rs index 81472a5cc..b07bc67ff 100644 --- a/crates/graphforge-api/src/lib.rs +++ b/crates/graphforge-api/src/lib.rs @@ -441,6 +441,9 @@ pub struct GraphForge { current_generation_uuid: Arc>, /// Authenticated UUID index handle cached for one topology generation. uuid_membership_index: Mutex>, + /// Exact generation-pinned ordinal destination identity authority shared + /// by every fixed-hop session. + ordinal_identities: Arc, /// Injected durable-write UTC microsecond clock. clock: Mutex Result + Send + Sync>>, /// Project directory backing topology/properties Parquet files. For an @@ -607,6 +610,7 @@ impl GraphForge { hydrate_graph_workspace(&resolved_generation, false)?; let property_inventory = property_inventory_for_hydrated_generation(&resolved_generation, &dir)?; + let ordinal_identities = ordinal_identity_resolver(&resolved_generation, &dir)?; Ok(Self { identity: GraphIdentity::new(), path: None, @@ -620,6 +624,7 @@ impl GraphForge { read_only: false, current_generation_uuid: Arc::new(Mutex::new(generation_uuid)), uuid_membership_index: Mutex::new(None), + ordinal_identities, clock: Mutex::new(Arc::new(system_time_micros)), adjacency_provider: Arc::new(graphforge_exec::PersistentAdjacencyProvider::new( dir.clone(), @@ -749,6 +754,7 @@ impl GraphForge { hydrate_graph_workspace(&resolved_generation, read_only)?; let property_inventory = property_inventory_for_hydrated_generation(&resolved_generation, &dir)?; + let ordinal_identities = ordinal_identity_resolver(&resolved_generation, &dir)?; let runtime_catalog = load_runtime_catalog(&dir)?; let semantic_storage_bindings = @@ -827,6 +833,7 @@ impl GraphForge { read_only, current_generation_uuid: Arc::new(Mutex::new(generation_uuid)), uuid_membership_index: Mutex::new(None), + ordinal_identities, clock: Mutex::new(Arc::new(system_time_micros)), adjacency_provider: Arc::new(graphforge_exec::PersistentAdjacencyProvider::new( dir.clone(), @@ -943,6 +950,7 @@ impl GraphForge { generation, )?, ); + let ordinal_replacement = ordinal_identity_handle(generation, &self.dir)?; *self .property_authority .lock() @@ -954,6 +962,7 @@ impl GraphForge { .current_generation_uuid .lock() .expect("generation UUID lock poisoned") = generation.generation_uuid(); + self.ordinal_identities.replace(ordinal_replacement); Ok(()) } @@ -1389,12 +1398,13 @@ impl GraphForge { execution_mode, )) }; - let session = ExecutionSession::new_with_target_provider_and_resources( + let session = ExecutionSession::new_with_target_provider_resources_and_identity( catalog, self.ontology.clone(), self.dir.clone(), execution_mode, adjacency_provider, + Some(Arc::clone(&self.ordinal_identities)), &self.session_resource_config(), )?; @@ -1789,6 +1799,12 @@ impl GraphForge { )); } + // Pin every generation-coupled session participant while publication is + // excluded. `install_property_generation` replaces the authenticated + // property inventory and ordinal identity authority as one publication + // transition; opening them without this guard could otherwise combine + // participants from adjacent generations. + let _read_visibility = self.graph_visibility.read()?; let catalog = { let rc = self .runtime_catalog @@ -1802,12 +1818,13 @@ impl GraphForge { ) .map_err(|e| GfError::Storage(e.to_string()))? }; - let session = ExecutionSession::new_with_target_provider_and_resources( + let session = ExecutionSession::new_with_target_provider_resources_and_identity( catalog, self.ontology.clone(), self.dir.clone(), self.ontology_mode, Arc::clone(&self.adjacency_provider), + Some(Arc::clone(&self.ordinal_identities)), &self.session_resource_config(), )?; @@ -3245,6 +3262,10 @@ impl GraphForge { let graph_ir = serde_json::to_string_pretty(&plan).map_err(|e| GfError::Plan(e.to_string()))?; + // EXPLAIN constructs a real physical session. Pin all of its + // generation-coupled authorities against same-instance publication, + // just like query execution does. + let _read_visibility = self.graph_visibility.read()?; // Open the catalog from the same snapshot so the logical and physical // stages resolve property names interned during this bind (a None // catalog would render them as `prop_` and fail to lower). @@ -3293,14 +3314,16 @@ impl GraphForge { explained.unwrap_or_else(|e| format!("(logical plan unavailable: {e})")) }; - let session = graphforge_exec::ExecutionSession::new_with_target_provider_and_resources( - catalog, - self.ontology.clone(), - self.dir.clone(), - self.ontology_mode, - Arc::clone(&self.adjacency_provider), - &self.session_resource_config(), - )?; + let session = + graphforge_exec::ExecutionSession::new_with_target_provider_resources_and_identity( + catalog, + self.ontology.clone(), + self.dir.clone(), + self.ontology_mode, + Arc::clone(&self.adjacency_provider), + Some(Arc::clone(&self.ordinal_identities)), + &self.session_resource_config(), + )?; let physical = self.block_on(async move { session.explain_physical(&plan).await })?; Ok(format!( @@ -3840,6 +3863,38 @@ fn property_inventory_for_hydrated_generation( Ok(Arc::new(admitted)) } +fn ordinal_identity_handle( + generation: &ResolvedProjectGeneration, + graph_root: &Path, +) -> Result, GfError> { + let Some(authority) = generation.authenticated_v4_ordinal_authority()? else { + return Ok(None); + }; + match authority + .open( + graph_root, + graphforge_storage::V4OrdinalIdentityLimits::default(), + ) + .map_err(|error| GfError::Storage(error.to_string()))? + { + graphforge_storage::V4OrdinalIdentityOpen::Ready(handle) => Ok(Some(*handle)), + graphforge_storage::V4OrdinalIdentityOpen::RebuildRequired { found_version } => { + Err(GfError::Validation(format!( + "selected graph generation requires ordinal identity rebuild from version {found_version}" + ))) + } + } +} + +fn ordinal_identity_resolver( + generation: &ResolvedProjectGeneration, + graph_root: &Path, +) -> Result, GfError> { + Ok(Arc::new(graphforge_exec::V4OrdinalIdentityResolver::new( + ordinal_identity_handle(generation, graph_root)?, + ))) +} + fn hydrate_graph_workspace( generation: &ResolvedProjectGeneration, read_only: bool, @@ -7007,6 +7062,63 @@ mod tests { } } + #[test] + fn streaming_and_explain_session_pins_wait_for_generation_publication() { + use std::sync::mpsc::{self, RecvTimeoutError}; + use std::time::Duration; + + let graph = GraphForge::new(None).expect("open ephemeral project"); + graph.execute("CREATE (:Person)").expect("seed graph"); + + std::thread::scope(|scope| { + let publication = graph.graph_visibility.lock().expect("publication lock"); + let ready = std::sync::Arc::new(std::sync::Barrier::new(2)); + let (sent, received) = mpsc::channel(); + let graph = &graph; + let child_ready = std::sync::Arc::clone(&ready); + scope.spawn(move || { + child_ready.wait(); + sent.send(graph.explain("MATCH (n:Person) RETURN n.node_uuid")) + .expect("send explain result"); + }); + ready.wait(); + assert!(matches!( + received.recv_timeout(Duration::from_millis(100)), + Err(RecvTimeoutError::Timeout) + )); + drop(publication); + received + .recv_timeout(Duration::from_secs(5)) + .expect("explain completes after publication") + .expect("explain session"); + }); + + std::thread::scope(|scope| { + let publication = graph.graph_visibility.lock().expect("publication lock"); + let ready = std::sync::Arc::new(std::sync::Barrier::new(2)); + let (sent, received) = mpsc::channel(); + let graph = &graph; + let child_ready = std::sync::Arc::clone(&ready); + scope.spawn(move || { + child_ready.wait(); + let result = graph + .execute_stream("MATCH (n:Person) RETURN n.node_uuid") + .map(drop); + sent.send(result).expect("send stream result"); + }); + ready.wait(); + assert!(matches!( + received.recv_timeout(Duration::from_millis(100)), + Err(RecvTimeoutError::Timeout) + )); + drop(publication); + received + .recv_timeout(Duration::from_secs(5)) + .expect("stream session completes after publication") + .expect("stream session"); + }); + } + #[test] fn persistent_open_creates_an_absent_final_target_through_storage() { let parent = tempfile::tempdir().unwrap(); diff --git a/crates/graphforge-api/tests/fixed_hop_limit.rs b/crates/graphforge-api/tests/fixed_hop_limit.rs index 35b7d4623..e8f0f63a9 100644 --- a/crates/graphforge-api/tests/fixed_hop_limit.rs +++ b/crates/graphforge-api/tests/fixed_hop_limit.rs @@ -6,17 +6,28 @@ use std::collections::HashMap; use std::path::Path; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; -use arrow::array::{FixedSizeBinaryArray, Int64Array, UInt64Array}; -use graphforge_api::GraphForge; +use arrow::array::{ + Array, ArrayRef, FixedSizeBinaryArray, FixedSizeBinaryBuilder, Int64Array, StringArray, + UInt64Array, +}; +use arrow::record_batch::RecordBatch; +use graphforge_api::{ + CONSTRUCTION_EDGE_SCHEMA, CONSTRUCTION_NODE_SCHEMA, GraphConstructionBudgets, GraphForge, + OperationId, PortableSelection, PortableV2ExportRequest, PortableV2ImportRequest, + PortableVerifyRequest, verify_portable_v2, +}; use graphforge_core::uuid::{Uuid, new_v7}; use graphforge_core::{OntologyMode, TypeId}; use graphforge_exec::demand::{self, DemandSnapshot}; use graphforge_ir::IrLiteral; use graphforge_storage::adjacency::build_adjacency_index; -use graphforge_storage::{GraphWriter, io_stats}; +use graphforge_storage::{ + GraphWriter, PortableV2Limits, PortableV2Mode, PortableV2Output, PortableV2SelectionProfile, + io_stats, +}; use tempfile::TempDir; #[path = "support/project_fixture.rs"] @@ -35,9 +46,50 @@ static IO_GUARD: Mutex<()> = Mutex::new(()); const ONE_HOP: &str = "MATCH (a)-[r]->(b) RETURN b.node_uuid AS id LIMIT 1000"; const TWO_HOP: &str = "MATCH (a)-[r1]->(b)-[r2]->(c) \ RETURN c.node_uuid AS id LIMIT 1000"; +const ORDERED_ONE_HOP: &str = "MATCH (a)-[r]->(b) RETURN b.node_uuid AS id ORDER BY id LIMIT 1000"; +const ORDERED_TWO_HOP: &str = + "MATCH (a)-[r1]->(b)-[r2]->(c) RETURN c.node_uuid AS id ORDER BY id LIMIT 1000"; + +fn measured_identity_query(forge: &GraphForge, query: &str) -> (Vec>, DemandSnapshot) { + io_stats::reset(); + demand::reset(); + let result = forge.execute(query).unwrap(); + demand::disable(); + let io = io_stats::snapshot(); + assert_eq!(io.edge_full_reads + io.edge_filtered_reads, 0, "{io:#?}"); + assert_eq!(io.node_full_reads + io.node_filtered_reads, 0, "{io:#?}"); + let snapshot = demand::snapshot(); + assert_eq!( + snapshot + .hops + .values() + .filter(|hop| hop.identity_revalidation_calls > 0) + .count(), + 1, + "{snapshot:#?}" + ); + assert!( + snapshot + .hops + .values() + .all(|hop| hop.identity_per_record_seeks == 0 + && hop.identity_peak_buffer_bytes <= 16 * 1024 * 1024 + && hop.identity_read_calls + <= hop + .identity_ranges_selected + .saturating_mul(2) + .saturating_add(2)), + "{snapshot:#?}" + ); + (fixed_binary_values(&result, "id"), snapshot) +} /// Deterministic ring: each node points to its next `fan_out` successors. -fn generate_graph(dir: &Path, nodes: usize, fan_out: usize) { +fn generate_graph(dir: &Path, nodes: usize, fan_out: usize, compact_v4: bool) { + if compact_v4 { + generate_bulk_graph(dir, nodes, fan_out); + return; + } assert!(nodes > fan_out); let workspace = TempDir::new().unwrap(); let uuids: Vec = (0..nodes).map(|_| new_v7()).collect(); @@ -74,6 +126,118 @@ fn generate_graph(dir: &Path, nodes: usize, fan_out: usize) { project_fixture::publish_graph_workspace(dir, workspace.path()); } +#[derive(Debug, PartialEq, Eq)] +struct BulkFixtureEvidence { + node_rows: usize, + edge_rows: usize, + node_batches: usize, + edge_batches: usize, + accepted_chunks: u64, + input_rows: u64, + peak_batch_rows: u64, +} + +/// Construct scale fixtures through the same bounded Arrow publication path +/// used by ordinary high-volume ingestion. Scalar `GraphWriter::create_edge` +/// deliberately checks its in-flight topology window for duplicate UUIDs and +/// is therefore not a realistic bulk-ingestion primitive. +fn generate_bulk_graph(dir: &Path, nodes: usize, fan_out: usize) -> BulkFixtureEvidence { + assert!(nodes > fan_out); + let forge = GraphForge::new(Some(dir.to_str().expect("temp path is UTF-8"))).unwrap(); + let mut session = forge + .begin_graph_construction(GraphConstructionBudgets { + max_batch_rows: WRITE_WINDOW, + max_run_records: 4 * WRITE_WINDOW, + ..GraphConstructionBudgets::default() + }) + .unwrap(); + + let mut node_batches = 0; + for start in (0..nodes).step_by(WRITE_WINDOW) { + let end = start.saturating_add(WRITE_WINDOW).min(nodes); + let rows = end - start; + let mut identities = FixedSizeBinaryBuilder::with_capacity(rows, 16); + for node in start..end { + identities + .append_value(fixture_node_uuid(node).as_bytes()) + .unwrap(); + } + let batch = RecordBatch::try_new( + Arc::clone(&CONSTRUCTION_NODE_SCHEMA), + vec![ + Arc::new(identities.finish()) as ArrayRef, + Arc::new(StringArray::from(vec!["Entity"; rows])), + ], + ) + .unwrap(); + session + .append_nodes(&format!("nodes-{start}"), &batch) + .unwrap(); + node_batches += 1; + } + + let edge_rows = nodes.saturating_mul(fan_out); + let mut edge_batches = 0; + for start in (0..edge_rows).step_by(WRITE_WINDOW) { + let end = start.saturating_add(WRITE_WINDOW).min(edge_rows); + let rows = end - start; + let mut identities = FixedSizeBinaryBuilder::with_capacity(rows, 16); + let mut sources = FixedSizeBinaryBuilder::with_capacity(rows, 16); + let mut targets = FixedSizeBinaryBuilder::with_capacity(rows, 16); + for edge in start..end { + let source = edge / fan_out; + let offset = edge % fan_out + 1; + identities + .append_value(fixture_edge_uuid(edge).as_bytes()) + .unwrap(); + sources + .append_value(fixture_node_uuid(source).as_bytes()) + .unwrap(); + targets + .append_value(fixture_node_uuid((source + offset) % nodes).as_bytes()) + .unwrap(); + } + let batch = RecordBatch::try_new( + Arc::clone(&CONSTRUCTION_EDGE_SCHEMA), + vec![ + Arc::new(identities.finish()) as ArrayRef, + Arc::new(StringArray::from(vec!["LINK"; rows])), + Arc::new(sources.finish()), + Arc::new(targets.finish()), + ], + ) + .unwrap(); + session + .append_edges(&format!("edges-{start}"), &batch) + .unwrap(); + edge_batches += 1; + } + + session.seal_and_publish().unwrap(); + let progress = session.progress(); + drop(session); + forge.index_adjacency().unwrap(); + drop(forge); + + BulkFixtureEvidence { + node_rows: nodes, + edge_rows, + node_batches, + edge_batches, + accepted_chunks: progress.accepted_chunks, + input_rows: progress.evidence.input_rows, + peak_batch_rows: progress.evidence.peak_batch_rows, + } +} + +fn fixture_node_uuid(index: usize) -> Uuid { + Uuid::from_u128(0x1000_0000_0000_0000_0000_0000_0000_0000 | index as u128 + 1) +} + +fn fixture_edge_uuid(index: usize) -> Uuid { + Uuid::from_u128(0x2000_0000_0000_0000_0000_0000_0000_0000 | index as u128 + 1) +} + fn open_forge(dir: &Path) -> GraphForge { GraphForge::new(Some(dir.to_str().expect("temp path is UTF-8"))).unwrap() } @@ -106,6 +270,43 @@ fn fixed_binary_values(result: &graphforge_api::ExecutionResult, column: &str) - values } +fn int64_values(result: &graphforge_api::ExecutionResult, column: &str) -> Vec { + result + .batches + .iter() + .flat_map(|batch| { + batch + .column_by_name(column) + .unwrap_or_else(|| panic!("missing {column} column")) + .as_any() + .downcast_ref::() + .unwrap_or_else(|| panic!("{column} must be Int64")) + .values() + .iter() + .copied() + .collect::>() + }) + .collect() +} + +fn string_values(result: &graphforge_api::ExecutionResult, column: &str) -> Vec { + result + .batches + .iter() + .flat_map(|batch| { + let values = batch + .column_by_name(column) + .unwrap_or_else(|| panic!("missing {column} column")) + .as_any() + .downcast_ref::() + .unwrap_or_else(|| panic!("{column} must be Utf8")); + (0..values.len()) + .map(|row| values.value(row).to_owned()) + .collect::>() + }) + .collect() +} + fn stable_fixture_uuid(kind: u8, ordinal: usize) -> Uuid { let mut bytes = [0u8; 16]; bytes[0] = kind; @@ -113,6 +314,45 @@ fn stable_fixture_uuid(kind: u8, ordinal: usize) -> Uuid { Uuid::from_bytes(bytes) } +fn generate_semantic_v4_graph(dir: &Path) -> Vec { + let workspace = TempDir::new().unwrap(); + let nodes = (1..=3) + .map(|ordinal| stable_fixture_uuid(3, ordinal)) + .collect::>(); + let mut writer = GraphWriter::open_at(workspace.path(), OntologyMode::Exploratory, TS).unwrap(); + for (node, name) in nodes.iter().zip(["A", "B", "C"]) { + writer.create_node(*node, NODE_TYPE).unwrap(); + writer + .set_properties( + node, + None, + HashMap::from([("name".to_owned(), IrLiteral::Str(name.to_owned()))]), + ) + .unwrap(); + } + for (ordinal, (source, destination, weight)) in + [(0_usize, 1_usize, 1_i64), (0, 1, 2), (0, 0, 3), (1, 2, 4)] + .into_iter() + .enumerate() + { + let edge = stable_fixture_uuid(4, ordinal + 1); + writer + .create_edge(edge, "LINK", &nodes[source], &nodes[destination]) + .unwrap(); + writer + .set_edge_properties( + &edge, + Some("LINK"), + HashMap::from([("weight".to_owned(), IrLiteral::Int(weight))]), + ) + .unwrap(); + } + writer.flush().unwrap(); + build_adjacency_index(workspace.path(), TS).unwrap(); + project_fixture::publish_graph_workspace_v4(dir, workspace.path()); + nodes +} + /// Build a graph whose first productive edge ids are localized but whose /// destination node ids are evenly scattered through the node table. Target /// UUIDs are stable across scales so the public results are directly @@ -211,7 +451,7 @@ struct ScaleResult { fn run_scale(nodes: usize, fan_out: usize) -> ScaleResult { let dir = TempDir::new().unwrap(); - generate_graph(dir.path(), nodes, fan_out); + generate_bulk_graph(dir.path(), nodes, fan_out); let forge = open_forge(dir.path()); let one_plan = forge.explain(ONE_HOP).unwrap(); @@ -260,6 +500,11 @@ fn assert_indexed_limit_io(io: &io_stats::IoSnapshot) { assert!(io.node_filtered_reads >= 1, "{io:?}"); } +fn assert_projected_identity_io(io: &io_stats::IoSnapshot) { + assert_eq!(io.edge_full_reads + io.edge_filtered_reads, 0, "{io:?}"); + assert_eq!(io.node_full_reads + io.node_filtered_reads, 0, "{io:?}"); +} + fn assert_bounded_demand(snapshot: &DemandSnapshot, expected_hops: usize, required: u64) { assert_eq!(snapshot.hops.len(), expected_hops, "{snapshot:#?}"); assert!(snapshot.cancellations >= 1, "{snapshot:#?}"); @@ -293,8 +538,8 @@ fn terminal_limit_keeps_fixed_hop_io_bounded_as_graph_grows() { println!("fixed-hop LIMIT structural smoke: small={small:?}, large={large:?}"); for scale in [&small, &large] { - assert_indexed_limit_io(&scale.one_hop_io); - assert_indexed_limit_io(&scale.two_hop_io); + assert_projected_identity_io(&scale.one_hop_io); + assert_projected_identity_io(&scale.two_hop_io); assert_bounded_demand(&scale.one_hop_demand, 1, LIMIT as u64); assert_bounded_demand(&scale.two_hop_demand, 2, LIMIT as u64); } @@ -319,6 +564,35 @@ fn terminal_limit_keeps_fixed_hop_io_bounded_as_graph_grows() { ); } +#[test] +fn scale_fixture_uses_bounded_bulk_publications() { + let _guard = IO_GUARD.lock().unwrap(); + let dir = TempDir::new().unwrap(); + let nodes = WRITE_WINDOW + 1; + let evidence = generate_bulk_graph(dir.path(), nodes, 2); + assert_eq!( + evidence, + BulkFixtureEvidence { + node_rows: nodes, + edge_rows: nodes * 2, + node_batches: 2, + edge_batches: 3, + accepted_chunks: 5, + input_rows: (nodes * 3) as u64, + peak_batch_rows: WRITE_WINDOW as u64, + } + ); + + let forge = open_forge(dir.path()); + let plan = forge.explain(ORDERED_ONE_HOP).unwrap(); + assert!(plan.contains("adjacency=hit"), "{plan}"); + assert!(plan.contains("identity=v4"), "{plan}"); + io_stats::reset(); + let result = forge.execute(ORDERED_ONE_HOP).unwrap(); + assert_eq!(result.stats.rows_produced, LIMIT as u64); + assert_projected_identity_io(&io_stats::snapshot()); +} + fn run_scattered_destination_scale( nodes: usize, ) -> ( @@ -390,11 +664,349 @@ fn scattered_node_hydration_is_neighborhood_proportional() { ); } +fn run_ordered_projection_scale(nodes: usize) -> (Vec>, DemandSnapshot) { + let dir = TempDir::new().unwrap(); + generate_graph(dir.path(), nodes, FAN_OUT, true); + let forge = open_forge(dir.path()); + let plan = forge.explain(ORDERED_ONE_HOP).unwrap(); + assert!(plan.contains("ExpandExec"), "{plan}"); + assert!(plan.contains("identity=v4"), "{plan}"); + assert!(plan.contains("SortExec"), "{plan}"); + assert!(plan.contains("projection=1"), "{plan}"); + + io_stats::reset(); + demand::reset(); + let first = forge.execute(ORDERED_ONE_HOP).unwrap(); + demand::disable(); + let io = io_stats::snapshot(); + let snapshot = demand::snapshot(); + let first_values = fixed_binary_values(&first, "id"); + assert_eq!(first_values.len(), LIMIT); + assert!(first_values.windows(2).all(|pair| pair[0] <= pair[1])); + let node_scan = forge + .execute("MATCH (b) RETURN b.node_uuid AS id ORDER BY id") + .unwrap(); + let expected = fixed_binary_values(&node_scan, "id") + .into_iter() + .flat_map(|uuid| std::iter::repeat_n(uuid, FAN_OUT)) + .take(LIMIT) + .collect::>(); + assert_eq!( + first_values, expected, + "ordered fixed hop differs from scan oracle" + ); + assert_eq!(io.edge_full_reads + io.edge_filtered_reads, 0, "{io:#?}"); + assert_eq!(io.node_full_reads + io.node_filtered_reads, 0, "{io:#?}"); + let hop = snapshot.hops.values().next().expect("one projected hop"); + assert!(hop.projected_chunks > 1, "{snapshot:#?}"); + assert_eq!(hop.projected_rows, (nodes * FAN_OUT) as u64); + assert_eq!(hop.projected_columns, 1); + assert_eq!(hop.identity_per_record_seeks, 0); + assert!(hop.identity_read_calls > 0, "{snapshot:#?}"); + assert!(hop.identity_bytes_read > 0, "{snapshot:#?}"); + assert!(hop.identity_peak_buffer_bytes <= 16 * 1024 * 1024); + + let repeated = forge.execute(ORDERED_ONE_HOP).unwrap(); + assert_eq!( + fixed_binary_values(&repeated, "id"), + first_values, + "ordered result changed across identical execution" + ); + (first_values, snapshot) +} + +#[test] +fn destination_uuid_projection_uses_authenticated_legacy_hydration_without_v4_authority() { + let _guard = IO_GUARD.lock().unwrap(); + let dir = TempDir::new().unwrap(); + generate_graph(dir.path(), 64, 4, false); + let result = open_forge(dir.path()).execute(ORDERED_ONE_HOP).unwrap(); + let values = fixed_binary_values(&result, "id"); + assert!(!values.is_empty()); + assert!( + values + .iter() + .all(|value| value.iter().any(|byte| *byte != 0)) + ); +} + +#[test] +fn ordered_destination_uuid_projection_is_exact_and_linear_at_1x_2x_4x() { + let _guard = IO_GUARD.lock().unwrap(); + let mut work = Vec::new(); + for nodes in [4_096, 8_192, 16_384] { + let (_, snapshot) = run_ordered_projection_scale(nodes); + let hop = snapshot.hops.values().next().unwrap(); + work.push(( + hop.projected_rows, + hop.identity_bytes_read, + hop.identity_read_calls, + hop.identity_revalidation_calls, + )); + } + for pair in work.windows(2) { + let (prior_rows, prior_bytes, prior_calls, prior_revalidation) = pair[0]; + let (next_rows, next_bytes, next_calls, next_revalidation) = pair[1]; + assert_eq!(next_rows, prior_rows * 2, "{work:?}"); + // Fixed block/range boundaries may add one coalesced read, but neither + // bytes nor calls may acquire a chunk-times-graph multiplier. + assert!(next_bytes <= prior_bytes * 2 + 2 * 1024 * 1024, "{work:?}"); + assert!(next_calls <= prior_calls * 2 + 2, "{work:?}"); + assert!( + next_revalidation <= prior_revalidation * 2 + 2, + "session authentication must be linear in retained artifacts: {work:?}" + ); + } +} + +#[test] +fn portable_v2_clean_import_preserves_projected_ordered_hops_and_io() { + let _guard = IO_GUARD.lock().unwrap(); + let root = TempDir::new().unwrap(); + let source_path = root.path().join("source"); + generate_graph(&source_path, 4_096, FAN_OUT, true); + let source = open_forge(&source_path); + let source_results = + [ORDERED_ONE_HOP, ORDERED_TWO_HOP].map(|query| measured_identity_query(&source, query).0); + + let limits = PortableV2Limits::default(); + let package = root.path().join("project.gfpb"); + let exported = source + .export_portable_v2( + &PortableV2ExportRequest { + selection: PortableSelection::Current, + output_path: package.clone(), + representation: PortableV2Output::Bundle, + profile: PortableV2SelectionProfile::Complete, + subset: None, + limits, + }, + None, + |_| {}, + ) + .unwrap(); + drop(source); + let verified = verify_portable_v2( + &PortableVerifyRequest { + input: package.clone(), + mode: PortableV2Mode::Full, + limits, + }, + None, + ) + .unwrap(); + assert_eq!(verified.package_digest, exported.package_digest); + + let imported_path = root.path().join("imported"); + GraphForge::import_portable_v2( + &imported_path, + &PortableV2ImportRequest { + input: package, + operation_id: OperationId(Uuid::from_u128(966)), + limits, + }, + None, + ) + .unwrap(); + let imported = open_forge(&imported_path); + for (query, expected) in [ORDERED_ONE_HOP, ORDERED_TWO_HOP] + .into_iter() + .zip(source_results) + { + assert_eq!( + measured_identity_query(&imported, query).0, + expected, + "{query}" + ); + } +} + +#[test] +fn optimized_v4_two_hop_direction_type_alias_and_quiescence_are_exact() { + let _guard = IO_GUARD.lock().unwrap(); + let dir = TempDir::new().unwrap(); + let nodes = 4_096; + generate_graph(dir.path(), nodes, FAN_OUT, true); + let forge = open_forge(dir.path()); + let scan = forge + .execute("MATCH (n) RETURN n.node_uuid AS id ORDER BY id") + .unwrap(); + let node_ids = fixed_binary_values(&scan, "id"); + + let cases = [ + ( + "MATCH (a)-[:LINK]->(b) RETURN b.node_uuid AS id ORDER BY id LIMIT 1000", + FAN_OUT, + true, + true, + ), + ( + "MATCH (a)<-[:LINK]-(b) RETURN a.node_uuid AS id ORDER BY id LIMIT 1000", + FAN_OUT, + true, + false, + ), + ( + "MATCH (a)-[:LINK]-(b) RETURN b.node_uuid AS id ORDER BY id LIMIT 1000", + FAN_OUT * 2, + false, + true, + ), + ( + "MATCH (a)-[:LINK]->(b)-[:LINK]->(c) RETURN c.node_uuid AS id ORDER BY id LIMIT 1000", + FAN_OUT * FAN_OUT, + true, + true, + ), + ]; + for (query, multiplicity, identity_only, expects_v4_lookup) in cases { + io_stats::reset(); + demand::reset(); + let result = forge.execute(query).unwrap(); + demand::disable(); + let expected = node_ids + .iter() + .flat_map(|uuid| std::iter::repeat_n(uuid.clone(), multiplicity)) + .take(LIMIT) + .collect::>(); + assert_eq!(fixed_binary_values(&result, "id"), expected, "{query}"); + let io = io_stats::snapshot(); + if identity_only { + assert_eq!( + io.edge_full_reads + io.edge_filtered_reads, + 0, + "{query}: {io:#?}" + ); + assert_eq!( + io.node_full_reads + io.node_filtered_reads, + 0, + "{query}: {io:#?}" + ); + } + let snapshot = demand::snapshot(); + assert!( + snapshot + .hops + .values() + .all(|hop| { hop.identity_per_record_seeks == 0 && hop.reads_after_cancel == 0 }), + "{query}: {snapshot:#?}" + ); + let pinned_hops = snapshot + .hops + .values() + .map(|hop| hop.identity_revalidation_calls) + .filter(|calls| *calls > 0) + .count(); + assert_eq!( + pinned_hops, + usize::from(expects_v4_lookup), + "a facade session pin is attributed exactly when destination identity lookup is required: {query}: {snapshot:#?}" + ); + } + + let alias = forge + .execute( + "MATCH (left)-[:LINK]->(right) RETURN right.node_uuid AS renamed ORDER BY renamed LIMIT 1000", + ) + .unwrap(); + let canonical = forge + .execute("MATCH (a)-[:LINK]->(b) RETURN b.node_uuid AS id ORDER BY id LIMIT 1000") + .unwrap(); + assert_eq!( + fixed_binary_values(&alias, "renamed"), + fixed_binary_values(&canonical, "id") + ); + + let empty = forge + .execute("MATCH (a)-[:MISSING]->(b) RETURN b.node_uuid AS id ORDER BY id LIMIT 1000") + .unwrap(); + assert_eq!(empty.stats.rows_produced, 0); +} + +#[test] +fn optimized_v4_preserves_parallel_self_loop_and_demanded_property_semantics() { + let _guard = IO_GUARD.lock().unwrap(); + let dir = TempDir::new().unwrap(); + let nodes = generate_semantic_v4_graph(dir.path()); + let forge = open_forge(dir.path()); + + let destination_query = "MATCH (a)-[:LINK]->(b) RETURN b.node_uuid AS id ORDER BY id"; + let destination_plan = forge.explain(destination_query).unwrap(); + assert!( + destination_plan.contains("identity=v4"), + "{destination_plan}" + ); + assert!( + destination_plan.contains("projection=1"), + "{destination_plan}" + ); + io_stats::reset(); + let destinations = forge.execute(destination_query).unwrap(); + let expected = [nodes[0], nodes[1], nodes[1], nodes[2]] + .into_iter() + .map(|uuid| uuid.as_bytes().to_vec()) + .collect::>(); + assert_eq!(fixed_binary_values(&destinations, "id"), expected); + let io = io_stats::snapshot(); + assert_eq!(io.edge_full_reads + io.edge_filtered_reads, 0, "{io:#?}"); + assert_eq!(io.node_full_reads + io.node_filtered_reads, 0, "{io:#?}"); + + let relationship = forge + .execute("MATCH (a)-[r:LINK]->(b) RETURN r.weight AS weight ORDER BY weight") + .unwrap(); + assert_eq!(int64_values(&relationship, "weight"), [1, 2, 3, 4]); + + let predicate = forge + .execute("MATCH (a)-[r:LINK]->(b) WHERE r.weight >= 2 RETURN b.node_uuid AS id ORDER BY id") + .unwrap(); + assert_eq!( + fixed_binary_values(&predicate, "id"), + [nodes[0], nodes[1], nodes[2]] + .into_iter() + .map(|uuid| uuid.as_bytes().to_vec()) + .collect::>() + ); + + let node_property = forge + .execute("MATCH (a)-[:LINK]->(b) RETURN b.name AS name ORDER BY name") + .unwrap(); + assert_eq!(string_values(&node_property, "name"), ["A", "B", "B", "C"]); + + let undirected = forge + .execute("MATCH (a)-[:LINK]-(b) RETURN b.node_uuid AS id ORDER BY id") + .unwrap(); + let values = fixed_binary_values(&undirected, "id"); + // The undirected self-loop is emitted once, while the two parallel LINK + // identities remain two distinct matches in each orientation. + assert_eq!(values.len(), 7); + assert_eq!( + values + .iter() + .filter(|value| value.as_slice() == nodes[0].as_bytes()) + .count(), + 3 + ); + assert_eq!( + values + .iter() + .filter(|value| value.as_slice() == nodes[1].as_bytes()) + .count(), + 3 + ); + assert_eq!( + values + .iter() + .filter(|value| value.as_slice() == nodes[2].as_bytes()) + .count(), + 1 + ); +} + #[test] fn limits_sweep_bounded_multi_hop_work_and_repartition() { let _guard = IO_GUARD.lock().unwrap(); let dir = TempDir::new().unwrap(); - generate_graph(dir.path(), 4_096, FAN_OUT); + generate_graph(dir.path(), 4_096, FAN_OUT, false); let forge = open_forge(dir.path()); for limit in [10_u64, 100, 1_000] { @@ -422,7 +1034,7 @@ fn limits_sweep_bounded_multi_hop_work_and_repartition() { fn selective_filter_tops_up_without_crossing_blockers() { let _guard = IO_GUARD.lock().unwrap(); let dir = TempDir::new().unwrap(); - generate_graph(dir.path(), 64, 4); + generate_graph(dir.path(), 64, 4, false); let forge = open_forge(dir.path()); let selective = "MATCH (a)-[r1]->(b)-[r2]->(c) \ @@ -454,7 +1066,8 @@ fn selective_filter_tops_up_without_crossing_blockers() { "MATCH ()-[r]->() RETURN count(r) AS total LIMIT 1", ] { let plan = forge.explain(query).unwrap(); - assert!(plan.contains("demand_batch=all, cancel=none"), "{plan}"); + assert!(plan.contains("demand_batch=all"), "{plan}"); + assert!(plan.contains("cancel=none"), "{plan}"); assert!(!plan.contains("DemandGuardExec"), "{plan}"); } @@ -462,10 +1075,8 @@ fn selective_filter_tops_up_without_crossing_blockers() { .explain("MATCH ()-[r1]->()-[r2]->() RETURN r1, r2") .unwrap(); assert!(!unlimited.contains("DemandGuardExec"), "{unlimited}"); - assert!( - unlimited.contains("demand_batch=all, cancel=none"), - "{unlimited}" - ); + assert!(unlimited.contains("demand_batch=all"), "{unlimited}"); + assert!(unlimited.contains("cancel=none"), "{unlimited}"); } #[test] @@ -499,7 +1110,7 @@ fn high_degree_source_resumes_without_losing_neighbors() { fn fixed_hop_limit_preserves_skip_parameters_filters_and_blockers() { let _guard = IO_GUARD.lock().unwrap(); let dir = TempDir::new().unwrap(); - generate_graph(dir.path(), 64, 4); + generate_graph(dir.path(), 64, 4, false); let forge = open_forge(dir.path()); io_stats::reset(); @@ -695,7 +1306,7 @@ fn release_livejournal_fixed_hop_limits() { fn release_scale(nodes: usize, fan_out: usize) -> ScaleResult { let dir = TempDir::new().unwrap(); - generate_graph(dir.path(), nodes, fan_out); + generate_bulk_graph(dir.path(), nodes, fan_out); let warm = open_forge(dir.path()); warm.execute(ONE_HOP).unwrap(); warm.execute(TWO_HOP).unwrap(); diff --git a/crates/graphforge-api/tests/support/project_fixture.rs b/crates/graphforge-api/tests/support/project_fixture.rs index 5f8b21aed..793c003d2 100644 --- a/crates/graphforge-api/tests/support/project_fixture.rs +++ b/crates/graphforge-api/tests/support/project_fixture.rs @@ -11,8 +11,10 @@ use arrow::record_batch::RecordBatch; use graphforge_core::canonical::{CANONICAL_CONTRACT_VERSION, CanonicalDomain, fingerprint}; use graphforge_core::uuid::Uuid; use graphforge_storage::{ - ProjectCapability, ProjectGenerationRequest, ProjectParticipant, ProjectParticipantEncoding, - ProjectStageOutcome, + GRAPH_CAPABILITY_ID, GRAPH_CAPABILITY_VERSION, ProjectCapability, ProjectGenerationRequest, + ProjectParticipant, ProjectParticipantEncoding, ProjectStageOutcome, UuidIndexBuildLimits, + capture_graph_files, rebuild_v4_ordinal_identity, resolve_project_generation, + stage_project_generation_with_graph_tree, }; const GRAPH_SNAPSHOT_SCHEMA_CANONICAL_BYTES: &[u8] = @@ -103,6 +105,51 @@ pub(crate) fn publish_graph_workspace(container: &Path, workspace: &Path) { .unwrap(); } +/// Publish through the compact graph-files v2 authority after constructing the +/// v4 ordinal identity artifacts in the exact workspace being committed. +pub(crate) fn publish_graph_workspace_v4(container: &Path, workspace: &Path) { + let _ = graphforge_storage::open_or_initialize_project(container).unwrap(); + rebuild_v4_ordinal_identity(workspace, UuidIndexBuildLimits::default()).unwrap(); + let parent = resolve_project_generation(container).unwrap(); + let expected_parent = parent.generation_uuid(); + drop(parent); + + let (_, graph_participant) = capture_graph_files(workspace).unwrap(); + let mut participants = graphforge_storage::empty_workspace_participants().unwrap(); + participants.insert(0, graph_participant); + let request = ProjectGenerationRequest { + transaction_uuid: Uuid::now_v7(), + generation_uuid: Uuid::now_v7(), + capabilities: vec![ + ProjectCapability { + capability_id: GRAPH_CAPABILITY_ID.into(), + capability_version: GRAPH_CAPABILITY_VERSION, + }, + ProjectCapability { + capability_id: "workspace".into(), + capability_version: 1, + }, + ], + participants, + }; + let ProjectStageOutcome::Staged(staged) = + stage_project_generation_with_graph_tree(container, &request, Some(workspace)).unwrap() + else { + panic!("fresh v4 fixture publication unexpectedly replayed"); + }; + staged + .validate( + |_| Ok(()), + |actual_parent, _| { + assert_eq!(actual_parent.generation_uuid(), expected_parent); + Ok(()) + }, + ) + .unwrap() + .publish() + .unwrap(); +} + fn collect_files(root: &Path, directory: &Path, files: &mut Vec<(String, Vec)>) { let mut entries = fs::read_dir(directory) .unwrap() diff --git a/crates/graphforge-exec/src/demand.rs b/crates/graphforge-exec/src/demand.rs index 1d1e51d2b..aedb334b2 100644 --- a/crates/graphforge-exec/src/demand.rs +++ b/crates/graphforge-exec/src/demand.rs @@ -6,7 +6,7 @@ //! the fixed-hop operators below that semantic boundary. Unknown and blocking //! operators are deliberately opaque: demand never crosses them. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::fmt; use std::pin::Pin; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -17,13 +17,15 @@ use arrow::datatypes::SchemaRef; use datafusion::common::config::ConfigOptions; use datafusion::common::{DataFusionError, Result}; use datafusion::execution::TaskContext; -use datafusion::physical_expr::ScalarFunctionExpr; +use datafusion::physical_expr::utils::collect_columns; +use datafusion::physical_expr::{PhysicalExpr, ScalarFunctionExpr}; use datafusion::physical_optimizer::PhysicalOptimizerRule; use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec; use datafusion::physical_plan::filter::FilterExec; use datafusion::physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; use datafusion::physical_plan::projection::ProjectionExec; use datafusion::physical_plan::repartition::RepartitionExec; +use datafusion::physical_plan::sorts::sort::SortExec; use datafusion::physical_plan::{ DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties, RecordBatchStream, SendableRecordBatchStream, @@ -88,6 +90,30 @@ pub struct HopSnapshot { pub node_validation_fallbacks: u64, /// Read attempts rejected after terminal cancellation. pub reads_after_cancel: u64, + /// Chunks served without edge or destination-node Parquet hydration. + pub projected_chunks: u64, + /// Candidate rows served by the destination-identity projection. + pub projected_rows: u64, + /// Required physical output columns at this hop. + pub projected_columns: u64, + /// Edge topology/property columns physically demanded by this hop. + pub edge_projected_columns: u64, + /// Destination-node columns physically demanded by this hop. + pub node_projected_columns: u64, + /// V4 ordinal ranges selected across bounded lookup batches. + pub identity_ranges_selected: u64, + /// Coalesced V4 ordinal/tombstone reads. + pub identity_read_calls: u64, + /// V4 ordinal/tombstone bytes read. + pub identity_bytes_read: u64, + /// Largest charged V4 request/cache/transient buffer. + pub identity_peak_buffer_bytes: u64, + /// Forbidden per-record seek count (must remain zero). + pub identity_per_record_seeks: u64, + /// Generation-authentication checks charged once per execution session. + pub identity_revalidation_calls: u64, + /// Artifact payload bytes read by session pinning. + pub identity_revalidation_bytes: u64, } /// Rows observed at one selective physical filter. @@ -163,6 +189,55 @@ pub(crate) fn record_candidates(edge_var: u32, rows: usize) { with_hop(edge_var, |hop| hop.candidates_generated += rows as u64); } +pub(crate) fn record_identity_projection( + edge_var: u32, + rows: usize, + projected_columns: usize, + metrics: &graphforge_storage::V4OrdinalLookupMetrics, +) { + with_hop(edge_var, |hop| { + hop.projected_chunks = hop.projected_chunks.saturating_add(1); + hop.projected_rows = hop.projected_rows.saturating_add(rows as u64); + hop.projected_columns = hop + .projected_columns + .max(projected_columns.try_into().unwrap_or(u64::MAX)); + hop.identity_ranges_selected = hop + .identity_ranges_selected + .saturating_add(metrics.ranges_selected); + hop.identity_read_calls = hop + .identity_read_calls + .saturating_add(metrics.sequential_read_calls); + hop.identity_bytes_read = hop.identity_bytes_read.saturating_add(metrics.bytes_read); + hop.identity_peak_buffer_bytes = hop + .identity_peak_buffer_bytes + .max(metrics.peak_buffer_bytes); + hop.identity_per_record_seeks = hop + .identity_per_record_seeks + .saturating_add(metrics.per_record_seeks); + hop.identity_revalidation_calls = hop + .identity_revalidation_calls + .saturating_add(metrics.revalidation_calls); + hop.identity_revalidation_bytes = hop + .identity_revalidation_bytes + .saturating_add(metrics.revalidation_bytes); + }); +} + +pub(crate) fn record_materialization_projection( + edge_var: u32, + edge_columns: usize, + node_columns: usize, +) { + with_hop(edge_var, |hop| { + hop.edge_projected_columns = hop + .edge_projected_columns + .max(edge_columns.try_into().unwrap_or(u64::MAX)); + hop.node_projected_columns = hop + .node_projected_columns + .max(node_columns.try_into().unwrap_or(u64::MAX)); + }); +} + pub(crate) fn record_emitted(edge_var: u32, rows: usize) { with_hop(edge_var, |hop| hop.rows_emitted += rows as u64); } @@ -368,6 +443,12 @@ impl PhysicalOptimizerRule for FixedHopDemandRule { plan: Arc, config: &ConfigOptions, ) -> Result> { + let plan = if contains_materializable_expand(&plan) { + let required = (0..plan.schema().fields().len()).collect::>(); + rewrite_materialization(plan, &required)? + } else { + plan + }; let Some(terminal) = find_terminal_demand(&plan) else { return Ok(plan); }; @@ -402,6 +483,107 @@ impl PhysicalOptimizerRule for FixedHopDemandRule { } } +fn collect_expr_columns(expr: &Arc, required: &mut BTreeSet) { + required.extend( + collect_columns(expr) + .into_iter() + .map(|column| column.index()), + ); +} + +/// Propagate exact physical output demand through operators whose column +/// dependency is explicit. Unknown and multi-input operators are conservative +/// barriers and require every child column. +fn rewrite_materialization( + plan: Arc, + required: &BTreeSet, +) -> Result> { + if let Some(projection) = plan.downcast_ref::() { + let mut child_required = BTreeSet::new(); + for &output in required { + if let Some(expr) = projection.expr().get(output) { + collect_expr_columns(&expr.expr, &mut child_required); + } + } + let child = rewrite_materialization(Arc::clone(projection.input()), &child_required)?; + return plan.with_new_children(vec![child]); + } + + let mut child_required = required.clone(); + if let Some(filter) = plan.downcast_ref::() { + // A FilterExec may carry DataFusion's own output projection. Its + // output ordinals are not input ordinals: map terminal demand through + // that projection before adding predicate dependencies. Treating the + // ordinals as identical silently selected an earlier same-named field + // in multi-hop plans (for example `a.node_uuid` instead of + // `c.node_uuid`) and allowed the demanded destination identity to be + // replaced with an unused placeholder. + if let Some(projection) = filter.projection() { + child_required = required + .iter() + .filter_map(|output| projection.get(*output).copied()) + .collect(); + } + collect_expr_columns(filter.predicate(), &mut child_required); + } + if let Some(sort) = plan.downcast_ref::() { + for expr in sort.expr() { + collect_expr_columns(&expr.expr, &mut child_required); + } + } + + if let Some(expand) = plan.downcast_ref::() { + let mut input_required = required + .iter() + .copied() + .filter(|index| *index < expand.input_width) + .collect::>(); + input_required.insert(expand.src_col_idx); + let child = rewrite_materialization(Arc::clone(&expand.input), &input_required)?; + let rebuilt = plan.with_new_children(vec![child])?; + let expand = rebuilt.downcast_ref::().ok_or_else(|| { + DataFusionError::Internal("ExpandExec rewrite changed physical type".into()) + })?; + let mask = (0..expand.schema().fields().len()) + .map(|index| required.contains(&index)) + .collect(); + return Ok(expand.with_required_output(mask)); + } + + let children = plan.children(); + if children.len() != 1 { + let rewritten = children + .into_iter() + .map(|child| { + let all = (0..child.schema().fields().len()).collect::>(); + rewrite_materialization(Arc::clone(child), &all) + }) + .collect::>>()?; + return if rewritten.is_empty() { + Ok(plan) + } else { + plan.with_new_children(rewritten) + }; + } + let child = children[0]; + let next = if materialization_transparent(plan.as_ref()) { + child_required + } else { + (0..child.schema().fields().len()).collect() + }; + let child = rewrite_materialization(Arc::clone(child), &next)?; + plan.with_new_children(vec![child]) +} + +fn materialization_transparent(plan: &dyn ExecutionPlan) -> bool { + plan.downcast_ref::().is_some() + || plan.downcast_ref::().is_some() + || plan.downcast_ref::().is_some() + || plan.downcast_ref::().is_some() + || plan.downcast_ref::().is_some() + || plan.downcast_ref::().is_some() +} + fn find_terminal_demand(plan: &Arc) -> Option { if let Some(limit) = plan.downcast_ref::() { let fetch = limit.fetch()?; @@ -447,6 +629,18 @@ fn contains_demand_expand(plan: &Arc) -> bool { is_fetch_transparent(plan.as_ref()) && plan.children().into_iter().any(contains_demand_expand) } +/// Whether the plan contains a fixed hop whose physical columns can be +/// narrowed. Materialization demand is independent of bounded-fetch demand: +/// it may safely cross an order-preserving sort even though cancellation must +/// stop at that blocking boundary. +fn contains_materializable_expand(plan: &Arc) -> bool { + plan.is::() + || plan + .children() + .into_iter() + .any(contains_materializable_expand) +} + fn rewrite_bounded( plan: Arc, batch_goal: usize, diff --git a/crates/graphforge-exec/src/lib.rs b/crates/graphforge-exec/src/lib.rs index d1785a1fd..96fe9cce9 100644 --- a/crates/graphforge-exec/src/lib.rs +++ b/crates/graphforge-exec/src/lib.rs @@ -226,12 +226,15 @@ mod write_driver; use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt; use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; use arrow::array::{ - Array, ArrayRef, FixedSizeBinaryArray, Int8Array, RecordBatch, StructArray, UInt64Array, + Array, ArrayRef, FixedSizeBinaryArray, FixedSizeBinaryBuilder, Int8Array, ListBuilder, + RecordBatch, StringArray, StructArray, TimestampMicrosecondArray, UInt32Array, UInt32Builder, + UInt64Array, new_null_array, }; -use arrow::datatypes::{DataType, SchemaRef}; +use arrow::datatypes::{DataType, Field, SchemaRef, TimeUnit}; use async_trait::async_trait; use datafusion::common::{DFSchema, DFSchemaRef, DataFusionError}; use datafusion::execution::context::{ExecutionProps, QueryPlanner, SessionState}; @@ -3004,9 +3007,14 @@ fn build_edge_prop_children( vec![rel_type_name.to_owned()] }; let mut prop_batches_by_rel = Vec::with_capacity(stems.len()); + let property_names = prop_fields + .iter() + .map(|field| field.name().clone()) + .collect::>(); for stem in &stems { - let batches = graphforge_storage::read_edge_properties(dir, stem) - .map_err(|e| exec_err(e.to_string()))?; + let batches = + graphforge_storage::read_edge_properties_projected(dir, stem, &property_names) + .map_err(|e| exec_err(e.to_string()))?; if let Some(first) = batches.first() { prop_batches_by_rel.push( concat_batches(&first.schema(), &batches).map_err(|e| exec_err(e.to_string()))?, @@ -3123,6 +3131,97 @@ impl ValueAt for arrow::array::UInt64Array { // ExpandExec — adjacency-backed single-hop expansion (#763) // --------------------------------------------------------------------------- +/// Generation-pinned destination-identity resolver shared by every query and +/// hop of one facade. Replacement is atomic with facade generation adoption; +/// execution never rediscovers mutable identity files by path. +#[derive(Debug, Default)] +pub struct V4OrdinalIdentityResolver { + handle: RwLock< + Option>>, + >, +} + +struct V4OrdinalIdentityPin { + session: Option>, + required: bool, +} + +impl V4OrdinalIdentityResolver { + /// Construct a resolver for an optional admitted generation facet. + #[must_use] + pub fn new( + handle: Option, + ) -> Self { + Self { + handle: RwLock::new(handle.map(|handle| Arc::new(Mutex::new(handle)))), + } + } + + /// Replace the exact generation served by subsequent sessions. + pub fn replace( + &self, + handle: Option, + ) { + *self.handle.write().expect("ordinal identity lock poisoned") = + handle.map(|handle| Arc::new(Mutex::new(handle))); + } + + fn pin(&self) -> Result { + let handle = self + .handle + .read() + .expect("ordinal identity lock poisoned") + .clone(); + let Some(handle) = handle else { + return Ok(V4OrdinalIdentityPin { + session: None, + required: false, + }); + }; + let revalidation = handle + .lock() + .expect("ordinal identity handle poisoned") + .revalidate_for_session() + .map_err(|error| GfError::Execution(error.to_string()))?; + Ok(V4OrdinalIdentityPin { + session: Some(Arc::new(V4OrdinalIdentitySession { + handle, + revalidation, + attribution_available: AtomicBool::new(true), + })), + required: true, + }) + } +} + +/// One exact, already-authenticated ordinal authority pinned for the lifetime +/// of an execution session. +#[derive(Debug)] +struct V4OrdinalIdentitySession { + handle: Arc>, + revalidation: graphforge_storage::V4OrdinalRevalidationMetrics, + attribution_available: AtomicBool, +} + +impl V4OrdinalIdentitySession { + fn lookup_node_uuids( + &self, + requested: &[u64], + ) -> Result { + let mut lookup = self + .handle + .lock() + .expect("ordinal identity handle poisoned") + .lookup_node_uuids_pinned(requested) + .map_err(|error| GfError::Execution(error.to_string()))?; + if self.attribution_available.swap(false, Ordering::AcqRel) { + lookup.metrics.revalidation_calls = self.revalidation.calls; + lookup.metrics.revalidation_bytes = self.revalidation.bytes_read; + } + Ok(lookup) + } +} + /// Physical node for adjacency-backed single-hop expansion, the physical /// counterpart of [`graphforge_plan::ExpandNode`]. /// @@ -3156,16 +3255,26 @@ pub struct ExpandExec { demand_batch: Option, /// Query-scoped terminal cancellation shared by the bounded hop chain. demand: Option>, + /// Exact output columns consumed above this operator. `None` preserves the + /// standalone full-schema contract. + required_output: Option>, + /// Facade-owned generation-pinned ordinal identity authority. + ordinal_identities: Option>, + /// A facade configured an identity authority, but admission may have + /// failed closed for this generation. Standalone sessions leave this false. + ordinal_identity_required: bool, } impl ExpandExec { /// Build the physical node from its logical counterpart, planned input, /// and the session's adjacency provider. #[must_use] - pub fn new( + pub(crate) fn new( node: &graphforge_plan::ExpandNode, input: Arc, provider: Arc, + ordinal_identities: Option>, + ordinal_identity_required: bool, ) -> Self { let schema: SchemaRef = Arc::new(node.schema().as_arrow().clone()); let props = Arc::new(PlanProperties::new( @@ -3196,6 +3305,9 @@ impl ExpandExec { edge_var: node.edge_var, demand_batch: None, demand: None, + required_output: None, + ordinal_identities, + ordinal_identity_required, } } @@ -3220,6 +3332,32 @@ impl ExpandExec { edge_var: self.edge_var, demand_batch: Some(batch_goal), demand: Some(demand), + required_output: self.required_output.clone(), + ordinal_identities: self.ordinal_identities.clone(), + ordinal_identity_required: self.ordinal_identity_required, + }) + } + + fn with_required_output(&self, required: Vec) -> Arc { + Arc::new(Self { + input: Arc::clone(&self.input), + rel_type_name: self.rel_type_name.clone(), + direction: self.direction, + dir: self.dir.clone(), + mode: self.mode, + src_col_idx: self.src_col_idx, + edge_prop_count: self.edge_prop_count, + input_width: self.input_width, + schema: Arc::clone(&self.schema), + props: Arc::clone(&self.props), + provider: Arc::clone(&self.provider), + fetch: self.fetch, + edge_var: self.edge_var, + demand_batch: self.demand_batch, + demand: self.demand.clone(), + required_output: Some(required.into()), + ordinal_identities: self.ordinal_identities.clone(), + ordinal_identity_required: self.ordinal_identity_required, }) } } @@ -3243,15 +3381,26 @@ impl DisplayAs for ExpandExec { }; write!( f, - "ExpandExec: rel={}, dir={arrow}, adjacency={}, fetch={}, demand_batch={}, cancel={}", + "ExpandExec: rel={}, dir={arrow}, adjacency={}, identity={}, fetch={}, demand_batch={}, projection={}, cancel={}", self.rel_type_name, self.provider .status(&self.rel_type_name, self.direction) .as_str(), + if self.ordinal_identities.is_some() { + "v4" + } else if self.ordinal_identity_required { + "required-missing" + } else { + "legacy" + }, self.fetch .map_or_else(|| "all".to_owned(), |n| n.to_string()), self.demand_batch .map_or_else(|| "all".to_owned(), |n| n.to_string()), + self.required_output.as_ref().map_or_else( + || "all".to_owned(), + |mask| mask.iter().filter(|needed| **needed).count().to_string(), + ), if self.demand.is_some() { "guarded" } else { @@ -3298,6 +3447,9 @@ impl ExecutionPlan for ExpandExec { edge_var: self.edge_var, demand_batch: self.demand_batch, demand: self.demand.clone(), + required_output: self.required_output.clone(), + ordinal_identities: self.ordinal_identities.clone(), + ordinal_identity_required: self.ordinal_identity_required, })) } @@ -3318,6 +3470,9 @@ impl ExecutionPlan for ExpandExec { edge_var: self.edge_var, demand_batch: self.demand_batch, demand: self.demand.clone(), + required_output: self.required_output.clone(), + ordinal_identities: self.ordinal_identities.clone(), + ordinal_identity_required: self.ordinal_identity_required, })) } @@ -3355,6 +3510,9 @@ impl ExecutionPlan for ExpandExec { provider: self.provider.clone(), edge_var: self.edge_var, demand: self.demand.clone(), + required_output: self.required_output.clone(), + ordinal_identities: self.ordinal_identities.clone(), + ordinal_identity_required: self.ordinal_identity_required, }; let schema = self.schema.clone(); let batch_size = context.session_config().batch_size(); @@ -3446,6 +3604,9 @@ struct SingleHopConfig { provider: Arc, edge_var: u32, demand: Option>, + required_output: Option>, + ordinal_identities: Option>, + ordinal_identity_required: bool, } /// Resumable position within one input batch. Keeping the raw adjacency offset @@ -3459,6 +3620,58 @@ struct SingleHopPosition { seen_edges: std::collections::HashSet, } +/// Build a schema-valid value column for an output that is provably unused by +/// every operator above this Expand. Nullable fields use Arrow nulls; required +/// physical fields receive inert values so the unchanged logical schema stays +/// valid without forcing their backing Parquet columns to be read. +fn unused_expand_column(field: &Field, rows: usize) -> Result { + if field.is_nullable() { + return Ok(new_null_array(field.data_type(), rows)); + } + let column: ArrayRef = match field.data_type() { + DataType::FixedSizeBinary(width) => Arc::new( + FixedSizeBinaryArray::try_from_iter( + (0..rows).map(|_| vec![0_u8; usize::try_from(*width).unwrap_or(0)]), + ) + .map_err(|error| GfError::Execution(error.to_string()))?, + ), + DataType::UInt64 => Arc::new(UInt64Array::from(vec![0_u64; rows])), + DataType::UInt32 => Arc::new(UInt32Array::from(vec![0_u32; rows])), + DataType::Utf8 => Arc::new(StringArray::from(vec![""; rows])), + DataType::Timestamp(TimeUnit::Microsecond, timezone) => { + let values = TimestampMicrosecondArray::from(vec![0_i64; rows]); + Arc::new(if let Some(timezone) = timezone { + values.with_timezone(Arc::clone(timezone)) + } else { + values + }) + } + DataType::List(item) if item.data_type() == &DataType::UInt32 => { + let mut builder = ListBuilder::new(UInt32Builder::new()).with_field(Arc::clone(item)); + for _ in 0..rows { + builder.append(true); + } + Arc::new(builder.finish()) + } + data_type => { + return Err(GfError::Execution(format!( + "Expand cannot synthesize unused non-nullable output '{}' with type {data_type}", + field.name() + ))); + } + }; + Ok(column) +} + +fn require_admitted_ordinal_identity(required: bool, admitted: bool) -> Result<(), GfError> { + if required && !admitted { + return Err(GfError::Execution( + "destination UUID projection requires admitted v4 ordinal identity".into(), + )); + } + Ok(()) +} + /// Execute the adjacency-backed single-hop expansion: for every input row's /// source node, emit one output row per adjacency entry, assembling input, /// edge-topology, edge-property (nullable), and destination-node columns in @@ -3534,6 +3747,132 @@ fn expand_single_hop_chunk( } demand::record_candidates(cfg.edge_var, triples.len()); + let dst_width = graphforge_storage::TOPOLOGY_NODES_SCHEMA.fields().len(); + let edge_end = cfg.out_schema.fields().len().saturating_sub(dst_width); + let required = cfg.required_output.as_deref(); + let edge_materialization_unused = required.is_some_and(|mask| { + mask.get(cfg.input_width..edge_end).is_some_and(|fields| { + fields.iter().enumerate().all(|(offset, needed)| { + !needed || cfg.out_schema.field(cfg.input_width + offset).name() == "edge_id" + }) + }) + }); + let destination_uuid_index = edge_end; + let destination_id_index = edge_end + 1; + let destination_identity_only = required.is_some_and(|mask| { + mask.iter() + .enumerate() + .skip(edge_end) + .all(|(index, needed)| { + !needed || index == destination_uuid_index || index == destination_id_index + }) + }); + let uuid_required = + required.is_some_and(|mask| mask.get(destination_uuid_index).copied().unwrap_or(false)); + if edge_materialization_unused && destination_identity_only && uuid_required { + require_admitted_ordinal_identity( + cfg.ordinal_identity_required, + cfg.ordinal_identities.is_some(), + )?; + } + if edge_materialization_unused + && destination_identity_only + && let Some(ordinal_identities) = cfg.ordinal_identities.as_ref() + { + let mut requested = reached.iter().copied().collect::>(); + requested.sort_unstable(); + let (resolved, identity_metrics) = if uuid_required { + let lookup = ordinal_identities.lookup_node_uuids(&requested)?; + (lookup.values, lookup.metrics) + } else { + ( + vec![None; requested.len()], + graphforge_storage::V4OrdinalLookupMetrics::default(), + ) + }; + let mut uuids = HashMap::with_capacity(requested.len()); + for (node_id, uuid) in requested.into_iter().zip(resolved) { + if uuid_required { + let uuid = uuid.ok_or_else(|| { + GfError::Execution(format!( + "Expand reached unknown destination node_id {node_id}" + )) + })?; + uuids.insert(node_id, *uuid.as_bytes()); + } + } + let src_take = arrow::array::UInt32Array::from( + triples + .iter() + .map(|(row, _, _)| { + u32::try_from(*row).map_err(|_| exec_err("row index exceeds u32".into())) + }) + .collect::, _>>()?, + ); + let mut columns = Vec::with_capacity(cfg.out_schema.fields().len()); + for column in input.columns() { + columns + .push(take(column, &src_take, None).map_err(|error| exec_err(error.to_string()))?); + } + for (offset, field) in cfg + .out_schema + .fields() + .iter() + .skip(cfg.input_width) + .take(edge_end.saturating_sub(cfg.input_width)) + .enumerate() + { + let index = cfg.input_width + offset; + if required.is_some_and(|mask| mask[index]) && field.name() == "edge_id" { + columns.push(Arc::new(UInt64Array::from( + triples + .iter() + .map(|(_, edge_id, _)| *edge_id) + .collect::>(), + ))); + } else { + columns.push(unused_expand_column(field, triples.len())?); + } + } + for (offset, field) in cfg.out_schema.fields().iter().skip(edge_end).enumerate() { + let index = edge_end + offset; + let column: ArrayRef = if required.is_some_and(|mask| mask[index]) + && index == destination_id_index + { + Arc::new(UInt64Array::from( + triples + .iter() + .map(|(_, _, neighbor)| *neighbor) + .collect::>(), + )) + } else if required.is_some_and(|mask| mask[index]) && index == destination_uuid_index { + let mut builder = FixedSizeBinaryBuilder::with_capacity(triples.len(), 16); + for (_, _, neighbor) in &triples { + builder + .append_value(uuids[neighbor]) + .map_err(|error| exec_err(error.to_string()))?; + } + Arc::new(builder.finish()) + } else { + unused_expand_column(field, triples.len())? + }; + columns.push(column); + } + let output = RecordBatch::try_new(cfg.out_schema.clone(), columns) + .map_err(|error| exec_err(error.to_string()))?; + let projected_columns = required.map_or(cfg.out_schema.fields().len(), |mask| { + mask.iter().filter(|needed| **needed).count() + }); + demand::record_identity_projection( + cfg.edge_var, + output.num_rows(), + projected_columns, + &identity_metrics, + ); + demand::record_emitted(cfg.edge_var, output.num_rows()); + return Ok(output); + } + // Edge rows keyed by edge_id, for the edge topology columns — read // lazily for the traversed ids only. let edge_permit = cfg @@ -3547,13 +3886,40 @@ fn expand_single_hop_chunk( Arc::new(demand::HopReadObserver::new(cfg.edge_var)) as Arc }); - let edge_batches = graphforge_storage::read_edges_filtered_observed( - &cfg.dir, - &cfg.rel_type_name, - cfg.mode, - &traversed, - edge_observer.as_ref(), - ) + let edge_topology_width = edge_end + .saturating_sub(cfg.input_width) + .saturating_sub(cfg.edge_prop_count); + let relationship_properties_required = required.is_none_or(|mask| { + mask[cfg.input_width + edge_topology_width..edge_end] + .iter() + .any(|needed| *needed) + }); + let mut edge_projection = (0..edge_topology_width) + .filter(|offset| required.is_none_or(|mask| mask[cfg.input_width + offset])) + .collect::>(); + // edge_id keys adjacency entries; edge_uuid keys demanded relationship + // properties. Storage adds edge_id automatically. + if relationship_properties_required { + edge_projection.push(0); + } + let edge_batches = if required.is_some() { + graphforge_storage::read_edges_filtered_projected_observed( + &cfg.dir, + &cfg.rel_type_name, + cfg.mode, + &traversed, + &edge_projection, + edge_observer.as_ref(), + ) + } else { + graphforge_storage::read_edges_filtered_observed( + &cfg.dir, + &cfg.rel_type_name, + cfg.mode, + &traversed, + edge_observer.as_ref(), + ) + } .map_err(|e| exec_err(e.to_string()))?; drop(edge_permit); let edge_schema = edge_batches @@ -3562,16 +3928,25 @@ fn expand_single_hop_chunk( .ok_or_else(|| exec_err("Expand: edge scan returned no batches".into()))?; let edge_batch = concat_batches(&edge_schema, &edge_batches).map_err(|e| exec_err(e.to_string()))?; - let edge_ids_col = u64_column(&edge_batch, 3)?; // edge_id + let edge_id_index = edge_batch + .schema() + .index_of("edge_id") + .map_err(|error| exec_err(error.to_string()))?; + let edge_ids_col = u64_column(&edge_batch, edge_id_index)?; let edge_row: HashMap = (0..edge_batch.num_rows()) .filter_map(|i| edge_ids_col.value_at(i).map(|id| (id, i))) .collect(); - let edge_uuids = edge_batch - .column(0) - .as_any() - .downcast_ref::() - .filter(|a| a.value_length() == 16) - .ok_or_else(|| exec_err("Expand: edge_uuid column is not FixedSizeBinary(16)".into()))?; + let edge_uuids = relationship_properties_required + .then(|| { + edge_batch + .column_by_name("edge_uuid") + .and_then(|column| column.as_any().downcast_ref::()) + .filter(|array| array.value_length() == 16) + .ok_or_else(|| { + exec_err("Expand: edge_uuid column is not FixedSizeBinary(16)".into()) + }) + }) + .transpose()?; // Destination node rows keyed by node_id — read lazily for the reached // neighbors only (#838), so an index Hit does not scan the whole node table. @@ -3586,11 +3961,43 @@ fn expand_single_hop_chunk( Arc::new(demand::HopReadObserver::new(cfg.edge_var)) as Arc }); - let node_batches = graphforge_storage::read_nodes_filtered_observed( - &cfg.dir, - &reached, - node_observer.as_ref(), - ) + let node_projection = (0..dst_width) + .filter(|offset| required.is_none_or(|mask| mask[edge_end + offset])) + .collect::>(); + let edge_key_index = + if matches!(cfg.mode, OntologyMode::Exploratory) || cfg.rel_type_name == "*" { + graphforge_storage::EXPLORATORY_EDGE_SCHEMA.index_of("edge_id") + } else { + graphforge_storage::TYPED_EDGE_SCHEMA.index_of("edge_id") + } + .map_err(|error| exec_err(error.to_string()))?; + let edge_key_already_demanded = usize::from(edge_projection.contains(&edge_key_index)); + let node_key_already_demanded = usize::from(node_projection.contains(&1)); + demand::record_materialization_projection( + cfg.edge_var, + edge_projection + .len() + .saturating_add(1_usize.saturating_sub(edge_key_already_demanded)) + .saturating_add(required.map_or(cfg.edge_prop_count, |mask| { + mask[cfg.input_width + edge_topology_width..edge_end] + .iter() + .filter(|needed| **needed) + .count() + })), + node_projection + .len() + .saturating_add(1_usize.saturating_sub(node_key_already_demanded)), + ); + let node_batches = if required.is_some() { + graphforge_storage::read_nodes_filtered_projected_observed( + &cfg.dir, + &reached, + &node_projection, + node_observer.as_ref(), + ) + } else { + graphforge_storage::read_nodes_filtered_observed(&cfg.dir, &reached, node_observer.as_ref()) + } .map_err(|e| exec_err(e.to_string()))?; drop(node_permit); let Some(first) = node_batches.first() else { @@ -3598,7 +4005,11 @@ fn expand_single_hop_chunk( }; let node_batch = concat_batches(&first.schema(), &node_batches).map_err(|e| exec_err(e.to_string()))?; - let node_ids = u64_column(&node_batch, 1)?; + let node_id_index = node_batch + .schema() + .index_of("node_id") + .map_err(|error| exec_err(error.to_string()))?; + let node_ids = u64_column(&node_batch, node_id_index)?; let node_row: HashMap = node_ids .iter() .enumerate() @@ -3628,14 +4039,16 @@ fn expand_single_hop_chunk( src_take.push(to_u32(row)?); edge_take.push(to_u32(edge_idx)?); dst_take.push(to_u32(dst_idx)?); - if edge_uuids.is_null(edge_idx) { - return Err(exec_err(format!( - "Expand: edge_id {edge_id} has a null edge_uuid" - ))); + if let Some(edge_uuids) = edge_uuids { + if edge_uuids.is_null(edge_idx) { + return Err(exec_err(format!( + "Expand: edge_id {edge_id} has a null edge_uuid" + ))); + } + let mut edge_uuid = [0u8; 16]; + edge_uuid.copy_from_slice(edge_uuids.value(edge_idx)); + output_edge_uuids.push(edge_uuid); } - let mut edge_uuid = [0u8; 16]; - edge_uuid.copy_from_slice(edge_uuids.value(edge_idx)); - output_edge_uuids.push(edge_uuid); } let src_take = arrow::array::UInt32Array::from(src_take); let edge_take = arrow::array::UInt32Array::from(edge_take); @@ -3643,30 +4056,79 @@ fn expand_single_hop_chunk( // Assemble columns in ExpandNode schema order: input ++ edge topology ++ // edge properties (nullable) ++ destination node. - let edge_topo_width = edge_batch.num_columns(); let prop_fields: Vec = cfg .out_schema .fields() .iter() - .skip(cfg.input_width + edge_topo_width) + .skip(cfg.input_width + edge_topology_width) .take(cfg.edge_prop_count) .cloned() .collect(); let mut columns = Vec::with_capacity(cfg.out_schema.fields().len()); - for col in input.columns() { - columns.push(take(col, &src_take, None).map_err(|e| exec_err(e.to_string()))?); + for column in input.columns() { + columns.push(take(column, &src_take, None).map_err(|error| exec_err(error.to_string()))?); } - for col in edge_batch.columns() { - columns.push(take(col, &edge_take, None).map_err(|e| exec_err(e.to_string()))?); + for (offset, field) in cfg + .out_schema + .fields() + .iter() + .skip(cfg.input_width) + .take(edge_topology_width) + .enumerate() + { + let index = cfg.input_width + offset; + columns.push(if required.is_none_or(|mask| mask[index]) { + let column = edge_batch.column_by_name(field.name()).ok_or_else(|| { + exec_err(format!( + "Expand: projected edge column {} is absent", + field.name() + )) + })?; + take(column, &edge_take, None).map_err(|error| exec_err(error.to_string()))? + } else { + unused_expand_column(field, triples.len())? + }); } - columns.extend(build_edge_prop_children( + let demanded_property_fields = prop_fields + .iter() + .enumerate() + .filter(|(offset, _)| { + required.is_none_or(|mask| mask[cfg.input_width + edge_topology_width + offset]) + }) + .map(|(_, field)| Arc::clone(field)) + .collect::>(); + let demanded_property_columns = build_edge_prop_children( &cfg.rel_type_name, &cfg.dir, - &prop_fields, + &demanded_property_fields, &output_edge_uuids, - )?); - for col in node_batch.columns() { - columns.push(take(col, &dst_take, None).map_err(|e| exec_err(e.to_string()))?); + )?; + let demanded_properties = demanded_property_fields + .iter() + .map(|field| field.name().clone()) + .zip(demanded_property_columns) + .collect::>(); + for field in &prop_fields { + columns.push( + demanded_properties + .get(field.name()) + .cloned() + .map_or_else(|| unused_expand_column(field, triples.len()), Ok)?, + ); + } + for (offset, field) in cfg.out_schema.fields().iter().skip(edge_end).enumerate() { + let index = edge_end + offset; + columns.push(if required.is_none_or(|mask| mask[index]) { + let column = node_batch.column_by_name(field.name()).ok_or_else(|| { + exec_err(format!( + "Expand: projected node column {} is absent", + field.name() + )) + })?; + take(column, &dst_take, None).map_err(|error| exec_err(error.to_string()))? + } else { + unused_expand_column(field, triples.len())? + }); } let output = RecordBatch::try_new(cfg.out_schema.clone(), columns) .map_err(|e| exec_err(e.to_string()))?; @@ -4198,6 +4660,47 @@ fn unwind_explode( /// [`GraphForgeExtensionPlanner`]. pub struct AdjacencyProviderExt(pub Arc); +/// `SessionConfig` extension carrying the facade's exact generation-pinned +/// ordinal identity authority. +struct OrdinalIdentityResolverExt(pub Option>); + +fn plan_expand_extension( + expand: &graphforge_plan::ExpandNode, + physical_inputs: &[Arc], + session_state: &SessionState, +) -> Result, DataFusionError> { + let input = physical_inputs + .first() + .cloned() + .ok_or_else(|| DataFusionError::Internal("Expand requires one physical input".into()))?; + let provider = session_state + .config() + .get_extension::() + .map_or_else( + || { + Arc::new(ScanBuildAdjacencyProvider::new( + expand.dir.clone(), + expand.mode, + )) as Arc + }, + |ext| Arc::clone(&ext.0), + ); + let identity_extension = session_state + .config() + .get_extension::(); + let ordinal_identity_required = identity_extension.is_some(); + let ordinal_identities = identity_extension + .as_ref() + .and_then(|extension| extension.0.as_ref().map(Arc::clone)); + Ok(Arc::new(ExpandExec::new( + expand, + input, + provider, + ordinal_identities, + ordinal_identity_required, + ))) +} + /// Plans GraphForge's custom logical [`Extension`](LogicalPlan::Extension) /// nodes into physical [`ExecutionPlan`]s. #[derive(Debug, Default)] @@ -4238,22 +4741,7 @@ impl ExtensionPlanner for GraphForgeExtensionPlanner { return Ok(Some(Arc::new(GraphRemoveExec::new(remove, input)))); } if let Some(expand) = node.as_any().downcast_ref::() { - let input = physical_inputs.first().cloned().ok_or_else(|| { - DataFusionError::Internal("Expand requires one physical input".into()) - })?; - let provider = session_state - .config() - .get_extension::() - .map_or_else( - || { - Arc::new(ScanBuildAdjacencyProvider::new( - expand.dir.clone(), - expand.mode, - )) as Arc - }, - |ext| Arc::clone(&ext.0), - ); - return Ok(Some(Arc::new(ExpandExec::new(expand, input, provider)))); + return plan_expand_extension(expand, physical_inputs, session_state).map(Some); } if let Some(var_len) = node.as_any().downcast_ref::() { let input = physical_inputs.first().cloned().ok_or_else(|| { @@ -4398,6 +4886,12 @@ pub struct ExecutionSession { relational_fixed_hop_reference: bool, } +#[derive(Default)] +struct OrdinalIdentityConfig { + session: Option>, + required: bool, +} + impl ExecutionSession { /// Create a read/query session. /// @@ -4413,6 +4907,7 @@ impl ExecutionSession { PathBuf::new(), OntologyMode::Exploratory, None, + OrdinalIdentityConfig::default(), &SessionResourceConfig::default(), )) } @@ -4433,6 +4928,7 @@ impl ExecutionSession { dir, mode, None, + OrdinalIdentityConfig::default(), &SessionResourceConfig::default(), )) } @@ -4475,12 +4971,43 @@ impl ExecutionSession { provider: Arc, resources: &SessionResourceConfig, ) -> Result { + Self::new_with_target_provider_resources_and_identity( + catalog, ontology, dir, mode, provider, None, resources, + ) + } + + /// Like [`Self::new_with_target_provider_and_resources`] with an exact + /// generation-pinned destination identity authority. + pub fn new_with_target_provider_resources_and_identity( + catalog: GraphCatalog, + ontology: Option, + dir: PathBuf, + mode: OntologyMode, + provider: Arc, + ordinal_identities: Option>, + resources: &SessionResourceConfig, + ) -> Result { + let identity = match ordinal_identities { + Some(resolver) => { + let pin = resolver.pin()?; + OrdinalIdentityConfig { + // Requirement and payload come from one resolver snapshot. + // Keep them separate so the expansion boundary remains + // fail-closed if a future planner/config rewrite loses the + // pinned authority. + required: pin.required, + session: pin.session, + } + } + None => OrdinalIdentityConfig::default(), + }; Ok(Self::build( catalog, ontology, dir, mode, Some(provider), + identity, resources, )) } @@ -4491,6 +5018,7 @@ impl ExecutionSession { dir: PathBuf, mode: OntologyMode, shared_provider: Option>, + identity: OrdinalIdentityConfig, resources: &SessionResourceConfig, ) -> Self { // The session-scoped adjacency provider (#761), threaded to the @@ -4515,6 +5043,9 @@ impl ExecutionSession { ))) .with_target_partitions(resources.target_partitions) .with_batch_size(resources.batch_size); + if identity.session.is_some() || identity.required { + config = config.with_extension(Arc::new(OrdinalIdentityResolverExt(identity.session))); + } // Authenticated overlay scans publish sound physical-row upper bounds. // Let DataFusion use those estimates so a small one-partition source is // not eagerly repartitioned merely because newest-wins makes its exact @@ -5394,6 +5925,17 @@ mod tests { ExecutionSession::new(catalog, None).unwrap() } + #[test] + fn unused_list_output_preserves_exact_non_nullable_child_schema() { + let item = Arc::new(Field::new("item", DataType::UInt32, false)); + let field = Field::new("type_ids", DataType::List(Arc::clone(&item)), false); + let column = unused_expand_column(&field, 3).unwrap(); + + assert_eq!(column.data_type(), field.data_type()); + assert_eq!(column.len(), 3); + assert_eq!(column.null_count(), 0); + } + #[test] fn session_uses_sound_row_estimates_for_partition_planning() { let session = make_session(); @@ -6517,6 +7059,19 @@ mod tests { assert!(u64_column(&utf8, 0).is_err()); } + #[test] + fn required_v4_destination_identity_fails_closed_without_admitted_session() { + let error = require_admitted_ordinal_identity(true, false) + .expect_err("required v4 identity must not fall back without its admitted session"); + assert!( + error + .to_string() + .contains("requires admitted v4 ordinal identity") + ); + require_admitted_ordinal_identity(true, true).expect("admitted v4 session"); + require_admitted_ordinal_identity(false, false).expect("legacy generation fallback"); + } + #[test] fn write_execs_expose_stable_plan_contracts_and_reject_invalid_shape() { let dir = TempDir::new().unwrap(); diff --git a/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_absent_expand_exec.snap b/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_absent_expand_exec.snap index b59501f8f..9a4a6f284 100644 --- a/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_absent_expand_exec.snap +++ b/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_absent_expand_exec.snap @@ -7,7 +7,7 @@ ProjectionExec: expr=[name@0 as bn] FilterExec: array_has(type_ids@1, 1073741824), projection=[node_uuid@0] RepartitionExec: partitioning=RoundRobinBatch(), input_partitions=1 ProjectionExec: expr=[node_uuid@15 as node_uuid, type_ids@18 as type_ids] - ExpandExec: rel=KNOWS, dir=->, adjacency=building, fetch=all, demand_batch=all, cancel=none + ExpandExec: rel=KNOWS, dir=->, adjacency=building, identity=legacy, fetch=all, demand_batch=all, projection=2, cancel=none HashJoinExec: mode=CollectLeft, join_type=Right, on=[(node_uuid@0, node_uuid@0)], projection=[node_uuid@2, node_id@3, type_id@4, type_ids@5, created_at@6, updated_at@7, name@1] PropertyOverlayExec: route=_untyped FilterExec: array_has(type_ids@3, 1073741824) diff --git a/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_present_expand_exec.snap b/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_present_expand_exec.snap index 4dcd1a21d..69223c892 100644 --- a/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_present_expand_exec.snap +++ b/crates/graphforge-exec/tests/explain_goldens/explain_snapshots__single_hop_index_present_expand_exec.snap @@ -7,7 +7,7 @@ ProjectionExec: expr=[name@0 as bn] FilterExec: array_has(type_ids@1, 1073741824), projection=[node_uuid@0] RepartitionExec: partitioning=RoundRobinBatch(), input_partitions=1 ProjectionExec: expr=[node_uuid@15 as node_uuid, type_ids@18 as type_ids] - ExpandExec: rel=KNOWS, dir=->, adjacency=hit, fetch=all, demand_batch=all, cancel=none + ExpandExec: rel=KNOWS, dir=->, adjacency=hit, identity=legacy, fetch=all, demand_batch=all, projection=2, cancel=none HashJoinExec: mode=CollectLeft, join_type=Right, on=[(node_uuid@0, node_uuid@0)], projection=[node_uuid@2, node_id@3, type_id@4, type_ids@5, created_at@6, updated_at@7, name@1] PropertyOverlayExec: route=_untyped FilterExec: array_has(type_ids@3, 1073741824) diff --git a/crates/graphforge-rel/src/expr.rs b/crates/graphforge-rel/src/expr.rs index 9dfdd4999..288d92eda 100644 --- a/crates/graphforge-rel/src/expr.rs +++ b/crates/graphforge-rel/src/expr.rs @@ -3971,6 +3971,7 @@ impl ScalarUDFImpl for CypherRelationshipDisjoint { fn relationship_ids(value: &ScalarValue, ids: &mut Vec>) { match value { ScalarValue::FixedSizeBinary(_, Some(uuid)) => ids.push(uuid.clone()), + ScalarValue::UInt64(Some(edge_id)) => ids.push(edge_id.to_le_bytes().to_vec()), ScalarValue::List(list) if !list.is_null(0) => { let values = list.value(0); for index in 0..values.len() { diff --git a/crates/graphforge-rel/src/lowerer.rs b/crates/graphforge-rel/src/lowerer.rs index a353bf884..02a730ab5 100644 --- a/crates/graphforge-rel/src/lowerer.rs +++ b/crates/graphforge-rel/src/lowerer.rs @@ -911,16 +911,28 @@ impl<'a> GraphPlanLowerer<'a> { let prior_alias = var_map .get(*prior) .ok_or(LoweringError::UnboundVar(prior.0))?; - let value = |alias: &str| { - if alias.ends_with(graphforge_plan::VAR_LEN_EDGE_LIST_FIELD) { + let edge_is_list = + edge_alias.ends_with(graphforge_plan::VAR_LEN_EDGE_LIST_FIELD); + let prior_is_list = + prior_alias.ends_with(graphforge_plan::VAR_LEN_EDGE_LIST_FIELD); + // Fixed-hop adjacency already carries the exact edge_id. + // Use that internal identity when both operands are fixed + // hops so uniqueness never forces an edge-Parquet UUID + // hydration. Mixed fixed/variable-length comparisons must + // remain on public UUID identity because path lists contain + // edge UUIDs rather than storage ordinals. + let value = |alias: &str, is_list: bool| { + if is_list { col(alias) + } else if !edge_is_list && !prior_is_list { + col(format!("{alias}.edge_id")) } else { col(format!("{alias}.edge_uuid")) } }; Ok(crate::expr::relationship_disjoint( - value(edge_alias), - value(prior_alias), + value(edge_alias, edge_is_list), + value(prior_alias, prior_is_list), )) }); let Some(mut predicate) = predicates.next().transpose()? else { diff --git a/crates/graphforge-storage/src/catalog.rs b/crates/graphforge-storage/src/catalog.rs index 48af5f30d..2c4b903c4 100644 --- a/crates/graphforge-storage/src/catalog.rs +++ b/crates/graphforge-storage/src/catalog.rs @@ -278,6 +278,66 @@ pub fn read_edges_filtered_observed( Ok(batches) } +/// Filter edge topology by `edge_id` while physically decoding only the +/// requested canonical columns plus the join key. +#[allow(clippy::implicit_hasher)] +#[doc(hidden)] +pub fn read_edges_filtered_projected_observed( + dir: &Path, + rel_name: &str, + mode: OntologyMode, + edge_ids: &std::collections::HashSet, + projection: &[usize], + observer: Option<&std::sync::Arc>, +) -> Result, DataFusionError> { + if rel_name == "*" && matches!(mode, OntologyMode::Advisory | OntologyMode::Strict) { + // Union normalization synthesizes rel_type_name. Normalize first, then + // shape the result; individual typed files still avoid node reads. + return read_edges_union(dir, Some(edge_ids), observer).and_then(|batches| { + project_batches_with_key(batches, &EXPLORATORY_EDGE_SCHEMA, projection, "edge_id") + }); + } + if matches!(mode, OntologyMode::Advisory | OntologyMode::Strict) { + let mut components = Path::new(rel_name).components(); + if !matches!(components.next(), Some(std::path::Component::Normal(_))) + || components.next().is_some() + { + return Err(DataFusionError::Execution(format!( + "invalid relation name {rel_name:?}: must be a plain file stem" + ))); + } + } + let (stem, schema) = match mode { + OntologyMode::Exploratory => ("_exploratory", EXPLORATORY_EDGE_SCHEMA.clone()), + OntologyMode::Advisory | OntologyMode::Strict => (rel_name, TYPED_EDGE_SCHEMA.clone()), + }; + let mut batches = Vec::new(); + for (_, path) in crate::mutator::edge_parquet_files(dir, Some(stem)) + .map_err(|error| DataFusionError::Execution(error.to_string()))? + { + let file_schema = admitted_parquet(&path)?.schema().clone(); + if file_schema.fields() != schema.fields() { + return Err(DataFusionError::Execution(format!( + "projected edge read requires canonical schema: {}", + path.display() + ))); + } + batches.extend(read_parquet_filtered_u64_projected( + &path, + schema.clone(), + "edge_id", + edge_ids, + FilteredReadKind::Edge, + observer, + projection, + )?); + } + if batches.is_empty() { + return project_batches_with_key(Vec::new(), &schema, projection, "edge_id"); + } + Ok(batches) +} + /// Read the union of every relation's edges (#823): the "all relation types" /// read for an untyped traversal in a typed project. Enumerates every /// `topology/edges/*.parquet` (stem order, for deterministic adjacency/BFS), @@ -701,7 +761,85 @@ fn read_parquet_filtered_u64( if ids.is_empty() || !path.exists() { return Ok(vec![RecordBatch::new_empty(fallback_schema)]); } - read_parquet_filtered_u64_attempt(path, fallback_schema, key_column, ids, kind, observer, true) + read_parquet_filtered_u64_attempt( + path, + fallback_schema, + key_column, + ids, + kind, + observer, + true, + None, + ) +} + +fn canonical_projection_with_key( + schema: &SchemaRef, + projection: &[usize], + key_column: &str, +) -> Result, DataFusionError> { + let mut indices = projection.to_vec(); + indices.push( + schema + .index_of(key_column) + .map_err(|error| DataFusionError::Execution(format!("filtered read: {error}")))?, + ); + indices.sort_unstable(); + indices.dedup(); + if indices.iter().any(|index| *index >= schema.fields().len()) { + return Err(DataFusionError::Execution( + "filtered read projection index is out of range".into(), + )); + } + Ok(indices) +} + +fn project_batches_with_key( + batches: Vec, + schema: &SchemaRef, + projection: &[usize], + key_column: &str, +) -> Result, DataFusionError> { + let indices = canonical_projection_with_key(schema, projection, key_column)?; + if batches.is_empty() { + let projected = Arc::new(arrow::datatypes::Schema::new( + indices + .iter() + .map(|index| schema.field(*index).clone()) + .collect::>(), + )); + return Ok(vec![RecordBatch::new_empty(projected)]); + } + batches + .into_iter() + .map(|batch| batch.project(&indices).map_err(Into::into)) + .collect() +} + +#[allow(clippy::too_many_arguments)] +fn read_parquet_filtered_u64_projected( + path: &Path, + fallback_schema: SchemaRef, + key_column: &str, + ids: &std::collections::HashSet, + kind: FilteredReadKind, + observer: Option<&std::sync::Arc>, + projection: &[usize], +) -> Result, DataFusionError> { + let projection = canonical_projection_with_key(&fallback_schema, projection, key_column)?; + if ids.is_empty() || !path.exists() { + return project_batches_with_key(Vec::new(), &fallback_schema, &projection, key_column); + } + read_parquet_filtered_u64_attempt( + path, + fallback_schema, + key_column, + ids, + kind, + observer, + true, + Some(&projection), + ) } #[allow(clippy::too_many_lines, clippy::too_many_arguments)] @@ -713,6 +851,7 @@ fn read_parquet_filtered_u64_attempt( kind: FilteredReadKind, observer: Option<&std::sync::Arc>, allow_dense_node_selection: bool, + projection: Option<&[usize]>, ) -> Result, DataFusionError> { use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::{ @@ -721,6 +860,17 @@ fn read_parquet_filtered_u64_attempt( use parquet::file::metadata::PageIndexPolicy; use parquet::file::statistics::Statistics; + let projected_schema = projection.map_or_else( + || fallback_schema.clone(), + |indices| { + Arc::new(arrow::datatypes::Schema::new( + indices + .iter() + .map(|index| fallback_schema.field(*index).clone()) + .collect::>(), + )) + }, + ); let mut observation = FilteredReadObservation::new(observer, kind); let file = File::open(path).map_err(|e| io_err(&e))?; // Optional, NOT required: with_page_index(true) errors on files lacking a @@ -768,13 +918,18 @@ fn read_parquet_filtered_u64_attempt( .map(|i| Some(!col.is_null(i) && ids.contains(&col.value(i)))) .collect() }; - filtered.push( - arrow::compute::filter_record_batch(batch, &mask) - .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?, - ); + let filtered_batch = arrow::compute::filter_record_batch(batch, &mask) + .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?; + filtered.push(if let Some(indices) = projection { + filtered_batch + .project(indices) + .map_err(|error| DataFusionError::ArrowError(Box::new(error), None))? + } else { + filtered_batch + }); } if filtered.is_empty() { - filtered.push(RecordBatch::new_empty(fallback_schema)); + filtered.push(RecordBatch::new_empty(projected_schema)); } record_pruning( kind, @@ -901,6 +1056,12 @@ fn read_parquet_filtered_u64_attempt( } else { builder }; + let builder = if let Some(indices) = projection { + let mask = ProjectionMask::roots(builder.parquet_schema(), indices.iter().copied()); + builder.with_projection(mask) + } else { + builder + }; let reader = builder .with_row_filter(RowFilter::new(vec![Box::new(predicate)])) .build() @@ -933,13 +1094,14 @@ fn read_parquet_filtered_u64_attempt( kind, observer, false, + projection, ); } } record_pruning(kind, &observation, pruning); observation.complete(returned, false); if batches.is_empty() { - return Ok(vec![RecordBatch::new_empty(fallback_schema)]); + return Ok(vec![RecordBatch::new_empty(projected_schema)]); } Ok(batches) } @@ -1176,6 +1338,62 @@ pub fn read_nodes_filtered_observed( Ok(batches) } +/// Filter destination nodes by `node_id` while physically decoding only the +/// requested canonical columns plus the join key. Legacy layouts retain full +/// normalization before projection. +#[allow(clippy::implicit_hasher)] +#[doc(hidden)] +pub fn read_nodes_filtered_projected_observed( + dir: &Path, + node_ids: &std::collections::HashSet, + projection: &[usize], + observer: Option<&std::sync::Arc>, +) -> Result, DataFusionError> { + let paths = crate::mutator::node_parquet_files(dir) + .map_err(|error| DataFusionError::Execution(error.to_string()))?; + let indices = canonical_projection_with_key(&TOPOLOGY_NODES_SCHEMA, projection, "node_id")?; + let mut batches = Vec::new(); + for path in paths { + let file_schema = admitted_parquet(&path)?.schema().clone(); + let canonical = file_schema.fields().len() == TOPOLOGY_NODES_SCHEMA.fields().len() + && file_schema + .fields() + .iter() + .zip(TOPOLOGY_NODES_SCHEMA.fields()) + .all(|(actual, expected)| actual.name() == expected.name()); + if canonical { + batches.extend(read_parquet_filtered_u64_projected( + &path, + TOPOLOGY_NODES_SCHEMA.clone(), + "node_id", + node_ids, + FilteredReadKind::Node, + observer, + &indices, + )?); + } else { + let normalized = normalize_topology_nodes(read_parquet_filtered_u64( + &path, + TOPOLOGY_NODES_SCHEMA.clone(), + "node_id", + node_ids, + FilteredReadKind::Node, + observer, + )?)?; + batches.extend( + normalized + .into_iter() + .map(|batch| batch.project(&indices).map_err(Into::into)) + .collect::, DataFusionError>>()?, + ); + } + } + if batches.is_empty() { + return project_batches_with_key(Vec::new(), &TOPOLOGY_NODES_SCHEMA, &indices, "node_id"); + } + Ok(batches) +} + /// Return the largest `edge_id` surrogate across every edge file under /// `topology/edges/` (both typed `.parquet` and `_exploratory.parquet`), /// or `0` if there are no edge files yet. @@ -1263,12 +1481,29 @@ fn read_property_overlay( dir: &Path, stem: &str, is_edge: bool, +) -> Result, DataFusionError> { + read_property_overlay_projected(dir, stem, is_edge, None) +} + +fn read_property_overlay_projected( + dir: &Path, + stem: &str, + is_edge: bool, + selected_properties: Option<&std::collections::BTreeSet>, ) -> Result, DataFusionError> { let mut batches = Vec::new(); - visit_property_overlay_batched(dir, stem, is_edge, 8_192, |batch| { - batches.push(batch.clone()); - Ok(true) - })?; + visit_property_overlay_batched_projected( + dir, + None, + stem, + is_edge, + 8_192, + selected_properties, + |batch| { + batches.push(batch.clone()); + Ok(true) + }, + )?; Ok(batches) } @@ -1282,7 +1517,7 @@ pub(crate) fn visit_property_overlay_batched( where F: FnMut(&RecordBatch) -> Result, { - visit_property_overlay_batched_with_inventory(dir, None, stem, is_edge, batch_size, visit) + visit_property_overlay_batched_projected(dir, None, stem, is_edge, batch_size, None, visit) } pub(crate) fn visit_property_overlay_batched_with_inventory( @@ -1291,6 +1526,22 @@ pub(crate) fn visit_property_overlay_batched_with_inventory( stem: &str, is_edge: bool, batch_size: usize, + visit: F, +) -> Result<(), DataFusionError> +where + F: FnMut(&RecordBatch) -> Result, +{ + visit_property_overlay_batched_projected(dir, inventory, stem, is_edge, batch_size, None, visit) +} + +#[allow(clippy::too_many_arguments)] +fn visit_property_overlay_batched_projected( + dir: &Path, + inventory: Option<&crate::AuthenticatedPropertyInventory>, + stem: &str, + is_edge: bool, + batch_size: usize, + selected_properties: Option<&std::collections::BTreeSet>, mut visit: F, ) -> Result<(), DataFusionError> where @@ -1318,11 +1569,12 @@ where let mut rows = Vec::with_capacity(batch_size.max(1)); let mut stopped = false; inventory - .visit_route( + .visit_route_projected( kind, stem, scratch.path(), crate::property_overlay::PropertyOverlayLimits::default(), + selected_properties, |row| { if stopped { return Ok(()); @@ -1338,6 +1590,8 @@ where graphforge_core::GfError::Storage("property batch disappeared".into()) })?; let batch = normalize_property_batch(batch, schema.as_ref())?; + let batch = + project_property_batch(batch, kind.uuid_field(), selected_properties)?; stopped = !visit(&batch) .map_err(|error| graphforge_core::GfError::Storage(error.to_string()))?; } @@ -1351,11 +1605,36 @@ where .ok_or_else(|| DataFusionError::Execution("property batch disappeared".into()))?; let batch = normalize_property_batch(batch, schema.as_ref()) .map_err(|error| DataFusionError::Execution(error.to_string()))?; + let batch = project_property_batch(batch, kind.uuid_field(), selected_properties) + .map_err(|error| DataFusionError::Execution(error.to_string()))?; let _ = visit(&batch)?; } Ok(()) } +fn project_property_batch( + batch: RecordBatch, + uuid_field: &str, + selected_properties: Option<&std::collections::BTreeSet>, +) -> Result { + let Some(selected_properties) = selected_properties else { + return Ok(batch); + }; + let indices = batch + .schema() + .fields() + .iter() + .enumerate() + .filter(|(_, field)| { + field.name() == uuid_field || selected_properties.contains(field.name()) + }) + .map(|(index, _)| index) + .collect::>(); + batch + .project(&indices) + .map_err(|error| graphforge_core::GfError::Storage(error.to_string())) +} + fn normalize_property_batch( batch: RecordBatch, schema: Option<&SchemaRef>, @@ -1883,6 +2162,18 @@ pub fn read_edge_properties(dir: &Path, stem: &str) -> Result, read_property_overlay(dir, stem, true) } +/// Read an authenticated newest-wins edge-property overlay while decoding +/// only the requested property names plus the mandatory edge UUID key. +#[doc(hidden)] +pub fn read_edge_properties_projected( + dir: &Path, + stem: &str, + property_names: &[String], +) -> Result, DataFusionError> { + let selected = property_names.iter().cloned().collect(); + read_property_overlay_projected(dir, stem, true, Some(&selected)) +} + /// Stems (relation names) of every `edge_properties/.parquet` under /// `dir`, **sorted** so schema unions built from them are deterministic /// (#1023). Empty when the directory is absent — a project with no persisted @@ -3295,6 +3586,30 @@ mod tests { assert_eq!(max_node_id(dir.path()).unwrap(), 2); } + #[test] + fn projected_node_reader_keeps_only_demand_and_join_key() { + let dir = TempDir::new().unwrap(); + write_nodes_parquet_value(&dir.path().join("topology/nodes.parquet"), 1, 1); + let ids = std::collections::HashSet::from([1]); + let batches = read_nodes_filtered_projected_observed( + dir.path(), + &ids, + &[TOPOLOGY_NODES_SCHEMA.index_of("node_uuid").unwrap()], + None, + ) + .unwrap(); + assert_eq!(total_rows(&batches), 1); + assert_eq!( + batches[0] + .schema() + .fields() + .iter() + .map(|field| field.name().as_str()) + .collect::>(), + ["node_uuid", "node_id"] + ); + } + #[test] fn legacy_scalar_node_labels_normalize_to_singleton_sets() { let dir = TempDir::new().unwrap(); @@ -3337,6 +3652,17 @@ mod tests { let values = labels.value(0); let values = values.as_any().downcast_ref::().unwrap(); assert_eq!(values.values(), &[7]); + + let projected = read_nodes_filtered_projected_observed( + dir.path(), + &std::collections::HashSet::from([1]), + &[TOPOLOGY_NODES_SCHEMA.index_of("type_ids").unwrap()], + None, + ) + .unwrap(); + assert_eq!(projected[0].num_columns(), 2); + assert!(projected[0].column_by_name("type_ids").is_some()); + assert!(projected[0].column_by_name("node_id").is_some()); } fn write_edge_parquet(path: &Path) { @@ -3838,6 +4164,56 @@ mod tests { edge_rel_pairs(&two), vec![(1, "KNOWS".to_owned()), (2, "OWNS".to_owned())] ); + + let projected = read_edges_filtered_projected_observed( + dir.path(), + "*", + OntologyMode::Strict, + &want, + &[EXPLORATORY_EDGE_SCHEMA.index_of("rel_type_name").unwrap()], + None, + ) + .unwrap(); + assert_eq!(total_rows(&projected), 1); + assert_eq!( + projected[0] + .schema() + .fields() + .iter() + .map(|field| field.name().as_str()) + .collect::>(), + ["edge_id", "rel_type_name"] + ); + } + + #[test] + fn projected_edge_read_rejects_reordered_physical_schema() { + let dir = TempDir::new().unwrap(); + let path = dir + .path() + .join("topology") + .join("edges") + .join("KNOWS.parquet"); + write_typed_edge(&path, 7, 1, 2); + let batches = read_parquet_or_empty(&path, TYPED_EDGE_SCHEMA.clone()).unwrap(); + let order = [3, 0, 1, 2, 4, 5, 6]; + let reordered = batches[0].project(&order).unwrap(); + let file = File::create(&path).unwrap(); + let mut writer = ArrowWriter::try_new(file, reordered.schema(), None).unwrap(); + writer.write(&reordered).unwrap(); + writer.close().unwrap(); + + let ids = [7].into_iter().collect(); + let error = read_edges_filtered_projected_observed( + dir.path(), + "KNOWS", + OntologyMode::Strict, + &ids, + &[TYPED_EDGE_SCHEMA.index_of("src_id").unwrap()], + None, + ) + .unwrap_err(); + assert!(error.to_string().contains("requires canonical schema")); } #[test] diff --git a/crates/graphforge-storage/src/lib.rs b/crates/graphforge-storage/src/lib.rs index e102fcd1c..6c9c01979 100644 --- a/crates/graphforge-storage/src/lib.rs +++ b/crates/graphforge-storage/src/lib.rs @@ -367,7 +367,7 @@ pub use ordinal_identity_v4::{ V4OrdinalFailureEvidence, V4OrdinalFailureKind, V4OrdinalIdentityDiscovery, V4OrdinalIdentityError, V4OrdinalIdentityLimits, V4OrdinalIdentityManifest, V4OrdinalIdentityOpen, V4OrdinalLookup, V4OrdinalLookupMetrics, V4OrdinalRange, - V4OrdinalTombstoneBlock, V4OrdinalTombstones, + V4OrdinalRevalidationMetrics, V4OrdinalTombstoneBlock, V4OrdinalTombstones, }; pub mod property_overlay; @@ -385,11 +385,12 @@ pub use catalog::{ AdmittedSourceFile, EdgePropertyTable, GraphCatalog, PropertyTable, TopologyNodeTable, TypedEdgeTable, UnionEdgeTable, list_edge_property_stems, list_property_stems, node_property_files, node_property_source_files, node_property_source_fragments, - node_topology_present, read_edge_properties, read_edges, read_edges_filtered, - read_edges_filtered_observed, read_nodes, read_nodes_filtered, read_nodes_filtered_observed, - read_properties, read_properties_batched, topology_node_files, visit_node_fragments_admitted, - visit_node_property_overlay_admitted, visit_nodes_batched, visit_properties_batched, - visit_property_fragments_admitted, + node_topology_present, read_edge_properties, read_edge_properties_projected, read_edges, + read_edges_filtered, read_edges_filtered_observed, read_edges_filtered_projected_observed, + read_nodes, read_nodes_filtered, read_nodes_filtered_observed, + read_nodes_filtered_projected_observed, read_properties, read_properties_batched, + topology_node_files, visit_node_fragments_admitted, visit_node_property_overlay_admitted, + visit_nodes_batched, visit_properties_batched, visit_property_fragments_admitted, }; pub mod runtime_entity_labels; diff --git a/crates/graphforge-storage/src/ordinal_identity_v4.rs b/crates/graphforge-storage/src/ordinal_identity_v4.rs index 0a253dfe5..45b0103a8 100644 --- a/crates/graphforge-storage/src/ordinal_identity_v4.rs +++ b/crates/graphforge-storage/src/ordinal_identity_v4.rs @@ -444,6 +444,21 @@ pub struct V4OrdinalLookupMetrics { pub transient_buffer_bytes: u64, /// Per-identity seeks are forbidden by contract. pub per_record_seeks: u64, + /// Generation-authentication file checks charged once to the pinned session. + pub revalidation_calls: u64, + /// Payload bytes read while revalidating the pinned session. Stamp and + /// identity checks read metadata only, so this is normally zero. + pub revalidation_bytes: u64, +} + +/// Aggregate-only work required to pin an already admitted v4 authority to one +/// execution session. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct V4OrdinalRevalidationMetrics { + /// Logical root/file identity and stamp checks performed. + pub calls: u64, + /// Artifact payload bytes read by those checks. + pub bytes_read: u64, } /// One caller-ordered lookup result and its sanitized evidence. @@ -764,6 +779,45 @@ impl V4OrdinalIdentityHandle { pub fn lookup_node_uuids( &mut self, requested: &[u64], + ) -> Result { + let revalidation = self.revalidate_for_session()?; + let mut lookup = self.lookup_node_uuids_pinned(requested)?; + lookup.metrics.revalidation_calls = revalidation.calls; + lookup.metrics.revalidation_bytes = revalidation.bytes_read; + Ok(lookup) + } + + /// Authenticate the retained authority once before sharing it with one + /// execution session. Subsequent lookups through that session use the + /// already-open immutable handles and must not repeat the artifact walk. + #[doc(hidden)] + pub fn revalidate_for_session( + &self, + ) -> Result { + self.revalidate()?; + let artifacts = self + .forward + .len() + .saturating_add(self.ranges.len()) + .saturating_add(self.tombstones.len()); + Ok(V4OrdinalRevalidationMetrics { + // Root identity, retained+named coordination, retained+named + // manifest, then retained+named checks for every artifact. + calls: 5_u64.saturating_add( + u64::try_from(artifacts) + .unwrap_or(u64::MAX) + .saturating_mul(2), + ), + bytes_read: 0, + }) + } + + /// Resolve through an authority already authenticated for the surrounding + /// execution session. + #[doc(hidden)] + pub fn lookup_node_uuids_pinned( + &mut self, + requested: &[u64], ) -> Result { if requested.len() > self.limits.max_requested { return Err(V4OrdinalIdentityError::RequestLimit { @@ -771,7 +825,6 @@ impl V4OrdinalIdentityHandle { maximum: self.limits.max_requested, }); } - self.revalidate()?; let mut metrics = V4OrdinalLookupMetrics { requested: requested.len() as u64, peak_buffer_bytes: (requested.len() as u64).saturating_mul(REQUEST_ENTRY_CHARGE), @@ -1847,6 +1900,27 @@ mod tests { assert!(!referenced.contains("planted-forward.uuidx")); } + #[test] + fn session_pin_authenticates_once_then_uses_retained_immutable_handles() { + let fixture = Fixture::new(&[4], &[]); + let mut handle = fixture.open(V4OrdinalIdentityLimits::default()); + let pin = handle.revalidate_for_session().unwrap(); + assert_eq!(pin.calls, 11); // root + coordination/manifest + three artifacts + assert_eq!(pin.bytes_read, 0); + + let manifest = fixture.root.path().join(INDEX_DIR).join(MANIFEST_NAME); + fs::write(&manifest, b"planted after the session pin").unwrap(); + let pinned = handle.lookup_node_uuids_pinned(&[1, 4]).unwrap(); + assert_eq!(pinned.metrics.revalidation_calls, 0); + assert_eq!(pinned.metrics.revalidation_bytes, 0); + assert!(pinned.values.iter().all(Option::is_some)); + + assert_eq!( + handle.lookup_node_uuids(&[1]).unwrap_err(), + V4OrdinalIdentityError::Authentication + ); + } + fn artifact( name: String, kind: V4OrdinalArtifactKind, diff --git a/crates/graphforge-storage/src/property_overlay.rs b/crates/graphforge-storage/src/property_overlay.rs index 6d81f564e..3f5fbc33a 100644 --- a/crates/graphforge-storage/src/property_overlay.rs +++ b/crates/graphforge-storage/src/property_overlay.rs @@ -60,7 +60,7 @@ impl PropertyRouteKind { } } - fn uuid_field(self) -> &'static str { + pub(crate) fn uuid_field(self) -> &'static str { match self { Self::Node => "node_uuid", Self::Edge => "edge_uuid", @@ -787,6 +787,23 @@ impl AuthenticatedPropertyInventory { limits: PropertyOverlayLimits, emit: F, ) -> Result + where + F: FnMut(PropertySnapshotRow) -> Result<(), GfError>, + { + self.visit_route_projected(kind, route, scratch, limits, None, emit) + } + + /// Visit one route while decoding only selected property columns plus the + /// UUID and tombstone keys required by newest-wins overlay semantics. + pub(crate) fn visit_route_projected( + &self, + kind: PropertyRouteKind, + route: &str, + scratch: &Path, + limits: PropertyOverlayLimits, + selected_properties: Option<&BTreeSet>, + emit: F, + ) -> Result where F: FnMut(PropertySnapshotRow) -> Result<(), GfError>, { @@ -799,45 +816,22 @@ impl AuthenticatedPropertyInventory { let authentication_read_calls = Arc::new(AtomicU64::new(0)); let budget = Arc::new(LiveByteBudget::new(limits.max_buffered_bytes)); let decoded = Arc::new(Mutex::new(DecodedRetention::default())); + let reader_context = ProjectedReaderContext { + inventory: self, + scratch, + limits, + kind, + route, + selected_properties, + counts: &counts, + budget: &budget, + decoded: &decoded, + authentication_bytes: &authentication_bytes, + authentication_block_equivalents: &authentication_block_equivalents, + authentication_read_calls: &authentication_read_calls, + }; let inputs = fragments.iter().map(|fragment| { - let reader = (|| { - let opened = self.open_fragment(fragment, scratch)?; - authentication_bytes.fetch_add(opened.authentication_bytes, Ordering::Relaxed); - authentication_block_equivalents - .fetch_add(opened.authentication_block_equivalents, Ordering::Relaxed); - authentication_read_calls - .fetch_add(opened.authentication_read_calls, Ordering::Relaxed); - let source = CountingChunkReader { - length: fragment.entry.byte_length, - file: Arc::clone(&opened.file), - counts: Arc::clone(&counts), - }; - let builder = - ParquetRecordBatchReaderBuilder::try_new(source).map_err(parquet_error)?; - validate_fragment_schema( - builder.schema().as_ref(), - fragment.id, - fragment.layout, - kind, - route, - )?; - let page_reservation_bytes = validate_parquet_resource_admission( - builder.metadata(), - limits, - opened.file.as_ref(), - &counts, - )?; - budget.charge(page_reservation_bytes)?; - { - let mut retention = decoded.lock().expect("property retention lock"); - retention.page_peak = retention.page_peak.max(page_reservation_bytes); - } - let reader = builder - .with_batch_size(admitted_batch_rows(limits)) - .build() - .map_err(parquet_error)?; - Ok((reader, page_reservation_bytes, opened.file, opened.handle)) - })(); + let reader = open_projected_fragment(fragment, &reader_context); let (reader, pending_error, page_reservation_bytes, handle) = match reader { Ok((reader, page_reservation_bytes, _file, handle)) => { (Some(reader), None, page_reservation_bytes, Some(handle)) @@ -868,48 +862,180 @@ impl AuthenticatedPropertyInventory { }); let mut metrics = visit_newest_property_snapshots(inputs, scratch, limits, budget.as_ref(), emit)?; - metrics.authentication_bytes = authentication_bytes.load(Ordering::Relaxed); - metrics.authentication_block_equivalents = - authentication_block_equivalents.load(Ordering::Relaxed); - metrics.authentication_read_calls = authentication_read_calls.load(Ordering::Relaxed); - metrics.property_authentication_bytes = metrics.authentication_bytes; - metrics.authenticated_snapshot_bytes = metrics.authentication_bytes; - metrics.authenticated_snapshot_peak_bytes = fragments - .iter() - .map(|fragment| fragment.entry.byte_length) - .max() - .unwrap_or(0); - metrics.property_authentication_block_equivalents = - metrics.authentication_block_equivalents; - metrics.property_authentication_read_calls = metrics.authentication_read_calls; - metrics.validation_bytes = counts.bytes.load(Ordering::Relaxed); - metrics.physical_bytes = metrics - .authentication_bytes - .saturating_add(metrics.validation_bytes); - metrics.read_calls = counts.blocks.load(Ordering::Relaxed); - metrics.validation_read_calls = metrics.read_calls; - metrics.physical_blocks = metrics - .authentication_read_calls - .saturating_add(metrics.read_calls); - metrics.range_seeks = counts.range_seeks.load(Ordering::Relaxed); - let decoded = decoded.lock().expect("property retention lock"); - metrics.decoder_peak_rows = decoded.peak_rows; - metrics.decoder_peak_bytes = decoded.peak_bytes; - metrics.decoder_page_reservation_bytes = decoded.page_peak; - metrics.emitted_batches = decoded.batches; - metrics.merge_peak_rows = metrics.peak_buffered_rows; - metrics.merge_peak_bytes = metrics.peak_buffered_bytes; - metrics.peak_buffered_rows = metrics - .decoder_peak_rows - .saturating_add(metrics.merge_peak_rows); - metrics.peak_buffered_bytes = metrics - .decoder_peak_bytes - .saturating_add(metrics.merge_peak_bytes); - metrics.peak_buffered_bytes = budget.peak(); + finalize_projected_metrics( + &mut metrics, + &ProjectedMetricSources { + counts: &counts, + authentication_bytes: &authentication_bytes, + authentication_block_equivalents: &authentication_block_equivalents, + authentication_read_calls: &authentication_read_calls, + decoded: &decoded, + budget: budget.as_ref(), + authenticated_snapshot_peak_bytes: fragments + .iter() + .map(|fragment| fragment.entry.byte_length) + .max() + .unwrap_or(0), + }, + ); Ok(metrics) } } +struct ProjectedReaderContext<'a> { + inventory: &'a AuthenticatedPropertyInventory, + scratch: &'a Path, + limits: PropertyOverlayLimits, + kind: PropertyRouteKind, + route: &'a str, + selected_properties: Option<&'a BTreeSet>, + counts: &'a Arc, + budget: &'a Arc, + decoded: &'a Arc>, + authentication_bytes: &'a Arc, + authentication_block_equivalents: &'a Arc, + authentication_read_calls: &'a Arc, +} + +fn open_projected_fragment( + fragment: &AuthenticatedPropertyFragment, + context: &ProjectedReaderContext<'_>, +) -> Result< + ( + ParquetRecordBatchReader, + u64, + Arc, + FragmentHandleGuard, + ), + GfError, +> { + let opened = context.inventory.open_fragment(fragment, context.scratch)?; + context + .authentication_bytes + .fetch_add(opened.authentication_bytes, Ordering::Relaxed); + context + .authentication_block_equivalents + .fetch_add(opened.authentication_block_equivalents, Ordering::Relaxed); + context + .authentication_read_calls + .fetch_add(opened.authentication_read_calls, Ordering::Relaxed); + let source = CountingChunkReader { + length: fragment.entry.byte_length, + file: Arc::clone(&opened.file), + counts: Arc::clone(context.counts), + }; + let builder = ParquetRecordBatchReaderBuilder::try_new(source).map_err(parquet_error)?; + validate_fragment_schema( + builder.schema().as_ref(), + fragment.id, + fragment.layout, + context.kind, + context.route, + )?; + let projected = projected_property_columns( + builder.schema().as_ref(), + context.kind, + context.selected_properties, + ); + let projection_mask = projected.as_ref().map(|roots| { + parquet::arrow::ProjectionMask::roots(builder.parquet_schema(), roots.iter().copied()) + }); + let projected_leaves = projection_mask.as_ref().map(|mask| { + (0..builder.parquet_schema().num_columns()) + .filter(|index| mask.leaf_included(*index)) + .collect::>() + }); + let page_reservation_bytes = validate_parquet_resource_admission( + builder.metadata(), + context.limits, + opened.file.as_ref(), + context.counts, + projected_leaves.as_ref(), + )?; + context.budget.charge(page_reservation_bytes)?; + { + let mut retention = context.decoded.lock().expect("property retention lock"); + retention.page_peak = retention.page_peak.max(page_reservation_bytes); + } + let builder = if let Some(mask) = projection_mask { + builder.with_projection(mask) + } else { + builder + }; + let reader = builder + .with_batch_size(admitted_batch_rows(context.limits)) + .build() + .map_err(parquet_error)?; + Ok((reader, page_reservation_bytes, opened.file, opened.handle)) +} + +struct ProjectedMetricSources<'a> { + counts: &'a ReadCounts, + authentication_bytes: &'a AtomicU64, + authentication_block_equivalents: &'a AtomicU64, + authentication_read_calls: &'a AtomicU64, + decoded: &'a Mutex, + budget: &'a LiveByteBudget, + authenticated_snapshot_peak_bytes: u64, +} + +fn projected_property_columns( + schema: &arrow::datatypes::Schema, + kind: PropertyRouteKind, + selected_properties: Option<&BTreeSet>, +) -> Option> { + selected_properties.map(|selected| { + schema + .fields() + .iter() + .enumerate() + .filter(|(_, field)| { + field.name() == kind.uuid_field() + || field.name() == PROPERTY_TOMBSTONE_FIELD + || selected.contains(field.name()) + }) + .map(|(index, _)| index) + .collect() + }) +} + +fn finalize_projected_metrics( + metrics: &mut PropertyOverlayMetrics, + sources: &ProjectedMetricSources<'_>, +) { + metrics.authentication_bytes = sources.authentication_bytes.load(Ordering::Relaxed); + metrics.authentication_block_equivalents = sources + .authentication_block_equivalents + .load(Ordering::Relaxed); + metrics.authentication_read_calls = sources.authentication_read_calls.load(Ordering::Relaxed); + metrics.property_authentication_bytes = metrics.authentication_bytes; + metrics.authenticated_snapshot_bytes = metrics.authentication_bytes; + metrics.authenticated_snapshot_peak_bytes = sources.authenticated_snapshot_peak_bytes; + metrics.property_authentication_block_equivalents = metrics.authentication_block_equivalents; + metrics.property_authentication_read_calls = metrics.authentication_read_calls; + metrics.validation_bytes = sources.counts.bytes.load(Ordering::Relaxed); + metrics.physical_bytes = metrics + .authentication_bytes + .saturating_add(metrics.validation_bytes); + metrics.read_calls = sources.counts.blocks.load(Ordering::Relaxed); + metrics.validation_read_calls = metrics.read_calls; + metrics.physical_blocks = metrics + .authentication_read_calls + .saturating_add(metrics.read_calls); + metrics.range_seeks = sources.counts.range_seeks.load(Ordering::Relaxed); + let decoded = sources.decoded.lock().expect("property retention lock"); + metrics.decoder_peak_rows = decoded.peak_rows; + metrics.decoder_peak_bytes = decoded.peak_bytes; + metrics.decoder_page_reservation_bytes = decoded.page_peak; + metrics.emitted_batches = decoded.batches; + metrics.merge_peak_rows = metrics.peak_buffered_rows; + metrics.merge_peak_bytes = metrics.peak_buffered_bytes; + metrics.peak_buffered_rows = metrics + .decoder_peak_rows + .saturating_add(metrics.merge_peak_rows); + metrics.peak_buffered_bytes = sources.budget.peak(); +} + fn decode_sha256(value: &str) -> Result<[u8; 32], GfError> { if value.len() != 64 { return Err(corrupt( @@ -2036,6 +2162,7 @@ pub fn read_authenticated_property_snapshots_for_inventory( PropertyOverlayLimits::default(), opened.file.as_ref(), &counts, + None, )?; metrics.decoder_page_reservation_bytes = metrics .decoder_page_reservation_bytes @@ -2346,6 +2473,7 @@ fn validate_parquet_resource_admission( limits: PropertyOverlayLimits, file: &File, counts: &Arc, + projected_columns: Option<&BTreeSet>, ) -> Result { const MAX_PAGE_HEADER_BYTES: usize = 64 * 1024; let max_page_bytes = limits.max_buffered_bytes / 4; @@ -2355,7 +2483,8 @@ fn validate_parquet_resource_admission( let mut largest_group_exposure = 0_u64; for group in metadata.row_groups() { let mut group_exposure = 0_u64; - for column in group.columns() { + for (column_index, column) in group.columns().iter().enumerate() { + let selected = projected_columns.is_none_or(|columns| columns.contains(&column_index)); let mut dictionary_exposure = 0_u64; let mut data_exposure = 0_u64; let _uncompressed = u64::try_from(column.uncompressed_size()) @@ -2400,15 +2529,17 @@ fn validate_parquet_resource_admission( let uncompressed_page = u64::try_from(header.uncompressed_page_size) .map_err(|_| corrupt("property page has negative uncompressed size"))?; if header_bytes == 0 - || uncompressed_page > max_page_bytes + || (selected && uncompressed_page > max_page_bytes) || compressed_page > compressed { return Err(corrupt("property page exceeds pre-decode byte admission")); } - if header.type_ == parquet::format::PageType::DICTIONARY_PAGE { - dictionary_exposure = dictionary_exposure.max(uncompressed_page); - } else { - data_exposure = data_exposure.max(uncompressed_page); + if selected { + if header.type_ == parquet::format::PageType::DICTIONARY_PAGE { + dictionary_exposure = dictionary_exposure.max(uncompressed_page); + } else { + data_exposure = data_exposure.max(uncompressed_page); + } } position = position .checked_add(header_bytes) @@ -2423,23 +2554,25 @@ fn validate_parquet_resource_admission( "property page sequence does not cover its column chunk", )); } - group_exposure = group_exposure - .checked_add(data_exposure) - .and_then(|bytes| { - dictionary_exposure - .checked_mul(u64::try_from(admitted_batch_rows(limits)).ok()?) - .and_then(|decoded_dictionary| bytes.checked_add(decoded_dictionary)) - }) - .and_then(|bytes| { - // Validity, offsets, and values buffers are live together. - // Sixteen bytes/value/column deliberately over-reserves the - // fixed Arrow bookkeeping before the builder allocates it. - u64::try_from(admitted_batch_rows(limits)) - .ok()? - .checked_mul(16) - .and_then(|overhead| bytes.checked_add(overhead)) - }) - .ok_or_else(|| corrupt("property projected page exposure overflows"))?; + if selected { + group_exposure = group_exposure + .checked_add(data_exposure) + .and_then(|bytes| { + dictionary_exposure + .checked_mul(u64::try_from(admitted_batch_rows(limits)).ok()?) + .and_then(|decoded_dictionary| bytes.checked_add(decoded_dictionary)) + }) + .and_then(|bytes| { + // Validity, offsets, and values buffers are live together. + // Sixteen bytes/value/column deliberately over-reserves the + // fixed Arrow bookkeeping before the builder allocates it. + u64::try_from(admitted_batch_rows(limits)) + .ok()? + .checked_mul(16) + .and_then(|overhead| bytes.checked_add(overhead)) + }) + .ok_or_else(|| corrupt("property projected page exposure overflows"))?; + } if group_exposure > max_page_bytes { return Err(corrupt( "property projected pages exceed pre-decode live-byte admission", @@ -4116,6 +4249,103 @@ mod tests { assert!(error.to_string().contains("GF_PROJECT_CORRUPT"), "{error}"); } + #[test] + fn projected_overlay_decodes_only_selected_values_and_mandatory_keys() { + let root = TempDir::new().unwrap(); + let scratch = TempDir::new().unwrap(); + let id = PropertyFragmentId { + generation: 1, + ordinal: 0, + }; + let route_dir = root.path().join("edge_properties/KNOWS"); + fs::create_dir_all(&route_dir).unwrap(); + let schema = Arc::new(Schema::new_with_metadata( + vec![ + Field::new("edge_uuid", DataType::FixedSizeBinary(16), false), + Field::new(PROPERTY_TOMBSTONE_FIELD, DataType::Boolean, false), + Field::new("keep", DataType::Utf8, true), + Field::new("unused", DataType::Utf8, true), + ], + HashMap::from([ + ( + PROPERTY_OVERLAY_FORMAT_KEY.into(), + PROPERTY_OVERLAY_FORMAT.into(), + ), + (PROPERTY_ROUTE_KEY.into(), "KNOWS".into()), + (PROPERTY_KIND_KEY.into(), "edge".into()), + (PROPERTY_GENERATION_KEY.into(), "1".into()), + (PROPERTY_ORDINAL_KEY.into(), "0".into()), + ]), + )); + let rows = 128; + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new( + FixedSizeBinaryArray::try_from_iter( + (0..rows).map(|row| vec![u8::try_from(row + 1).unwrap(); 16]), + ) + .unwrap(), + ), + Arc::new(BooleanArray::from(vec![false; rows])), + Arc::new(StringArray::from(vec![Some("kept"); rows])), + Arc::new(StringArray::from(vec![Some("x".repeat(8_192)); rows])), + ], + ) + .unwrap(); + let path = route_dir.join(id.file_name()); + let mut writer = ArrowWriter::try_new(File::create(&path).unwrap(), schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let bytes = fs::read(&path).unwrap(); + let inventory = AuthenticatedPropertyInventory::from_entries_at_root( + root.path(), + vec![crate::GraphFileEntry { + relative_path: format!("edge_properties/KNOWS/{}", id.file_name()), + byte_length: u64::try_from(bytes.len()).unwrap(), + content_sha256: digest_hex(&Sha256::digest(&bytes)), + role: crate::GraphFileRole::Properties, + }], + ) + .unwrap(); + + let mut full_rows = Vec::new(); + let full = inventory + .visit_route( + PropertyRouteKind::Edge, + "KNOWS", + scratch.path(), + PropertyOverlayLimits::default(), + |row| { + full_rows.push(row); + Ok(()) + }, + ) + .unwrap(); + let mut projected_rows = Vec::new(); + let projected = inventory + .visit_route_projected( + PropertyRouteKind::Edge, + "KNOWS", + scratch.path(), + PropertyOverlayLimits::default(), + Some(&BTreeSet::from(["keep".to_owned()])), + |row| { + projected_rows.push(row); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(projected_rows.len(), full_rows.len()); + assert!( + projected_rows.iter().all(|row| { + row.values.contains_key("keep") && !row.values.contains_key("unused") + }) + ); + assert!(projected.validation_bytes < full.validation_bytes); + assert_eq!(projected.per_record_seeks, 0); + } + #[test] fn hostile_authenticated_property_matrix_fails_closed_before_projection_or_limit() { fn write_fragment( diff --git a/docs/book/architecture/execution-model.md b/docs/book/architecture/execution-model.md index 6fcede5d0..ffc929220 100644 --- a/docs/book/architecture/execution-model.md +++ b/docs/book/architecture/execution-model.md @@ -113,6 +113,31 @@ execution paths consume it through a single `AdjacencyProvider` abstraction: | `VarLenExpandExec` | Iterative BFS over the `AdjacencyProvider` for `*min..max` patterns (replaces the per-query in-memory adjacency build) | | `ExpandExec` | Adjacency-backed single-hop expansion: chosen at lowering time when the provider reports a `hit` for a typed relation (any direction; undirected wraps in `DISTINCT`, mirroring the join path's union+distinct). Exploratory single-hop and uncovered patterns keep the DataFusion join chain. | +`ExpandExec` receives exact physical column demand through projections, filters, +sorts, and limits; unknown or multi-input physical operators are conservative +materialization barriers. A destination-identity-only hop carries the CSR +neighbor `node_id` directly and resolves `node_uuid` through the facade's exact +generation-pinned v4 ordinal authority. It opens neither edge topology nor +destination-node Parquet. Lookup batches are sorted and deduplicated within the +operator batch, coalesced by the v4 reader, and restored to adjacency order. +There is no graph-sized identity map and no path-based rediscovery of mutable +authority. The facade authenticates and pins the selected immutable generation +once when it creates an execution session. Every hop in that session then reads +through the retained authenticated file handles; it does not repeat a complete +artifact-name/stamp walk for every bounded expansion chunk. A later session +revalidates the names and identities again, so a planted replacement cannot be +adopted as authority. + +When relationship or destination-node data is required, filtered readers decode +only the demanded canonical columns plus their join key. Legacy node layouts and +typed wildcard unions retain their normalization path before result shaping. +Aggregate diagnostics report projected columns/chunks/rows and v4 ranges, +coalesced calls, bytes, peak charged buffers, and forbidden per-record seeks; +session revalidation calls/bytes are accounted exactly once rather than hidden +from lookup evidence. Diagnostics never report identities or paths. Global `ORDER BY ... LIMIT` still examines +the complete unordered candidate stream and does not use invalid early +cancellation. + **Selection is a planner choice, not an IR change.** The Graph IR is unchanged: variable-length traversal is still encoded on `Expand { …, min_hops, max_hops }`. A lowering rule selects an adjacency-backed physical node when the provider covers the relation type + direction, and falls diff --git a/docs/development/perf-g500-ladder.md b/docs/development/perf-g500-ladder.md index 49e1d66d9..bd191997c 100644 --- a/docs/development/perf-g500-ladder.md +++ b/docs/development/perf-g500-ladder.md @@ -143,6 +143,19 @@ checkpoints, from the authenticated receipt chain. Deterministic 1x/2x/4x tests require exact chunk counts and bounded peak windows while aggregate merge work remains linear in accepted rows. +Query-phase ordered-LIMIT evidence must distinguish complete candidate +examination from materialization amplification. For the canonical fixed-hop +destination-UUID query, record expansion chunks/candidates/projected columns, +edge and node Parquet calls/rows, and v4 ordinal ranges/coalesced calls/bytes/ +peak charged buffer/per-record seeks plus session revalidation calls/bytes. +Generation authentication is charged once per execution session and must scale +linearly with retained immutable artifacts, never expansion chunks multiplied +by artifact count. Edge and node Parquet calls are exactly +zero, per-record seeks are exactly zero, ordered source and imported results +are byte-identical, and 1x/2x/4x ordinal work remains a linear constant factor +of candidates examined. A bounded RSS result does not pass if logical reads +grow as expansion chunks multiplied by graph size. + ## Commands Always-on CI (SCALE-10 smoke + all reconciliation / determinism / bounded /