diff --git a/crates/graphforge-api/src/composite_publish.rs b/crates/graphforge-api/src/composite_publish.rs index 66ab86e59..69ff5f698 100644 --- a/crates/graphforge-api/src/composite_publish.rs +++ b/crates/graphforge-api/src/composite_publish.rs @@ -1011,99 +1011,7 @@ fn require_capabilities( Ok(()) } -/// Cumulative probe I/O and serial decoder peaks. The identity containers are -/// reported separately: decoder counters are not a bound on process memory. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -struct EdgeOwnerProbeWork { - candidate_routes: usize, - target_memberships: usize, - resolved_targets: usize, - route_name_bytes: usize, - physical_bytes: u64, - authentication_bytes: u64, - authenticated_snapshot_bytes: u64, - authenticated_snapshot_peak_bytes: u64, - physical_rows: u64, - physical_blocks: u64, - fragments_considered: u64, - row_groups_considered: u64, - row_groups_selected: u64, - decoder_peak_bytes: u64, - decoder_page_reservation_bytes: u64, -} - -/// Construction and ordinary mutation have different exploratory property -/// owners. Probe only the candidate routes for requested identities, once per -/// route, using authenticated row presence (including a newest tombstone). -fn resolve_existing_edge_property_owners( - inventory: &graphforge_storage::AuthenticatedPropertyInventory, - owners: &mut BTreeMap, -) -> Result { - use graphforge_storage::PropertyRouteKind; - let mut candidates = BTreeMap::>::new(); - for (uuid, route) in owners.iter() { - for candidate in [route.as_str(), "_exploratory"] { - if inventory - .route_schema(PropertyRouteKind::Edge, candidate) - .is_some() - { - if let Some(targets) = candidates.get_mut(candidate) { - targets.insert(uuid.into_bytes()); - } else { - candidates.insert(candidate.to_owned(), BTreeSet::from([uuid.into_bytes()])); - } - } - } - } - let mut resolved = BTreeMap::new(); - let mut total = EdgeOwnerProbeWork { - candidate_routes: candidates.len(), - target_memberships: candidates.values().map(BTreeSet::len).sum(), - route_name_bytes: candidates.keys().map(String::len).sum(), - ..EdgeOwnerProbeWork::default() - }; - for (route, targets) in &candidates { - let (present, work) = - graphforge_storage::read_authenticated_property_presence_for_inventory( - inventory, - PropertyRouteKind::Edge, - route, - targets, - )?; - total.physical_bytes += work.physical_bytes; - total.authentication_bytes += work.authentication_bytes; - total.authenticated_snapshot_bytes += work.authenticated_snapshot_bytes; - total.authenticated_snapshot_peak_bytes = total - .authenticated_snapshot_peak_bytes - .max(work.authenticated_snapshot_peak_bytes); - total.physical_rows += work.physical_rows; - total.physical_blocks += work.physical_blocks; - total.fragments_considered += work.fragments_considered; - total.row_groups_considered += work.row_groups_considered; - total.row_groups_selected += work.row_groups_selected; - total.decoder_peak_bytes = total.decoder_peak_bytes.max(work.decoder_peak_bytes); - total.decoder_page_reservation_bytes = total - .decoder_page_reservation_bytes - .max(work.decoder_page_reservation_bytes); - for uuid in present { - if resolved - .insert(Uuid::from_bytes(uuid), route.as_str()) - .is_some() - { - return Err(GfError::Storage( - "composite edge property owner is ambiguous".into(), - )); - } - } - } - // No property row means ordinary writer ownership, already derived from - // topology. Same-request creations are added by the caller afterwards. - total.resolved_targets = resolved.len(); - for (uuid, route) in resolved { - owners.insert(uuid, route.to_owned()); - } - Ok(total) -} +use graphforge_storage::resolve_existing_edge_property_owners; #[allow(clippy::too_many_lines)] fn build_validation_snapshot( diff --git a/crates/graphforge-api/tests/permanent_storage_budgets.rs b/crates/graphforge-api/tests/permanent_storage_budgets.rs index 6db668dcf..dadb1f305 100644 --- a/crates/graphforge-api/tests/permanent_storage_budgets.rs +++ b/crates/graphforge-api/tests/permanent_storage_budgets.rs @@ -5986,3 +5986,383 @@ fn selective_fixed_paths_preserve_public_mutation_and_portable_lifecycle() { verify_selective_fixed_paths(&reopened, &nodes, &edges); verify_graph(&reopened, fixture, &nodes, &edges); } + +#[test] +fn constructed_edge_cypher_set_has_one_authenticated_owner() { + let root = tempfile::tempdir().unwrap(); + let source = root.path().join("source"); + let fixture = Fixture { + name: "constructed_edge_cypher_owner", + nodes: 33, + edges: 129, + routes: 2, + identifiers: Identifiers::Random, + properties: true, + adjacency: true, + heterogeneous: false, + }; + let (nodes, mut edges) = rows(fixture); + construct(&source, fixture, &nodes, &edges); + let graph = GraphForge::new(source.to_str()).unwrap(); + verify_graph(&graph, fixture, &nodes, &edges); + graph + .execute("MATCH ()-[r]->() WHERE r.weight IS NOT NULL SET r.weight = r.weight + 1") + .unwrap(); + for edge in &mut edges { + edge.4 = edge.4.map(|weight| weight + 1); + } + verify_graph(&graph, fixture, &nodes, &edges); +} + +#[test] +fn constructed_edge_cypher_maps_and_removals_preserve_portable_lifecycle() { + use futures::TryStreamExt as _; + for node_count in [33, 4097] { + let root = tempfile::tempdir().unwrap(); + let source = root.path().join("source"); + let fixture = Fixture { + name: "constructed_edge_cypher_lifecycle", + nodes: node_count, + edges: 129, + routes: 2, + identifiers: Identifiers::Random, + properties: true, + adjacency: true, + heterogeneous: false, + }; + let (mut nodes, mut edges) = rows(fixture); + construct(&source, fixture, &nodes, &edges); + let graph = GraphForge::new(source.to_str()).unwrap(); + verify_graph(&graph, fixture, &nodes, &edges); + let created = graph.execute("CREATE (a:NodeExtra)-[r:REL0 {weight: 3, text: 'ordinary'}]->(b:NodeExtra), (a)-[s:REL1]->(b) RETURN a.node_uuid, b.node_uuid, r.edge_uuid, s.edge_uuid").unwrap(); + assert_eq!( + created + .batches + .iter() + .map(RecordBatch::num_rows) + .sum::(), + 1 + ); + let created = &created.batches[0]; + let a = uuid_at(created, 0, 0); + let b = uuid_at(created, 1, 0); + nodes.extend([(a, "NodeExtra".into(), None), (b, "NodeExtra".into(), None)]); + edges.extend([ + ( + uuid_at(created, 2, 0), + a, + b, + "REL0".into(), + Some(3), + Some("ordinary".into()), + ), + (uuid_at(created, 3, 0), a, b, "REL1".into(), None, None), + ]); + verify_graph(&graph, fixture, &nodes, &edges); + let old_query = "MATCH ()-[r]->() RETURN r.edge_uuid, r.weight"; + let snapshot = graph.execute_stream(old_query).unwrap(); + let mut old_expected: Vec<_> = edges.iter().map(|edge| (edge.0, edge.4)).collect(); + old_expected.sort(); + graph + .execute("MATCH ()-[r]->() WHERE r.weight IS NOT NULL SET r.weight = r.weight + 1") + .unwrap(); + for edge in &mut edges { + edge.4 = edge.4.map(|weight| weight + 1); + } + verify_graph(&graph, fixture, &nodes, &edges); + graph + .execute("MATCH ()-[r]->() WHERE r.weight IS NOT NULL SET r += {weight: r.weight + 2}") + .unwrap(); + for edge in &mut edges { + edge.4 = edge.4.map(|weight| weight + 2); + } + verify_graph(&graph, fixture, &nodes, &edges); + graph + .execute("MATCH ()-[r:REL0]->() SET r.text = null") + .unwrap(); + for edge in edges.iter_mut().filter(|edge| edge.3 == "REL0") { + edge.5 = None; + } + verify_graph(&graph, fixture, &nodes, &edges); + graph + .execute("MATCH ()-[r:REL1]->() REMOVE r.weight") + .unwrap(); + for edge in edges.iter_mut().filter(|edge| edge.3 == "REL1") { + edge.4 = None; + } + verify_graph(&graph, fixture, &nodes, &edges); + graph + .execute("MATCH ()-[r:REL0]->() SET r = {weight: 7, text: 'replacement'}") + .unwrap(); + for edge in edges.iter_mut().filter(|edge| edge.3 == "REL0") { + edge.4 = Some(7); + edge.5 = Some("replacement".into()); + } + verify_graph(&graph, fixture, &nodes, &edges); + graph.execute("MATCH ()-[r:REL1]->() SET r = {}").unwrap(); + for edge in edges.iter_mut().filter(|edge| edge.3 == "REL1") { + edge.4 = None; + edge.5 = None; + } + verify_graph(&graph, fixture, &nodes, &edges); + graph + .execute("MATCH ()-[r:REL1]->() SET r.weight = 11") + .unwrap(); + for edge in edges.iter_mut().filter(|edge| edge.3 == "REL1") { + edge.4 = Some(11); + } + verify_graph(&graph, fixture, &nodes, &edges); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let old: Vec = runtime.block_on(snapshot.try_collect()).unwrap(); + let mut actual = Vec::new(); + for batch in old { + for row in 0..batch.num_rows() { + actual.push((uuid_at(&batch, 0, row), int_at(&batch, 1, row))); + } + } + actual.sort(); + assert_eq!(actual, old_expected); + drop(graph); + round_trip(root.path(), &source, fixture, &nodes, &edges); + let imported_path = root.path().join("imported"); + let imported = GraphForge::new(imported_path.to_str()).unwrap(); + imported + .execute("MATCH ()-[r:REL0]->() REMOVE r.text") + .unwrap(); + for edge in edges.iter_mut().filter(|edge| edge.3 == "REL0") { + edge.5 = None; + } + imported + .execute("MATCH ()-[r:REL1]->() SET r += {weight: 13}") + .unwrap(); + for edge in edges.iter_mut().filter(|edge| edge.3 == "REL1") { + edge.4 = Some(13); + } + verify_graph(&imported, fixture, &nodes, &edges); + drop(imported); + let reopened = GraphForge::new(imported_path.to_str()).unwrap(); + verify_graph(&reopened, fixture, &nodes, &edges); + } +} + +#[test] +fn cypher_edge_owner_work_accounts_for_multiple_batches_and_items() { + let root = tempfile::tempdir().unwrap(); + let source = root.path().join("source"); + let fixture = Fixture { + name: "cypher_owner_statement_work", + nodes: 33, + edges: 8193, + routes: 2, + identifiers: Identifiers::Random, + properties: true, + adjacency: true, + heterogeneous: false, + }; + let (nodes, mut edges) = rows(fixture); + construct(&source, fixture, &nodes, &edges); + let graph = GraphForge::new(source.to_str()).unwrap(); + let (result, work) = graphforge_exec::demand::capture(|| { + graph.execute("MATCH ()-[r]->() SET r.weight = 1, r.text = 'changed'") + }); + result.unwrap(); + assert_eq!( + work.property_writes["owner_targets"], + 2 * fixture.edges as u64 + ); + assert!( + work.property_writes["owner_batches"] >= 4, + "{:?}", + work.property_writes + ); + assert!(work.property_writes["owner_target_peak"] <= 8192); + for edge in &mut edges { + edge.4 = Some(1); + edge.5 = Some("changed".into()); + } + verify_graph(&graph, fixture, &nodes, &edges); + let (result, replacement) = + graphforge_exec::demand::capture(|| graph.execute("MATCH ()-[r]->() SET r = {weight: 2}")); + result.unwrap(); + assert_eq!( + replacement.property_writes["owner_targets"], + fixture.edges as u64 + ); + assert_eq!( + replacement.property_writes["replacement_targets"], + fixture.edges as u64 + ); + for edge in &mut edges { + edge.4 = Some(2); + edge.5 = None; + } + verify_graph(&graph, fixture, &nodes, &edges); + // Measured fixed-fixture envelopes cover every batch and item, including + // repeated authenticated reads. They bound this resolver's work, not the + // entire mutation or process RSS. Replacement probes and key reads are + // both charged, even when they consume the same physical fragments. + let count = fixture.edges as u64; + assert!(work.property_writes["owner_rows"] <= 7 * count); + assert!(work.property_writes["owner_physical_bytes"] <= 320 * count); + assert!(work.property_writes["owner_authentication_bytes"] <= 128 * count); + for prefix in ["owner", "replacement"] { + assert!(replacement.property_writes[&format!("{prefix}_rows")] <= 6 * count); + assert!(replacement.property_writes[&format!("{prefix}_physical_bytes")] <= 256 * count); + assert!( + replacement.property_writes[&format!("{prefix}_authentication_bytes")] <= 128 * count + ); + assert!( + replacement.property_writes[&format!("{prefix}_snapshot_peak_bytes")] <= 256 * 1024 + ); + assert!(replacement.property_writes[&format!("{prefix}_decoder_peak_bytes")] <= 4096); + } + println!( + "CYPHER_OWNER_WORK {}", + json!({"scalar":work.property_writes,"replacement":replacement.property_writes}) + ); +} + +#[test] +fn qualified_cypher_edge_owners_preserve_constructed_and_pending_values() { + let root = tempfile::tempdir().unwrap(); + let source = root.path().join("source"); + let fixture = Fixture { + name: "qualified_cypher_owners", + nodes: 33, + edges: 129, + routes: 1, + identifiers: Identifiers::Random, + properties: false, + adjacency: false, + heterogeneous: false, + }; + let (base_nodes, base_edges) = rows(fixture); + construct(&source, fixture, &base_nodes, &base_edges); + configure_composite_ontology(root.path(), &source, true); + let a = id(3, 0, true); + let b = id(3, 1, true); + let edge = id(4, 0, true); + let nodes = vec![(a, "NewNode".into(), None), (b, "NewNode".into(), None)]; + let edges = vec![(edge, a, b, "NEW_TYPED".into(), None, None)]; + construct(&source, fixture, &nodes, &edges); + let graph = GraphForge::new(source.to_str()).unwrap(); + let query = "MATCH (a:`mixed:NewNode`)-[r:`mixed:NEW_TYPED`]->(b:`mixed:NewNode`) RETURN r.edge_uuid, a.node_uuid, b.node_uuid, r.weight"; + let mut expected = vec![(edge, a, b, None)]; + let verify = |graph: &GraphForge, expected: &[(Uuid, Uuid, Uuid, Option)]| { + for suffix in [ + "RETURN type(r) AS kind, r AS relationship", + "WITH r AS x RETURN type(x) AS kind, x AS relationship", + ] { + let values = graph + .execute(&format!("MATCH ()-[r:`mixed:NEW_TYPED`]->() {suffix}")) + .unwrap_or_else(|error| panic!("{suffix}: {error}")); + assert_eq!( + values + .batches + .iter() + .map(RecordBatch::num_rows) + .sum::(), + expected.len() + ); + for batch in &values.batches { + let kinds = batch + .column_by_name("kind") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let relationships = batch + .column_by_name("relationship") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let types = relationships + .column_by_name("rel_type") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + assert_eq!(kinds.value(row), "mixed:NEW_TYPED", "{suffix}"); + assert_eq!(types.value(row), "mixed:NEW_TYPED", "{suffix}"); + } + } + } + let mut actual = Vec::new(); + for batch in graph.execute(query).unwrap().batches { + for row in 0..batch.num_rows() { + actual.push(( + uuid_at(&batch, 0, row), + uuid_at(&batch, 1, row), + uuid_at(&batch, 2, row), + int_at(&batch, 3, row), + )); + } + } + actual.sort(); + let mut expected = expected.to_vec(); + expected.sort(); + assert_eq!(actual, expected); + digest_hex(&serde_json::to_vec(&actual).unwrap()) + }; + verify(&graph, &expected); + drop(graph); + let graph = GraphForge::new(source.to_str()).unwrap(); + graph + .execute("MATCH ()-[r:`mixed:NEW_TYPED`]->() SET r.weight = 7") + .unwrap(); + expected[0].3 = Some(7); + verify(&graph, &expected); + drop(graph); + let graph = GraphForge::new(source.to_str()).unwrap(); + graph + .execute("MATCH ()-[r:`mixed:NEW_TYPED`]->() SET r = {weight: 8}") + .unwrap(); + expected[0].3 = Some(8); + verify(&graph, &expected); + drop(graph); + let graph = GraphForge::new(source.to_str()).unwrap(); + graph + .execute("MATCH ()-[r:`mixed:NEW_TYPED`]->() REMOVE r.weight") + .unwrap(); + expected[0].3 = None; + verify(&graph, &expected); + drop(graph); + let graph = GraphForge::new(source.to_str()).unwrap(); + let created = graph.execute("CREATE (a:`mixed:NewNode` {score: 1})-[r:`mixed:NEW_TYPED`]->(b:`mixed:NewNode` {score: 2}) SET r.weight = 9 REMOVE r.weight SET r += {weight: 10} RETURN r.edge_uuid, a.node_uuid, b.node_uuid").unwrap(); + assert_eq!( + created + .batches + .iter() + .map(RecordBatch::num_rows) + .sum::(), + 1 + ); + let batch = &created.batches[0]; + expected.push(( + uuid_at(batch, 0, 0), + uuid_at(batch, 1, 0), + uuid_at(batch, 2, 0), + Some(10), + )); + verify(&graph, &expected); + drop(graph); + let graph = GraphForge::new(source.to_str()).unwrap(); + drop(graph); + round_trip_checked(root.path(), &source, |graph| verify(graph, &expected)); + let imported_path = root.path().join("imported"); + let imported = GraphForge::new(imported_path.to_str()).unwrap(); + imported + .execute("MATCH ()-[r:`mixed:NEW_TYPED`]->() SET r += {weight: 11}") + .unwrap(); + for edge in &mut expected { + edge.3 = Some(11); + } + verify(&imported, &expected); + drop(imported); + verify(&GraphForge::new(imported_path.to_str()).unwrap(), &expected); +} diff --git a/crates/graphforge-exec/src/demand.rs b/crates/graphforge-exec/src/demand.rs index b72137f91..007b48a09 100644 --- a/crates/graphforge-exec/src/demand.rs +++ b/crates/graphforge-exec/src/demand.rs @@ -190,6 +190,9 @@ pub struct DemandSnapshot { pub max_in_flight_reads: u64, /// Query-owned path hydration work, using physical operator metric names. pub hydration: BTreeMap, + /// Statement-wide owner-resolution and replacement-key work. Decoder peaks + /// and identity counts are logical accounting, not process-memory bounds. + pub property_writes: BTreeMap, /// Query memory-pool reservation before physical execution. pub memory_reserved_before: u64, /// Query memory-pool reservation after every operator stream was dropped. @@ -442,6 +445,27 @@ pub(crate) fn record_plan_completion( state.snapshot.execution_batch_rows = execution_batch_rows as u64; } +pub(crate) fn record_property_write_work(sums: &[(&str, u64)], peaks: &[(&str, u64)]) { + let epoch = bound_capture_session(); + if epoch == 0 || !capture_enabled() { + return; + } + let mut state = CAPTURE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !session_active_locked(&state, epoch) { + return; + } + let work = &mut state.snapshot.property_writes; + for (name, value) in sums { + *work.entry((*name).to_owned()).or_default() += value; + } + for (name, value) in peaks { + let current = work.entry((*name).to_owned()).or_default(); + *current = (*current).max(*value); + } +} + fn with_hop(epoch: u64, edge_var: u32, update: impl FnOnce(&mut HopSnapshot)) { let mut state = CAPTURE .lock() @@ -1614,6 +1638,65 @@ fn hydration_metrics(plan: &Arc, result: &mut BTreeMap bool { + self.nodes.is_empty() && self.edges.is_empty() + } + fn record( &mut self, is_edge: bool, @@ -1797,13 +1801,28 @@ impl SetAccumulator { &self, staged: &mut graphforge_storage::RewriteBatch, dir: &Path, + inventory: Option<&graphforge_storage::AuthenticatedPropertyInventory>, ) -> Result { + if self.is_empty() { + return Ok(0); + } + let captured; + let inventory = if let Some(inventory) = inventory { + inventory + } else { + captured = graphforge_storage::AuthenticatedPropertyInventory::capture(dir)?; + &captured + }; let mut total = 0u64; for (stem, updates) in &self.nodes { - total += graphforge_storage::stage_set_node_properties(staged, dir, stem, updates)?; + total += graphforge_storage::stage_set_node_properties_authenticated( + staged, dir, inventory, stem, updates, + )?; } for (stem, updates) in &self.edges { - total += graphforge_storage::stage_set_edge_properties(staged, dir, stem, updates)?; + total += graphforge_storage::stage_set_edge_properties_authenticated( + staged, dir, inventory, stem, updates, + )?; } Ok(total) } @@ -1813,7 +1832,7 @@ impl SetAccumulator { /// returning the total number of distinct entities written. fn apply(&self, dir: &Path) -> Result { let mut staged = graphforge_storage::RewriteBatch::new(); - let total = self.stage_into(&mut staged, dir)?; + let total = self.stage_into(&mut staged, dir, None)?; staged.commit_at(dir)?; Ok(total) } @@ -1839,6 +1858,10 @@ pub(crate) struct RemoveAccumulator { } impl RemoveAccumulator { + fn is_empty(&self) -> bool { + self.nodes.is_empty() && self.edges.is_empty() + } + fn record(&mut self, is_edge: bool, stem: String, uuid: [u8; 16], prop: String) { let map = if is_edge { &mut self.edges @@ -1876,13 +1899,28 @@ impl RemoveAccumulator { &self, staged: &mut graphforge_storage::RewriteBatch, dir: &Path, + inventory: Option<&graphforge_storage::AuthenticatedPropertyInventory>, ) -> Result { + if self.is_empty() { + return Ok(0); + } + let captured; + let inventory = if let Some(inventory) = inventory { + inventory + } else { + captured = graphforge_storage::AuthenticatedPropertyInventory::capture(dir)?; + &captured + }; let mut total = 0u64; for (stem, removals) in &self.nodes { - total += graphforge_storage::stage_remove_node_properties(staged, dir, stem, removals)?; + total += graphforge_storage::stage_remove_node_properties_authenticated( + staged, dir, inventory, stem, removals, + )?; } for (stem, removals) in &self.edges { - total += graphforge_storage::stage_remove_edge_properties(staged, dir, stem, removals)?; + total += graphforge_storage::stage_remove_edge_properties_authenticated( + staged, dir, inventory, stem, removals, + )?; } Ok(total) } @@ -1890,7 +1928,7 @@ impl RemoveAccumulator { /// One staged batch across all stems, like [`SetAccumulator::apply`]. fn apply(&self, dir: &Path) -> Result { let mut staged = graphforge_storage::RewriteBatch::new(); - let total = self.stage_into(&mut staged, dir)?; + let total = self.stage_into(&mut staged, dir, None)?; staged.commit_at(dir)?; Ok(total) } @@ -5789,6 +5827,7 @@ impl ExecutionSession { params, &wctx.writer.pending_nodes_batch()?, )?; + let property_inventory = env.inventory.clone(); drop(env); drop(lowerer); let terminal_result = match terminal_logical { @@ -5799,7 +5838,7 @@ impl ExecutionSession { }; let c = wctx.mutation.counters; let mutation_receipt = Some(wctx.mutation_receipt()); - transaction.prepare_statement(wctx, &resource)?; + transaction.prepare_statement(wctx, &resource, property_inventory.as_deref())?; let side_effects = Some(SideEffects { nodes_created: c.nodes_created, nodes_deleted: c.nodes_deleted, diff --git a/crates/graphforge-exec/src/mutation.rs b/crates/graphforge-exec/src/mutation.rs index 33c2f08c8..4372fc164 100644 --- a/crates/graphforge-exec/src/mutation.rs +++ b/crates/graphforge-exec/src/mutation.rs @@ -305,11 +305,13 @@ impl MutationTransaction { &mut self, mut context: crate::write_driver::StatementWriteContext, resource: &crate::write_resource::BoundWriteResource, + inventory: Option<&graphforge_storage::AuthenticatedPropertyInventory>, ) -> Result<(), graphforge_core::GfError> { self.admit(resource)?; self.ensure_unprepared()?; self.prepared = true; - self.staged = crate::write_driver::stage_statement(&mut context, resource.directory())?; + self.staged = + crate::write_driver::stage_statement(&mut context, resource.directory(), inventory)?; self.prepared = true; self.state = context.mutation; self.topology = Some(PreparedTopology { diff --git a/crates/graphforge-exec/src/write_driver.rs b/crates/graphforge-exec/src/write_driver.rs index b4f90caf2..c34c15ff1 100644 --- a/crates/graphforge-exec/src/write_driver.rs +++ b/crates/graphforge-exec/src/write_driver.rs @@ -2743,6 +2743,83 @@ fn resolve_kind(schema: &DFSchema, var: VarId, clause: &str) -> Result, + keys: HashMap<[u8; 16], HashSet>, +} + +fn resolve_property_write_batch( + env: &PhaseEnv<'_>, + ctx: &StatementWriteContext, + col: &WriteCol, + batch: &RecordBatch, + selected: Option<&[bool]>, + replacement: bool, +) -> Result { + use graphforge_storage::{AuthenticatedPropertyInventory, PropertyRouteKind}; + use std::collections::{BTreeMap, BTreeSet}; + let mut stems = HashMap::new(); + let mut committed = BTreeMap::new(); + for row in 0..batch.num_rows() { + if selected.is_some_and(|mask| !mask[row]) || batch.column(col.uuid_idx).is_null(row) { + continue; + } + let uuid = fixed_binary_uuid(batch, col.uuid_idx, row)?; + let bytes = to_bytes(&uuid); + let stem = col.stem_for_row(batch, row, env.mode, &env.type_map)?; + stems.insert(bytes, stem.clone()); + let pending = if col.is_edge { + ctx.writer.contains_pending_edge(&bytes) + } else { + ctx.writer.contains_pending_node(&bytes) + }; + if !pending { + committed.insert(uuid, stem); + } + } + let mut keys = HashMap::new(); + if committed.is_empty() || (!col.is_edge && !replacement) { + return Ok(PropertyWriteBatch { stems, keys }); + } + let captured; + let inventory = if let Some(inventory) = env.inventory.as_deref() { + inventory + } else { + captured = AuthenticatedPropertyInventory::capture(env.dir)?; + &captured + }; + if col.is_edge { + let work = + graphforge_storage::resolve_existing_edge_property_owners(inventory, &mut committed)?; + crate::demand::record_edge_owner_work(&work, committed.len()); + } + let mut routes: BTreeMap> = BTreeMap::new(); + for (uuid, stem) in committed { + let bytes = uuid.into_bytes(); + stems.insert(bytes, stem.clone()); + if replacement { + routes.entry(stem).or_default().insert(bytes); + } + } + let kind = if col.is_edge { + PropertyRouteKind::Edge + } else { + PropertyRouteKind::Node + }; + for (route, targets) in routes { + let (rows, work) = graphforge_storage::read_authenticated_property_snapshots_for_inventory( + inventory, kind, &route, &targets, + )?; + crate::demand::record_replacement_key_work(&work, targets.len()); + for (uuid, row) in rows { + keys.insert(uuid, row.values.into_keys().collect()); + } + } + Ok(PropertyWriteBatch { stems, keys }) +} + /// SET phase: evaluate each item's value per frontier row; route the write to /// the pending buffer (created entities), reject deleted targets, accumulate /// the rest for the commit-time rewrite. @@ -2808,6 +2885,7 @@ fn run_set_phase_masked( .and_then(|cv| cv.into_array(n)) .map_err(GfError::from_execution_error)?; let selected = mask.map(|mask| &mask[offset..offset + n]); + let owners = resolve_property_write_batch(env, ctx, &col, batch, selected, false)?; let overlay_values = if let Some(selected) = selected { let selection = BooleanArray::from(selected.to_vec()); let previous = frontier @@ -2849,7 +2927,7 @@ fn run_set_phase_masked( } let scalar = ScalarValue::try_from_array(&values, row) .map_err(GfError::from_execution_error)?; - let stem = col.stem_for_row(batch, row, env.mode, &env.type_map)?; + let stem = owners.stems[&uuid].clone(); if scalar.is_null() { let present = property_is_present( &frontier.df_schema, @@ -2998,16 +3076,27 @@ fn run_set_map_phase_with_input( .ok_or_else(|| { GfError::Execution("SET map expression must evaluate to a map".into()) })?; + let selected: Vec<_> = (0..batch.num_rows()) + .map(|row| !maps.is_null(row)) + .collect(); + let owners = resolve_property_write_batch( + env, + ctx, + &identity, + batch, + Some(&selected), + item.replace, + )?; if maps .column_by_name(graphforge_value::heterogeneous::TAG) .is_some() { for row in 0..batch.num_rows() { - if maps.is_null(row) { + if maps.is_null(row) || batch.column(identity.uuid_idx).is_null(row) { continue; } let uuid = to_bytes(&fixed_binary_uuid(batch, identity.uuid_idx, row)?); - let stem = identity.stem_for_row(batch, row, env.mode, &env.type_map)?; + let stem = owners.stems[&uuid].clone(); let updates = decode_tagged_map_updates(maps, row)?; let mut present: HashSet<_> = existing_names .iter() @@ -3020,9 +3109,7 @@ fn run_set_map_phase_with_input( && !ctx.writer.contains_pending_node(&uuid) && !ctx.writer.contains_pending_edge(&uuid) { - present.extend(graphforge_storage::read_entity_property_keys( - env.dir, &stem, &uuid, is_edge, - )?); + present.extend(owners.keys.get(&uuid).into_iter().flatten().cloned()); } let removals = if item.replace { present @@ -3071,7 +3158,7 @@ fn run_set_map_phase_with_input( "cannot SET properties on an entity deleted in this statement".into(), )); } - let stem = identity.stem_for_row(batch, row, env.mode, &env.type_map)?; + let stem = owners.stems[&uuid].clone(); let mut updates = HashMap::with_capacity(maps.num_columns()); let mut null_keys = HashSet::new(); for (field, column) in maps.fields().iter().zip(maps.columns()) { @@ -3096,9 +3183,7 @@ fn run_set_map_phase_with_input( && !ctx.writer.contains_pending_node(&uuid) && !ctx.writer.contains_pending_edge(&uuid) { - present.extend(graphforge_storage::read_entity_property_keys( - env.dir, &stem, &uuid, is_edge, - )?); + present.extend(owners.keys.get(&uuid).into_iter().flatten().cloned()); } let (removals, replaced) = map_removals(item.replace, &present, &updates, &null_keys); @@ -3321,6 +3406,19 @@ fn run_remove_phase( }); let mut overlay = Vec::with_capacity(frontier.batches.len()); for batch in &frontier.batches { + let selected: Vec<_> = (0..batch.num_rows()) + .map(|row| { + property_is_present( + &frontier.df_schema, + batch, + item.target, + &item.prop_name, + row, + ) + }) + .collect(); + let owners = + resolve_property_write_batch(env, ctx, &col, batch, Some(&selected), false)?; let id_col = batch.column(col.uuid_idx); for row in 0..batch.num_rows() { if id_col.is_null(row) { @@ -3341,7 +3439,7 @@ fn run_remove_phase( ) { continue; } - let stem = col.stem_for_row(batch, row, env.mode, &env.type_map)?; + let stem = owners.stems[&uuid].clone(); let keys = HashSet::from([item.prop_name.clone()]); if is_edge && ctx.writer.contains_pending_edge(&uuid) { ctx.writer.remove_pending_edge_props(&uuid, &keys); @@ -3495,15 +3593,24 @@ fn run_label_phase( pub(crate) fn stage_statement( ctx: &mut StatementWriteContext, dir: &Path, + inventory: Option<&graphforge_storage::AuthenticatedPropertyInventory>, ) -> Result { // Writes to entities deleted later in the statement are unobservable — // they must not resurrect rows in the rewrite. ctx.set_acc.scrub(&ctx.deleted); ctx.remove_acc.scrub(&ctx.deleted); + // Property values/counts belong to the writable workspace; the pinned + // generation contributes only declared semantic owners for absent routes. + let workspace_inventory = if ctx.set_acc.is_empty() && ctx.remove_acc.is_empty() { + None + } else { + Some(graphforge_storage::AuthenticatedPropertyInventory::capture_workspace(dir, inventory)?) + }; + let inventory = workspace_inventory.as_ref(); let mut staged = graphforge_storage::RewriteBatch::new(); - ctx.set_acc.stage_into(&mut staged, dir)?; - ctx.remove_acc.stage_into(&mut staged, dir)?; + ctx.set_acc.stage_into(&mut staged, dir, inventory)?; + ctx.remove_acc.stage_into(&mut staged, dir, inventory)?; graphforge_storage::stage_mutate_node_labels( &mut staged, dir, @@ -4189,6 +4296,88 @@ mod tests { } } + #[test] + fn ambiguous_edge_owner_refuses_before_accumulation_or_publication() { + let dir = tempfile::tempdir().unwrap(); + let mut writer = + graphforge_storage::GraphWriter::open_at(dir.path(), OntologyMode::Exploratory, 1) + .unwrap(); + for route in ["KNOWS", "_exploratory"] { + writer + .set_edge_properties( + &graphforge_core::uuid::Uuid::from_bytes([7; 16]), + Some(route), + HashMap::from([("score".into(), IrLiteral::Int(1))]), + ) + .unwrap(); + } + writer.flush().unwrap(); + let before = graphforge_storage::capture_graph_files(dir.path()) + .unwrap() + .0; + let catalog = graphforge_storage::GraphCatalog::open( + dir.path(), + None, + &graphforge_ir::RuntimeCatalog::new(), + ) + .unwrap(); + let lowerer = GraphPlanLowerer::new_for_writes( + &graphforge_storage::lowering_snapshot(Some(&catalog), Some(dir.path())).unwrap(), + None, + OntologyMode::Exploratory, + ) + .unwrap(); + let mut exprs = ExprArena::new(); + let value = exprs.push(IrExpr::Literal(IrLiteral::Int(42))); + let params = HashMap::new(); + let env = phase_env(&lowerer, &exprs, dir.path(), ¶ms); + let mut var_map = VarMap::new(); + var_map.insert(VarId(1), "var_1"); + for remove in [false, true] { + let mut ctx = + StatementWriteContext::new(dir.path(), OntologyMode::Exploratory).unwrap(); + let mut frontier = write_frontier(true); + let result = if remove { + run_remove_phase( + &env, + &[RemovePropItem { + target: VarId(1), + prop: graphforge_value::PropertyId::ontology(PropId(9)).unwrap(), + prop_name: "score".into(), + }], + &mut frontier, + &mut ctx, + ) + } else { + run_set_phase( + &env, + &[SetPropItem { + target: VarId(1), + prop: graphforge_value::PropertyId::ontology(PropId(9)).unwrap(), + prop_name: "score".into(), + value, + }], + &mut frontier, + &var_map, + &mut ctx, + ) + }; + assert!( + result + .unwrap_err() + .to_string() + .contains("edge property owner is ambiguous") + ); + assert!(ctx.set_acc.nodes.is_empty() && ctx.set_acc.edges.is_empty()); + assert!(ctx.remove_acc.nodes.is_empty() && ctx.remove_acc.edges.is_empty()); + assert_eq!(ctx.mutation.counters, WriteCounters::default()); + let after = graphforge_storage::capture_graph_files(dir.path()) + .unwrap() + .0; + assert_eq!(before, after); + } + } + #[test] fn set_accumulates_nodes_and_edges_once_across_duplicate_and_null_rows() { for is_edge in [false, true] { diff --git a/crates/graphforge-rel/src/lowerer.rs b/crates/graphforge-rel/src/lowerer.rs index f5b509413..865f6a2d2 100644 --- a/crates/graphforge-rel/src/lowerer.rs +++ b/crates/graphforge-rel/src/lowerer.rs @@ -2375,8 +2375,8 @@ impl GraphPlanLowerer { /// A node var carries a `var_.node_uuid` column; an edge var a /// `var_.edge_uuid` column. An edge var must also carry the /// `var_.rel_type_name` column — the per-row file stem for edge property - /// writes (present in Exploratory mode; a typed-only Strict-mode edge has no - /// such column and is rejected here, a documented #791 follow-up). Returns + /// writes (stored for exploratory routes and projected from the authenticated + /// catalog route for typed scans). Returns /// `is_edge` or a lowering error. fn resolve_write_kind( schema: &datafusion::common::DFSchemaRef, @@ -3245,6 +3245,27 @@ fn lower_node_scan( builder.build().map_unsupported_expr() } +/// Carry the catalog-resolved physical route into the write frontier. Typed +/// topology omits this constant on disk; mutation ownership still needs it. +fn project_typed_edge_route( + scan: LogicalPlan, + alias: &str, + route: &str, +) -> Result { + let mut columns: Vec<_> = scan + .schema() + .columns() + .into_iter() + .map(DfExpr::Column) + .collect(); + columns + .push(datafusion::logical_expr::lit(route).alias_qualified(Some(alias), "rel_type_name")); + LogicalPlanBuilder::from(scan) + .project(columns) + .and_then(LogicalPlanBuilder::build) + .map_unsupported_expr() +} + fn lower_typed_edge_scan( var: VarId, rel_ty: RelationTypeId, @@ -3282,8 +3303,8 @@ fn lower_typed_edge_scan( rel_ty.encode() )) })?; - return LogicalPlanBuilder::scan( - alias, + let scan = LogicalPlanBuilder::scan( + alias.clone(), graphforge_plan::GraphReadSource::new( graphforge_plan::GraphReadTable::SemanticEdges(rel_ty), &provider, @@ -3294,7 +3315,8 @@ fn lower_typed_edge_scan( None, ) .and_then(LogicalPlanBuilder::build) - .map_unsupported_expr(); + .map_unsupported_expr()?; + return project_typed_edge_route(scan, &alias, rel_name); } // Check if the catalog has a typed edge table for this relation. @@ -3313,9 +3335,10 @@ fn lower_typed_edge_scan( let src = edge_scan_source(dir, rel_name, &TYPED_EDGE_SCHEMA, mode); // Use alias as the scan qualifier so downstream join predicates // (var_map.get(edge) → "var_N") can resolve edge columns correctly. - LogicalPlanBuilder::scan(alias, src, None) + let scan = LogicalPlanBuilder::scan(alias.clone(), src, None) .and_then(LogicalPlanBuilder::build) - .map_unsupported_expr() + .map_unsupported_expr()?; + project_typed_edge_route(scan, &alias, rel_name) } } @@ -4077,7 +4100,7 @@ fn try_lower_provider_expand( let node = graphforge_plan::ExpandNode::new( Arc::new(input.clone()), - rel_name, + rel_name.clone(), src.0, traversal_dst.0, edge.0, @@ -4097,6 +4120,9 @@ fn try_lower_provider_expand( let mut base = LogicalPlan::Extension(datafusion::logical_expr::Extension { node: Arc::new(node), }); + if rel_ty.is_some() && !matches!(mode, OntologyMode::Exploratory) { + base = project_typed_edge_route(base, &var_alias(edge), &rel_name)?; + } if let Some(bound_dst) = bound_dst { use datafusion::logical_expr::col; @@ -6558,6 +6584,30 @@ relation_types: (tmp, catalog, plan) } + fn assert_typed_route_projection(plan: &DfLogicalPlan) -> &Extension { + let DfLogicalPlan::Projection(projection) = plan else { + panic!("expected typed route projection, got {plan:?}"); + }; + let mut expected = projection + .input + .schema() + .columns() + .into_iter() + .map(DfExpr::Column) + .collect::>(); + expected.push( + datafusion::logical_expr::lit("KNOWS").alias_qualified(Some("var_1"), "rel_type_name"), + ); + assert_eq!( + projection.expr, expected, + "preserve all input columns and exact route" + ); + let DfLogicalPlan::Extension(extension) = projection.input.as_ref() else { + panic!("expected ExpandNode directly beneath route projection, got {plan:?}"); + }; + extension + } + #[test] fn project_backed_single_hop_emits_expand_extension_node() { use datafusion::logical_expr::UserDefinedLogicalNodeCore; @@ -6571,9 +6621,7 @@ relation_types: .unwrap(); let lp = lowerer.lower_plan(&plan).unwrap(); - let DfLogicalPlan::Extension(ext) = &lp else { - panic!("expected Extension (ExpandNode), got {lp:?}"); - }; + let ext = assert_typed_route_projection(&lp); let node = ext .node .as_any() @@ -6623,9 +6671,7 @@ relation_types: // No DISTINCT wrapper: the self-loop dedup happens inside ExpandExec // (a wrapping Distinct trips DataFusion's duplicate-field-name // disambiguation against the extension's multi-var schema). - let DfLogicalPlan::Extension(ext) = &lp else { - panic!("expected Extension (ExpandNode), got {lp:?}"); - }; + let ext = assert_typed_route_projection(&lp); let node = ext .node .as_any() diff --git a/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__named_path_fixed_segment_and_return_p.snap b/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__named_path_fixed_segment_and_return_p.snap index 0fcf9d47b..57e4c0695 100644 --- a/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__named_path_fixed_segment_and_return_p.snap +++ b/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__named_path_fixed_segment_and_return_p.snap @@ -1,9 +1,10 @@ --- source: crates/graphforge-rel/tests/logical_plan_golden.rs -assertion_line: 320 +assertion_line: 339 --- Projection: CASE WHEN var_1.edge_uuid IS NOT NULL THEN UInt64(1) ELSE UInt64(NULL) END AS hops, CASE WHEN var_1.edge_uuid IS NOT NULL THEN make_array(CASE WHEN var_0.node_uuid IS NOT NULL THEN named_struct(Utf8("node_uuid"), var_0.node_uuid, Utf8("labels"), array_concat(CASE WHEN array_has(var_0.type_ids, UInt32(0)) THEN make_array(Utf8("Person")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(1)) THEN make_array(Utf8("Employee")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(2)) THEN make_array(Utf8("Manager")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(3)) THEN make_array(Utf8("Department")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(4)) THEN make_array(Utf8("Project")) ELSE List([]) END)) ELSE NULL END, CASE WHEN var_2.node_uuid IS NOT NULL THEN named_struct(Utf8("node_uuid"), var_2.node_uuid, Utf8("labels"), array_concat(CASE WHEN array_has(var_2.type_ids, UInt32(0)) THEN make_array(Utf8("Person")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(1)) THEN make_array(Utf8("Employee")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(2)) THEN make_array(Utf8("Manager")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(3)) THEN make_array(Utf8("Department")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(4)) THEN make_array(Utf8("Project")) ELSE List([]) END)) ELSE NULL END) ELSE NULL END AS ns, CASE WHEN var_1.edge_uuid IS NOT NULL THEN make_array(named_struct(Utf8("edge_uuid"), var_1.edge_uuid, Utf8("src_uuid"), var_1.src_uuid, Utf8("dst_uuid"), var_1.dst_uuid, Utf8("rel_type"), CAST(Utf8("IS_FRIEND_OF") AS Utf8))) ELSE NULL END AS rels, CASE WHEN CASE WHEN var_1.edge_uuid IS NOT NULL THEN make_array(CASE WHEN var_0.node_uuid IS NOT NULL THEN named_struct(Utf8("node_uuid"), var_0.node_uuid, Utf8("labels"), array_concat(CASE WHEN array_has(var_0.type_ids, UInt32(0)) THEN make_array(Utf8("Person")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(1)) THEN make_array(Utf8("Employee")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(2)) THEN make_array(Utf8("Manager")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(3)) THEN make_array(Utf8("Department")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(4)) THEN make_array(Utf8("Project")) ELSE List([]) END)) ELSE NULL END, CASE WHEN var_2.node_uuid IS NOT NULL THEN named_struct(Utf8("node_uuid"), var_2.node_uuid, Utf8("labels"), array_concat(CASE WHEN array_has(var_2.type_ids, UInt32(0)) THEN make_array(Utf8("Person")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(1)) THEN make_array(Utf8("Employee")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(2)) THEN make_array(Utf8("Manager")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(3)) THEN make_array(Utf8("Department")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(4)) THEN make_array(Utf8("Project")) ELSE List([]) END)) ELSE NULL END) ELSE NULL END IS NOT NULL THEN named_struct(Utf8("nodes"), CASE WHEN var_1.edge_uuid IS NOT NULL THEN make_array(CASE WHEN var_0.node_uuid IS NOT NULL THEN named_struct(Utf8("node_uuid"), var_0.node_uuid, Utf8("labels"), array_concat(CASE WHEN array_has(var_0.type_ids, UInt32(0)) THEN make_array(Utf8("Person")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(1)) THEN make_array(Utf8("Employee")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(2)) THEN make_array(Utf8("Manager")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(3)) THEN make_array(Utf8("Department")) ELSE List([]) END, CASE WHEN array_has(var_0.type_ids, UInt32(4)) THEN make_array(Utf8("Project")) ELSE List([]) END)) ELSE NULL END, CASE WHEN var_2.node_uuid IS NOT NULL THEN named_struct(Utf8("node_uuid"), var_2.node_uuid, Utf8("labels"), array_concat(CASE WHEN array_has(var_2.type_ids, UInt32(0)) THEN make_array(Utf8("Person")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(1)) THEN make_array(Utf8("Employee")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(2)) THEN make_array(Utf8("Manager")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(3)) THEN make_array(Utf8("Department")) ELSE List([]) END, CASE WHEN array_has(var_2.type_ids, UInt32(4)) THEN make_array(Utf8("Project")) ELSE List([]) END)) ELSE NULL END) ELSE NULL END, Utf8("relationships"), CASE WHEN var_1.edge_uuid IS NOT NULL THEN make_array(named_struct(Utf8("edge_uuid"), var_1.edge_uuid, Utf8("src_uuid"), var_1.src_uuid, Utf8("dst_uuid"), var_1.dst_uuid, Utf8("rel_type"), CAST(Utf8("IS_FRIEND_OF") AS Utf8))) ELSE NULL END) ELSE NULL END AS p [hops:UInt64;N, ns:List(Struct("node_uuid": FixedSizeBinary(16), "labels": List(Utf8)));N, rels:List(Struct("edge_uuid": FixedSizeBinary(16), "src_uuid": FixedSizeBinary(16), "dst_uuid": FixedSizeBinary(16), "rel_type": Utf8));N, p:Struct("nodes": List(Struct("node_uuid": FixedSizeBinary(16), "labels": List(Utf8))), "relationships": List(Struct("edge_uuid": FixedSizeBinary(16), "src_uuid": FixedSizeBinary(16), "dst_uuid": FixedSizeBinary(16), "rel_type": Utf8)));N] - Filter: array_has(var_2.type_ids, UInt32(0)) [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] - Expand: rel=IS_FRIEND_OF, dir=->, edge=var_1 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] - Filter: array_has(var_0.type_ids, UInt32(0)) [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] - TableScan: var_0 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] + Filter: array_has(var_2.type_ids, UInt32(0)) [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), rel_type_name:Utf8] + Projection: var_0.node_uuid, var_0.node_id, var_0.type_id, var_0.type_ids, var_0.created_at, var_0.updated_at, var_1.edge_uuid, var_1.src_uuid, var_1.dst_uuid, var_1.edge_id, var_1.src_id, var_1.dst_id, var_1.created_at, var_2.node_uuid, var_2.node_id, var_2.type_id, var_2.type_ids, var_2.created_at, var_2.updated_at, Utf8("IS_FRIEND_OF") AS rel_type_name [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), rel_type_name:Utf8] + Expand: rel=IS_FRIEND_OF, dir=->, edge=var_1 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] + Filter: array_has(var_0.type_ids, UInt32(0)) [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] + TableScan: var_0 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] diff --git a/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__optional_match.snap b/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__optional_match.snap index 510e16814..7b088b79d 100644 --- a/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__optional_match.snap +++ b/crates/graphforge-rel/tests/logical_plan_goldens/logical_plan_golden__optional_match.snap @@ -1,11 +1,12 @@ --- source: crates/graphforge-rel/tests/logical_plan_golden.rs -assertion_line: 336 +assertion_line: 355 --- Projection: Int64(1) AS one [one:Int64] - OptionalMatch: keys=1 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16);N, src_uuid:FixedSizeBinary(16);N, dst_uuid:FixedSizeBinary(16);N, edge_id:UInt64;N, src_id:UInt64;N, dst_id:UInt64;N, created_at:Timestamp(µs, "UTC");N, node_uuid:FixedSizeBinary(16);N, node_id:UInt64;N, type_id:UInt32;N, type_ids:List(non-null UInt32);N, created_at:Timestamp(µs, "UTC");N, updated_at:Timestamp(µs, "UTC");N] + OptionalMatch: keys=1 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16);N, src_uuid:FixedSizeBinary(16);N, dst_uuid:FixedSizeBinary(16);N, edge_id:UInt64;N, src_id:UInt64;N, dst_id:UInt64;N, created_at:Timestamp(µs, "UTC");N, node_uuid:FixedSizeBinary(16);N, node_id:UInt64;N, type_id:UInt32;N, type_ids:List(non-null UInt32);N, created_at:Timestamp(µs, "UTC");N, updated_at:Timestamp(µs, "UTC");N, rel_type_name:Utf8;N] Filter: array_has(var_0.type_ids, UInt32(1)) [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] TableScan: var_0 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] - Filter: array_has(var_2.type_ids, UInt32(2)) [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] - Expand: rel=REPORTS_TO, dir=->, edge=var_1 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] - TableScan: var_0 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] + Filter: array_has(var_2.type_ids, UInt32(2)) [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), rel_type_name:Utf8] + Projection: var_0.node_uuid, var_0.node_id, var_0.type_id, var_0.type_ids, var_0.created_at, var_0.updated_at, var_1.edge_uuid, var_1.src_uuid, var_1.dst_uuid, var_1.edge_id, var_1.src_id, var_1.dst_id, var_1.created_at, var_2.node_uuid, var_2.node_id, var_2.type_id, var_2.type_ids, var_2.created_at, var_2.updated_at, Utf8("REPORTS_TO") AS rel_type_name [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), rel_type_name:Utf8] + Expand: rel=REPORTS_TO, dir=->, edge=var_1 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC"), edge_uuid:FixedSizeBinary(16), src_uuid:FixedSizeBinary(16), dst_uuid:FixedSizeBinary(16), edge_id:UInt64, src_id:UInt64, dst_id:UInt64, created_at:Timestamp(µs, "UTC"), node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] + TableScan: var_0 [node_uuid:FixedSizeBinary(16), node_id:UInt64, type_id:UInt32, type_ids:List(non-null UInt32), created_at:Timestamp(µs, "UTC"), updated_at:Timestamp(µs, "UTC")] diff --git a/crates/graphforge-storage/src/lib.rs b/crates/graphforge-storage/src/lib.rs index a7cc75aec..47e09120c 100644 --- a/crates/graphforge-storage/src/lib.rs +++ b/crates/graphforge-storage/src/lib.rs @@ -396,13 +396,14 @@ pub use ordinal_identity_v4::{ pub mod property_overlay; mod property_scan; pub use property_overlay::{ - AuthenticatedPropertyInventory, PROPERTY_OVERLAY_FORMAT, PROPERTY_OVERLAY_FORMAT_KEY, - PROPERTY_TOMBSTONE_FIELD, PropertyFragmentId, PropertyInventoryOpenMetrics, - PropertyOverlayLimits, PropertyOverlayMetrics, PropertyRouteKind, PropertySnapshotRow, - PropertyTargetSnapshots, enumerate_property_fragments, + AuthenticatedPropertyInventory, EdgeOwnerProbeWork, PROPERTY_OVERLAY_FORMAT, + PROPERTY_OVERLAY_FORMAT_KEY, PROPERTY_TOMBSTONE_FIELD, PropertyFragmentId, + PropertyInventoryOpenMetrics, PropertyOverlayLimits, PropertyOverlayMetrics, PropertyRouteKind, + PropertySnapshotRow, PropertyTargetSnapshots, enumerate_property_fragments, read_authenticated_property_presence_for_inventory, read_authenticated_property_snapshots_for, read_authenticated_property_snapshots_for_inventory, - read_authenticated_property_targets_for_inventory, visit_authenticated_property_snapshots, + read_authenticated_property_targets_for_inventory, resolve_existing_edge_property_owners, + visit_authenticated_property_snapshots, }; pub mod catalog; diff --git a/crates/graphforge-storage/src/property_overlay.rs b/crates/graphforge-storage/src/property_overlay.rs index 0154a37f0..f5dc7b69c 100644 --- a/crates/graphforge-storage/src/property_overlay.rs +++ b/crates/graphforge-storage/src/property_overlay.rs @@ -22,6 +22,7 @@ use parquet::file::reader::{ChunkReader, Length}; use parquet::thrift::TSerializable; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use uuid::Uuid; /// On-disk property overlay format marker. pub const PROPERTY_OVERLAY_FORMAT: &str = "full-snapshot-v1"; @@ -372,6 +373,12 @@ struct RouteSchemaBuilder { } impl AuthenticatedPropertyInventory { + /// Authenticate the current property authority of a project or materialized + /// workspace. Sessions should retain and share the resulting inventory. + pub fn capture(project: &Path) -> Result { + authenticated_property_inventory(project) + } + /// Return semantic relation names and their admitted physical payload paths. /// The inventory retains the generation lease and route authority. #[must_use] @@ -956,6 +963,17 @@ impl AuthenticatedPropertyInventory { }) } + /// Admit the writable workspace's current property rows and statistics, + /// retaining declared semantic owners from the transaction's pinned generation. + /// The generation's historical live counts are not workspace statistics. + pub fn capture_workspace(project: &Path, pinned: Option<&Self>) -> Result { + let mut inventory = Self::capture(project)?; + if let Some(generation) = pinned.and_then(|pinned| pinned.generation_lease.as_ref()) { + inventory.seed_semantic_property_schemas(generation)?; + } + Ok(inventory) + } + // A declared owner may not have a property payload yet. Its authenticated // binding still owns the metadata of the first fragment we publish. fn seed_semantic_property_schemas( @@ -2449,6 +2467,114 @@ pub fn read_authenticated_property_snapshots_for_inventory( Ok((result.rows, result.metrics)) } +/// Cumulative probe I/O and serial decoder peaks. The identity containers are +/// reported separately: decoder counters are not a bound on process memory. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct EdgeOwnerProbeWork { + /// Candidate routes. + pub candidate_routes: usize, + /// Target memberships. + pub target_memberships: usize, + /// Resolved targets. + pub resolved_targets: usize, + /// Route name bytes. + pub route_name_bytes: usize, + /// Physical bytes. + pub physical_bytes: u64, + /// Authentication bytes. + pub authentication_bytes: u64, + /// Authenticated snapshot bytes. + pub authenticated_snapshot_bytes: u64, + /// Authenticated snapshot peak bytes. + pub authenticated_snapshot_peak_bytes: u64, + /// Physical rows. + pub physical_rows: u64, + /// Physical blocks. + pub physical_blocks: u64, + /// Fragments considered. + pub fragments_considered: u64, + /// Row groups considered. + pub row_groups_considered: u64, + /// Row groups selected. + pub row_groups_selected: u64, + /// Decoder peak bytes. + pub decoder_peak_bytes: u64, + /// Decoder page reservation bytes. + pub decoder_page_reservation_bytes: u64, +} + +/// Construction and ordinary mutation have different exploratory property +/// owners. Probe only the candidate routes for requested identities, once per +/// route, using authenticated row presence (including a newest tombstone). +pub fn resolve_existing_edge_property_owners( + inventory: &crate::AuthenticatedPropertyInventory, + owners: &mut BTreeMap, +) -> Result { + use crate::PropertyRouteKind; + let mut candidates = BTreeMap::>::new(); + for (uuid, route) in owners.iter() { + for candidate in [route.as_str(), "_exploratory"] { + if inventory + .route_schema(PropertyRouteKind::Edge, candidate) + .is_some() + { + if let Some(targets) = candidates.get_mut(candidate) { + targets.insert(uuid.into_bytes()); + } else { + candidates.insert(candidate.to_owned(), BTreeSet::from([uuid.into_bytes()])); + } + } + } + } + let mut resolved = BTreeMap::new(); + let mut total = EdgeOwnerProbeWork { + candidate_routes: candidates.len(), + target_memberships: candidates.values().map(BTreeSet::len).sum(), + route_name_bytes: candidates.keys().map(String::len).sum(), + ..EdgeOwnerProbeWork::default() + }; + for (route, targets) in &candidates { + let (present, work) = crate::read_authenticated_property_presence_for_inventory( + inventory, + PropertyRouteKind::Edge, + route, + targets, + )?; + total.physical_bytes += work.physical_bytes; + total.authentication_bytes += work.authentication_bytes; + total.authenticated_snapshot_bytes += work.authenticated_snapshot_bytes; + total.authenticated_snapshot_peak_bytes = total + .authenticated_snapshot_peak_bytes + .max(work.authenticated_snapshot_peak_bytes); + total.physical_rows += work.physical_rows; + total.physical_blocks += work.physical_blocks; + total.fragments_considered += work.fragments_considered; + total.row_groups_considered += work.row_groups_considered; + total.row_groups_selected += work.row_groups_selected; + total.decoder_peak_bytes = total.decoder_peak_bytes.max(work.decoder_peak_bytes); + total.decoder_page_reservation_bytes = total + .decoder_page_reservation_bytes + .max(work.decoder_page_reservation_bytes); + for uuid in present { + if resolved + .insert(Uuid::from_bytes(uuid), route.as_str()) + .is_some() + { + return Err(GfError::Storage( + "composite edge property owner is ambiguous".into(), + )); + } + } + } + // No property row means ordinary writer ownership, already derived from + // topology. Same-request creations are added by the caller afterwards. + total.resolved_targets = resolved.len(); + for (uuid, route) in resolved { + owners.insert(uuid, route.to_owned()); + } + Ok(total) +} + /// Selected logical values and tombstone-inclusive ownership from one read. pub struct PropertyTargetSnapshots { /// Latest live snapshots for requested UUIDs only. diff --git a/docs/book/architecture/storage.md b/docs/book/architecture/storage.md index 6552093a9..0f354796d 100644 --- a/docs/book/architecture/storage.md +++ b/docs/book/architecture/storage.md @@ -1208,3 +1208,53 @@ temporary-disk bounds. The separate point census includes retained generations. See [`bounded-csr-1205.json`](../../development/evidence/bounded-csr-1205.json) for frozen source/executable hashes, exact observations, commands, decoded bounds, CPU/I/O costs and limitations, including the superseded incomplete baseline trace. + +### Ordinary Cypher property ownership + +Committed edge SET, map updates and REMOVE resolve the authenticated property +owner before accumulating effects. Each input batch probes only its logical +relation routes and `_exploratory`, including newest tombstones. Multiple owners +are refused; an entity without a property row keeps its logical route. Edges +created in the same statement retain the pending writer's route. Replacement +maps read existing keys in batches from the resolved owner, so omitted properties +are removed from the same authority that receives the replacement. + +Typed scans and fixed expansions carry their catalog-resolved route as an +internal constant column. This uses the generation's storage binding, without +reinterpreting ontology IDs as runtime IDs. During staging, current property +rows and live counts come from the writable workspace. The pinned generation +supplies declared semantic schema metadata for a route's first property write; +its historical counts and generation authority are not substituted for the +workspace's. SET/REMOVE and pending appends therefore share workspace authority. Immediate query success is +insufficient: ownership fixtures also reopen after qualified mutations and +exercise full portable verification, clean import and later mutation. + +The `property_writes` demand diagnostics sum owner probes and replacement-key +reads across every input batch and SET item. Repeated reads are counted each +time. Decoder and authenticated-snapshot peaks describe those readers; target +counts describe retained identities. These counters neither measure the entire +publication nor bound process RSS. The representative multi-batch regression +checks exact results and explicit cumulative work ceilings. + +Frozen-release measurements cover the entire test process, including fixture +construction, mutations, exact queries and portable lifecycle work. The original +baseline refuses corruption before completing that lifecycle, so these are +corrected-path costs rather than a speedup comparison. + +| Fixture | CPU seconds (3 runs) | Maximum RSS (KiB) | Successful syscall read / write bytes | Sampled allocated bytes | +|---|---:|---:|---:|---:| +| Mixed construction and ordinary mutation, 33 / 4,097 nodes | 4.20–4.21 | 120,536 | 340,327,349 / 69,275,673 | 14,802,944 | +| 8,193-edge multi-batch owner and replacement work | 1.57–1.60 | 166,980 | 114,304,045 / 25,674,019 | 5,394,432 | +| Qualified mutation, reopen and portable lifecycle | 1.07–1.16 | 90,456 | 28,610,264 / 2,582,102 | 2,129,920 | + +Syscall bytes include non-file descriptors and count a file copy once as a read +and once as a write; they are not physical disk I/O. The evidence records GNU +filesystem counters separately. Disk sampling deduplicates shared device/inode +pairs across the private workspace and all coexisting lifecycle artifacts. Its +largest observed gap was 15.10 ms despite a requested 5 ms interval; it excludes +directories and unlinked open files and is not a hard bound. Native process RSS +is separate from the logical reader-buffer ceilings above. + +See [Cypher ownership evidence](../../development/evidence/cypher-property-ownership-1224.json) +for executable/source hashes, all observations, deterministic budgets, reproduction +instructions, failed prototypes and the known local #1192 fixture limitation. diff --git a/docs/development/evidence/cypher-property-ownership-1224.json b/docs/development/evidence/cypher-property-ownership-1224.json new file mode 100644 index 000000000..877f696d3 --- /dev/null +++ b/docs/development/evidence/cypher-property-ownership-1224.json @@ -0,0 +1,376 @@ +{ + "issue": 1224, + "scope": "Ordinary Cypher property ownership; public complete lifecycle and source-bound resource cost, not a speedup claim.", + "runtime_source": "055e34574b7bf140a6ce138f8a503ce3257988c0", + "executable_sha256": "36aef0711bdc1a1ef9900f18d8cfbedeac65085a8301eed4ba24d536f615e867", + "later_changes": "e95a796c and 119d76ec update only cfg(test) route-projection assertions and two plan goldens; runtime implementation is unchanged from the frozen source.", + "environment": { + "machine": "AMD Ryzen 7 3800X; 16 logical CPUs", + "native_project_filesystem": "ext4", + "tmpfs_tmp": "The separately tracked #1192 fixture hardcodes /tmp and fails admission; measured projects use private native-volume TMPDIRs.", + "rustc": "rustc 1.96.0 (ac68faa20 2026-05-25)" + }, + "protocol": { + "repetitions": 3, + "fresh_processes": true, + "case_order": "Rotate lifecycle/owners/qualified across repetitions", + "freeze": "Successful release compiler artifact copied to a private executable; SHA-256 checked before and after processes. No overlapping builds/tests during measurements.", + "timing": "GNU time whole-process user/system CPU, elapsed time, max RSS and filesystem counters. Sampling and syscall tracing are separate runs; their timings are not latency evidence.", + "sampling": "Requested 5ms, non-atomic recursive regular-file walk of the entire private TMPDIR; globally deduplicate (st_dev,st_ino), sum st_blocks*512. Includes coexisting source, retained generations, workspace temporaries, portable package and clean import. Excludes directories and unlinked open files; sampled peak is not a hard bound.", + "io": "Successful read/pread/readv/preadv/write/pwrite/writev/pwritev/copy_file_range/sendfile returns across the tracee process tree, including non-file descriptors. Copy/sendfile contributes one read and one write. Does not count memory accesses or mapped-file page faults as syscall bytes. GNU filesystem counters are reported separately, without equating them to syscall bytes." + }, + "cases": { + "lifecycle": { + "test": "constructed_edge_cypher_maps_and_removals_preserve_portable_lifecycle", + "observations": [ + { + "user_seconds": 2.19, + "system_seconds": 2.01, + "peak_rss_kib": 118592, + "filesystem_input_blocks": 32504, + "filesystem_output_blocks": 157560, + "major_page_faults": 0, + "minor_page_faults": 20975, + "cpu_seconds": 4.2, + "gnu_elapsed": "0:04.31" + }, + { + "user_seconds": 2.23, + "system_seconds": 1.97, + "peak_rss_kib": 118676, + "filesystem_input_blocks": 32504, + "filesystem_output_blocks": 157560, + "major_page_faults": 0, + "minor_page_faults": 19524, + "cpu_seconds": 4.2, + "gnu_elapsed": "0:04.31" + }, + { + "user_seconds": 2.17, + "system_seconds": 2.04, + "peak_rss_kib": 120536, + "filesystem_input_blocks": 32504, + "filesystem_output_blocks": 157560, + "major_page_faults": 0, + "minor_page_faults": 20191, + "cpu_seconds": 4.21, + "gnu_elapsed": "0:04.32" + } + ], + "cpu_seconds_range": [ + 4.2, + 4.21 + ], + "maximum_observed_rss_kib": 120536, + "sample": { + "samples": 474, + "sampled_allocated_peak": 14802944, + "sampled_logical_peak": 12763965, + "max_sample_gap_seconds": 0.015093212015926838 + }, + "trace": { + "read_bytes": 340327349, + "write_bytes": 69275673, + "syscalls": { + "read": { + "calls": 76476, + "bytes": 244933093, + "errors": 0 + }, + "pread64": { + "calls": 221389, + "bytes": 81289043, + "errors": 0 + }, + "write": { + "calls": 11855, + "bytes": 55170460, + "errors": 0 + }, + "fsync": { + "calls": 5859, + "bytes": 0, + "errors": 0 + }, + "copy_file_range": { + "calls": 2640, + "bytes": 14105213, + "errors": 0 + } + }, + "unfinished_calls": 34670, + "resumed_calls": 34670, + "unparsed_results": 0 + }, + "all_oracles_passed": true + }, + "owners": { + "test": "cypher_edge_owner_work_accounts_for_multiple_batches_and_items", + "observations": [ + { + "user_seconds": 1.03, + "system_seconds": 0.57, + "peak_rss_kib": 157316, + "filesystem_input_blocks": 101000, + "filesystem_output_blocks": 55808, + "major_page_faults": 0, + "minor_page_faults": 47308, + "cpu_seconds": 1.6, + "gnu_elapsed": "0:01.91" + }, + { + "user_seconds": 1.0, + "system_seconds": 0.57, + "peak_rss_kib": 166980, + "filesystem_input_blocks": 101000, + "filesystem_output_blocks": 55808, + "major_page_faults": 0, + "minor_page_faults": 52330, + "cpu_seconds": 1.57, + "gnu_elapsed": "0:01.86" + }, + { + "user_seconds": 1.02, + "system_seconds": 0.57, + "peak_rss_kib": 163100, + "filesystem_input_blocks": 101000, + "filesystem_output_blocks": 55808, + "major_page_faults": 0, + "minor_page_faults": 42104, + "cpu_seconds": 1.59, + "gnu_elapsed": "0:01.88" + } + ], + "cpu_seconds_range": [ + 1.57, + 1.6 + ], + "maximum_observed_rss_kib": 166980, + "sample": { + "samples": 269, + "sampled_allocated_peak": 5394432, + "sampled_logical_peak": 5052849, + "max_sample_gap_seconds": 0.009272310067899525 + }, + "trace": { + "read_bytes": 114304045, + "write_bytes": 25674019, + "syscalls": { + "read": { + "calls": 11605, + "bytes": 78762257, + "errors": 0 + }, + "pread64": { + "calls": 47568, + "bytes": 33971541, + "errors": 0 + }, + "write": { + "calls": 1461, + "bytes": 24103772, + "errors": 0 + }, + "fsync": { + "calls": 2181, + "bytes": 0, + "errors": 0 + }, + "copy_file_range": { + "calls": 170, + "bytes": 1570247, + "errors": 0 + } + }, + "unfinished_calls": 1102, + "resumed_calls": 1102, + "unparsed_results": 0 + }, + "all_oracles_passed": true + }, + "qualified": { + "test": "qualified_cypher_edge_owners_preserve_constructed_and_pending_values", + "observations": [ + { + "user_seconds": 0.4, + "system_seconds": 0.75, + "peak_rss_kib": 90456, + "filesystem_input_blocks": 8464, + "filesystem_output_blocks": 13808, + "major_page_faults": 0, + "minor_page_faults": 5526, + "cpu_seconds": 1.15, + "gnu_elapsed": "0:01.49" + }, + { + "user_seconds": 0.33, + "system_seconds": 0.74, + "peak_rss_kib": 89176, + "filesystem_input_blocks": 8464, + "filesystem_output_blocks": 13808, + "major_page_faults": 0, + "minor_page_faults": 5982, + "cpu_seconds": 1.07, + "gnu_elapsed": "0:01.40" + }, + { + "user_seconds": 0.36, + "system_seconds": 0.8, + "peak_rss_kib": 89564, + "filesystem_input_blocks": 8464, + "filesystem_output_blocks": 13808, + "major_page_faults": 0, + "minor_page_faults": 5374, + "cpu_seconds": 1.16, + "gnu_elapsed": "0:01.50" + } + ], + "cpu_seconds_range": [ + 1.07, + 1.16 + ], + "maximum_observed_rss_kib": 90456, + "sample": { + "samples": 186, + "sampled_allocated_peak": 2129920, + "sampled_logical_peak": 764396, + "max_sample_gap_seconds": 0.010875510983169079 + }, + "trace": { + "read_bytes": 28610264, + "write_bytes": 2582102, + "syscalls": { + "read": { + "calls": 30294, + "bytes": 22281337, + "errors": 0 + }, + "pread64": { + "calls": 16319, + "bytes": 5680172, + "errors": 0 + }, + "write": { + "calls": 2572, + "bytes": 1933347, + "errors": 0 + }, + "fsync": { + "calls": 3135, + "bytes": 0, + "errors": 0 + }, + "copy_file_range": { + "calls": 844, + "bytes": 648755, + "errors": 0 + } + }, + "unfinished_calls": 718, + "resumed_calls": 718, + "unparsed_results": 0 + }, + "all_oracles_passed": true + } + }, + "deterministic_work": { + "replacement": { + "owner_authentication_bytes": 686158, + "owner_batches": 2, + "owner_candidate_routes": 2, + "owner_decoder_peak_bytes": 701, + "owner_fragments": 36, + "owner_physical_bytes": 1747572, + "owner_row_groups": 36, + "owner_rows": 49158, + "owner_snapshot_bytes": 686158, + "owner_snapshot_peak_bytes": 133060, + "owner_target_memberships": 8193, + "owner_target_peak": 8192, + "owner_targets": 8193, + "replacement_authentication_bytes": 686158, + "replacement_batches": 2, + "replacement_decoder_peak_bytes": 701, + "replacement_fragments": 36, + "replacement_physical_bytes": 1747572, + "replacement_rows": 49158, + "replacement_snapshot_bytes": 686158, + "replacement_snapshot_peak_bytes": 133060, + "replacement_target_peak": 8192, + "replacement_targets": 8193 + }, + "scalar": { + "owner_authentication_bytes": 840076, + "owner_batches": 4, + "owner_candidate_routes": 4, + "owner_decoder_peak_bytes": 701, + "owner_fragments": 68, + "owner_physical_bytes": 2327176, + "owner_row_groups": 68, + "owner_rows": 50192, + "owner_snapshot_bytes": 840076, + "owner_snapshot_peak_bytes": 14382, + "owner_target_memberships": 16386, + "owner_target_peak": 8192, + "owner_targets": 16386 + } + }, + "deterministic_budgets": { + "fixture_edges": 8193, + "fixture_nodes": 33, + "routes": 2, + "scalar_items": 2, + "scalar_owner_batches": 4, + "scalar_owner_target_memberships": 16386, + "replacement_owner_batches": 2, + "replacement_key_batches": 2, + "target_peak": 8192, + "scalar_owner_rows_max": 57351, + "scalar_owner_physical_bytes_max": 2621760, + "scalar_owner_authentication_bytes_max": 1048704, + "replacement_owner_and_key_each_rows_max": 49158, + "replacement_owner_and_key_each_physical_bytes_max": 2097408, + "replacement_owner_and_key_each_authentication_bytes_max": 1048704, + "replacement_each_snapshot_peak_bytes_max": 262144, + "decoder_peak_bytes_max": 4096, + "scope": "Owner probing and replacement-key reads only. Cumulative counters count repeated reads/items/batches again. Snapshot/decoder peaks are logical reader buffers, not full-publication memory or native RSS. Workspace admission and existing complete transaction costs are included only in whole-process observations." + }, + "correctness": { + "lifecycle": "33 and 4097 original nodes, 129 constructed edges, two routes plus two ordinary-created edges (including one initially propertyless). Scalar SET, map merge/replacement, nulling and REMOVE; exact identities/endpoints/types/nullable values; retained old stream; reopen, export/full verify/clean import and imported mutation/reopen.", + "qualified": "Mixed exploratory/qualified construction, first qualified property publication, reopen after mutations, same-statement qualified CREATE/SET/REMOVE/map merge, public relationship type through direct and WITH returns, portable verification/import and subsequent mutation.", + "refusal": "Ambiguous authenticated owners refuse SET and REMOVE before accumulation; exact graph-file inventory remains unchanged. Existing malformed-route, mask, null, pending-create and deletion regressions retained." + }, + "verification": { + "write_driver": "27 passed", + "rel_units": "244 passed after exact projection assertions", + "plan_goldens": "15 passed after reviewing two route-only diffs; all prior plan content preserved", + "public_boundary_tests": "4 passed; all nine uninstrumented and six separately instrumented resource runs passed their exact oracles", + "fast_gate": "make pre-push-fast passed; exact new public symbols recorded in property-overlay contract ledger", + "initial_ci": { + "run": 34551535010, + "head": "055e34574b7bf140a6ce138f8a503ce3257988c0", + "result": "FAILURE: 99 of 101 Bazel targets passed; rel unit and golden targets contained old plan-shape expectations. All independent Python/Node/platform/quality/policy lanes passed. Structural expectations corrected in later test-only commits." + }, + "local_full": "Full workspace stopped at old lowerer assertions. Corrected rel unit/golden runs passed. Remaining-crate --no-fail-fast census and corrected full pre-push encountered only known #1192 UnsupportedFilesystem in the hardcoded /tmp socket fixture. Neither full command is claimed green." + }, + "failed_and_superseded": [ + "Main c7a8cdd0 plus test-only 3983d97e reproduced successful constructed-edge SET followed by immediate duplicate-owner corruption refusal. That incomplete lifecycle cannot be a latency baseline.", + "Earlier qualified-route prototype omitted the provider-backed route column; a subsequent attempt failed reopen semantic metadata validation. Neither is counted as successful lifecycle evidence.", + "Passing pinned historical property counts directly into writer staging caused live-schema/physical-row-bound and mixed-generation errors. The uncommitted prototype was discarded; current workspace counts plus pinned missing-route declarations passed.", + "Broader relationship-container diagnostics failed on both main and the candidate. They do not establish an internal-route leak or general list/UNWIND conformance. Required direct and WITH-qualified value assertions pass.", + "Earlier frozen caa58d48/18d4b01f executables were superseded before resource acceptance. Only the stated 055e3457 executable supplies these observations.", + "Supplementary process-FD sampling did not produce accepted evidence: the first collector failed with a missing stat import; the corrected collector encountered PermissionError reading a child process /proc/PID/fd. Both runs are excluded. Reported disk peaks retain the explicit regular-file-only scope and exclude unlinked open files." + ], + "reproduce": [ + "Build cargo test --release -p graphforge-api --test permanent_storage_budgets --no-run --message-format=json with isolated CARGO_TARGET_DIR and native-volume TMPDIR. Copy the successful compiler-artifact executable and hash it before any subprocess.", + "For each named case, run /usr/bin/time -v FROZEN_EXECUTABLE --exact TEST --nocapture --test-threads=1 three times with separate private native TMPDIRs; rotate case order.", + "Run separate strace -f -qq -yy -s 0 -e trace=read,pread64,readv,preadv,write,pwrite64,writev,pwritev,copy_file_range,sendfile,fsync,fdatasync executions and sum successful return lengths. Retain unfinished/resumed and unparsed-result counts.", + "Separately sample all private TMPDIR regular files by unique device/inode as specified above; retain actual maximum sampling gap." + ], + "collector_sha256": { + "gf1224-measure.py": "ea8505062100c4452de86fa1cb2c8cab1c940c5ead869d186b269cb07df63fdb", + "gf1224-trace.py": "31d840b148aa527a4169a8772b66cc610d1859e247e875ab2386be29e90276bf", + "gf1224-run-measurements.py": "c3f456b24c63149f6f8f3084d8b039635a2abd8507d2828961de70dc19d794f9", + "gf1224-run-instrumented.py": "1eae0b5991b7680892ea00def73c497b0f153abce14ba60eda90bdfffb3641ea" + } +} diff --git a/tests/contracts/property-overlay-v1.json b/tests/contracts/property-overlay-v1.json index 35823f8f0..468785fd7 100644 --- a/tests/contracts/property-overlay-v1.json +++ b/tests/contracts/property-overlay-v1.json @@ -22,6 +22,7 @@ }, "exports": { "property_overlay": [ + "EdgeOwnerProbeWork", "resolve_existing_edge_property_owners", "AuthenticatedPropertyInventory", "PROPERTY_OVERLAY_FORMAT", "PROPERTY_OVERLAY_FORMAT_KEY", "PROPERTY_TOMBSTONE_FIELD", "PropertyFragmentId", "PropertyInventoryOpenMetrics", "PropertyOverlayLimits", @@ -33,6 +34,8 @@ "visit_authenticated_property_snapshots" ], "property_overlay_module": [ + "AuthenticatedPropertyInventory::capture", "AuthenticatedPropertyInventory::capture_workspace", + "EdgeOwnerProbeWork", "resolve_existing_edge_property_owners", "AuthenticatedPropertyInventory", "AuthenticatedPropertyInventory::edge_files", "AuthenticatedPropertyInventory::from_materialized_generation", "AuthenticatedPropertyInventory::from_materialized_inventory",