From 0235a257b8a241f979955ae18a5f8e2bcf47aab1 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 5 Sep 2026 01:33:05 +0000 Subject: [PATCH 1/3] fix(exec): bound recount and ordered traversal work (#1119) --- crates/graphforge-api/src/query_evidence.rs | 3 + .../graphforge-api/tests/fixed_hop_limit.rs | 473 ++++++++++++++++-- crates/graphforge-exec/src/adjacency.rs | 87 ++++ crates/graphforge-exec/src/demand.rs | 57 ++- crates/graphforge-exec/src/edge_count.rs | 321 ++++++++++++ crates/graphforge-exec/src/lib.rs | 16 + crates/graphforge-exec/src/ordered_one_hop.rs | 332 ++++++++++++ crates/graphforge-exec/src/ordered_two_hop.rs | 89 ++-- crates/graphforge-rel/src/expr.rs | 7 + crates/graphforge-storage/src/adjacency.rs | 117 ++++- .../src/ordinal_identity_v4.rs | 90 ++++ crates/graphforge-storage/src/parquet_scan.rs | 29 ++ 12 files changed, 1539 insertions(+), 82 deletions(-) create mode 100644 crates/graphforge-exec/src/edge_count.rs create mode 100644 crates/graphforge-exec/src/ordered_one_hop.rs diff --git a/crates/graphforge-api/src/query_evidence.rs b/crates/graphforge-api/src/query_evidence.rs index 2a6d95eb1..56d8bdb8d 100644 --- a/crates/graphforge-api/src/query_evidence.rs +++ b/crates/graphforge-api/src/query_evidence.rs @@ -51,6 +51,8 @@ pub struct QueryHopEvidence { pub input_rows: u64, /// Adjacency candidates examined before projection. pub candidates_generated: u64, + /// Adjacency degree probes, including empty destinations, in ordered shortcuts. + pub adjacency_rows_examined: u64, /// Rows emitted by the hop. pub rows_emitted: u64, /// Chunks served by destination-only projection. @@ -261,6 +263,7 @@ impl From for QueryExecutionEvidence { input_batches: hop.input_batches, input_rows: hop.input_rows, candidates_generated: hop.candidates_generated, + adjacency_rows_examined: hop.adjacency_rows_examined, rows_emitted: hop.rows_emitted, projected_chunks: hop.projected_chunks, projected_rows: hop.projected_rows, diff --git a/crates/graphforge-api/tests/fixed_hop_limit.rs b/crates/graphforge-api/tests/fixed_hop_limit.rs index 35376f7bc..a023120a3 100644 --- a/crates/graphforge-api/tests/fixed_hop_limit.rs +++ b/crates/graphforge-api/tests/fixed_hop_limit.rs @@ -56,10 +56,19 @@ fn measured_identity_query(forge: &GraphForge, query: &str) -> (Vec>, De 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 optimized_two_hop = query == ORDERED_TWO_HOP; - if optimized_two_hop { + let optimized_ordered = query == ORDERED_ONE_HOP || query == ORDERED_TWO_HOP; + if optimized_ordered { assert!(snapshot.sorts.is_empty(), "{snapshot:#?}"); - assert!(snapshot.operator_rss.is_empty(), "{snapshot:#?}"); + assert_eq!(snapshot.operator_rss.len(), 1, "{snapshot:#?}"); + assert_eq!( + snapshot.operator_rss[0].operator, + if query == ORDERED_ONE_HOP { + "ordered_one_hop" + } else { + "ordered_two_hop" + }, + "{snapshot:#?}" + ); } else { assert_eq!(snapshot.sorts.len(), 1, "{snapshot:#?}"); let sort = &snapshot.sorts[0]; @@ -343,10 +352,14 @@ fn stable_fixture_uuid(kind: u8, ordinal: usize) -> Uuid { } 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::>(); + generate_semantic_v4_graph_with_nodes(dir, nodes) +} + +fn generate_semantic_v4_graph_with_nodes(dir: &Path, nodes: Vec) -> Vec { + let workspace = TempDir::new().unwrap(); 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(); @@ -381,6 +394,332 @@ fn generate_semantic_v4_graph(dir: &Path) -> Vec { nodes } +#[test] +fn regression1094_property_free_shortcuts() { + let _guard = IO_GUARD.lock().unwrap(); + for ordinals in [[3, 1, 2], [1, 2, 3]] { + let dir = TempDir::new().unwrap(); + let workspace = TempDir::new().unwrap(); + let nodes = ordinals.map(|ordinal| stable_fixture_uuid(3, ordinal)); + let mut writer = + GraphWriter::open_at(workspace.path(), OntologyMode::Exploratory, TS).unwrap(); + for node in nodes { + writer.create_node(node, NODE_TYPE).unwrap(); + } + for (ordinal, (source, destination)) in + [(0, 1), (0, 1), (0, 0), (1, 2)].into_iter().enumerate() + { + writer + .create_edge( + stable_fixture_uuid(4, ordinal + 1), + "LINK", + &nodes[source], + &nodes[destination], + ) + .unwrap(); + } + writer.flush().unwrap(); + build_adjacency_index(workspace.path(), TS).unwrap(); + project_fixture::publish_graph_workspace_v4(dir.path(), workspace.path()); + let forge = open_forge(dir.path()); + let mut failures = Vec::new(); + for (query, expected) in [ + ("MATCH ()-[r]->() RETURN count(r) AS n", 4), + ("MATCH ()-[r]->() RETURN count(NULL) AS n", 0), + ("MATCH ()-[r]->() RETURN count(1) AS n", 4), + ("MATCH (a:Missing)-[r]->() RETURN count(r) AS n", 0), + ("MATCH ()-[r]-() RETURN count(r) AS n", 7), + ( + "MATCH ()-[r]->() RETURN count(CASE WHEN true THEN NULL ELSE 1 END) AS n", + 0, + ), + ( + "MATCH (a) WITH a LIMIT 0 MATCH (a)-[r]->(b) RETURN count(r) AS n", + 0, + ), + ] { + let plan = forge.explain(query).unwrap(); + let should_optimize = query == "MATCH ()-[r]->() RETURN count(r) AS n" + || query == "MATCH ()-[r]->() RETURN count(1) AS n"; + assert_eq!(plan.contains("EdgeCountExec"), should_optimize, "{plan}"); + let result = forge.execute(query).unwrap(); + let actual = int64_values(&result, "n"); + if actual != vec![expected] { + failures.push(format!("{query}: {actual:?} != {expected}; {plan}")); + } + } + for query in [ + "MATCH (a)-[r]->(b) RETURN b.node_uuid AS renamed ORDER BY renamed LIMIT 2", + "MATCH (a)-[r]->(b)-[s]->(c) RETURN c.node_uuid AS renamed ORDER BY renamed LIMIT 2", + ] { + let plan = forge.explain(query).unwrap(); + assert_eq!( + plan.contains("OrderedOneHopExec") || plan.contains("OrderedTwoHopPathCountExec"), + ordinals == [1, 2, 3], + "{plan}" + ); + let (result, evidence) = demand::capture(|| forge.execute(query)); + let result = result.unwrap(); + if ordinals == [1, 2, 3] && query.contains("[s]") { + // First destination has only its self-loop pair, which must be + // counted as examined even though disjointness rejects it. + assert_eq!( + evidence.hops.values().next().unwrap().candidates_generated, + 3 + ); + } + if result + .batches + .iter() + .any(|batch| batch.column_by_name("renamed").is_none()) + { + failures.push(format!("{query}: wrong batch schema; {plan}")); + } + let actual = result + .batches + .iter() + .flat_map(|batch| { + let values = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|row| values.value(row).to_vec()) + .collect::>() + }) + .collect::>(); + let destinations = if query.contains("[s]") { + [nodes[2], nodes[2], nodes[1], nodes[1]] + } else { + [nodes[1], nodes[1], nodes[0], nodes[2]] + }; + let mut expected = destinations.map(|uuid| uuid.as_bytes().to_vec()).to_vec(); + expected.sort(); + expected.truncate(2); + if actual != expected { + failures.push(format!("{query}: {actual:?} != {expected:?}; {plan}")); + } + } + if ordinals == [1, 2, 3] { + for (query, destinations) in [ + ( + "MATCH (a)-[r]->(b) RETURN b.node_uuid AS id ORDER BY id LIMIT 9223372036854775807", + [nodes[0], nodes[1], nodes[1], nodes[2]], + ), + ( + "MATCH (a)-[r]->(b)-[s]->(c) RETURN c.node_uuid AS id ORDER BY id LIMIT 9223372036854775807", + [nodes[1], nodes[1], nodes[2], nodes[2]], + ), + ] { + let plan = forge.explain(query).unwrap(); + assert!( + plan.contains("OrderedOneHopExec") + || plan.contains("OrderedTwoHopPathCountExec"), + "{plan}" + ); + let result = forge.execute(query).unwrap(); + assert_eq!( + fixed_binary_values(&result, "id"), + destinations.map(|uuid| uuid.as_bytes().to_vec()) + ); + } + } + assert!(failures.is_empty(), "{}", failures.join("\n")); + } +} + +#[test] +fn regression1094_recount_and_ordered_frontier_semantics() { + let _guard = IO_GUARD.lock().unwrap(); + let dir = TempDir::new().unwrap(); + let nodes = generate_semantic_v4_graph_with_nodes( + dir.path(), + [3, 1, 2] + .map(|ordinal| stable_fixture_uuid(3, ordinal)) + .to_vec(), + ); + let forge = open_forge(dir.path()); + let mut failures = Vec::new(); + for (query, expected) in [ + ("MATCH ()-[r:LINK]->() RETURN count(r) AS n", 4), + ("MATCH ()-[r:LINK]->() RETURN count(NULL) AS n", 0), + ("MATCH ()-[r:LINK]->() RETURN count(1) AS n", 4), + ( + "MATCH (a)-[r:LINK]->() WHERE a.name = 'B' RETURN count(r) AS n", + 1, + ), + ("MATCH (a:Missing)-[r:LINK]->() RETURN count(r) AS n", 0), + ("MATCH ()-[r:LINK]-() RETURN count(r) AS n", 7), + ] { + let plan = forge.explain(query).unwrap(); + match forge.execute(query) { + Ok(result) => { + let actual = int64_values(&result, "n"); + if actual != vec![expected] { + failures.push(format!("{query}: {actual:?} != {expected}; {plan}")); + } + } + Err(error) => failures.push(format!("{query}: {error}; {plan}")), + } + } + for (query, expected) in [ + ( + "MATCH (a)-[:LINK]->(b) RETURN b.node_uuid AS renamed ORDER BY renamed LIMIT 2", + vec![nodes[1], nodes[1]], + ), + ( + "MATCH (a)-[:LINK]->(b) WHERE a.name = 'B' RETURN b.node_uuid AS renamed ORDER BY renamed LIMIT 2", + vec![nodes[2]], + ), + ( + "MATCH (a:Missing)-[:LINK]->(b) RETURN b.node_uuid AS renamed ORDER BY renamed LIMIT 2", + vec![], + ), + ( + "MATCH (a)-[:LINK]->(b)-[:LINK]->(c) RETURN c.node_uuid AS renamed ORDER BY renamed LIMIT 2", + vec![nodes[1], nodes[1]], + ), + ( + "MATCH (a)-[:LINK]->(b)-[:LINK]->(c) WHERE a.name = 'B' RETURN c.node_uuid AS renamed ORDER BY renamed LIMIT 2", + vec![], + ), + ] { + let plan = forge.explain(query).unwrap(); + match forge.execute(query) { + Ok(result) => { + if result + .batches + .iter() + .any(|batch| batch.column_by_name("renamed").is_none()) + { + failures.push(format!("{query}: incorrect batch schema; {plan}")); + } + let actual = result + .batches + .iter() + .flat_map(|batch| { + let values = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|row| values.value(row).to_vec()) + .collect::>() + }) + .collect::>(); + let expected = expected + .iter() + .map(|uuid| uuid.as_bytes().to_vec()) + .collect::>(); + if actual != expected { + failures.push(format!("{query}: {actual:?} != {expected:?}; {plan}")); + } + } + Err(error) => failures.push(format!("{query}: {error}; {plan}")), + } + } + assert!(failures.is_empty(), "{}", failures.join("\n")); +} + +#[test] +fn regression1094_identity_lookup_limit_is_chunked() { + let _guard = IO_GUARD.lock().unwrap(); + let dir = TempDir::new().unwrap(); + let limit = graphforge_storage::V4OrdinalIdentityLimits::default().max_requested + 1; + generate_graph(dir.path(), limit + 1, 1, true); + let forge = open_forge(dir.path()); + let query = + format!("MATCH (a)-[r]->(b) RETURN b.node_uuid AS renamed ORDER BY renamed LIMIT {limit}"); + assert!(forge.explain(&query).unwrap().contains("OrderedOneHopExec")); + let (result, evidence) = demand::capture(|| forge.execute(&query)); + let expected = (0..limit) + .map(|index| fixture_node_uuid(index).as_bytes().to_vec()) + .collect::>(); + assert_eq!(fixed_binary_values(&result.unwrap(), "renamed"), expected); + let hop = evidence.hops.values().next().unwrap(); + assert_eq!(hop.projected_chunks, 2); + assert_eq!(hop.projected_rows, limit as u64); + assert_eq!(hop.candidates_generated, limit as u64); +} + +#[test] +fn regression1094_sharded_hubs_keep_one_and_two_hop_work_bounded() { + let _guard = IO_GUARD.lock().unwrap(); + for degree in [16_384, 32_768] { + let dir = TempDir::new().unwrap(); + let workspace = TempDir::new().unwrap(); + let nodes = [stable_fixture_uuid(3, 1), stable_fixture_uuid(3, 2)]; + let mut writer = + GraphWriter::open_at(workspace.path(), OntologyMode::Exploratory, TS).unwrap(); + for node in nodes { + writer.create_node(node, NODE_TYPE).unwrap(); + } + writer.flush().unwrap(); + for start in (0..degree * 2).step_by(WRITE_WINDOW) { + let mut writer = + GraphWriter::open_at(workspace.path(), OntologyMode::Exploratory, TS).unwrap(); + writer.register_existing_endpoints(&nodes).unwrap(); + for index in start..(start + WRITE_WINDOW).min(degree * 2) { + let source = usize::from(index >= degree); + writer + .create_edge( + stable_fixture_uuid(4, index + 1), + "LINK", + &nodes[source], + &nodes[1 - source], + ) + .unwrap(); + } + writer.flush().unwrap(); + } + let mut options = graphforge_storage::adjacency::AdjacencyBuildOptions::default(); + options.shard_max_edges = 4_096; + let (_, build) = graphforge_storage::adjacency::build_adjacency_index_into_with_metrics( + workspace.path(), + workspace.path(), + TS, + &options, + || Ok(()), + ) + .unwrap(); + assert!(build.peak_shard_edges <= 4_096); + project_fixture::publish_graph_workspace_v4(dir.path(), workspace.path()); + let forge = open_forge(dir.path()); + for query in [ORDERED_ONE_HOP, ORDERED_TWO_HOP] { + let plan = forge.explain(query).unwrap(); + assert!( + !plan.contains("SortExec") && !plan.contains("ExpandExec"), + "{plan}" + ); + let (result, evidence) = demand::capture(|| forge.execute(query)); + assert_eq!( + fixed_binary_values(&result.unwrap(), "id"), + vec![nodes[0].as_bytes().to_vec(); LIMIT] + ); + let hop = evidence.hops.values().next().unwrap(); + assert_eq!(hop.candidates_generated, LIMIT as u64); + assert!(hop.adjacency_rows_examined <= 2); + assert_eq!( + (hop.input_batches, hop.input_rows, hop.projected_rows), + (0, 0, 1) + ); + assert_eq!(evidence.operator_rss.len(), 1); + assert!(evidence.operator_rss[0].before_bytes > 0 || !cfg!(target_os = "linux")); + println!( + "hub degree={degree} query={query} rss={:?}", + evidence.operator_rss + ); + } + let reopened = open_forge(dir.path()); + let result = reopened + .execute("MATCH ()-[r]->() RETURN count(r) AS total") + .unwrap(); + assert_eq!(int64_values(&result, "total"), vec![(degree * 2) as i64]); + } +} + /// 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 @@ -613,11 +952,12 @@ fn scale_fixture_uses_bounded_bulk_publications() { 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}"); + assert!(plan.contains("OrderedOneHopExec"), "{plan}"); io_stats::reset(); - let result = forge.execute(ORDERED_ONE_HOP).unwrap(); + let (result, demand) = demand::capture(|| forge.execute(ORDERED_ONE_HOP)); + let result = result.unwrap(); assert_eq!(result.stats.rows_produced, LIMIT as u64); + assert!(demand.hops.values().all(|hop| hop.identity_read_calls > 0)); assert_projected_identity_io(&io_stats::snapshot()); } @@ -697,10 +1037,11 @@ fn run_ordered_projection_scale(nodes: usize) -> (Vec>, DemandSnapshot) 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}"); + assert!(plan.contains("OrderedOneHopExec"), "{plan}"); + assert!( + !plan.contains("ExpandExec") && !plan.contains("SortExec"), + "{plan}" + ); io_stats::reset(); demand::reset(); @@ -726,14 +1067,63 @@ fn run_ordered_projection_scale(nodes: usize) -> (Vec>, DemandSnapshot) 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_chunks, 1, "{snapshot:#?}"); + assert_eq!(hop.projected_rows, LIMIT.div_ceil(FAN_OUT) as u64); + assert_eq!(hop.candidates_generated, LIMIT as u64); + // Bulk construction assigns IDs from one; the first degree probe is empty. + assert_eq!( + hop.adjacency_rows_examined, + LIMIT.div_ceil(FAN_OUT) as u64 + 1 + ); + assert_eq!(hop.input_batches, 0); + assert_eq!(hop.input_rows, 0); + assert!(snapshot.sorts.is_empty()); + assert_eq!(snapshot.operator_rss.len(), 1); + assert_eq!(snapshot.operator_rss[0].operator, "ordered_one_hop"); 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 recount_query = "MATCH ()-[r]->() RETURN count(r) AS total"; + assert!( + forge + .explain(recount_query) + .unwrap() + .contains("EdgeCountExec") + ); + io_stats::reset(); + let (recount, recount_demand) = demand::capture(|| forge.execute(recount_query)); + assert_eq!( + int64_values(&recount.unwrap(), "total"), + vec![(nodes * FAN_OUT) as i64] + ); + let recount_io = io_stats::snapshot(); + assert_eq!( + recount_io.edge_full_reads + + recount_io.edge_filtered_reads + + recount_io.node_full_reads + + recount_io.node_filtered_reads, + 0 + ); + let count_hop = recount_demand.hops.values().next().unwrap(); + assert_eq!( + ( + count_hop.candidates_generated, + count_hop.input_batches, + count_hop.input_rows, + count_hop.rows_emitted + ), + (0, 0, 0, 1) + ); + assert_eq!(recount_demand.operator_rss.len(), 1); + assert_eq!(recount_demand.operator_rss[0].operator, "edge_count"); + println!( + "recount nodes={nodes} rss={:?}", + recount_demand.operator_rss + ); + let repeated = forge.execute(ORDERED_ONE_HOP).unwrap(); assert_eq!( fixed_binary_values(&repeated, "id"), @@ -759,7 +1149,7 @@ fn destination_uuid_projection_uses_authenticated_legacy_hydration_without_v4_au } #[test] -fn ordered_destination_uuid_projection_is_exact_and_linear_at_1x_2x_4x() { +fn ordered_destination_uuid_projection_is_exact_and_bounded_at_1x_2x_4x() { let _guard = IO_GUARD.lock().unwrap(); let mut work = Vec::new(); for nodes in [4_096, 8_192, 16_384] { @@ -767,30 +1157,15 @@ fn ordered_destination_uuid_projection_is_exact_and_linear_at_1x_2x_4x() { let hop = snapshot.hops.values().next().unwrap(); work.push(( hop.projected_rows, - hop.identity_bytes_read, - hop.identity_read_calls, - hop.identity_revalidation_calls, - snapshot.sorts[0].fetch, - snapshot.sorts[0].retained_bytes, + hop.candidates_generated, + hop.adjacency_rows_examined, )); - } - for pair in work.windows(2) { - let (prior_rows, prior_bytes, prior_calls, prior_revalidation, prior_fetch, prior_retained) = - pair[0]; - let (next_rows, next_bytes, next_calls, next_revalidation, next_fetch, next_retained) = - 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_eq!((prior_fetch, next_fetch), (Some(LIMIT), Some(LIMIT))); - assert_eq!((prior_retained, next_retained), (0, 0)); - assert!( - next_revalidation <= prior_revalidation * 2 + 2, - "session authentication must be linear in retained artifacts: {work:?}" + println!( + "ordered nodes={nodes} identity_bytes={} identity_calls={} rss={:?}", + hop.identity_bytes_read, hop.identity_read_calls, snapshot.operator_rss ); } + assert!(work.windows(2).all(|pair| pair[0] == pair[1]), "{work:?}"); } #[test] @@ -818,13 +1193,20 @@ fn ordinary_streaming_sink_exposes_deterministic_query_evidence() { .unwrap(); assert_eq!(receipt.evidence.contract, "graphforge-query-evidence/1"); assert_eq!(receipt.evidence.hops.len(), 1); - if ordinal == 0 { - assert_eq!(receipt.evidence.sorts.len(), 1); - assert_eq!(receipt.evidence.sorts[0].fetch_rows, Some(LIMIT)); - assert_eq!(receipt.evidence.sorts[0].retained_bytes, 0); - } else { - assert!(receipt.evidence.sorts.is_empty()); - } + assert!(receipt.evidence.sorts.is_empty()); + assert_eq!(receipt.evidence.operator_rss.len(), 1); + assert_eq!( + receipt.evidence.operator_rss[0].operator, + if ordinal == 0 { + "ordered_one_hop" + } else { + "ordered_two_hop" + } + ); + println!( + "ordered evidence {}", + serde_json::to_string(&receipt.evidence).unwrap() + ); assert!( receipt.evidence.memory_reserved_after <= receipt @@ -1209,7 +1591,6 @@ fn selective_filter_tops_up_without_crossing_blockers() { for query in [ "MATCH ()-[r]->(b) RETURN b.node_id AS id ORDER BY id DESC LIMIT 5", "MATCH ()-[r]->(b) RETURN DISTINCT b.node_id AS id LIMIT 5", - "MATCH ()-[r]->() RETURN count(r) AS total LIMIT 1", ] { let plan = forge.explain(query).unwrap(); assert!(plan.contains("demand_batch=all"), "{plan}"); @@ -1217,6 +1598,12 @@ fn selective_filter_tops_up_without_crossing_blockers() { assert!(!plan.contains("DemandGuardExec"), "{plan}"); } + let recount = forge + .explain("MATCH ()-[r]->() RETURN count(r) AS total LIMIT 1") + .unwrap(); + assert!(recount.contains("EdgeCountExec"), "{recount}"); + assert!(!recount.contains("ExpandExec"), "{recount}"); + let unlimited = forge .explain("MATCH ()-[r1]->()-[r2]->() RETURN r1, r2") .unwrap(); diff --git a/crates/graphforge-exec/src/adjacency.rs b/crates/graphforge-exec/src/adjacency.rs index e53abe865..2459a2bdf 100644 --- a/crates/graphforge-exec/src/adjacency.rs +++ b/crates/graphforge-exec/src/adjacency.rs @@ -273,6 +273,47 @@ impl Adjacency { self.inner.is_empty() } + /// Exclusive node ID extent. CSR and overlay views read retained metadata; + /// scan-built maps inspect their keys without traversing edge entries. + pub(crate) fn node_extent(&self) -> u64 { + self.inner.node_extent() + } + + /// Count neighbors without materializing a sharded or undirected payload. + pub(crate) fn degree(&self, node_id: u64) -> Result { + self.inner.degree(node_id) + } + + /// Copy a bounded chunk of one directed row, without materializing the rest. + pub(crate) fn neighbor_chunk( + &self, + node_id: u64, + skip: usize, + limit: usize, + ) -> Result, GfError> { + match &self.inner { + AdjacencyInner::Sharded(base) => base.row_chunk(node_id, skip, limit), + AdjacencyInner::ShardedOverlay { base, replaced, .. } => match replaced.get(&node_id) { + Some(row) => Ok(row.iter().copied().skip(skip).take(limit).collect()), + None => base.row_chunk(node_id, skip, limit), + }, + AdjacencyInner::Undirected { .. } => Err(GfError::Execution( + "bounded ordered traversal requires a directed adjacency view".into(), + )), + _ => Ok(self + .neighbors(node_id) + .iter() + .skip(skip) + .take(limit) + .collect()), + } + } + + /// Total adjacency entries, including both orientations in undirected views. + pub fn edge_entry_count(&self) -> Result { + self.inner.edge_entry_count() + } + /// Visit every node that may have adjacency entries. /// /// CSR-native backings yield dense `0..node_extent` ids (empty rows @@ -285,6 +326,52 @@ impl Adjacency { } impl AdjacencyInner { + fn edge_entry_count(&self) -> Result { + Ok(match self { + Self::Empty => 0, + Self::Map(map) => map.values().map(|row| row.len() as u64).sum(), + Self::Csr(csr) => csr.edge_count(), + Self::Sharded(csr) => csr.edge_count(), + Self::ShardedOverlay { base, replaced, .. } => { + let mut total = base.edge_count(); + for (node, row) in replaced { + total = total - base.row_len(*node)? + row.len() as u64; + } + total + } + Self::Overlay(overlay) => { + let mut total = overlay.base.edge_count(); + for (node, row) in &overlay.replaced { + total = total - overlay.base.row(*node).len() as u64 + row.len() as u64; + } + total + } + Self::Undirected { out, inbound } => out + .edge_entry_count()? + .saturating_add(inbound.edge_entry_count()?), + }) + } + + fn degree(&self, node_id: u64) -> Result { + Ok(match self { + Self::Empty => 0, + Self::Map(map) => map.get(&node_id).map_or(0, |row| row.len() as u64), + Self::Csr(csr) => csr.row(node_id).len() as u64, + Self::Sharded(csr) => csr.row_len(node_id)?, + Self::ShardedOverlay { base, replaced, .. } => match replaced.get(&node_id) { + Some(row) => row.len() as u64, + None => base.row_len(node_id)?, + }, + Self::Overlay(overlay) => match overlay.row(node_id) { + graphforge_storage::adjacency_delta::OverlayRow::Base(row) => row.len() as u64, + graphforge_storage::adjacency_delta::OverlayRow::Replaced(row) => row.len() as u64, + }, + Self::Undirected { out, inbound } => out + .degree(node_id)? + .saturating_add(inbound.degree(node_id)?), + }) + } + fn is_empty(&self) -> bool { match self { Self::Empty => true, diff --git a/crates/graphforge-exec/src/demand.rs b/crates/graphforge-exec/src/demand.rs index 6111b9906..0de2553e1 100644 --- a/crates/graphforge-exec/src/demand.rs +++ b/crates/graphforge-exec/src/demand.rs @@ -21,6 +21,7 @@ use datafusion::execution::TaskContext; use datafusion::physical_expr::utils::collect_columns; use datafusion::physical_expr::{PhysicalExpr, ScalarFunctionExpr}; use datafusion::physical_optimizer::PhysicalOptimizerRule; +use datafusion::physical_plan::aggregates::AggregateExec; use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec; use datafusion::physical_plan::filter::FilterExec; use datafusion::physical_plan::limit::{GlobalLimitExec, LocalLimitExec}; @@ -45,6 +46,8 @@ pub struct HopSnapshot { pub input_rows: u64, /// Adjacency candidates selected before record materialization. pub candidates_generated: u64, + /// Destination adjacency rows probed by an ordered shortcut, including empty rows. + pub adjacency_rows_examined: u64, /// Rows emitted by the hop. pub rows_emitted: u64, /// Edge filtered reads scheduled. @@ -451,6 +454,12 @@ pub(crate) fn record_input(epoch: u64, edge_var: u32, rows: usize) { }); } +pub(crate) fn record_adjacency_row(epoch: u64, edge_var: u32) { + with_hop(epoch, edge_var, |hop| { + hop.adjacency_rows_examined += 1; + }); +} + pub(crate) fn record_candidates(epoch: u64, edge_var: u32, rows: usize) { with_hop(epoch, edge_var, |hop| { hop.candidates_generated += rows as u64; @@ -758,6 +767,8 @@ impl PhysicalOptimizerRule for FixedHopDemandRule { plan }; let plan = crate::ordered_two_hop::try_rewrite_ordered_two_hop(plan)?; + let plan = crate::ordered_one_hop::try_rewrite_ordered_one_hop(plan)?; + let plan = crate::edge_count::try_rewrite_edge_count(plan)?; let Some(terminal) = find_terminal_demand(&plan) else { // RSS probes are diagnostics only; skip when this thread is not the // capturing session so concurrent executes cannot append operator_rss. @@ -827,6 +838,21 @@ fn instrument_operator_rss( }; let operator = if rebuilt.downcast_ref::().is_some() { Some("expand") + } else if rebuilt + .downcast_ref::() + .is_some() + { + Some("edge_count") + } else if rebuilt + .downcast_ref::() + .is_some() + { + Some("ordered_one_hop") + } else if rebuilt + .downcast_ref::() + .is_some() + { + Some("ordered_two_hop") } else if rebuilt.downcast_ref::().is_some() { Some("sort") } else { @@ -929,9 +955,9 @@ impl ExecutionPlan for RssProbeExec { partition: usize, context: Arc, ) -> Result { - let stream = self.input.execute(partition, context)?; let epoch = self.capture_epoch; let before = current_rss_bytes(); + let stream = self.input.execute(partition, context)?; { let mut state = CAPTURE .lock() @@ -959,6 +985,7 @@ impl ExecutionPlan for RssProbeExec { } } } + record_operator_rss(epoch, self.ordinal, false); Ok(Box::pin(RssProbeStream { schema: stream.schema(), inner: stream, @@ -1104,6 +1131,34 @@ fn rewrite_materialization( return Ok(expand.with_required_output(mask)); } + if let Some(aggregate) = plan.downcast_ref::() { + let mut child_required = BTreeSet::new(); + if aggregate.group_expr().is_empty() + && matches!( + aggregate.mode(), + datafusion::physical_plan::aggregates::AggregateMode::Single + | datafusion::physical_plan::aggregates::AggregateMode::SinglePartitioned + | datafusion::physical_plan::aggregates::AggregateMode::Partial + ) + { + for aggr in aggregate.aggr_expr() { + for expr in aggr.expressions() { + collect_expr_columns(&expr, &mut child_required); + } + for order in aggr.order_bys() { + collect_expr_columns(&order.expr, &mut child_required); + } + } + for filter in aggregate.filter_expr().iter().flatten() { + collect_expr_columns(filter, &mut child_required); + } + } else { + child_required = (0..aggregate.input().schema().fields().len()).collect(); + } + let child = rewrite_materialization(Arc::clone(aggregate.input()), &child_required)?; + return plan.with_new_children(vec![child]); + } + let children = plan.children(); if children.len() != 1 { let rewritten = children diff --git a/crates/graphforge-exec/src/edge_count.rs b/crates/graphforge-exec/src/edge_count.rs new file mode 100644 index 000000000..84ca73752 --- /dev/null +++ b/crates/graphforge-exec/src/edge_count.rs @@ -0,0 +1,321 @@ +//! CSR/catalog edge-count short-circuit for unconstrained `count(r)` (#1094). +//! +//! `MATCH ()-[r]->() RETURN count(r)` otherwise expands every adjacency entry +//! into Arrow batches before aggregating. That retains process RSS ~linear in +//! edge count and fails the progressive S18→S19 plateau gate. When the physical +//! plan is a global nonnull literal / compiler row-marker count over a single +//! unconstrained outward Expand (including a validated Partial/Final pair), +//! replace it with the adjacency view's edge-entry count. + +use std::fmt; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use arrow::array::{ArrayRef, Int64Array}; +use arrow::datatypes::SchemaRef; +use datafusion::common::{DataFusionError, Result}; +use datafusion::execution::TaskContext; +use datafusion::physical_expr::ScalarFunctionExpr; +use datafusion::physical_expr::expressions::{Column, Literal}; +use datafusion::physical_plan::Partitioning; +use datafusion::physical_plan::aggregates::{AggregateExec, AggregateMode}; +use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec; +use datafusion::physical_plan::limit::GlobalLimitExec; +use datafusion::physical_plan::projection::ProjectionExec; +use datafusion::physical_plan::repartition::RepartitionExec; +use datafusion::physical_plan::{ + DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, RecordBatchStream, + SendableRecordBatchStream, +}; +use futures::Stream; +use graphforge_ir::Direction; + +use crate::ExpandExec; +use crate::adjacency::AdjacencyProvider; +use crate::demand; + +/// Rewrite a global edge `count` over one Expand into an adjacency edge count. +pub(crate) fn try_rewrite_edge_count( + plan: Arc, +) -> Result> { + let Some(spec) = detect_edge_count(&plan) else { + return Ok(plan); + }; + let replacement = Arc::new(EdgeCountExec::new(spec)) as Arc; + replace_peeled(&plan, replacement) +} + +fn replace_peeled( + plan: &Arc, + replacement: Arc, +) -> Result> { + if let Some(limit) = plan.downcast_ref::() { + return Arc::clone(plan) + .with_new_children(vec![replace_peeled(limit.input(), replacement)?]); + } + if let Some(coalesce) = plan.downcast_ref::() { + return Arc::clone(plan) + .with_new_children(vec![replace_peeled(coalesce.input(), replacement)?]); + } + if let Some(repartition) = plan.downcast_ref::() { + return Arc::clone(plan) + .with_new_children(vec![replace_peeled(repartition.input(), replacement)?]); + } + Ok(replacement) +} + +struct EdgeCountSpec { + schema: SchemaRef, + props: Arc, + rel_type_name: String, + direction: Direction, + provider: Arc, +} + +fn peel_transport(plan: &Arc) -> Arc { + if let Some(limit) = plan.downcast_ref::() { + return peel_transport(limit.input()); + } + if let Some(coalesce) = plan.downcast_ref::() { + return peel_transport(coalesce.input()); + } + if let Some(repartition) = plan.downcast_ref::() { + return peel_transport(repartition.input()); + } + Arc::clone(plan) +} + +fn peel_expand_input(plan: &Arc) -> Arc { + if let Some(coalesce) = plan.downcast_ref::() { + return peel_expand_input(coalesce.input()); + } + if let Some(repartition) = plan.downcast_ref::() { + return peel_expand_input(repartition.input()); + } + if let Some(projection) = plan.downcast_ref::() { + return peel_expand_input(projection.input()); + } + Arc::clone(plan) +} + +/// Trace only row-preserving column projections/transports to an actual complete +/// topology scan. A node name or absence of Filter alone is not sufficient. +pub(crate) fn has_complete_frontier(expand: &ExpandExec) -> bool { + fn trace(plan: &Arc, column: usize) -> bool { + if let Some(projection) = plan.downcast_ref::() { + return projection + .expr() + .get(column) + .and_then(|expr| expr.expr.downcast_ref::()) + .is_some_and(|column| trace(projection.input(), column.index())); + } + if let Some(coalesce) = plan.downcast_ref::() { + return trace(coalesce.input(), column); + } + if let Some(repartition) = plan.downcast_ref::() { + return trace(repartition.input(), column); + } + graphforge_storage::parquet_scan::is_complete_node_id_scan(plan.as_ref(), column) + } + expand.fetch.is_none() && trace(&expand.input, expand.src_col_idx) +} + +fn is_row_count(aggregate: &AggregateExec) -> bool { + aggregate.group_expr().is_empty() + && !aggregate.aggr_expr().is_empty() + && aggregate.filter_expr().iter().all(Option::is_none) + && aggregate.aggr_expr().iter().all(|expr| { + if expr + .fun() + .inner() + .downcast_ref::() + .is_none() + || expr.is_distinct() + || !expr.order_bys().is_empty() + { + return false; + } + expr.expressions().iter().all(|arg| { + if let Some(literal) = arg.downcast_ref::() { + return !literal.value().is_null(); + } + arg.downcast_ref::() + .is_some_and(|function| { + graphforge_rel::expr::is_cypher_row_marker(function.fun()) + && function.args().len() == 1 + && function.args()[0].downcast_ref::().is_some() + }) + }) + }) +} + +fn detect_edge_count(plan: &Arc) -> Option { + let plan = peel_transport(plan); + let aggregate = plan.downcast_ref::()?; + if !is_row_count(aggregate) { + return None; + } + let input = match aggregate.mode() { + AggregateMode::Single => Arc::clone(aggregate.input()), + AggregateMode::Final => { + let input = peel_expand_input_without_projection(aggregate.input()); + let partial = input.downcast_ref::()?; + if *partial.mode() != AggregateMode::Partial + || !is_row_count(partial) + || aggregate.aggr_expr() != partial.aggr_expr() + { + return None; + } + Arc::clone(partial.input()) + } + _ => return None, + }; + let expand_plan = peel_expand_input(&input); + let expand = expand_plan.downcast_ref::()?; + if expand.direction() != Direction::Out || !has_complete_frontier(expand) { + return None; + } + Some(EdgeCountSpec { + schema: plan.schema(), + props: Arc::new( + plan.properties() + .as_ref() + .clone() + .with_partitioning(Partitioning::UnknownPartitioning(1)), + ), + rel_type_name: expand.rel_type_name().to_owned(), + direction: expand.direction(), + provider: Arc::clone(expand.provider()), + }) +} + +fn peel_expand_input_without_projection(plan: &Arc) -> Arc { + if let Some(coalesce) = plan.downcast_ref::() { + return peel_expand_input_without_projection(coalesce.input()); + } + if let Some(repartition) = plan.downcast_ref::() { + return peel_expand_input_without_projection(repartition.input()); + } + Arc::clone(plan) +} + +pub(crate) struct EdgeCountExec { + schema: SchemaRef, + props: Arc, + rel_type_name: String, + direction: Direction, + provider: Arc, + capture_epoch: u64, +} + +impl EdgeCountExec { + fn new(spec: EdgeCountSpec) -> Self { + Self { + schema: spec.schema, + props: spec.props, + rel_type_name: spec.rel_type_name, + direction: spec.direction, + provider: spec.provider, + capture_epoch: demand::stamp_capture_epoch().unwrap_or(0), + } + } +} + +impl fmt::Debug for EdgeCountExec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("EdgeCountExec") + .field("rel", &self.rel_type_name) + .finish_non_exhaustive() + } +} + +impl DisplayAs for EdgeCountExec { + fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "EdgeCountExec: rel={}, dir={:?}", + self.rel_type_name, self.direction + ) + } +} + +impl ExecutionPlan for EdgeCountExec { + fn name(&self) -> &str { + "EdgeCountExec" + } + + fn properties(&self) -> &Arc { + &self.props + } + + fn children(&self) -> Vec<&Arc> { + Vec::new() + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> Result> { + if children.is_empty() { + Ok(self) + } else { + Err(DataFusionError::Internal( + "EdgeCountExec has no children".into(), + )) + } + } + + fn execute( + &self, + partition: usize, + _context: Arc, + ) -> Result { + if partition != 0 { + return Err(DataFusionError::Internal(format!( + "EdgeCountExec only has partition 0, got {partition}" + ))); + } + let adjacency = self + .provider + .adjacency(&self.rel_type_name, self.direction) + .map_err(|error| DataFusionError::External(Box::new(error)))?; + let count = adjacency + .edge_entry_count() + .map_err(|error| DataFusionError::External(Box::new(error)))?; + // No edge candidates are materialized by the cardinality query. + demand::record_candidates(self.capture_epoch, 0, 0); + demand::record_emitted(self.capture_epoch, 0, 1); + let mut columns: Vec = Vec::with_capacity(self.schema.fields().len()); + for _ in self.schema.fields() { + let value = i64::try_from(count) + .map_err(|_| DataFusionError::Internal("edge count exceeds Int64".into()))?; + columns.push(Arc::new(Int64Array::from(vec![value]))); + } + let batch = arrow::record_batch::RecordBatch::try_new(Arc::clone(&self.schema), columns) + .map_err(|error| DataFusionError::ArrowError(Box::new(error), None))?; + Ok(Box::pin(EdgeCountStream { + schema: Arc::clone(&self.schema), + batch: Some(batch), + })) + } +} + +struct EdgeCountStream { + schema: SchemaRef, + batch: Option, +} + +impl Stream for EdgeCountStream { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(self.batch.take().map(Ok)) + } +} + +impl RecordBatchStream for EdgeCountStream { + fn schema(&self) -> SchemaRef { + Arc::clone(&self.schema) + } +} diff --git a/crates/graphforge-exec/src/lib.rs b/crates/graphforge-exec/src/lib.rs index 30a1bd396..275699aa5 100644 --- a/crates/graphforge-exec/src/lib.rs +++ b/crates/graphforge-exec/src/lib.rs @@ -189,6 +189,8 @@ pub(crate) mod algorithm_similar_knn; pub(crate) mod algorithm_weighted_undirected; #[doc(hidden)] pub mod demand; +mod edge_count; +mod ordered_one_hop; mod ordered_two_hop; pub use adjacency::{ Adjacency, AdjacencyBacking, AdjacencyProvider, AdjacencyStatus, PersistentAdjacencyProvider, @@ -3205,6 +3207,20 @@ pub(crate) struct V4OrdinalIdentitySession { } impl V4OrdinalIdentitySession { + pub(crate) fn max_requested_ids(&self) -> usize { + self.handle + .lock() + .expect("ordinal identity handle poisoned") + .max_requested_ids() + } + + pub(crate) fn uuid_order_matches_ordinals(&self) -> bool { + self.handle + .lock() + .expect("ordinal identity handle poisoned") + .uuid_order_matches_ordinals() + } + pub(crate) fn lookup_node_uuids( &self, requested: &[u64], diff --git a/crates/graphforge-exec/src/ordered_one_hop.rs b/crates/graphforge-exec/src/ordered_one_hop.rs new file mode 100644 index 000000000..58888909f --- /dev/null +++ b/crates/graphforge-exec/src/ordered_one_hop.rs @@ -0,0 +1,332 @@ +//! Order-aware one-hop emission for `ORDER BY destination_uuid LIMIT K` (#1094). +//! +//! Canonical ladder one-hop queries are identity-only expands into TopK Sort. +//! Streaming every adjacency entry before the limit retains process RSS ~linear +//! in edge count. Mirror the two-hop rewrite: walk destinations in node-id +//! order (UUID order for monotonic ordinal identity), emit each destination's +//! inbound multiplicity until the limit is satisfied, then stop. + +use std::fmt; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use arrow::array::{ArrayRef, FixedSizeBinaryBuilder}; +use arrow::datatypes::SchemaRef; +use datafusion::common::{DataFusionError, Result}; +use datafusion::execution::TaskContext; +use datafusion::physical_expr::expressions::Column; +use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec; +use datafusion::physical_plan::limit::GlobalLimitExec; +use datafusion::physical_plan::projection::ProjectionExec; +use datafusion::physical_plan::repartition::RepartitionExec; +use datafusion::physical_plan::sorts::sort::SortExec; +use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; +use datafusion::physical_plan::{ + DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties, RecordBatchStream, + SendableRecordBatchStream, +}; +use futures::Stream; +use graphforge_ir::Direction; + +use crate::ExpandExec; +use crate::adjacency::AdjacencyProvider; +use crate::demand; + +/// Rewrite ordered one-hop identity-only plans when matched. +pub fn try_rewrite_ordered_one_hop(plan: Arc) -> Result> { + let Some(spec) = detect_ordered_one_hop(&plan) else { + return Ok(plan); + }; + let replacement = Arc::new(OrderedOneHopExec::new(spec)) as Arc; + replace_peeled_projection(&plan, replacement) +} + +fn replace_peeled_projection( + plan: &Arc, + replacement: Arc, +) -> Result> { + if let Some(limit) = plan.downcast_ref::() { + return Arc::clone(plan) + .with_new_children(vec![replace_peeled_projection(limit.input(), replacement)?]); + } + if let Some(coalesce) = plan.downcast_ref::() { + return Arc::clone(plan).with_new_children(vec![replace_peeled_projection( + coalesce.input(), + replacement, + )?]); + } + if let Some(repartition) = plan.downcast_ref::() { + return Arc::clone(plan).with_new_children(vec![replace_peeled_projection( + repartition.input(), + replacement, + )?]); + } + Ok(replacement) +} + +struct OrderedOneHopSpec { + schema: SchemaRef, + props: Arc, + fetch: usize, + rel_type_name: String, + direction: Direction, + provider: Arc, + ordinal_identities: Arc, +} + +fn peel_plan(plan: &Arc) -> Arc { + if let Some(limit) = plan.downcast_ref::() { + return peel_plan(limit.input()); + } + if let Some(coalesce) = plan.downcast_ref::() { + return peel_plan(coalesce.input()); + } + if let Some(repartition) = plan.downcast_ref::() { + return peel_plan(repartition.input()); + } + Arc::clone(plan) +} + +fn peel_expand_transport(plan: &Arc) -> Arc { + if let Some(coalesce) = plan.downcast_ref::() { + return peel_expand_transport(coalesce.input()); + } + if let Some(repartition) = plan.downcast_ref::() { + return peel_expand_transport(repartition.input()); + } + if let Some(projection) = plan.downcast_ref::() { + return peel_expand_transport(projection.input()); + } + Arc::clone(plan) +} + +fn detect_ordered_one_hop(plan: &Arc) -> Option { + let plan = peel_plan(plan); + let projection = plan.downcast_ref::()?; + if projection.expr().len() != 1 { + return None; + } + let projected_column = projection.expr()[0].expr.downcast_ref::()?; + if projection + .input() + .schema() + .field(projected_column.index()) + .name() + != "node_uuid" + { + return None; + } + let projection_children = projection.children(); + let sort_input = projection_children.first()?; + let sort = if let Some(merge) = sort_input.downcast_ref::() { + merge.input().downcast_ref::()? + } else { + sort_input.downcast_ref::()? + }; + let fetch = sort.fetch()?; + if sort.expr().len() != 1 || sort.expr()[0].options.descending { + return None; + } + let sort_column = sort.expr()[0].expr.downcast_ref::()?; + if sort.input().schema().field(sort_column.index()).name() != "node_uuid" { + return None; + } + let expand_plan = peel_expand_transport(sort.children().first()?); + let expand = expand_plan.downcast_ref::()?; + // Two-hop plans are handled separately; refuse stacked expands here. + if expand.children().first().is_some_and(|child| { + let peeled = peel_expand_transport(child); + peeled.downcast_ref::().is_some() + }) { + return None; + } + if !expand.is_destination_identity_only() || expand.direction() != Direction::Out { + return None; + } + let ordinal_identities = expand.ordinal_identities()?; + if !crate::edge_count::has_complete_frontier(expand) + || !ordinal_identities.uuid_order_matches_ordinals() + { + return None; + } + Some(OrderedOneHopSpec { + schema: plan.schema(), + props: Arc::new( + plan.properties() + .as_ref() + .clone() + .with_partitioning(Partitioning::UnknownPartitioning(1)), + ), + fetch, + rel_type_name: expand.rel_type_name().to_owned(), + direction: expand.direction(), + provider: Arc::clone(expand.provider()), + ordinal_identities, + }) +} + +pub struct OrderedOneHopExec { + schema: SchemaRef, + props: Arc, + fetch: usize, + rel_type_name: String, + direction: Direction, + provider: Arc, + ordinal_identities: Arc, + capture_epoch: u64, +} + +impl OrderedOneHopExec { + fn new(spec: OrderedOneHopSpec) -> Self { + Self { + schema: spec.schema, + props: spec.props, + fetch: spec.fetch, + rel_type_name: spec.rel_type_name, + direction: spec.direction, + provider: spec.provider, + ordinal_identities: spec.ordinal_identities, + capture_epoch: demand::stamp_capture_epoch().unwrap_or(0), + } + } +} + +impl fmt::Debug for OrderedOneHopExec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OrderedOneHopExec") + .field("fetch", &self.fetch) + .field("rel", &self.rel_type_name) + .finish_non_exhaustive() + } +} + +impl DisplayAs for OrderedOneHopExec { + fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "OrderedOneHopExec: rel={}, dir={:?}, fetch={}", + self.rel_type_name, self.direction, self.fetch + ) + } +} + +impl ExecutionPlan for OrderedOneHopExec { + fn name(&self) -> &str { + "OrderedOneHopExec" + } + + fn properties(&self) -> &Arc { + &self.props + } + + fn children(&self) -> Vec<&Arc> { + Vec::new() + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> Result> { + if children.is_empty() { + Ok(self) + } else { + Err(DataFusionError::Internal( + "OrderedOneHopExec has no children".into(), + )) + } + } + + fn execute( + &self, + partition: usize, + _context: Arc, + ) -> Result { + if partition != 0 { + return Err(DataFusionError::Internal(format!( + "OrderedOneHopExec only has partition 0, got {partition}" + ))); + } + // Inbound view: destinations ordered by node id, multiplicity = in-degree. + let inbound = self + .provider + .adjacency(&self.rel_type_name, Direction::In) + .map_err(|error| DataFusionError::External(Box::new(error)))?; + let node_extent = inbound.node_extent(); + + let mut remaining = self.fetch; + let mut destinations = Vec::new(); + let mut candidates = 0_u64; + let epoch = self.capture_epoch; + for destination in 0..node_extent { + if remaining == 0 { + break; + } + demand::record_adjacency_row(self.capture_epoch, 0); + let in_degree = inbound + .degree(destination) + .map_err(|error| DataFusionError::External(Box::new(error)))?; + if in_degree == 0 { + continue; + } + let emit = usize_from_u64(in_degree).min(remaining); + candidates = candidates.saturating_add(emit as u64); + destinations.push((destination, emit)); + remaining -= emit; + } + + demand::record_candidates(epoch, 0, usize_from_u64(candidates)); + demand::record_emitted(epoch, 0, self.fetch - remaining); + + let mut builder = FixedSizeBinaryBuilder::with_capacity(self.fetch - remaining, 16); + for destinations in destinations.chunks(self.ordinal_identities.max_requested_ids()) { + let requested = destinations.iter().map(|(id, _)| *id).collect::>(); + let lookup = self + .ordinal_identities + .lookup_node_uuids(&requested) + .map_err(|error| DataFusionError::External(Box::new(error)))?; + demand::record_identity_projection(epoch, 0, requested.len(), 1, &lookup.metrics); + for ((_, copies), uuid) in destinations.iter().zip(&lookup.values) { + let uuid = uuid + .as_ref() + .ok_or_else(|| DataFusionError::Internal("missing destination uuid".into()))?; + for _ in 0..*copies { + builder + .append_value(uuid.as_bytes()) + .map_err(|error| DataFusionError::External(Box::new(error)))?; + } + } + } + let column: ArrayRef = Arc::new(builder.finish()); + let batch = + arrow::record_batch::RecordBatch::try_new(Arc::clone(&self.schema), vec![column]) + .map_err(|error| DataFusionError::ArrowError(Box::new(error), None))?; + Ok(Box::pin(OrderedOneHopStream { + schema: Arc::clone(&self.schema), + batch: Some(batch), + })) + } +} + +fn usize_from_u64(value: u64) -> usize { + usize::try_from(value).unwrap_or(usize::MAX) +} + +struct OrderedOneHopStream { + schema: SchemaRef, + batch: Option, +} + +impl Stream for OrderedOneHopStream { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(self.batch.take().map(Ok)) + } +} + +impl RecordBatchStream for OrderedOneHopStream { + fn schema(&self) -> SchemaRef { + Arc::clone(&self.schema) + } +} diff --git a/crates/graphforge-exec/src/ordered_two_hop.rs b/crates/graphforge-exec/src/ordered_two_hop.rs index e45e304d0..cfc1bc877 100644 --- a/crates/graphforge-exec/src/ordered_two_hop.rs +++ b/crates/graphforge-exec/src/ordered_two_hop.rs @@ -14,7 +14,7 @@ use std::sync::Arc; use std::task::{Context, Poll}; use arrow::array::{ArrayRef, FixedSizeBinaryBuilder}; -use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use arrow::datatypes::SchemaRef; use datafusion::common::{DataFusionError, Result}; use datafusion::execution::TaskContext; use datafusion::physical_expr::ScalarFunctionExpr; @@ -27,7 +27,7 @@ use datafusion::physical_plan::repartition::RepartitionExec; use datafusion::physical_plan::sorts::sort::SortExec; use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; use datafusion::physical_plan::{ - DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, RecordBatchStream, + DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties, RecordBatchStream, SendableRecordBatchStream, }; use futures::Stream; @@ -74,7 +74,6 @@ struct OrderedTwoHopSpec { props: Arc, fetch: usize, rel_type_name: String, - direction: Direction, provider: Arc, ordinal_identities: Arc, require_edge_disjoint: bool, @@ -162,12 +161,21 @@ fn detect_ordered_two_hop(plan: &Arc) -> Option, fetch: usize, rel_type_name: String, - direction: Direction, provider: Arc, ordinal_identities: Arc, require_edge_disjoint: bool, @@ -193,7 +200,6 @@ impl OrderedTwoHopPathCountExec { props: spec.props, fetch: spec.fetch, rel_type_name: spec.rel_type_name, - direction: spec.direction, provider: spec.provider, ordinal_identities: spec.ordinal_identities, require_edge_disjoint: spec.require_edge_disjoint, @@ -256,36 +262,32 @@ impl ExecutionPlan for OrderedTwoHopPathCountExec { "OrderedTwoHopPathCountExec only has partition 0, got {partition}" ))); } - let outbound = self - .provider - .adjacency(&self.rel_type_name, self.direction) - .map_err(|error| DataFusionError::External(Box::new(error)))?; let inbound = self .provider .adjacency(&self.rel_type_name, Direction::In) .map_err(|error| DataFusionError::External(Box::new(error)))?; - let mut max_node = 0_u64; - outbound.for_each_row(|node_id, _| { - max_node = max_node.max(node_id); - }); - inbound.for_each_row(|node_id, _| { - max_node = max_node.max(node_id); - }); + let node_extent = inbound.node_extent(); let mut remaining = self.fetch; - let mut uuids = Vec::with_capacity(self.fetch); + let mut uuids = Vec::new(); let mut candidates = 0_u64; let epoch = self.capture_epoch; - for destination in 0..=max_node { + for destination in 0..node_extent { if remaining == 0 { break; } - let path_count = - count_two_hop_paths_to(&inbound, destination, self.require_edge_disjoint); + demand::record_adjacency_row(self.capture_epoch, 1); + let (path_count, examined) = count_two_hop_paths_to( + &inbound, + destination, + self.require_edge_disjoint, + remaining, + ) + .map_err(|error| DataFusionError::External(Box::new(error)))?; + candidates = candidates.saturating_add(examined); if path_count == 0 { continue; } - candidates = candidates.saturating_add(path_count); let lookup = self .ordinal_identities .lookup_node_uuids(&[destination]) @@ -303,8 +305,6 @@ impl ExecutionPlan for OrderedTwoHopPathCountExec { demand::record_candidates(epoch, 1, usize_from_u64(candidates)); demand::record_emitted(epoch, 1, uuids.len()); - let field = Field::new("id", DataType::FixedSizeBinary(16), true); - let schema = Arc::new(Schema::new(vec![field])); let mut builder = FixedSizeBinaryBuilder::with_capacity(uuids.len(), 16); for uuid in uuids { builder @@ -312,8 +312,9 @@ impl ExecutionPlan for OrderedTwoHopPathCountExec { .map_err(|error| DataFusionError::External(Box::new(error)))?; } let column: ArrayRef = Arc::new(builder.finish()); - let batch = arrow::record_batch::RecordBatch::try_new(schema, vec![column]) - .map_err(|error| DataFusionError::ArrowError(Box::new(error), None))?; + let batch = + arrow::record_batch::RecordBatch::try_new(Arc::clone(&self.schema), vec![column]) + .map_err(|error| DataFusionError::ArrowError(Box::new(error), None))?; Ok(Box::pin(OrderedTwoHopStream { schema: Arc::clone(&self.schema), batch: Some(batch), @@ -329,16 +330,38 @@ fn count_two_hop_paths_to( inbound: &crate::Adjacency, destination: u64, require_edge_disjoint: bool, -) -> u64 { + remaining: usize, +) -> std::result::Result<(u64, u64), graphforge_core::GfError> { + const CHUNK: usize = 256; let mut count = 0_u64; - for (r2, middle) in inbound.neighbors(destination).iter() { - for (r1, _) in inbound.neighbors(middle).iter() { - if !require_edge_disjoint || r1 != r2 { - count = count.saturating_add(1); + let mut examined = 0_u64; + let mut outer_offset = 0; + loop { + let outer = inbound.neighbor_chunk(destination, outer_offset, CHUNK)?; + if outer.is_empty() { + return Ok((count, examined)); + } + outer_offset += outer.len(); + for (r2, middle) in outer { + let mut inner_offset = 0; + loop { + let inner = inbound.neighbor_chunk(middle, inner_offset, CHUNK)?; + if inner.is_empty() { + break; + } + inner_offset += inner.len(); + for (r1, _) in inner { + examined = examined.saturating_add(1); + if !require_edge_disjoint || r1 != r2 { + count = count.saturating_add(1); + if count >= remaining as u64 { + return Ok((count, examined)); + } + } + } } } } - count } struct OrderedTwoHopStream { diff --git a/crates/graphforge-rel/src/expr.rs b/crates/graphforge-rel/src/expr.rs index 288d92eda..4b9a1ac84 100644 --- a/crates/graphforge-rel/src/expr.rs +++ b/crates/graphforge-rel/src/expr.rs @@ -5855,6 +5855,13 @@ static CYPHER_SIZE: LazyLock = pub(crate) static CYPHER_ROW_MARKER: LazyLock = LazyLock::new(|| ScalarUDF::new_from_impl(CypherRowMarker::new())); +/// Identify the compiler-owned, always-true per-row marker by implementation. +/// Execution rewrites must not accept a user function with the same name. +#[must_use] +pub fn is_cypher_row_marker(function: &ScalarUDF) -> bool { + function.inner().downcast_ref::().is_some() +} + #[derive(Debug, PartialEq, Eq, Hash)] struct CypherRowMarker { signature: Signature, diff --git a/crates/graphforge-storage/src/adjacency.rs b/crates/graphforge-storage/src/adjacency.rs index 78ec71b0d..a21b24bdf 100644 --- a/crates/graphforge-storage/src/adjacency.rs +++ b/crates/graphforge-storage/src/adjacency.rs @@ -219,16 +219,83 @@ impl ShardedCsrIndex { } let mut output = Vec::new(); for record in &self.manifest.shards[start..end] { - output.extend(self.read_record_row(record, node_id)?); + output.extend( + self.map_record_row(record, node_id, |row| row.iter().collect::>())?, + ); } Ok(output) } - fn read_record_row( + /// Count a logical row without concatenating its neighbor payload. A hub can + /// span many shards; only one authenticated bounded shard is retained. + pub fn row_len(&self, node_id: u64) -> Result { + if node_id >= self.manifest.node_count { + return Ok(0); + } + let end = self + .manifest + .shards + .partition_point(|record| record.first_node <= node_id); + let mut total = 0_u64; + for record in self.manifest.shards[..end].iter().rev() { + if node_id >= record.first_node.saturating_add(record.node_count) { + break; + } + let count = self.map_record_row(record, node_id, |row| row.len() as u64)?; + total = total + .checked_add(count) + .ok_or_else(|| GfError::Storage("CSR degree exceeds u64".into()))?; + } + Ok(total) + } + + /// Copy at most `limit` entries after a logical row offset. This preserves + /// shard order and never concatenates a whole high-degree row. + pub fn row_chunk( &self, - record: &CsrShardRecord, node_id: u64, + mut skip: usize, + limit: usize, ) -> Result, GfError> { + if node_id >= self.manifest.node_count || limit == 0 { + return Ok(Vec::new()); + } + let end = self + .manifest + .shards + .partition_point(|record| record.first_node <= node_id); + let mut start = end; + while start > 0 + && node_id + < self.manifest.shards[start - 1] + .first_node + .saturating_add(self.manifest.shards[start - 1].node_count) + { + start -= 1; + } + let mut output = Vec::new(); + for record in &self.manifest.shards[start..end] { + self.map_record_row(record, node_id, |row| { + if skip >= row.len() { + skip -= row.len(); + return; + } + output.extend(row.iter().skip(skip).take(limit - output.len())); + skip = 0; + })?; + if output.len() == limit { + break; + } + } + Ok(output) + } + + fn map_record_row( + &self, + record: &CsrShardRecord, + node_id: u64, + map: impl FnOnce(CsrRow<'_>) -> T, + ) -> Result { let mut cache = self .cache .lock() @@ -236,7 +303,7 @@ impl ShardedCsrIndex { if let Some((file, csr)) = cache.as_ref() && file == &record.file { - return Ok(csr.row(node_id - record.first_node).iter().collect()); + return Ok(map(csr.row(node_id - record.first_node))); } let path = self.root.join(&record.file); let bytes = std::fs::read(&path).map_err(|error| { @@ -255,7 +322,7 @@ impl ShardedCsrIndex { record.file ))); } - let output = csr.row(node_id - record.first_node).iter().collect(); + let output = map(csr.row(node_id - record.first_node)); *cache = Some((record.file.clone(), csr)); Ok(output) } @@ -2488,6 +2555,20 @@ mod tests { let original = std::fs::read(&first).unwrap(); std::fs::write(&first, b"corrupt").unwrap(); assert!(reader.row(0).unwrap_err().to_string().contains("checksum")); + assert!( + reader + .row_len(0) + .unwrap_err() + .to_string() + .contains("checksum") + ); + assert!( + reader + .row_chunk(0, 0, 1) + .unwrap_err() + .to_string() + .contains("checksum") + ); std::fs::write(&first, original).unwrap(); std::fs::remove_file(&first).unwrap(); assert!( @@ -2497,6 +2578,20 @@ mod tests { .to_string() .contains("missing CSR shard") ); + assert!( + reader + .row_len(0) + .unwrap_err() + .to_string() + .contains("missing CSR shard") + ); + assert!( + reader + .row_chunk(0, 0, 1) + .unwrap_err() + .to_string() + .contains("missing CSR shard") + ); } #[test] @@ -2575,6 +2670,18 @@ mod tests { reader.row(0).unwrap(), expected.row(0).iter().collect::>() ); + assert_eq!(reader.row_len(0).unwrap(), 7); + assert_eq!(reader.row_len(1).unwrap(), 0); + let mut chunks = Vec::new(); + for offset in (0..7).step_by(3) { + let chunk = reader.row_chunk(0, offset, 3).unwrap(); + assert!(chunk.len() <= 3); + chunks.extend(chunk); + let cache = reader.cache.lock().unwrap(); + assert!(cache.as_ref().unwrap().1.edge_count() <= 2); + } + assert_eq!(chunks, expected.row(0).iter().collect::>()); + assert!(reader.row_chunk(0, 7, 3).unwrap().is_empty()); } #[test] diff --git a/crates/graphforge-storage/src/ordinal_identity_v4.rs b/crates/graphforge-storage/src/ordinal_identity_v4.rs index 45b0103a8..4fbe5d561 100644 --- a/crates/graphforge-storage/src/ordinal_identity_v4.rs +++ b/crates/graphforge-storage/src/ordinal_identity_v4.rs @@ -553,6 +553,7 @@ pub struct V4OrdinalIdentityHandle { tombstone_cache: TombstoneBlockCache, limits: V4OrdinalIdentityLimits, admission: V4OrdinalAdmissionMetrics, + uuid_order_matches_ordinals: bool, } impl V4OrdinalIdentityHandle { @@ -705,6 +706,8 @@ impl V4OrdinalIdentityHandle { } validate_unique_forward_runs(&mut forward, &mut admission)?; let mut ranges = Vec::with_capacity(manifest.ordinal_ranges.len()); + let mut prior_uuid = None; + let mut uuid_order_matches_ordinals = true; for range in &manifest.ordinal_ranges { if !names.insert(range.artifact.name.clone()) { return Err(V4OrdinalIdentityError::InvalidDescriptor( @@ -718,6 +721,8 @@ impl V4OrdinalIdentityHandle { range, &mut ordinal_commitment, &mut admission, + &mut prior_uuid, + &mut uuid_order_matches_ordinals, )?, }); } @@ -760,6 +765,7 @@ impl V4OrdinalIdentityHandle { tombstone_cache: TombstoneBlockCache::default(), limits, admission, + uuid_order_matches_ordinals, }))) } @@ -775,6 +781,20 @@ impl V4OrdinalIdentityHandle { self.topology_generation } + /// Whether authenticated UUIDs increase strictly across all ordinal ranges. + /// This conservative proof includes tombstoned identities: removing any of + /// them preserves ordering. It requires no additional reads or durable data. + #[must_use] + pub const fn uuid_order_matches_ordinals(&self) -> bool { + self.uuid_order_matches_ordinals + } + + /// Maximum request entries accepted by this admitted handle per lookup. + #[must_use] + pub const fn max_requested_ids(&self) -> usize { + self.limits.max_requested + } + /// Resolve a bounded caller batch while preserving caller order. pub fn lookup_node_uuids( &mut self, @@ -1186,6 +1206,8 @@ fn admit_ordinal_artifact( range: &V4OrdinalRange, commitment: &mut MappingCommitment, metrics: &mut V4OrdinalAdmissionMetrics, + prior_uuid: &mut Option<[u8; UUID_WIDTH_USIZE]>, + uuid_order_matches_ordinals: &mut bool, ) -> Result { let mut artifact = open_admission_file(root, &range.artifact)?; let block_bytes = @@ -1226,6 +1248,10 @@ fn admit_ordinal_artifact( .first_node_id .checked_add(ordinal) .ok_or(V4OrdinalIdentityError::Authentication)?; + if prior_uuid.is_some_and(|prior| prior >= uuid) { + *uuid_order_matches_ordinals = false; + } + *prior_uuid = Some(uuid); commitment.add(&uuid, node_id); ordinal = ordinal.saturating_add(1); } @@ -2343,6 +2369,70 @@ mod tests { let _handle = fixture.open(V4OrdinalIdentityLimits::default()); } + #[test] + fn admitted_uuid_order_proof_spans_sparse_ranges_and_tombstones() { + for (uuids, expected) in [ + ([1_u128, 2, 3, 4], true), + ([1, 2, 4, 3], false), + ([3, 4, 1, 2], false), + ] { + let mut fixture = Fixture::new(&[2, 2], &[2]); + let index = fixture.root.path().join(INDEX_DIR); + let mut mappings = Vec::new(); + for (range, values) in fixture + .manifest + .ordinal_ranges + .iter_mut() + .zip(uuids.chunks_exact(2)) + { + let bytes = values + .iter() + .flat_map(|value| Uuid::from_u128(*value).into_bytes()) + .collect::>(); + fs::write(index.join(&range.artifact.name), &bytes).unwrap(); + range.artifact = artifact( + range.artifact.name.clone(), + V4OrdinalArtifactKind::OrdinalUuids, + 7, + &bytes, + ); + range.blocks = ordinal_blocks(&bytes); + mappings.extend( + values + .iter() + .enumerate() + .map(|(offset, uuid)| (*uuid, range.first_node_id + offset as u64)), + ); + } + mappings.sort_unstable(); + let bytes = mappings + .iter() + .flat_map(|(uuid, id)| { + Uuid::from_u128(*uuid) + .into_bytes() + .into_iter() + .chain(id.to_be_bytes()) + }) + .collect::>(); + fs::write(index.join("forward.uuidx"), &bytes).unwrap(); + fixture.manifest.forward_identities = vec![artifact( + "forward.uuidx".into(), + V4OrdinalArtifactKind::ForwardIdentities, + 7, + &bytes, + )]; + fixture.publish(); + let mut handle = fixture.open(V4OrdinalIdentityLimits::default()); + assert_eq!(handle.uuid_order_matches_ordinals(), expected); + let lookup = handle.lookup_node_uuids(&[1, 2, 6, 7]).unwrap(); + assert!( + lookup.values[1].is_none(), + "tombstoned identity remains absent" + ); + assert_eq!(lookup.values[0], Some(Uuid::from_u128(uuids[0]))); + } + } + #[test] fn forward_runs_reject_cross_generation_duplicates_and_noncanonical_order() { let mut fixture = Fixture::new(&[4], &[]); diff --git a/crates/graphforge-storage/src/parquet_scan.rs b/crates/graphforge-storage/src/parquet_scan.rs index bf3bfc2c4..d0144a3b7 100644 --- a/crates/graphforge-storage/src/parquet_scan.rs +++ b/crates/graphforge-storage/src/parquet_scan.rs @@ -89,6 +89,35 @@ fn footer_num_rows(path: &Path) -> Option { usize::try_from(builder.metadata().file_metadata().num_rows()).ok() } +/// Prove that a physical leaf emits every topology node ID exactly once. +/// Only this module's unbounded topology scans and ordered partition wrapper +/// qualify; filters, joins, limits and arbitrary sources do not. +#[must_use] +pub fn is_complete_node_id_scan(plan: &dyn ExecutionPlan, column: usize) -> bool { + if let Some(ordered) = plan.downcast_ref::() { + return is_complete_node_id_scan(ordered.input.as_ref(), column); + } + let Some(scan) = plan.downcast_ref::() else { + return false; + }; + let base_column = match &scan.projection { + Some(indices) => indices.get(column).copied(), + None => Some(column), + }; + scan.limit.is_none() + && scan.base_schema.as_ref() == crate::TOPOLOGY_NODES_SCHEMA.as_ref() + && base_column.is_some_and(|index| { + scan.base_schema + .fields() + .get(index) + .is_some_and(|field| field.name() == "node_id") + }) + && scan + .fragments + .iter() + .all(|fragment| fragment.normalize_topology) +} + /// Session extension carrying the #337 I/O concurrency semaphore. #[derive(Clone, Debug)] pub struct IoConcurrencyExt(pub Arc); From a77eebe27db51b6e50abf9c652d34a9c19b639b4 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 5 Sep 2026 06:22:17 +0000 Subject: [PATCH 2/3] test(scale): validate actual ordered leaf evidence --- .../graphforge-api/tests/scale_g500_ladder.rs | 401 ++++++++++++++++-- 1 file changed, 361 insertions(+), 40 deletions(-) diff --git a/crates/graphforge-api/tests/scale_g500_ladder.rs b/crates/graphforge-api/tests/scale_g500_ladder.rs index 5b1dc6b20..0eeba5fa2 100644 --- a/crates/graphforge-api/tests/scale_g500_ladder.rs +++ b/crates/graphforge-api/tests/scale_g500_ladder.rs @@ -717,6 +717,7 @@ fn query_work_evidence(snapshot: &DemandSnapshot) -> Value { "input_batches": hop.input_batches, "input_rows": hop.input_rows, "candidates_generated": hop.candidates_generated, + "adjacency_rows_examined": hop.adjacency_rows_examined, "rows_emitted": hop.rows_emitted, "edge_reads_started": hop.edge_reads_started, "edge_reads_completed": hop.edge_reads_completed, @@ -764,10 +765,10 @@ fn query_work_evidence(snapshot: &DemandSnapshot) -> Value { const MAX_IDENTITY_BUFFER_BYTES: u64 = 16 * 1024 * 1024; fn released_memory_is_bounded(snapshot: &DemandSnapshot) -> bool { - snapshot.memory_reserved_after - <= snapshot - .memory_reserved_before - .saturating_add(snapshot.returned_batch_bytes) + snapshot + .memory_reserved_before + .checked_add(snapshot.returned_batch_bytes) + .is_some_and(|bound| snapshot.memory_reserved_after <= bound) } fn has_no_materializing_reads(hop: &demand::HopSnapshot) -> bool { @@ -787,35 +788,66 @@ fn has_no_materializing_reads(hop: &demand::HopSnapshot) -> bool { && hop.node_projected_columns == 0 } -fn bounded_counting_ordered_two_hop(snapshot: &DemandSnapshot, limit: usize) -> bool { +fn bounded_ordered_leaf( + snapshot: &DemandSnapshot, + expected_hops: usize, + limit: usize, + live_nodes: u64, + expected_rows: u64, +) -> bool { let Some(hop) = snapshot.hops.values().next() else { return false; }; + let [operator] = snapshot.operator_rss.as_slice() else { + return false; + }; + let expected_operator = match expected_hops { + 1 => "ordered_one_hop", + 2 => "ordered_two_hop", + _ => return false, + }; snapshot.hops.len() == 1 + && snapshot.filters.is_empty() && snapshot.sorts.is_empty() - && snapshot.operator_rss.is_empty() + && operator.ordinal == 0 + && operator.operator == expected_operator + && operator.peak_bytes >= operator.before_bytes + && operator.peak_bytes >= operator.after_bytes + && (operator.after_bytes > 0 || !cfg!(target_os = "linux")) && snapshot.execution_batch_rows > 0 && hop.input_batches == 0 && hop.input_rows == 0 - && hop.rows_emitted == limit as u64 + && hop.rows_emitted == expected_rows && hop.candidates_generated >= hop.rows_emitted - && hop.candidates_generated <= (limit as u64).saturating_add(snapshot.execution_batch_rows) + && (expected_hops != 1 || hop.candidates_generated == hop.rows_emitted) + // Each optimized leaf probes destinations once in ordinal order. + // Empty rows count as probes, not emitted candidates, and may exceed K. + && hop.adjacency_rows_examined > 0 + && hop.adjacency_rows_examined <= live_nodes + && (limit as u64) + .checked_add(snapshot.execution_batch_rows) + .is_some_and(|bound| hop.candidates_generated <= bound) && has_no_materializing_reads(hop) && hop.projected_chunks > 0 - && hop.projected_chunks == hop.projected_rows + && hop.projected_chunks <= hop.projected_rows + && (expected_hops != 2 || hop.projected_chunks == hop.projected_rows) + && hop.projected_rows <= hop.adjacency_rows_examined && hop.projected_rows <= limit as u64 && hop.projected_columns == 1 && hop.identity_ranges_selected > 0 && hop.identity_ranges_selected <= hop.projected_rows - && hop.identity_read_calls - <= hop - .identity_ranges_selected - .saturating_mul(2) - .saturating_add(2) - && hop.identity_bytes_read - <= hop - .identity_read_calls - .saturating_mul(MAX_IDENTITY_BUFFER_BYTES) + && hop.identity_read_calls > 0 + && hop.identity_bytes_read > 0 + && hop.identity_peak_buffer_bytes > 0 + && hop + .identity_ranges_selected + .checked_mul(2) + .and_then(|bound| bound.checked_add(2)) + .is_some_and(|bound| hop.identity_read_calls <= bound) + && hop + .identity_read_calls + .checked_mul(MAX_IDENTITY_BUFFER_BYTES) + .is_some_and(|bound| hop.identity_bytes_read <= bound) && hop.identity_peak_buffer_bytes <= MAX_IDENTITY_BUFFER_BYTES && hop.identity_per_record_seeks == 0 && hop.reads_after_cancel == 0 @@ -824,6 +856,44 @@ fn bounded_counting_ordered_two_hop(snapshot: &DemandSnapshot, limit: usize) -> && released_memory_is_bounded(snapshot) } +fn bounded_no_sort_one_hop(snapshot: &DemandSnapshot, limit: usize, expected_rows: u64) -> bool { + let Some(hop) = snapshot.hops.values().next() else { + return false; + }; + let canonical_batch_rows = + u64::try_from(graphforge_exec::SessionResourceConfig::default().batch_size).ok(); + let row_bound = canonical_batch_rows.and_then(|batch_rows| { + (snapshot.execution_batch_rows == batch_rows) + .then_some(batch_rows) + .and_then(|batch_rows| u64::try_from(limit).ok()?.checked_add(batch_rows)) + }); + let Some(row_bound) = row_bound else { + return false; + }; + let [operator] = snapshot.operator_rss.as_slice() else { + return false; + }; + snapshot.hops.len() == 1 + && snapshot.filters.is_empty() + && snapshot.sorts.is_empty() + && operator.ordinal == 0 + && operator.operator == "expand" + && operator.peak_bytes >= operator.before_bytes + && operator.peak_bytes >= operator.after_bytes + && (operator.after_bytes > 0 || !cfg!(target_os = "linux")) + && hop.input_batches > 0 + && hop.input_batches <= hop.input_rows + && hop.input_rows <= row_bound + && hop.rows_emitted >= expected_rows + && hop.rows_emitted <= row_bound + && hop.candidates_generated >= hop.rows_emitted + && hop.candidates_generated <= row_bound + && hop.reads_after_cancel == 0 + && snapshot.cancellations == 0 + && snapshot.max_in_flight_reads == 0 + && released_memory_is_bounded(snapshot) +} + fn bounded_streaming_ordered_limit( snapshot: &DemandSnapshot, expected_hops: usize, @@ -836,10 +906,13 @@ fn bounded_streaming_ordered_limit( // DataFusion's baseline output counter is charged before its final // fetch wrapper slices the terminal batch, so it may include at most // one physical batch beyond the returned TopK rows. - && snapshot.sorts[0].output_rows - <= (limit as u64).saturating_add(snapshot.execution_batch_rows) + && (limit as u64) + .checked_add(snapshot.execution_batch_rows) + .is_some_and(|bound| snapshot.sorts[0].output_rows <= bound) && snapshot.sorts[0].retained_bytes == 0 - && snapshot.operator_rss.len() == expected_hops.saturating_add(1) + && expected_hops + .checked_add(1) + .is_some_and(|expected| snapshot.operator_rss.len() == expected) && snapshot.operator_rss.iter().all(|operator| { operator.peak_bytes >= operator.before_bytes && operator.peak_bytes >= operator.after_bytes @@ -852,29 +925,154 @@ fn bounded_streaming_ordered_limit( .all(|hop| hop.reads_after_cancel == 0) } -fn bounded_ordered_limit(snapshot: &DemandSnapshot, expected_hops: usize, limit: usize) -> bool { - let optimized_two_hop_shape = expected_hops == 2 - && snapshot.hops.len() == 1 - && snapshot.sorts.is_empty() - && snapshot.operator_rss.is_empty(); - if optimized_two_hop_shape { - bounded_counting_ordered_two_hop(snapshot, limit) +fn bounded_ordered_limit( + snapshot: &DemandSnapshot, + expected_hops: usize, + limit: usize, + live_nodes: u64, + live_edges: u64, +) -> bool { + let expected_rows = if expected_hops == 1 { + (limit as u64).min(live_edges) } else { - bounded_streaming_ordered_limit(snapshot, expected_hops, limit) + limit as u64 + }; + if snapshot.sorts.is_empty() { + if snapshot.operator_rss.first().is_some_and(|operator| { + matches!(operator.operator, "ordered_one_hop" | "ordered_two_hop") + }) { + return bounded_ordered_leaf(snapshot, expected_hops, limit, live_nodes, expected_rows); + } + return expected_hops == 1 && bounded_no_sort_one_hop(snapshot, limit, expected_rows); + } + bounded_streaming_ordered_limit(snapshot, expected_hops, limit) +} + +fn no_sort_one_hop_snapshot(limit: usize) -> DemandSnapshot { + let batch_rows = u64::try_from(graphforge_exec::SessionResourceConfig::default().batch_size) + .expect("canonical execution batch rows fit u64"); + let mut snapshot = DemandSnapshot { + execution_batch_rows: batch_rows, + returned_batch_bytes: u64::try_from(limit) + .expect("synthetic limit fits u64") + .checked_mul(16) + .expect("synthetic returned-byte count overflow"), + operator_rss: vec![demand::OperatorRssSnapshot { + ordinal: 0, + operator: "expand", + before_bytes: 100, + peak_bytes: 120, + after_bytes: 110, + }], + ..DemandSnapshot::default() + }; + snapshot.hops.insert( + 1, + demand::HopSnapshot { + input_batches: 1, + input_rows: 1_024, + // Truthful no-sort one-hop observation. 4_310 was the stale mix + // 3_300 (late one-hop) + 1_010 (fused two-hop) and must not be + // treated as a legitimate one-hop baseline. + candidates_generated: 3_300, + rows_emitted: 3_300, + ..demand::HopSnapshot::default() + }, + ); + snapshot +} + +#[test] +fn no_sort_one_hop_budget_requires_exact_expand_and_bounded_rows() { + let limit = 1_000; + let baseline = no_sort_one_hop_snapshot(limit); + assert!(bounded_ordered_limit(&baseline, 1, limit, 16_384, 16_384)); + assert!( + !bounded_ordered_limit(&baseline, 2, limit, 16_384, 16_384), + "the no-sort one-hop policy must not weaken optimized two-hop classification" + ); + + macro_rules! reject_mutation { + ($name:literal, $mutate:expr) => {{ + let mut snapshot = baseline.clone(); + $mutate(&mut snapshot); + assert!( + !bounded_ordered_limit(&snapshot, 1, limit, 16_384, 16_384), + "accepted {}: {snapshot:#?}", + $name + ); + }}; } + reject_mutation!("missing expand", |s: &mut DemandSnapshot| { + s.operator_rss.clear(); + }); + reject_mutation!("extra expand", |s: &mut DemandSnapshot| { + s.operator_rss.push(demand::OperatorRssSnapshot { + ordinal: 1, + operator: "expand", + before_bytes: 100, + peak_bytes: 120, + after_bytes: 110, + }); + }); + reject_mutation!("wrong operator", |s: &mut DemandSnapshot| { + s.operator_rss[0].operator = "sort"; + }); + reject_mutation!("unexpected sort", |s: &mut DemandSnapshot| { + s.sorts.push(demand::SortSnapshot { + fetch: Some(limit), + ..demand::SortSnapshot::default() + }); + }); + let row_bound = u64::try_from(limit) + .expect("synthetic limit fits u64") + .checked_add(baseline.execution_batch_rows) + .expect("synthetic row bound"); + reject_mutation!("input row bound exceeded", |s: &mut DemandSnapshot| { + s.hops.get_mut(&1).unwrap().input_rows = + row_bound.checked_add(1).expect("synthetic excess input"); + }); + reject_mutation!("candidate row bound exceeded", |s: &mut DemandSnapshot| { + s.hops.get_mut(&1).unwrap().candidates_generated = row_bound + .checked_add(1) + .expect("synthetic excess candidates"); + }); + reject_mutation!("emitted row bound exceeded", |s: &mut DemandSnapshot| { + let excess = row_bound.checked_add(1).expect("synthetic excess output"); + let hop = s.hops.get_mut(&1).unwrap(); + hop.rows_emitted = excess; + hop.candidates_generated = excess; + }); + reject_mutation!("row bound overflow", |s: &mut DemandSnapshot| { + s.execution_batch_rows = u64::MAX; + }); } fn counting_snapshot(limit: usize) -> DemandSnapshot { + // Fused two-hop work retains actual rejected candidate visits and identity + // reads, while its leaf consumes no Arrow input batches. let mut snapshot = DemandSnapshot { execution_batch_rows: 8_192, - returned_batch_bytes: (limit as u64).saturating_mul(16), + operator_rss: vec![demand::OperatorRssSnapshot { + ordinal: 0, + operator: "ordered_two_hop", + before_bytes: 100, + peak_bytes: 120, + after_bytes: 110, + }], + returned_batch_bytes: (limit as u64) + .checked_mul(16) + .expect("synthetic returned-byte count overflow"), ..DemandSnapshot::default() }; snapshot.hops.insert( 1, demand::HopSnapshot { - candidates_generated: (limit as u64).saturating_add(10), + candidates_generated: (limit as u64) + .checked_add(10) + .expect("synthetic candidate count overflow"), rows_emitted: limit as u64, + adjacency_rows_examined: 53, projected_chunks: 52, projected_rows: 52, projected_columns: 1, @@ -888,23 +1086,101 @@ fn counting_snapshot(limit: usize) -> DemandSnapshot { snapshot } +#[test] +fn optimized_leaf_budget_requires_rss_probes_and_actual_identity_work() { + let limit = 1_000; + let live_nodes = 16_384; + for expected_hops in [1, 2] { + let mut baseline = counting_snapshot(limit); + if expected_hops == 1 { + baseline.operator_rss[0].operator = "ordered_one_hop"; + let hop = baseline.hops.get_mut(&1).unwrap(); + hop.candidates_generated = limit as u64; + hop.projected_chunks = 1; + } + // Sparse leading rows may outnumber K without generating candidates. + baseline.hops.get_mut(&1).unwrap().adjacency_rows_examined = 12_000; + assert!(bounded_ordered_limit( + &baseline, + expected_hops, + limit, + live_nodes, + 16_384 + )); + for mutation in 0..11 { + let mut invalid = baseline.clone(); + match mutation { + 0 => invalid.operator_rss.clear(), + 1 => invalid.operator_rss[0].operator = "edge_count", + 2 => invalid.operator_rss[0].peak_bytes = 0, + 3 => invalid.hops.get_mut(&1).unwrap().adjacency_rows_examined = 0, + 4 => invalid.hops.get_mut(&1).unwrap().adjacency_rows_examined = live_nodes + 1, + 5 => invalid.hops.get_mut(&1).unwrap().input_batches = 1, + 6 => invalid.hops.get_mut(&1).unwrap().input_rows = 1, + 7 => invalid.hops.get_mut(&1).unwrap().identity_read_calls = 0, + 8 => invalid.hops.get_mut(&1).unwrap().identity_bytes_read = 0, + 9 => invalid.hops.get_mut(&1).unwrap().identity_peak_buffer_bytes = 0, + 10 => invalid.hops.get_mut(&1).unwrap().node_full_reads = 1, + _ => unreachable!(), + } + assert!( + !bounded_ordered_limit(&invalid, expected_hops, limit, live_nodes, 16_384), + "accepted optimized family {expected_hops} mutation {mutation}: {invalid:?}" + ); + } + assert!(!bounded_ordered_limit( + &baseline, + expected_hops, + limit, + 0, + 16_384 + )); + } +} + +#[test] +fn optimized_one_hop_underfilled_limit_uses_reopened_edge_count() { + let mut snapshot = counting_snapshot(1_000); + snapshot.operator_rss[0].operator = "ordered_one_hop"; + let hop = snapshot.hops.get_mut(&1).unwrap(); + hop.candidates_generated = 929; + hop.rows_emitted = 929; + hop.adjacency_rows_examined = 997; + hop.projected_chunks = 1; + assert!(bounded_ordered_limit(&snapshot, 1, 1_000, 1_024, 929)); + for incorrect in [928, 930] { + let mut invalid = snapshot.clone(); + let hop = invalid.hops.get_mut(&1).unwrap(); + hop.candidates_generated = incorrect; + hop.rows_emitted = incorrect; + assert!(!bounded_ordered_limit(&invalid, 1, 1_000, 1_024, 929)); + } + assert!(!bounded_ordered_limit(&snapshot, 1, 1_000, 1_024, 1_000)); +} + #[test] fn optimized_two_hop_budget_requires_complete_counting_evidence() { let limit = 1_000; let baseline = counting_snapshot(limit); - assert!(bounded_ordered_limit(&baseline, 2, limit)); + assert!(bounded_ordered_limit(&baseline, 2, limit, 16_384, 16_384)); assert!( - !bounded_ordered_limit(&baseline, 1, limit), + !bounded_ordered_limit(&baseline, 1, limit, 16_384, 16_384), "the optimized exception must not weaken the one-hop budget" ); - assert!(!bounded_ordered_limit(&DemandSnapshot::default(), 2, limit)); + assert!(!bounded_ordered_limit( + &DemandSnapshot::default(), + 2, + limit, + 16_384, + 16_384 + )); macro_rules! reject_mutation { ($name:literal, $mutate:expr) => {{ let mut snapshot = baseline.clone(); $mutate(&mut snapshot); assert!( - !bounded_ordered_limit(&snapshot, 2, limit), + !bounded_ordered_limit(&snapshot, 2, limit, 16_384, 16_384), "accepted {}: {snapshot:#?}", $name ); @@ -941,6 +1217,23 @@ fn optimized_two_hop_budget_requires_complete_counting_evidence() { reject_mutation!("node materialization", |s: &mut DemandSnapshot| { s.hops.get_mut(&1).unwrap().node_rows_scanned = 1; }); + reject_mutation!("stale one-hop mix", |s: &mut DemandSnapshot| { + // Process-global capture previously accepted late one-hop events into + // the next two-hop snapshot (4_310 = 3_300 + 1_010) together with + // leftover expand RSS and scanned rows. + let hop = s.hops.get_mut(&1).unwrap(); + hop.candidates_generated = 4_310; + hop.input_batches = 1; + hop.input_rows = 1_024; + hop.edge_rows_scanned = 3_300; + s.operator_rss.push(demand::OperatorRssSnapshot { + ordinal: 0, + operator: "expand", + before_bytes: 100, + peak_bytes: 120, + after_bytes: 110, + }); + }); reject_mutation!("unexpected cancellation", |s: &mut DemandSnapshot| { s.cancellations = 1; }); @@ -1594,7 +1887,13 @@ fn run_rung( violation = Some("execution_failure"); } else if rows != 1_000 { violation = Some("result_mismatch"); - } else if !bounded_ordered_limit(&work, expected_hops, 1_000) { + } else if !bounded_ordered_limit( + &work, + expected_hops, + 1_000, + node_count, + edge_count, + ) { violation = Some("operator_budget_violation"); } if let Some(class) = violation { @@ -3027,9 +3326,14 @@ fn run_integrated_certification_with_edge_factor( let (source_1hop_result, source_1hop_work) = certification_query(&graph, ONE_HOP, "source_query_1hop", phase, &mut journal); assert!( - bounded_ordered_limit(&source_1hop_work, 1, 1_000), + bounded_ordered_limit(&source_1hop_work, 1, 1_000, source_nodes, source_edges), "{source_1hop_work:#?}" ); + assert_eq!( + row_count(&source_1hop_result) as u64, + 1_000_u64.min(source_edges), + "unrestricted directed one-hop returns one row per live edge up to LIMIT" + ); let source_1hop = result_fingerprint(&source_1hop_result); journal.pass_with_detail( "source_query_1hop", @@ -3041,7 +3345,7 @@ fn run_integrated_certification_with_edge_factor( let (source_2hop_result, source_2hop_work) = certification_query(&graph, TWO_HOP, "source_query_2hop", phase, &mut journal); assert!( - bounded_ordered_limit(&source_2hop_work, 2, 1_000), + bounded_ordered_limit(&source_2hop_work, 2, 1_000, source_nodes, source_edges), "{source_2hop_work:#?}" ); let source_2hop = result_fingerprint(&source_2hop_result); @@ -3167,9 +3471,20 @@ fn run_integrated_certification_with_edge_factor( &mut journal, ); assert!( - bounded_ordered_limit(&imported_1hop_work, 1, 1_000), + bounded_ordered_limit( + &imported_1hop_work, + 1, + 1_000, + imported_nodes, + imported_edges + ), "{imported_1hop_work:#?}" ); + assert_eq!( + row_count(&imported_1hop_result) as u64, + 1_000_u64.min(imported_edges), + "unrestricted directed one-hop returns one row per live edge up to LIMIT" + ); let imported_1hop = result_fingerprint(&imported_1hop_result); assert_eq!(source_1hop, imported_1hop); journal.pass_with_detail( @@ -3187,7 +3502,13 @@ fn run_integrated_certification_with_edge_factor( &mut journal, ); assert!( - bounded_ordered_limit(&imported_2hop_work, 2, 1_000), + bounded_ordered_limit( + &imported_2hop_work, + 2, + 1_000, + imported_nodes, + imported_edges + ), "{imported_2hop_work:#?}" ); let imported_2hop = result_fingerprint(&imported_2hop_result); From e838eafa5ae458a267c6f7820de8a69d34f1de28 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 5 Sep 2026 07:06:59 +0000 Subject: [PATCH 3/3] fix(bench): retain adjacency probes in query receipts --- benchmarks/runners/certify/src/lib.rs | 62 ++++++++++++++++++++++++++- 1 file changed, 61 insertions(+), 1 deletion(-) diff --git a/benchmarks/runners/certify/src/lib.rs b/benchmarks/runners/certify/src/lib.rs index f78b9284a..f9a7d69c2 100644 --- a/benchmarks/runners/certify/src/lib.rs +++ b/benchmarks/runners/certify/src/lib.rs @@ -1198,11 +1198,12 @@ fn sanitized_query_evidence(value: &serde_json::Value) -> bool { && sanitized_operator_rss(object.get("operator_rss")) } -const QUERY_HOP_KEYS: [&str; 25] = [ +const QUERY_HOP_KEYS: [&str; 26] = [ "ordinal", "input_batches", "input_rows", "candidates_generated", + "adjacency_rows_examined", "rows_emitted", "projected_chunks", "projected_rows", @@ -1862,6 +1863,65 @@ mod tests { assert!(parse_receipts(&encoded, true).is_err()); } + #[test] + fn query_receipts_preserve_typed_adjacency_probe_evidence() { + let query = serde_json::json!({ + "contract": "graphforge-result-sink/2", + "destination": "/private/query.arrow", + "format": "ArrowIpc", "rows": 2, "batches": 1, "bytes": 64, + "complete": true, "result_sha256": "a".repeat(64), "scalar_u64": null, + "query_evidence": { + "contract": "graphforge-query-evidence/1", + "hops": [{ + "ordinal": 0, "input_batches": 0, "input_rows": 0, + "candidates_generated": 2, "adjacency_rows_examined": 3, + "rows_emitted": 2, "projected_chunks": 1, "projected_rows": 2, + "projected_columns": 1, "edge_projected_columns": 0, + "node_projected_columns": 0, "edge_reader_calls": 0, + "edge_rows_returned": 0, "edge_logical_rows_scanned": 0, + "edge_full_reads": 0, "node_reader_calls": 0, + "node_rows_returned": 0, "node_logical_rows_scanned": 0, + "node_full_reads": 0, "identity_reader_calls": 1, + "identity_logical_bytes": 64, "identity_ranges_selected": 1, + "identity_peak_buffer_bytes": 64, "identity_per_record_seeks": 0, + "identity_revalidation_calls": 1, "identity_revalidation_bytes": 0 + }], + "sorts": [], + "operator_rss": [{ + "ordinal": 0, "operator": "ordered_one_hop", + "before_bytes": 100, "peak_bytes": 120, "after_bytes": 110 + }], + "max_in_flight_reads": 0, "memory_reserved_before": 0, + "memory_reserved_after": 0, "returned_batch_bytes": 32, + "execution_batch_rows": 8192, "peak_rss_bytes": 120, + "rss_after_release_bytes": 110 + } + }); + let accepted = parse_receipts(&serde_json::to_vec(&query).unwrap(), true).unwrap(); + assert_eq!(accepted[0]["query_evidence"], query["query_evidence"]); + assert!(accepted[0].get("destination").is_none()); + for bad_value in [ + serde_json::json!(-1), + serde_json::json!("3"), + serde_json::Value::Null, + ] { + let mut invalid = query.clone(); + invalid["query_evidence"]["hops"][0]["adjacency_rows_examined"] = bad_value; + assert!(parse_receipts(&serde_json::to_vec(&invalid).unwrap(), true).is_err()); + } + for field in ["adjacency_rows_examined", "identity_reader_calls"] { + let mut invalid = query.clone(); + invalid["query_evidence"]["hops"][0] + .as_object_mut() + .unwrap() + .remove(field); + assert!(parse_receipts(&serde_json::to_vec(&invalid).unwrap(), true).is_err()); + } + let mut invalid = query; + invalid["query_evidence"]["hops"][0]["project_path"] = serde_json::json!("/private"); + assert!(parse_receipts(&serde_json::to_vec(&invalid).unwrap(), true).is_err()); + } + #[test] fn storage_receipts_are_closed_and_preserve_only_semantic_attribution() { let totals = serde_json::json!({