From 8c6be85e643cc2977a8950d78f4cb1c2e0883198 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Sat, 29 Aug 2026 10:20:48 -0600 Subject: [PATCH 1/2] fix(query): expose ordinary execution evidence --- crates/graphforge-api/src/lib.rs | 5 + crates/graphforge-api/src/query_evidence.rs | 368 ++++++++++++++++++ .../graphforge-api/tests/fixed_hop_limit.rs | 87 ++++- .../tests/non-cypher-parity-policy.json | 11 +- .../tests/non_cypher_release.py | 6 +- crates/graphforge-cli/src/portable_cli.rs | 49 ++- crates/graphforge-exec/src/lib.rs | 83 +++- docs/book/architecture/execution-model.md | 8 + scripts/ci/test-non-cypher-surface-gate.py | 2 +- tests/contracts/non-cypher-rust-surface.json | 7 +- 10 files changed, 590 insertions(+), 36 deletions(-) create mode 100644 crates/graphforge-api/src/query_evidence.rs diff --git a/crates/graphforge-api/src/lib.rs b/crates/graphforge-api/src/lib.rs index 28c85ba8..ca5b485b 100644 --- a/crates/graphforge-api/src/lib.rs +++ b/crates/graphforge-api/src/lib.rs @@ -127,6 +127,7 @@ mod provider_embedding_execution; mod provider_find; mod provider_rerank; mod provider_session; +mod query_evidence; mod repository; mod resource_policy; mod resumable_construction; @@ -207,6 +208,10 @@ pub use import_session::{ GraphImportSession, ImportConstructionEvidence, ImportPhase, ImportProgress, ImportSessionLimits, ImportSourceKind, }; +pub use query_evidence::{ + QueryExecutionEvidence, QueryHopEvidence, QueryOperatorRssEvidence, QuerySinkEvidenceReceipt, + QuerySortEvidence, +}; // The Arrow-backed result of [`GraphForge::execute`]. pub use generation_diff::{ CommittedGenerationIdentity, GenerationDiffDisposition, GenerationDiffLimits, diff --git a/crates/graphforge-api/src/query_evidence.rs b/crates/graphforge-api/src/query_evidence.rs new file mode 100644 index 00000000..f1a81b3c --- /dev/null +++ b/crates/graphforge-api/src/query_evidence.rs @@ -0,0 +1,368 @@ +//! Sanitized ordinary-query work evidence. + +use arrow::array::{Array, Int64Array, UInt64Array}; +use arrow::ipc::{reader::StreamReader, writer::StreamWriter}; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use serde::Serialize; +use sha2::{Digest, Sha256}; +use std::fs::File; +use std::io::{BufReader, Write}; + +use crate::{ + CancellationToken, GfError, GraphForge, IrLiteral, ResultSinkFormat, ResultSinkOptions, + ResultSinkReceipt, +}; + +/// Versioned aggregate-only query evidence emitted by ordinary execution. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct QueryExecutionEvidence { + /// Evidence schema identifier. + pub contract: &'static str, + /// Per-hop physical work in stable plan order. + pub hops: Vec, + /// Fetch-aware physical sort work. + pub sorts: Vec, + /// Sanitized operator RSS samples. + pub operator_rss: Vec, + /// Maximum concurrent filtered reads. + pub max_in_flight_reads: u64, + /// Query memory reservation before execution. + pub memory_reserved_before: u64, + /// Query memory reservation after every stream was released. + pub memory_reserved_after: u64, + /// Arrow bytes retained by returned batches while the sink consumed them. + pub returned_batch_bytes: u64, + /// Configured physical execution batch-row bound. + pub execution_batch_rows: u64, + /// Largest observed operator RSS sample. + pub peak_rss_bytes: u64, + /// Largest operator RSS sample after stream release. + pub rss_after_release_bytes: u64, +} + +/// Aggregate deterministic counters for one fixed hop. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct QueryHopEvidence { + /// Stable hop ordinal, independent of internal variable identifiers. + pub ordinal: usize, + /// Input batches pulled by the hop. + pub input_batches: u64, + /// Input rows pulled by the hop. + pub input_rows: u64, + /// Adjacency candidates examined before projection. + pub candidates_generated: u64, + /// Rows emitted by the hop. + pub rows_emitted: u64, + /// Chunks served by destination-only projection. + pub projected_chunks: u64, + /// Rows served by destination-only projection. + pub projected_rows: u64, + /// Required output columns at this hop. + pub projected_columns: u64, + /// Required edge columns. + pub edge_projected_columns: u64, + /// Required destination-node columns. + pub node_projected_columns: u64, + /// Edge reader calls opened. + pub edge_reader_calls: u64, + /// Edge rows returned by readers. + pub edge_rows_returned: u64, + /// Edge rows evaluated by physical readers. + pub edge_logical_rows_scanned: u64, + /// Edge reads that fell back to full materialization. + pub edge_full_reads: u64, + /// Node reader calls opened. + pub node_reader_calls: u64, + /// Node rows returned by readers. + pub node_rows_returned: u64, + /// Node rows evaluated by physical readers. + pub node_logical_rows_scanned: u64, + /// Node reads that fell back to full materialization. + pub node_full_reads: u64, + /// Bounded identity reader calls. + pub identity_reader_calls: u64, + /// Logical identity bytes read. + pub identity_logical_bytes: u64, + /// Coalesced ordinal ranges selected. + pub identity_ranges_selected: u64, + /// Largest identity request or transient buffer. + pub identity_peak_buffer_bytes: u64, + /// Forbidden per-record identity seeks. + pub identity_per_record_seeks: u64, + /// Generation-authority validation calls. + pub identity_revalidation_calls: u64, + /// Bytes read while validating identity authority. + pub identity_revalidation_bytes: u64, +} + +/// Aggregate deterministic counters for one fetch-aware sort. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct QuerySortEvidence { + /// Stable sort ordinal. + pub ordinal: usize, + /// Physical TopK bound, or `None` for an unbounded sort. + pub fetch_rows: Option, + /// Rows observed by the sort. + pub output_rows: u64, + /// Spill operations performed. + pub spill_count: u64, + /// Rows written to spill files. + pub spilled_rows: u64, + /// Bytes written to spill files. + pub spilled_bytes: u64, + /// Sort memory retained after stream release. + pub retained_bytes: u64, +} + +/// Content-free RSS lifetime for one physical operator class. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct QueryOperatorRssEvidence { + /// Stable operator ordinal. + pub ordinal: usize, + /// Sanitized physical operator class. + pub operator: &'static str, + /// RSS before the operator stream was created. + pub before_bytes: u64, + /// Largest RSS sample while the stream lived. + pub peak_bytes: u64, + /// RSS after the operator stream was released. + pub after_bytes: u64, +} + +/// Atomic result-sink publication plus ordinary-query evidence. +#[derive(Debug)] +pub struct QuerySinkEvidenceReceipt { + /// Atomic result-sink publication receipt. + pub sink: ResultSinkReceipt, + /// SHA-256 of the atomically published result artifact. + pub result_sha256: String, + /// Exact unsigned scalar for a one-row integer result representable as `u64`. + pub scalar_u64: Option, + /// Sanitized ordinary physical-query evidence. + pub evidence: QueryExecutionEvidence, +} + +struct DigestWriter<'a>(&'a mut Sha256); + +impl Write for DigestWriter<'_> { + fn write(&mut self, buffer: &[u8]) -> std::io::Result { + self.0.update(buffer); + Ok(buffer.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +fn published_result_metadata( + path: &std::path::Path, + format: ResultSinkFormat, +) -> Result<(Option, String), GfError> { + let mut rows = 0usize; + let mut scalar = None; + let mut scalar_shape = true; + let mut digest = Sha256::new(); + let mut logical_schema = None; + let mut observe = |batch: arrow::record_batch::RecordBatch| -> Result<(), GfError> { + let schema = logical_schema.get_or_insert_with(|| { + std::sync::Arc::new(arrow::datatypes::Schema::new( + batch.schema().fields().clone(), + )) + }); + let logical_batch = arrow::record_batch::RecordBatch::try_new( + std::sync::Arc::clone(schema), + batch.columns().to_vec(), + ) + .map_err(|error| GfError::Storage(error.to_string()))?; + let mut writer = StreamWriter::try_new(DigestWriter(&mut digest), schema) + .map_err(|error| GfError::Storage(error.to_string()))?; + writer + .write(&logical_batch) + .map_err(|error| GfError::Storage(error.to_string()))?; + writer + .finish() + .map_err(|error| GfError::Storage(error.to_string()))?; + if batch.num_columns() != 1 { + scalar_shape = false; + } + let before = rows; + rows = rows.saturating_add(batch.num_rows()); + if rows > 1 { + scalar_shape = false; + scalar = None; + return Ok(()); + } + if before == 0 && batch.num_rows() == 1 && scalar_shape { + let column = batch.column(0); + scalar = column + .as_any() + .downcast_ref::() + .filter(|array| !array.is_null(0)) + .map(|array| array.value(0)) + .or_else(|| { + column + .as_any() + .downcast_ref::() + .filter(|array| !array.is_null(0)) + .and_then(|array| u64::try_from(array.value(0)).ok()) + }); + } + Ok(()) + }; + match format { + ResultSinkFormat::Parquet => { + let reader = ParquetRecordBatchReaderBuilder::try_new( + File::open(path).map_err(|error| GfError::Storage(error.to_string()))?, + ) + .map_err(|error| GfError::Storage(error.to_string()))? + .build() + .map_err(|error| GfError::Storage(error.to_string()))?; + for batch in reader { + observe(batch.map_err(|error| GfError::Storage(error.to_string()))?)?; + } + } + ResultSinkFormat::ArrowIpc => { + let reader = StreamReader::try_new( + BufReader::new( + File::open(path).map_err(|error| GfError::Storage(error.to_string()))?, + ), + None, + ) + .map_err(|error| GfError::Storage(error.to_string()))?; + for batch in reader { + observe(batch.map_err(|error| GfError::Storage(error.to_string()))?)?; + } + } + } + Ok(( + (rows == 1 && scalar_shape).then_some(scalar).flatten(), + hex_digest(&digest.finalize()), + )) +} + +fn hex_digest(bytes: &[u8]) -> String { + use std::fmt::Write as _; + let mut output = String::with_capacity(bytes.len() * 2); + for byte in bytes { + write!(output, "{byte:02x}").expect("writing to String cannot fail"); + } + output +} + +impl From for QueryExecutionEvidence { + fn from(snapshot: graphforge_exec::demand::DemandSnapshot) -> Self { + let hops = snapshot + .hops + .into_values() + .enumerate() + .map(|(ordinal, hop)| QueryHopEvidence { + ordinal, + input_batches: hop.input_batches, + input_rows: hop.input_rows, + candidates_generated: hop.candidates_generated, + rows_emitted: hop.rows_emitted, + projected_chunks: hop.projected_chunks, + projected_rows: hop.projected_rows, + projected_columns: hop.projected_columns, + edge_projected_columns: hop.edge_projected_columns, + node_projected_columns: hop.node_projected_columns, + edge_reader_calls: hop.edge_reads_started, + edge_rows_returned: hop.edge_rows_returned, + edge_logical_rows_scanned: hop.edge_rows_scanned, + edge_full_reads: hop.edge_full_reads, + node_reader_calls: hop.node_reads_started, + node_rows_returned: hop.node_rows_returned, + node_logical_rows_scanned: hop.node_rows_scanned, + node_full_reads: hop.node_full_reads, + identity_reader_calls: hop.identity_read_calls, + identity_logical_bytes: hop.identity_bytes_read, + identity_ranges_selected: hop.identity_ranges_selected, + identity_peak_buffer_bytes: hop.identity_peak_buffer_bytes, + identity_per_record_seeks: hop.identity_per_record_seeks, + identity_revalidation_calls: hop.identity_revalidation_calls, + identity_revalidation_bytes: hop.identity_revalidation_bytes, + }) + .collect(); + let sorts = snapshot + .sorts + .into_iter() + .map(|sort| QuerySortEvidence { + ordinal: sort.ordinal, + fetch_rows: sort.fetch, + output_rows: sort.output_rows, + spill_count: sort.spill_count, + spilled_rows: sort.spilled_rows, + spilled_bytes: sort.spilled_bytes, + retained_bytes: sort.retained_bytes, + }) + .collect(); + let operator_rss = snapshot + .operator_rss + .into_iter() + .map(|operator| QueryOperatorRssEvidence { + ordinal: operator.ordinal, + operator: operator.operator, + before_bytes: operator.before_bytes, + peak_bytes: operator.peak_bytes, + after_bytes: operator.after_bytes, + }) + .collect::>(); + let peak_rss_bytes = operator_rss + .iter() + .map(|operator| operator.peak_bytes) + .max() + .unwrap_or(0); + let rss_after_release_bytes = operator_rss + .iter() + .map(|operator| operator.after_bytes) + .max() + .unwrap_or(0); + Self { + contract: "graphforge-query-evidence/1", + hops, + sorts, + operator_rss, + max_in_flight_reads: snapshot.max_in_flight_reads, + memory_reserved_before: snapshot.memory_reserved_before, + memory_reserved_after: snapshot.memory_reserved_after, + returned_batch_bytes: snapshot.returned_batch_bytes, + execution_batch_rows: snapshot.execution_batch_rows, + peak_rss_bytes, + rss_after_release_bytes, + } + } +} + +impl GraphForge { + /// Execute an ordinary streaming query sink and retain sanitized physical evidence. + pub fn execute_to_result_sink_with_evidence( + &self, + cypher: &str, + params: &std::collections::HashMap, + path: &str, + format: ResultSinkFormat, + options: &ResultSinkOptions, + cancellation: Option<&CancellationToken>, + ) -> Result { + let (sink, evidence) = graphforge_exec::demand::capture(|| { + self.execute_to_result_sink_with_params( + cypher, + params, + path, + format, + options, + cancellation, + ) + }); + let sink = sink?; + let (scalar_u64, result_sha256) = + published_result_metadata(&sink.destination, sink.format)?; + Ok(QuerySinkEvidenceReceipt { + sink, + result_sha256, + scalar_u64, + evidence: evidence.into(), + }) + } +} diff --git a/crates/graphforge-api/tests/fixed_hop_limit.rs b/crates/graphforge-api/tests/fixed_hop_limit.rs index 6162907d..91d44176 100644 --- a/crates/graphforge-api/tests/fixed_hop_limit.rs +++ b/crates/graphforge-api/tests/fixed_hop_limit.rs @@ -17,7 +17,7 @@ use arrow::record_batch::RecordBatch; use graphforge_api::{ CONSTRUCTION_EDGE_SCHEMA, CONSTRUCTION_NODE_SCHEMA, GraphConstructionBudgets, GraphForge, OperationId, PortableSelection, PortableV2ExportRequest, PortableV2ImportRequest, - PortableVerifyRequest, verify_portable_v2, + PortableVerifyRequest, ResultSinkFormat, ResultSinkOptions, verify_portable_v2, }; use graphforge_core::uuid::{Uuid, new_v7}; use graphforge_core::{OntologyMode, TypeId}; @@ -792,6 +792,91 @@ fn ordered_destination_uuid_projection_is_exact_and_linear_at_1x_2x_4x() { } } +#[test] +fn ordinary_streaming_sink_exposes_deterministic_query_evidence() { + let _guard = IO_GUARD.lock().unwrap(); + let root = TempDir::new().unwrap(); + generate_graph(root.path(), 4_096, FAN_OUT, true); + let forge = open_forge(root.path()); + let outputs = TempDir::new().unwrap(); + let options = ResultSinkOptions::default(); + let params = HashMap::new(); + + let mut fingerprints = Vec::new(); + for (ordinal, query) in [ORDERED_ONE_HOP, ORDERED_TWO_HOP].into_iter().enumerate() { + let output = outputs.path().join(format!("query-{ordinal}.parquet")); + let receipt = forge + .execute_to_result_sink_with_evidence( + query, + ¶ms, + output.to_str().unwrap(), + ResultSinkFormat::Parquet, + &options, + None, + ) + .unwrap(); + assert_eq!(receipt.evidence.contract, "graphforge-query-evidence/1"); + assert_eq!(receipt.evidence.hops.len(), ordinal + 1); + 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); + assert!( + receipt.evidence.memory_reserved_after + <= receipt + .evidence + .memory_reserved_before + .saturating_add(receipt.evidence.returned_batch_bytes) + ); + assert!( + receipt.evidence.hops.iter().all(|hop| { + hop.edge_reader_calls == 0 + && hop.node_reader_calls == 0 + && hop.identity_per_record_seeks == 0 + }), + "{:?}", + receipt.evidence.hops + ); + assert!( + receipt + .evidence + .hops + .iter() + .map(|hop| hop.identity_reader_calls) + .sum::() + > 0, + "the query must exercise the bounded reusable identity reader" + ); + assert_eq!(receipt.scalar_u64, None); + fingerprints.push(receipt.result_sha256); + } + assert_ne!(fingerprints[0], fingerprints[1]); + let repeated_output = outputs.path().join("query-repeat.parquet"); + let repeated = forge + .execute_to_result_sink_with_evidence( + ORDERED_ONE_HOP, + ¶ms, + repeated_output.to_str().unwrap(), + ResultSinkFormat::Parquet, + &options, + None, + ) + .unwrap(); + assert_eq!(repeated.result_sha256, fingerprints[0]); + + let count_output = outputs.path().join("count.parquet"); + let count = forge + .execute_to_result_sink_with_evidence( + "MATCH (n) RETURN count(n) AS total", + ¶ms, + count_output.to_str().unwrap(), + ResultSinkFormat::Parquet, + &options, + None, + ) + .unwrap(); + assert_eq!(count.scalar_u64, Some(4_096)); +} + #[test] fn portable_v2_clean_import_preserves_projected_ordered_hops_and_io() { let _guard = IO_GUARD.lock().unwrap(); diff --git a/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json b/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json index 679be16c..23dc0362 100644 --- a/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json +++ b/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json @@ -1,8 +1,8 @@ { "contractVersion": 1, "rustManifest": "../../../tests/contracts/non-cypher-rust-surface.json", - "releaseSurfaceCount": 259, - "releaseSurfaceDigest": "5b0c6f3b4995545f6527b8be3b3fea7cb7aca0d070ad280ae6ec80e24ac9e938", + "releaseSurfaceCount": 260, + "releaseSurfaceDigest": "00576480be38cfe444cb74a0da55d944e4c6cb5a4f1788397b212177f9ea2bdd", "requiredEquivalent": [ "GraphForge.adopt_ontology", "GraphForge.clear_ontology", @@ -260,6 +260,13 @@ ], "reason": "Node binds parameters on PlanHandle before the asynchronous bounded Parquet sink." }, + "GraphForge.execute_to_result_sink_with_evidence": { + "nodeMembers": [ + "GraphForge.plan", + "PlanHandle.sinkParquet" + ], + "reason": "The aggregate qualification receipt is currently a Rust facade and gf CLI certification surface; Node retains its asynchronous result sink without the certification envelope." + }, "GraphForge.configure_provider_find_runtime": { "nodeMembers": [ "GraphForge.configureOpenrouter" diff --git a/crates/graphforge-bindings-py/tests/non_cypher_release.py b/crates/graphforge-bindings-py/tests/non_cypher_release.py index 25c63b7a..9a2c6606 100644 --- a/crates/graphforge-bindings-py/tests/non_cypher_release.py +++ b/crates/graphforge-bindings-py/tests/non_cypher_release.py @@ -23,8 +23,8 @@ RUST_MANIFEST = ROOT / "tests/contracts/non-cypher-rust-surface.json" RUST_GATE = ROOT / "scripts/ci/non-cypher-surface-gate.py" PYO3_SOURCE = ROOT / "crates/graphforge-bindings-py/src/lib.rs" -EXPECTED_RUST_DIGEST = "499f902e4713931d7b3591fbaf978e6694c3029fde08c5c5476547563e5776e5" -EXPECTED_RELEASE_DIGEST = "5b0c6f3b4995545f6527b8be3b3fea7cb7aca0d070ad280ae6ec80e24ac9e938" +EXPECTED_RUST_DIGEST = "72510094f55ba39627bda6c3c5c0df6845b6c90c602aac54ab3e628274240cab" +EXPECTED_RELEASE_DIGEST = "00576480be38cfe444cb74a0da55d944e4c6cb5a4f1788397b212177f9ea2bdd" PYTHON_ONLY_METHODS = frozenset( { @@ -257,7 +257,7 @@ def _classification_report() -> dict[str, object]: for group in manifest["method_evidence_groups"].values() for method_id in group["ids"] } - assert len(release_methods) == 259 + assert len(release_methods) == 260 assert _digest(release_methods) == EXPECTED_RELEASE_DIGEST assert set(EVIDENCE) == set(manifest["method_evidence_groups"]) diff --git a/crates/graphforge-cli/src/portable_cli.rs b/crates/graphforge-cli/src/portable_cli.rs index 36c3fdb5..273d4594 100644 --- a/crates/graphforge-cli/src/portable_cli.rs +++ b/crates/graphforge-cli/src/portable_cli.rs @@ -491,32 +491,31 @@ pub(crate) fn run_query( graphforge_api::GfError::Validation("query --output must be valid UTF-8".into()) })?; let params = std::collections::HashMap::new(); - let receipt = match args.format { - QuerySinkFormat::Parquet => graph.execute_to_parquet_stream_with_params( - &args.cypher, - ¶ms, - path, - &options, - None, - )?, - QuerySinkFormat::ArrowIpc => graph.execute_to_arrow_ipc_stream_with_params( - &args.cypher, - ¶ms, - path, - &options, - None, - )?, + let format = match args.format { + QuerySinkFormat::Parquet => graphforge_api::ResultSinkFormat::Parquet, + QuerySinkFormat::ArrowIpc => graphforge_api::ResultSinkFormat::ArrowIpc, }; + let receipt = graph.execute_to_result_sink_with_evidence( + &args.cypher, + ¶ms, + path, + format, + &options, + None, + )?; if json { write_json( &serde_json::json!({ - "contract": "graphforge-result-sink/1", - "destination": receipt.destination, - "format": format!("{:?}", receipt.format), - "rows": receipt.progress.rows, - "batches": receipt.progress.batches, - "bytes": receipt.progress.bytes, - "complete": receipt.progress.complete, + "contract": "graphforge-result-sink/2", + "destination": receipt.sink.destination, + "format": format!("{:?}", receipt.sink.format), + "rows": receipt.sink.progress.rows, + "batches": receipt.sink.progress.batches, + "bytes": receipt.sink.progress.bytes, + "complete": receipt.sink.progress.complete, + "result_sha256": receipt.result_sha256, + "scalar_u64": receipt.scalar_u64, + "query_evidence": receipt.evidence, }), output, )?; @@ -524,9 +523,9 @@ pub(crate) fn run_query( writeln!( output, "wrote {} rows={} bytes={}", - receipt.destination.display(), - receipt.progress.rows, - receipt.progress.bytes + receipt.sink.destination.display(), + receipt.sink.progress.rows, + receipt.sink.progress.bytes ) .map_err(|error| graphforge_api::GfError::Execution(error.to_string()))?; } diff --git a/crates/graphforge-exec/src/lib.rs b/crates/graphforge-exec/src/lib.rs index 66a3c198..76e1fe3e 100644 --- a/crates/graphforge-exec/src/lib.rs +++ b/crates/graphforge-exec/src/lib.rs @@ -226,8 +226,10 @@ mod write_driver; use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt; use std::path::{Path, PathBuf}; +use std::pin::Pin; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, RwLock}; +use std::task::{Context, Poll}; use arrow::array::{ Array, ArrayRef, FixedSizeBinaryArray, FixedSizeBinaryBuilder, Int8Array, ListBuilder, @@ -252,7 +254,7 @@ use datafusion::physical_plan::{ use datafusion::physical_planner::{DefaultPhysicalPlanner, ExtensionPlanner, PhysicalPlanner}; use datafusion::prelude::SessionContext; use datafusion::scalar::ScalarValue; -use futures::StreamExt; +use futures::{Stream, StreamExt}; pub use graphforge_core::GfError; use graphforge_core::OntologyMode; @@ -4857,6 +4859,66 @@ impl Default for SessionResourceConfig { } } +struct QueryEvidenceStream { + inner: Option, + physical: Arc, + task_ctx: Arc, + memory_reserved_before: usize, + returned_batch_bytes: usize, + finalized: bool, +} + +impl QueryEvidenceStream { + fn finalize(&mut self) { + if self.finalized { + return; + } + self.finalized = true; + drop(self.inner.take()); + demand::record_plan_completion( + &self.physical, + self.memory_reserved_before, + self.task_ctx.memory_pool().reserved(), + self.returned_batch_bytes, + self.task_ctx.session_config().batch_size(), + ); + } +} + +impl Stream for QueryEvidenceStream { + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); + let Some(inner) = this.inner.as_mut() else { + return Poll::Ready(None); + }; + match inner.as_mut().poll_next(cx) { + Poll::Ready(Some(Ok(batch))) => { + this.returned_batch_bytes = this + .returned_batch_bytes + .saturating_add(batch.get_array_memory_size()); + Poll::Ready(Some(Ok(batch))) + } + Poll::Ready(Some(Err(error))) => { + this.finalize(); + Poll::Ready(Some(Err(error))) + } + Poll::Ready(None) => { + this.finalize(); + Poll::Ready(None) + } + Poll::Pending => Poll::Pending, + } + } +} + +impl Drop for QueryEvidenceStream { + fn drop(&mut self) { + self.finalize(); + } +} + /// A configured DataFusion [`SessionContext`] ready to execute [`GraphPlan`]s. /// /// Construct via [`ExecutionSession::new`] (read/query) or @@ -5730,8 +5792,23 @@ impl ExecutionSession { params: &HashMap, ) -> Result { let (physical, _) = self.plan_physical(plan, params).await?; - datafusion::physical_plan::execute_stream(physical, self.ctx.task_ctx()) - .map_err(|e| GfError::Execution(e.to_string())) + let task_ctx = self.ctx.task_ctx(); + let memory_reserved_before = task_ctx.memory_pool().reserved(); + let stream = + datafusion::physical_plan::execute_stream(Arc::clone(&physical), Arc::clone(&task_ctx)) + .map_err(|e| GfError::Execution(e.to_string()))?; + let schema = stream.schema(); + Ok(Box::pin(RecordBatchStreamAdapter::new( + schema, + QueryEvidenceStream { + inner: Some(stream), + physical, + task_ctx, + memory_reserved_before, + returned_batch_bytes: 0, + finalized: false, + }, + ))) } /// Render the physical plan for a [`GraphPlan`] (indented, one line per diff --git a/docs/book/architecture/execution-model.md b/docs/book/architecture/execution-model.md index ffc92922..ccd5b270 100644 --- a/docs/book/architecture/execution-model.md +++ b/docs/book/architecture/execution-model.md @@ -138,6 +138,14 @@ from lookup evidence. Diagnostics never report identities or paths. Global `ORDE the complete unordered candidate stream and does not use invalid early cancellation. +Ordinary streaming result sinks retain the same aggregate evidence through the +terminal stream boundary. `gf --json query` emits `graphforge-result-sink/2` +with nested `graphforge-query-evidence/1`: named hop reader, logical-row, +projection, identity-byte, TopK/spill, memory-release, and operator-RSS fields. +The receipt also includes the SHA-256 of the atomically published result and an +optional `scalar_u64` only for an exact one-row, one-`UInt64` result. Evidence is +content-free: it contains no graph values, identities, paths, or provider names. + **Selection is a planner choice, not an IR change.** The Graph IR is unchanged: variable-length traversal is still encoded on `Expand { …, min_hops, max_hops }`. A lowering rule selects an adjacency-backed physical node when the provider covers the relation type + direction, and falls diff --git a/scripts/ci/test-non-cypher-surface-gate.py b/scripts/ci/test-non-cypher-surface-gate.py index d1155b81..ae077444 100644 --- a/scripts/ci/test-non-cypher-surface-gate.py +++ b/scripts/ci/test-non-cypher-surface-gate.py @@ -29,7 +29,7 @@ def validate(self, manifest: dict) -> list[str]: def test_checked_in_inventory_is_complete(self) -> None: self.assertEqual(GATE.validate(), []) - self.assertEqual(len(GATE.public_methods()), 374) + self.assertEqual(len(GATE.public_methods()), 375) self.assertEqual(len(GATE.algorithm_registry()), 94) def test_new_or_removed_public_method_fails_frozen_digest(self) -> None: diff --git a/tests/contracts/non-cypher-rust-surface.json b/tests/contracts/non-cypher-rust-surface.json index 8dc8ee04..a98824d9 100644 --- a/tests/contracts/non-cypher-rust-surface.json +++ b/tests/contracts/non-cypher-rust-surface.json @@ -1,7 +1,7 @@ { "contract_version": 1, "scope": "Rust non-Cypher public release surface", - "public_method_digest": "499f902e4713931d7b3591fbaf978e6694c3029fde08c5c5476547563e5776e5", + "public_method_digest": "72510094f55ba39627bda6c3c5c0df6845b6c90c602aac54ab3e628274240cab", "method_policy": { "receiver_defaults": { "GraphForge": "release-tested", @@ -818,6 +818,7 @@ "GraphForge.execute_to_arrow_ipc_stream_with_params", "GraphForge.execute_to_parquet", "GraphForge.execute_to_parquet_stream_with_params", + "GraphForge.execute_to_result_sink_with_evidence", "GraphForge.execute_to_parquet_with_params", "GraphForge.execute_with_params", "GraphForge.explain", @@ -826,6 +827,10 @@ "GraphForge.schema" ], "test_refs": [ + { + "path": "crates/graphforge-api/tests/fixed_hop_limit.rs", + "symbol": "ordinary_streaming_sink_exposes_deterministic_query_evidence" + }, { "path": "crates/graphforge-api/src/composition_binding_tests.rs", "symbol": "facade_executes_qualified_and_unique_composed_queries" From b19a9267011e3d31ed78d7e384914d6402c9d937 Mon Sep 17 00:00:00 2001 From: David Spencer <1526975+DecisionNerd@users.noreply.github.com> Date: Sat, 29 Aug 2026 10:42:56 -0600 Subject: [PATCH 2/2] fix(query): clarify logical result receipt parity --- crates/graphforge-api/src/query_evidence.rs | 2 +- .../tests/non-cypher-parity-policy.json | 3 ++- docs/book/architecture/execution-model.md | 7 ++++--- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/crates/graphforge-api/src/query_evidence.rs b/crates/graphforge-api/src/query_evidence.rs index f1a81b3c..2a6d95eb 100644 --- a/crates/graphforge-api/src/query_evidence.rs +++ b/crates/graphforge-api/src/query_evidence.rs @@ -134,7 +134,7 @@ pub struct QueryOperatorRssEvidence { pub struct QuerySinkEvidenceReceipt { /// Atomic result-sink publication receipt. pub sink: ResultSinkReceipt, - /// SHA-256 of the atomically published result artifact. + /// SHA-256 of the published result's bounded logical Arrow encoding. pub result_sha256: String, /// Exact unsigned scalar for a one-row integer result representable as `u64`. pub scalar_u64: Option, diff --git a/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json b/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json index 23dc0362..14c5af2f 100644 --- a/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json +++ b/crates/graphforge-bindings-node/tests/non-cypher-parity-policy.json @@ -242,7 +242,8 @@ "GraphForge.execute_to_parquet": { "nodeMembers": [ "GraphForge.plan", - "PlanHandle.sinkParquet" + "PlanHandle.sinkParquet", + "PlanHandle.sinkArrowIpc" ], "reason": "Node uses the asynchronous PlanHandle Parquet sink." }, diff --git a/docs/book/architecture/execution-model.md b/docs/book/architecture/execution-model.md index ccd5b270..602a3ad7 100644 --- a/docs/book/architecture/execution-model.md +++ b/docs/book/architecture/execution-model.md @@ -142,9 +142,10 @@ Ordinary streaming result sinks retain the same aggregate evidence through the terminal stream boundary. `gf --json query` emits `graphforge-result-sink/2` with nested `graphforge-query-evidence/1`: named hop reader, logical-row, projection, identity-byte, TopK/spill, memory-release, and operator-RSS fields. -The receipt also includes the SHA-256 of the atomically published result and an -optional `scalar_u64` only for an exact one-row, one-`UInt64` result. Evidence is -content-free: it contains no graph values, identities, paths, or provider names. +The receipt also includes the SHA-256 of a bounded logical Arrow encoding of the +published result and an optional `scalar_u64` only for an exact one-row integer +result representable as `u64`. Evidence is content-free: it contains no graph +values, identities, paths, or provider names. **Selection is a planner choice, not an IR change.** The Graph IR is unchanged: variable-length traversal is still encoded on `Expand { …, min_hops, max_hops }`. A lowering rule selects an