1use std::collections::{BTreeSet, HashMap};
8use std::fmt;
9use std::sync::Arc;
10
11#[cfg(test)]
12use std::cell::Cell;
13
14use xlog_core::{RelId, Result, ScalarType, Schema};
15use xlog_ir::{ConstValue, ExecutionPlan, Expr, JoinType, ProjectExpr, RirNode};
16
17const RESIDENT_GRAPH_MAX_INTERMEDIATE_ARITY: usize = 17;
18
19#[derive(Debug, Clone, Default)]
21pub struct ResidentGraphSchemaCatalog {
22 by_relation: HashMap<RelId, Vec<(String, Schema)>>,
23}
24
25impl ResidentGraphSchemaCatalog {
26 pub fn from_named_schemas(entries: impl IntoIterator<Item = (String, RelId, Schema)>) -> Self {
28 let mut by_relation: HashMap<RelId, Vec<(String, Schema)>> = HashMap::new();
29 for (name, relation, schema) in entries {
30 by_relation
31 .entry(relation)
32 .or_default()
33 .push((name, schema));
34 }
35 for aliases in by_relation.values_mut() {
36 aliases.sort_by(|left, right| left.0.cmp(&right.0));
37 aliases.dedup();
38 }
39 Self { by_relation }
40 }
41
42 fn descriptor(&self, relation: RelId) -> Option<String> {
43 let aliases = self.by_relation.get(&relation)?;
44 Some(
45 aliases
46 .iter()
47 .map(|(name, schema)| format!("{name}={schema:#?}"))
48 .collect::<Vec<_>>()
49 .join("|"),
50 )
51 }
52
53 pub(crate) fn schema(&self, relation: RelId) -> Option<&Schema> {
54 let aliases = self.by_relation.get(&relation)?;
55 let (_, first) = aliases.first()?;
56 aliases
57 .iter()
58 .all(|(_, schema)| schema == first)
59 .then_some(first)
60 }
61}
62
63#[derive(Debug, Clone, PartialEq, Eq)]
65pub enum ResidentGraphDeclineReason {
66 MissingScanSchema { relation: RelId },
68 UnsupportedNode { path: String, node: &'static str },
70 UnsupportedJoin { path: String, join_type: JoinType },
72 FullStoreRequested,
75 NonOrdinaryPlan,
77 ImportedInputUnsupported { relation: String },
80 SourceSetUncertified { relation: String },
82 ConditionalGraphUnavailable { detail: String },
84 WorkspaceUnbounded { detail: String },
86}
87
88#[derive(Debug, Clone)]
90pub struct ResidentGraphRouteCertificate {
91 covered_route_descriptors: BTreeSet<String>,
92 declines: Vec<ResidentGraphDeclineReason>,
93 schema_catalog: ResidentGraphSchemaCatalog,
94 plan_fingerprint: u64,
95}
96
97#[derive(Debug, Clone)]
103pub struct ResidentGraphCertifiedPlan {
104 plan: Arc<ExecutionPlan>,
105 certificate: ResidentGraphRouteCertificate,
106}
107
108impl ResidentGraphCertifiedPlan {
109 pub fn inspect(plan: Arc<ExecutionPlan>, catalog: &ResidentGraphSchemaCatalog) -> Result<Self> {
111 let certificate = ResidentGraphRouteCertificate::inspect(&plan, catalog)?;
112 Ok(Self { plan, certificate })
113 }
114
115 pub fn plan(&self) -> &ExecutionPlan {
117 &self.plan
118 }
119
120 pub fn certificate(&self) -> &ResidentGraphRouteCertificate {
122 &self.certificate
123 }
124}
125
126#[cfg(test)]
127thread_local! {
128 static RESIDENT_ROUTE_INSPECTION_COUNT: Cell<usize> = const { Cell::new(0) };
129}
130
131#[cfg(test)]
132pub(crate) fn reset_resident_route_inspection_count() {
133 RESIDENT_ROUTE_INSPECTION_COUNT.with(|count| count.set(0));
134}
135
136#[cfg(test)]
137pub(crate) fn resident_route_inspection_count() -> usize {
138 RESIDENT_ROUTE_INSPECTION_COUNT.with(Cell::get)
139}
140
141impl ResidentGraphRouteCertificate {
142 pub fn inspect(plan: &ExecutionPlan, catalog: &ResidentGraphSchemaCatalog) -> Result<Self> {
144 #[cfg(test)]
145 RESIDENT_ROUTE_INSPECTION_COUNT.with(|count| count.set(count.get() + 1));
146 let mut certificate = Self {
147 covered_route_descriptors: BTreeSet::new(),
148 declines: Vec::new(),
149 schema_catalog: catalog.clone(),
150 plan_fingerprint: 0,
151 };
152 certificate
153 .covered_route_descriptors
154 .insert(format!("plan;scc_count={}", plan.sccs.len()));
155 for (scc_index, scc) in plan.sccs.iter().enumerate() {
156 certificate.covered_route_descriptors.insert(format!(
157 "plan;scc={scc_index};id={};recursive={};predicate_count={}",
158 scc.id,
159 scc.is_recursive,
160 scc.predicates.len()
161 ));
162 for (predicate_index, predicate) in scc.predicates.iter().enumerate() {
163 certificate.covered_route_descriptors.insert(format!(
164 "plan;scc={scc_index};predicate={predicate_index};name={predicate:?}"
165 ));
166 }
167 }
168 certificate
169 .covered_route_descriptors
170 .insert(format!("plan;stratum_count={}", plan.strata.len()));
171 for (stratum_index, stratum) in plan.strata.iter().enumerate() {
172 certificate.covered_route_descriptors.insert(format!(
173 "plan;stratum={stratum_index};id={};scc_count={}",
174 stratum.id,
175 stratum.sccs.len()
176 ));
177 for (scc_position, scc_id) in stratum.sccs.iter().enumerate() {
178 certificate.covered_route_descriptors.insert(format!(
179 "plan;stratum={stratum_index};scc={scc_position};id={scc_id}"
180 ));
181 }
182 }
183 certificate.covered_route_descriptors.insert(format!(
184 "plan;rules_by_scc_count={}",
185 plan.rules_by_scc.len()
186 ));
187 certificate.covered_route_descriptors.insert(format!(
188 "plan;generated_query_rule_count={}",
189 plan.generated_query_rules.len()
190 ));
191 for (position, provenance) in plan.generated_query_rules.iter().enumerate() {
192 certificate.covered_route_descriptors.insert(format!(
193 "plan;generated_query_rule={position};query_index={};scc={};rule={}",
194 provenance.query_index, provenance.scc_index, provenance.rule_index
195 ));
196 }
197 for (scc_index, rules) in plan.rules_by_scc.iter().enumerate() {
198 certificate.covered_route_descriptors.insert(format!(
199 "plan;rules_by_scc={scc_index};rule_count={}",
200 rules.len()
201 ));
202 for (rule_index, rule) in rules.iter().enumerate() {
203 certificate.covered_route_descriptors.insert(format!(
204 "plan;rules_by_scc={scc_index};rule={rule_index};head={:?};meta={:#?};body={:#?}",
205 rule.head, rule.meta, rule.body
206 ));
207 }
208 }
209 certificate
210 .covered_route_descriptors
211 .insert(format!("plan;est_memory_peak={}", plan.est_memory_peak));
212 certificate
213 .covered_route_descriptors
214 .insert(format!("plan;rel_arity_count={}", plan.rel_arities.len()));
215 let mut rel_arities = plan
216 .rel_arities
217 .iter()
218 .map(|(&relation, &arity)| (relation, arity))
219 .collect::<Vec<_>>();
220 rel_arities.sort_unstable_by_key(|(relation, _)| *relation);
221 for (position, (relation, arity)) in rel_arities.into_iter().enumerate() {
222 certificate.covered_route_descriptors.insert(format!(
223 "plan;rel_arity={position};relation={};arity={arity}",
224 relation.0
225 ));
226 }
227 for (scc_index, scc) in plan.sccs.iter().enumerate() {
228 let Some(rules) = plan.rules_by_scc.get(scc_index) else {
229 certificate
230 .declines
231 .push(ResidentGraphDeclineReason::UnsupportedNode {
232 path: format!("scc={scc_index}"),
233 node: "missing_rule_vector",
234 });
235 continue;
236 };
237 for (rule_index, rule) in rules.iter().enumerate() {
238 certificate.visit(
239 catalog,
240 scc_index,
241 rule_index,
242 scc.is_recursive,
243 &rule.body,
244 "primary/root",
245 );
246 let identity = format!(
247 "scc={scc_index};rule={rule_index};head={};schema={:#?}",
248 rule.head, rule.meta.schema
249 );
250 certificate
251 .covered_route_descriptors
252 .insert(format!("{identity};implicit=rule_result_union"));
253 certificate
254 .covered_route_descriptors
255 .insert(format!("{identity};implicit=full_row_dedup"));
256 if scc.is_recursive {
257 certificate
258 .covered_route_descriptors
259 .insert(format!("{identity};implicit=novel_tuple_difference"));
260 certificate
261 .covered_route_descriptors
262 .insert(format!("{identity};implicit=device_convergence"));
263 }
264 }
265 }
266 certificate.plan_fingerprint = stable_descriptor_fingerprint(
267 certificate
268 .covered_route_descriptors
269 .iter()
270 .map(String::as_bytes),
271 );
272 Ok(certificate)
273 }
274
275 pub fn is_supported(&self) -> bool {
277 self.declines.is_empty()
278 }
279
280 pub fn covered_route_descriptors(&self) -> &BTreeSet<String> {
282 &self.covered_route_descriptors
283 }
284
285 pub fn declines(&self) -> &[ResidentGraphDeclineReason] {
287 &self.declines
288 }
289
290 pub fn plan_fingerprint(&self) -> u64 {
293 self.plan_fingerprint
294 }
295
296 pub fn matches_plan(&self, plan: &ExecutionPlan) -> Result<bool> {
299 let inspected = Self::inspect(plan, &self.schema_catalog)?;
300 Ok(inspected.plan_fingerprint == self.plan_fingerprint
301 && inspected.covered_route_descriptors == self.covered_route_descriptors
302 && inspected.declines == self.declines)
303 }
304
305 pub(crate) fn schema_for(&self, relation: RelId) -> Option<&Schema> {
306 self.schema_catalog.schema(relation)
307 }
308
309 pub(crate) fn node_schema(&self, node: &RirNode) -> Option<Schema> {
310 node_schema(&self.schema_catalog, node)
311 }
312
313 fn visit(
314 &mut self,
315 catalog: &ResidentGraphSchemaCatalog,
316 scc_index: usize,
317 rule_index: usize,
318 recursive: bool,
319 node: &RirNode,
320 path: &str,
321 ) {
322 self.validate_node_route(catalog, node, path);
323 let scan_schema = match node {
324 RirNode::Scan { rel } => match catalog.descriptor(*rel) {
325 Some(descriptor) => descriptor,
326 None => {
327 self.declines
328 .push(ResidentGraphDeclineReason::MissingScanSchema { relation: *rel });
329 String::new()
330 }
331 },
332 _ => String::new(),
333 };
334 self.covered_route_descriptors.insert(format!(
335 "scc={scc_index};rule={rule_index};recursive={recursive};path={path};node={node:#?};scan_schema={scan_schema}"
336 ));
337 match node {
338 RirNode::Unit | RirNode::Scan { .. } => {}
339 RirNode::Filter { input, .. }
340 | RirNode::Project { input, .. }
341 | RirNode::Distinct { input, .. } => self.visit(
342 catalog,
343 scc_index,
344 rule_index,
345 recursive,
346 input,
347 &format!("{path}/input"),
348 ),
349 RirNode::Join {
350 left,
351 right,
352 join_type,
353 ..
354 } => {
355 if !matches!(join_type, JoinType::Inner | JoinType::Semi) {
356 self.declines
357 .push(ResidentGraphDeclineReason::UnsupportedJoin {
358 path: path.to_string(),
359 join_type: *join_type,
360 });
361 }
362 self.visit(
363 catalog,
364 scc_index,
365 rule_index,
366 recursive,
367 left,
368 &format!("{path}/left"),
369 );
370 self.visit(
371 catalog,
372 scc_index,
373 rule_index,
374 recursive,
375 right,
376 &format!("{path}/right"),
377 );
378 }
379 RirNode::ChainJoin {
380 left,
381 right,
382 fallback,
383 ..
384 } => {
385 self.visit(
386 catalog,
387 scc_index,
388 rule_index,
389 recursive,
390 left,
391 &format!("{path}/primary/left"),
392 );
393 self.visit(
394 catalog,
395 scc_index,
396 rule_index,
397 recursive,
398 right,
399 &format!("{path}/primary/right"),
400 );
401 self.visit(
402 catalog,
403 scc_index,
404 rule_index,
405 recursive,
406 fallback,
407 &format!("{path}/alternative/captured_fallback"),
408 );
409 }
410 RirNode::Union { inputs } => {
411 for (index, input) in inputs.iter().enumerate() {
412 self.visit(
413 catalog,
414 scc_index,
415 rule_index,
416 recursive,
417 input,
418 &format!("{path}/input[{index}]"),
419 );
420 }
421 }
422 RirNode::Diff { left, right } => {
423 self.visit(
424 catalog,
425 scc_index,
426 rule_index,
427 recursive,
428 left,
429 &format!("{path}/left"),
430 );
431 self.visit(
432 catalog,
433 scc_index,
434 rule_index,
435 recursive,
436 right,
437 &format!("{path}/right"),
438 );
439 }
440 RirNode::Fixpoint {
441 base,
442 recursive: step,
443 ..
444 } => {
445 self.visit(
446 catalog,
447 scc_index,
448 rule_index,
449 recursive,
450 base,
451 &format!("{path}/base"),
452 );
453 self.visit(
454 catalog,
455 scc_index,
456 rule_index,
457 recursive,
458 step,
459 &format!("{path}/recursive"),
460 );
461 }
462 RirNode::MultiWayJoin {
463 inputs, fallback, ..
464 } => {
465 self.declines
466 .push(ResidentGraphDeclineReason::UnsupportedNode {
467 path: path.to_string(),
468 node: "multi_way_join",
469 });
470 for (index, input) in inputs.iter().enumerate() {
471 self.visit(
472 catalog,
473 scc_index,
474 rule_index,
475 recursive,
476 input,
477 &format!("{path}/primary/input[{index}]"),
478 );
479 }
480 self.visit(
481 catalog,
482 scc_index,
483 rule_index,
484 recursive,
485 fallback,
486 &format!("{path}/alternative/captured_fallback"),
487 );
488 }
489 RirNode::GroupBy { input, .. } => {
490 self.declines
491 .push(ResidentGraphDeclineReason::UnsupportedNode {
492 path: path.to_string(),
493 node: "group_by",
494 });
495 self.visit(
496 catalog,
497 scc_index,
498 rule_index,
499 recursive,
500 input,
501 &format!("{path}/input"),
502 );
503 }
504 RirNode::TensorMaskedJoin { .. } => {
505 self.declines
506 .push(ResidentGraphDeclineReason::UnsupportedNode {
507 path: path.to_string(),
508 node: "tensor_masked_join",
509 })
510 }
511 }
512 }
513
514 fn validate_node_route(
515 &mut self,
516 catalog: &ResidentGraphSchemaCatalog,
517 node: &RirNode,
518 path: &str,
519 ) {
520 let schema = node_schema(catalog, node);
521 let unsupported = schema
522 .as_ref()
523 .is_some_and(|schema| schema.arity() > RESIDENT_GRAPH_MAX_INTERMEDIATE_ARITY)
524 .then_some("intermediate_arity")
525 .or_else(|| {
526 schema
527 .as_ref()
528 .is_some_and(|schema| {
529 schema
530 .columns
531 .iter()
532 .any(|(_, scalar)| !resident_scalar_supported(*scalar))
533 })
534 .then_some("scalar_type")
535 })
536 .or_else(|| match node {
537 RirNode::Filter { input, predicate } => node_schema(catalog, input)
538 .filter(|schema| resident_predicate_supported(predicate, schema))
539 .is_none()
540 .then_some("filter_expression"),
541 RirNode::Project { input, columns } => node_schema(catalog, input)
542 .filter(|schema| resident_projection_supported(columns, schema))
543 .is_none()
544 .then_some("project_expression"),
545 RirNode::Join {
546 left,
547 right,
548 left_keys,
549 right_keys,
550 join_type,
551 } if matches!(join_type, JoinType::Inner | JoinType::Semi) => {
552 let compatible = (|| {
553 let left_schema = node_schema(catalog, left)?;
554 let right_schema = node_schema(catalog, right)?;
555 if left_keys.len() != 1 || right_keys.len() != 1 {
556 return None;
557 }
558 let left_ty = left_schema.column_type(left_keys[0])?;
559 let right_ty = right_schema.column_type(right_keys[0])?;
560 (left_ty == right_ty && resident_scalar_supported(left_ty)).then_some(())
561 })()
562 .is_some();
563 (!compatible).then_some("join_key_layout")
564 }
565 RirNode::ChainJoin {
566 left,
567 right,
568 left_key,
569 right_key,
570 output_columns,
571 ..
572 } => {
573 let compatible = (|| {
574 let left_schema = node_schema(catalog, left)?;
575 let right_schema = node_schema(catalog, right)?;
576 let left_ty = left_schema.column_type(*left_key)?;
577 let right_ty = right_schema.column_type(*right_key)?;
578 if left_ty != right_ty || !resident_scalar_supported(left_ty) {
579 return None;
580 }
581 let joined = joined_schema(&left_schema, &right_schema);
582 resident_projection_supported(output_columns, &joined).then_some(())
583 })()
584 .is_some();
585 (!compatible).then_some("chain_join_layout")
586 }
587 RirNode::Distinct { input, key_cols } => node_schema(catalog, input)
588 .filter(|schema| key_cols.iter().copied().eq(0..schema.arity()))
589 .is_none()
590 .then_some("partial_row_distinct"),
591 _ => None,
592 });
593 if let Some(node) = unsupported {
594 self.declines
595 .push(ResidentGraphDeclineReason::UnsupportedNode {
596 path: path.to_string(),
597 node,
598 });
599 }
600 }
601}
602
603fn resident_scalar_supported(scalar: ScalarType) -> bool {
604 matches!(
605 scalar,
606 ScalarType::Symbol | ScalarType::U32 | ScalarType::U64
607 )
608}
609
610fn constant_scalar(value: &ConstValue) -> ScalarType {
611 match value {
612 ConstValue::U32(_) => ScalarType::U32,
613 ConstValue::U64(_) => ScalarType::U64,
614 ConstValue::I32(_) => ScalarType::I32,
615 ConstValue::I64(_) => ScalarType::I64,
616 ConstValue::F32(_) => ScalarType::F32,
617 ConstValue::F64(_) => ScalarType::F64,
618 ConstValue::Bool(_) => ScalarType::Bool,
619 ConstValue::Symbol(_) => ScalarType::Symbol,
620 }
621}
622
623fn scalar_expression_type(expression: &Expr, schema: &Schema) -> Option<ScalarType> {
624 match expression {
625 Expr::Column(index) => schema.column_type(*index),
626 Expr::Const(value) => Some(constant_scalar(value)),
627 _ => None,
628 }
629}
630
631fn resident_predicate_supported(expression: &Expr, schema: &Schema) -> bool {
632 match expression {
633 Expr::Compare { left, right, .. } => {
634 let Some(left_ty) = scalar_expression_type(left, schema) else {
635 return false;
636 };
637 let Some(right_ty) = scalar_expression_type(right, schema) else {
638 return false;
639 };
640 left_ty == right_ty && resident_scalar_supported(left_ty)
641 }
642 Expr::And(expressions) => {
643 !expressions.is_empty()
644 && expressions
645 .iter()
646 .all(|expression| resident_predicate_supported(expression, schema))
647 }
648 _ => false,
649 }
650}
651
652fn resident_projection_supported(expressions: &[ProjectExpr], schema: &Schema) -> bool {
653 expressions.iter().all(|expression| match expression {
654 ProjectExpr::Column(index) => schema
655 .column_type(*index)
656 .is_some_and(resident_scalar_supported),
657 ProjectExpr::Computed(Expr::Const(value), declared) => {
658 *declared == constant_scalar(value) && resident_scalar_supported(*declared)
659 }
660 _ => false,
661 })
662}
663
664fn joined_schema(left: &Schema, right: &Schema) -> Schema {
665 Schema::new(
666 left.columns
667 .iter()
668 .chain(&right.columns)
669 .enumerate()
670 .map(|(index, (_, ty))| (format!("column_{index}"), *ty))
671 .collect(),
672 )
673}
674
675pub(crate) fn node_schema(catalog: &ResidentGraphSchemaCatalog, node: &RirNode) -> Option<Schema> {
676 match node {
677 RirNode::Unit => Some(Schema::new(Vec::new())),
678 RirNode::Scan { rel } => catalog.schema(*rel).cloned(),
679 RirNode::Filter { input, .. } | RirNode::Distinct { input, .. } => {
680 node_schema(catalog, input)
681 }
682 RirNode::Project { input, columns } => {
683 let input = node_schema(catalog, input)?;
684 let output = columns
685 .iter()
686 .enumerate()
687 .map(|(index, expression)| {
688 let scalar = match expression {
689 ProjectExpr::Column(column) => input.column_type(*column),
690 ProjectExpr::Computed(_, scalar) => Some(*scalar),
691 };
692 scalar.map(|scalar| (format!("column_{index}"), scalar))
693 })
694 .collect::<Option<Vec<_>>>()?;
695 Some(Schema::new(output))
696 }
697 RirNode::Join {
698 left,
699 right,
700 join_type,
701 ..
702 } => {
703 let left = node_schema(catalog, left)?;
704 if matches!(join_type, JoinType::Semi | JoinType::Anti) {
705 Some(left)
706 } else {
707 Some(joined_schema(&left, &node_schema(catalog, right)?))
708 }
709 }
710 RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
711 node_schema(catalog, fallback)
712 }
713 RirNode::Union { inputs } => inputs.first().and_then(|input| node_schema(catalog, input)),
714 RirNode::Diff { left, .. } => node_schema(catalog, left),
715 RirNode::Fixpoint { base, .. } => node_schema(catalog, base),
716 RirNode::GroupBy { .. } | RirNode::TensorMaskedJoin { .. } => None,
717 }
718}
719
720fn stable_descriptor_fingerprint<'a>(descriptors: impl IntoIterator<Item = &'a [u8]>) -> u64 {
721 let mut hash = 0xcbf2_9ce4_8422_2325u64;
722 for descriptor in descriptors {
723 for byte in (descriptor.len() as u64)
724 .to_le_bytes()
725 .iter()
726 .chain(descriptor)
727 {
728 hash ^= u64::from(*byte);
729 hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
730 }
731 }
732 hash
733}
734
735#[derive(Debug, Clone, PartialEq, Eq)]
737pub enum ResidentGraphDeviceStatus {
738 Success { iterations: u32 },
740 IterationLimit { limit: u32, completed: u32 },
742 CapacityOverflow {
744 op_id: u32,
745 required: u64,
746 capacity: u64,
747 },
748 ResourceExhausted {
750 op_id: u32,
751 resource: &'static str,
752 required: u64,
753 capacity: u64,
754 },
755}
756
757#[derive(Debug, Clone, PartialEq, Eq)]
759pub enum ResidentGraphExecutionError {
760 Declined(ResidentGraphDeclineReason),
762 IterationLimit { limit: u32, completed: u32 },
764 CapacityOverflow {
766 op_id: u32,
767 required: u64,
768 capacity: u64,
769 },
770 ResourceExhausted {
772 op_id: u32,
773 resource: &'static str,
774 required: u64,
775 capacity: u64,
776 },
777 Runtime(String),
779}
780
781impl fmt::Display for ResidentGraphExecutionError {
782 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
783 write!(formatter, "{self:?}")
784 }
785}
786
787impl std::error::Error for ResidentGraphExecutionError {}
788
789#[derive(Debug, Clone)]
792pub struct ResidentGraphDeviceStatusTestInjection {
793 pub(crate) after_op: u32,
794 pub(crate) status: ResidentGraphDeviceStatus,
795}
796
797impl ResidentGraphDeviceStatusTestInjection {
798 pub fn device_kernel_after_op(after_op: u32, status: ResidentGraphDeviceStatus) -> Self {
800 Self { after_op, status }
801 }
802}
803
804#[derive(Debug, Clone, Default)]
806pub struct ResidentGraphPrepareOptions {
807 pub(crate) test_device_status: Option<ResidentGraphDeviceStatusTestInjection>,
808 pub(crate) latency_diagnostic_sample: Option<u64>,
809}
810
811impl ResidentGraphPrepareOptions {
812 pub fn with_test_device_status(
814 mut self,
815 injection: ResidentGraphDeviceStatusTestInjection,
816 ) -> Self {
817 self.test_device_status = Some(injection);
818 self
819 }
820
821 #[doc(hidden)]
823 pub fn with_latency_diagnostic_sample(mut self, sample: u64) -> Self {
824 self.latency_diagnostic_sample = Some(sample);
825 self
826 }
827}
828
829#[doc(hidden)]
831#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
832pub struct ResidentGraphPrepareDiagnosticSnapshot {
833 pub(crate) sample: u64,
834 pub(crate) total_ns: u64,
835 pub(crate) admission_and_source_snapshot_ns: u64,
836 pub(crate) execution_domain_and_build_setup_ns: u64,
837 pub(crate) logical_schedule_planning_ns: u64,
838 pub(crate) manifest_compact_construction_ns: u64,
839 pub(crate) schedule_lowering_ns: u64,
840 pub(crate) reservation_ns: u64,
841 pub(crate) relation_preparation_ns: u64,
842 pub(crate) count_initialization_ns: u64,
843 pub(crate) workspace_preparation_ns: u64,
844 pub(crate) metadata_binding_construction_ns: u64,
845 pub(crate) metadata_preparation_ns: u64,
846 pub(crate) reservation_validation_and_release_ns: u64,
847 pub(crate) pinned_receipt_ns: u64,
848 pub(crate) graph_body_capture_ns: u64,
849 pub(crate) graph_instantiate_ns: u64,
850 pub(crate) validation_owner_assembly_ns: u64,
851 pub(crate) unattributed_ns: u64,
852 pub(crate) required_reservation_bytes: u64,
853 pub(crate) logical_relation_values: u64,
854 pub(crate) physical_relation_slots: u64,
855 pub(crate) relation_device_allocation_calls: u64,
856 pub(crate) relation_reserved_bytes: u64,
857 pub(crate) relation_slot_preparation_ns_max: u64,
858 pub(crate) count_memset_calls: u64,
859 pub(crate) count_memset_bytes: u64,
860 pub(crate) count_initialization_ns_max: u64,
861 pub(crate) workspace_provider_calls: u64,
862 pub(crate) workspace_reserved_bytes: u64,
863 pub(crate) filter_scratch_preparation_ns: u64,
864 pub(crate) filter_scratch_reserved_bytes: u64,
865 pub(crate) set_workspace_preparation_ns: u64,
866 pub(crate) set_workspace_reserved_bytes: u64,
867 pub(crate) join_workspace_preparation_ns: u64,
868 pub(crate) join_workspace_reserved_bytes: u64,
869 pub(crate) control_preparation_ns: u64,
870 pub(crate) control_reserved_bytes: u64,
871 pub(crate) metadata_provider_calls: u64,
872 pub(crate) metadata_reserved_bytes: u64,
873 pub(crate) metadata_initial_htod_calls: u64,
874 pub(crate) metadata_initial_htod_bytes: u64,
875 pub(crate) device_trace_preparation_ns: u64,
876 pub(crate) device_trace_reserved_bytes: u64,
877 pub(crate) device_trace_initial_htod_calls: u64,
878 pub(crate) device_trace_initial_htod_bytes: u64,
879 pub(crate) schema_winners_preparation_ns: u64,
880 pub(crate) schema_winners_reserved_bytes: u64,
881 pub(crate) schema_winners_initial_htod_calls: u64,
882 pub(crate) schema_winners_initial_htod_bytes: u64,
883 pub(crate) receipt_preparation_ns: u64,
884 pub(crate) receipt_reserved_bytes: u64,
885 pub(crate) receipt_initial_htod_calls: u64,
886 pub(crate) receipt_initial_htod_bytes: u64,
887 pub(crate) schedule_program_preparation_ns: u64,
888 pub(crate) schedule_program_reserved_bytes: u64,
889 pub(crate) schedule_program_initial_htod_calls: u64,
890 pub(crate) schedule_program_initial_htod_bytes: u64,
891 pub(crate) compact_ops: u64,
892 pub(crate) compact_waves: u64,
893 pub(crate) compact_regions: u64,
894 pub(crate) conditional_regions: u64,
895 pub(crate) parent_graph_nodes: u64,
896 pub(crate) conditional_body_nodes: u64,
897}
898
899impl ResidentGraphPrepareDiagnosticSnapshot {
900 #[doc(hidden)]
902 pub fn format_line(self) -> String {
903 let Self {
904 sample,
905 total_ns,
906 admission_and_source_snapshot_ns,
907 execution_domain_and_build_setup_ns,
908 logical_schedule_planning_ns,
909 manifest_compact_construction_ns,
910 schedule_lowering_ns,
911 reservation_ns,
912 relation_preparation_ns,
913 count_initialization_ns,
914 workspace_preparation_ns,
915 metadata_binding_construction_ns,
916 metadata_preparation_ns,
917 reservation_validation_and_release_ns,
918 pinned_receipt_ns,
919 graph_body_capture_ns,
920 graph_instantiate_ns,
921 validation_owner_assembly_ns,
922 unattributed_ns,
923 required_reservation_bytes,
924 logical_relation_values,
925 physical_relation_slots,
926 relation_device_allocation_calls,
927 relation_reserved_bytes,
928 relation_slot_preparation_ns_max,
929 count_memset_calls,
930 count_memset_bytes,
931 count_initialization_ns_max,
932 workspace_provider_calls,
933 workspace_reserved_bytes,
934 filter_scratch_preparation_ns,
935 filter_scratch_reserved_bytes,
936 set_workspace_preparation_ns,
937 set_workspace_reserved_bytes,
938 join_workspace_preparation_ns,
939 join_workspace_reserved_bytes,
940 control_preparation_ns,
941 control_reserved_bytes,
942 metadata_provider_calls,
943 metadata_reserved_bytes,
944 metadata_initial_htod_calls,
945 metadata_initial_htod_bytes,
946 device_trace_preparation_ns,
947 device_trace_reserved_bytes,
948 device_trace_initial_htod_calls,
949 device_trace_initial_htod_bytes,
950 schema_winners_preparation_ns,
951 schema_winners_reserved_bytes,
952 schema_winners_initial_htod_calls,
953 schema_winners_initial_htod_bytes,
954 receipt_preparation_ns,
955 receipt_reserved_bytes,
956 receipt_initial_htod_calls,
957 receipt_initial_htod_bytes,
958 schedule_program_preparation_ns,
959 schedule_program_reserved_bytes,
960 schedule_program_initial_htod_calls,
961 schedule_program_initial_htod_bytes,
962 compact_ops,
963 compact_waves,
964 compact_regions,
965 conditional_regions,
966 parent_graph_nodes,
967 conditional_body_nodes,
968 } = self;
969 format!(
970 "resident prepare phases: sample={sample} total_ns={total_ns} admission_and_source_snapshot_ns={admission_and_source_snapshot_ns} execution_domain_and_build_setup_ns={execution_domain_and_build_setup_ns} logical_schedule_planning_ns={logical_schedule_planning_ns} manifest_compact_construction_ns={manifest_compact_construction_ns} schedule_lowering_ns={schedule_lowering_ns} reservation_ns={reservation_ns} relation_preparation_ns={relation_preparation_ns} count_initialization_ns={count_initialization_ns} workspace_preparation_ns={workspace_preparation_ns} metadata_binding_construction_ns={metadata_binding_construction_ns} metadata_preparation_ns={metadata_preparation_ns} reservation_validation_and_release_ns={reservation_validation_and_release_ns} pinned_receipt_ns={pinned_receipt_ns} graph_body_capture_ns={graph_body_capture_ns} graph_instantiate_ns={graph_instantiate_ns} validation_owner_assembly_ns={validation_owner_assembly_ns} unattributed_ns={unattributed_ns} required_reservation_bytes={required_reservation_bytes} logical_relation_values={logical_relation_values} physical_relation_slots={physical_relation_slots} relation_device_allocation_calls={relation_device_allocation_calls} relation_reserved_bytes={relation_reserved_bytes} relation_slot_preparation_ns_max={relation_slot_preparation_ns_max} count_memset_calls={count_memset_calls} count_memset_bytes={count_memset_bytes} count_initialization_ns_max={count_initialization_ns_max} workspace_provider_calls={workspace_provider_calls} workspace_reserved_bytes={workspace_reserved_bytes} filter_scratch_preparation_ns={filter_scratch_preparation_ns} filter_scratch_reserved_bytes={filter_scratch_reserved_bytes} set_workspace_preparation_ns={set_workspace_preparation_ns} set_workspace_reserved_bytes={set_workspace_reserved_bytes} join_workspace_preparation_ns={join_workspace_preparation_ns} join_workspace_reserved_bytes={join_workspace_reserved_bytes} control_preparation_ns={control_preparation_ns} control_reserved_bytes={control_reserved_bytes} metadata_provider_calls={metadata_provider_calls} metadata_reserved_bytes={metadata_reserved_bytes} metadata_initial_htod_calls={metadata_initial_htod_calls} metadata_initial_htod_bytes={metadata_initial_htod_bytes} device_trace_preparation_ns={device_trace_preparation_ns} device_trace_reserved_bytes={device_trace_reserved_bytes} device_trace_initial_htod_calls={device_trace_initial_htod_calls} device_trace_initial_htod_bytes={device_trace_initial_htod_bytes} schema_winners_preparation_ns={schema_winners_preparation_ns} schema_winners_reserved_bytes={schema_winners_reserved_bytes} schema_winners_initial_htod_calls={schema_winners_initial_htod_calls} schema_winners_initial_htod_bytes={schema_winners_initial_htod_bytes} receipt_preparation_ns={receipt_preparation_ns} receipt_reserved_bytes={receipt_reserved_bytes} receipt_initial_htod_calls={receipt_initial_htod_calls} receipt_initial_htod_bytes={receipt_initial_htod_bytes} schedule_program_preparation_ns={schedule_program_preparation_ns} schedule_program_reserved_bytes={schedule_program_reserved_bytes} schedule_program_initial_htod_calls={schedule_program_initial_htod_calls} schedule_program_initial_htod_bytes={schedule_program_initial_htod_bytes} compact_ops={compact_ops} compact_waves={compact_waves} compact_regions={compact_regions} conditional_regions={conditional_regions} parent_graph_nodes={parent_graph_nodes} conditional_body_nodes={conditional_body_nodes} deallocation_calls=unavailable"
971 )
972 }
973}
974
975#[cfg(test)]
976fn resident_graph_prepare_diagnostic_line(
977 snapshot: Option<ResidentGraphPrepareDiagnosticSnapshot>,
978) -> Option<String> {
979 snapshot.map(ResidentGraphPrepareDiagnosticSnapshot::format_line)
980}
981
982#[derive(Debug, Clone, Copy, PartialEq, Eq)]
984pub enum ResidentGraphSelectionKind {
985 ExistingGpu,
987 ResidentConditionalGraph,
989}
990
991#[derive(Debug, Clone, Default)]
993pub struct ResidentGraphCoreTransferStats {
994 pub tracked_htod_calls: u64,
995 pub tracked_htod_bytes: u64,
996 pub tracked_dtoh_calls: u64,
997 pub tracked_dtoh_bytes: u64,
998 pub provider_dtoh_calls: u64,
999 pub untracked_metadata_dtoh_calls: u64,
1000}
1001
1002#[derive(Debug, Clone, Default)]
1004pub struct ResidentGraphFinalObservationStats {
1005 pub dtoh_calls: u64,
1006 pub dtoh_bytes: u64,
1007 pub pinned_receipts: u64,
1008}
1009
1010#[derive(Debug, Clone, Default)]
1012pub struct ResidentGraphDeferredProfile {
1013 pub timed_scan_filter_invocations: u64,
1014 pub device_elapsed_ns: u64,
1015 pub final_sync_misattributed_ns: u64,
1016}
1017
1018#[derive(Debug, Clone)]
1020pub struct ResidentGraphExecutionStats {
1021 pub selection: ResidentGraphSelectionKind,
1022 pub decline: Option<ResidentGraphDeclineReason>,
1023 pub conditional_graph_launches: u64,
1024 pub terminal_synchronizations: u64,
1025 pub host_iterations: u64,
1026 pub host_allocations: u64,
1027 pub host_status_injections: u64,
1028 pub deterministic_d2h_violations: u64,
1029 pub host_dispatched_scan_ops: u64,
1030 pub host_dispatched_filter_ops: u64,
1031 pub device_scan_invocations: u64,
1033 pub device_filter_invocations: u64,
1035 pub semantic_scan_invocations: u64,
1038 pub semantic_filter_invocations: u64,
1041 pub staged_store_mutations: u64,
1042 pub deferred_profile: ResidentGraphDeferredProfile,
1043 pub core_transfers: ResidentGraphCoreTransferStats,
1044 pub final_observation: ResidentGraphFinalObservationStats,
1045}
1046
1047impl ResidentGraphExecutionStats {
1048 pub fn declined(reason: ResidentGraphDeclineReason) -> Self {
1050 Self {
1051 selection: ResidentGraphSelectionKind::ExistingGpu,
1052 decline: Some(reason),
1053 conditional_graph_launches: 0,
1054 terminal_synchronizations: 0,
1055 host_iterations: 0,
1056 host_allocations: 0,
1057 host_status_injections: 0,
1058 deterministic_d2h_violations: 0,
1059 host_dispatched_scan_ops: 0,
1060 host_dispatched_filter_ops: 0,
1061 device_scan_invocations: 0,
1062 device_filter_invocations: 0,
1063 semantic_scan_invocations: 0,
1064 semantic_filter_invocations: 0,
1065 staged_store_mutations: 0,
1066 deferred_profile: ResidentGraphDeferredProfile::default(),
1067 core_transfers: ResidentGraphCoreTransferStats::default(),
1068 final_observation: ResidentGraphFinalObservationStats::default(),
1069 }
1070 }
1071}
1072
1073#[cfg(test)]
1074mod tests {
1075 use super::*;
1076 use std::sync::Arc;
1077 use xlog_ir::{CompiledRule, GeneratedQueryRuleProvenance, RirMeta, Scc, Stratum};
1078
1079 fn schema(columns: &[(&str, ScalarType)]) -> Schema {
1080 Schema::new(
1081 columns
1082 .iter()
1083 .map(|(name, scalar)| ((*name).to_owned(), *scalar))
1084 .collect(),
1085 )
1086 }
1087
1088 fn rule(head: &str, body: RirNode, columns: &[(&str, ScalarType)]) -> CompiledRule {
1089 let mut meta = RirMeta::default();
1090 meta.schema = schema(columns);
1091 CompiledRule {
1092 head: head.to_owned(),
1093 body,
1094 meta,
1095 }
1096 }
1097
1098 fn chain_join() -> RirNode {
1099 RirNode::ChainJoin {
1100 left: Box::new(RirNode::Scan { rel: RelId(1) }),
1101 right: Box::new(RirNode::Scan { rel: RelId(1) }),
1102 left_key: 0,
1103 right_key: 0,
1104 output_columns: vec![ProjectExpr::Column(0)],
1105 fallback: Box::new(RirNode::Join {
1106 left: Box::new(RirNode::Scan { rel: RelId(1) }),
1107 right: Box::new(RirNode::Scan { rel: RelId(1) }),
1108 left_keys: vec![0],
1109 right_keys: vec![0],
1110 join_type: JoinType::Inner,
1111 }),
1112 }
1113 }
1114
1115 fn representative_plan() -> ExecutionPlan {
1116 let mut rel_arities = HashMap::new();
1117 rel_arities.insert(RelId(9), 0);
1118 rel_arities.insert(RelId(1), 2);
1119 ExecutionPlan {
1120 sccs: vec![
1121 Scc {
1122 id: 10,
1123 predicates: vec!["alpha".to_owned(), "beta".to_owned()],
1124 is_recursive: true,
1125 },
1126 Scc {
1127 id: 20,
1128 predicates: Vec::new(),
1129 is_recursive: false,
1130 },
1131 Scc {
1132 id: 30,
1133 predicates: vec!["gamma".to_owned()],
1134 is_recursive: false,
1135 },
1136 Scc {
1137 id: 40,
1138 predicates: Vec::new(),
1139 is_recursive: false,
1140 },
1141 ],
1142 strata: vec![
1143 Stratum {
1144 id: 0,
1145 sccs: vec![10],
1146 },
1147 Stratum {
1148 id: 1,
1149 sccs: Vec::new(),
1150 },
1151 Stratum {
1152 id: 2,
1153 sccs: vec![20, 30, 40],
1154 },
1155 ],
1156 rules_by_scc: vec![
1157 vec![
1158 rule(
1159 "alpha",
1160 RirNode::Scan { rel: RelId(1) },
1161 &[("left", ScalarType::U32), ("right", ScalarType::Symbol)],
1162 ),
1163 rule("beta", chain_join(), &[("value", ScalarType::U64)]),
1164 ],
1165 Vec::new(),
1166 vec![rule("gamma", RirNode::Unit, &[])],
1167 Vec::new(),
1168 ],
1169 generated_query_rules: vec![],
1170 est_memory_peak: 4_096,
1171 rel_arities,
1172 }
1173 }
1174
1175 fn catalog() -> ResidentGraphSchemaCatalog {
1176 ResidentGraphSchemaCatalog::from_named_schemas([(
1177 "input".to_owned(),
1178 RelId(1),
1179 schema(&[("left", ScalarType::U32), ("right", ScalarType::Symbol)]),
1180 )])
1181 }
1182
1183 fn assert_mutation_rejected(mutate: impl FnOnce(&mut ExecutionPlan)) {
1184 let plan = representative_plan();
1185 let certificate = ResidentGraphRouteCertificate::inspect(&plan, &catalog()).unwrap();
1186 let mut mutated = plan.clone();
1187 mutate(&mut mutated);
1188 assert!(!certificate.matches_plan(&mutated).unwrap());
1189 }
1190
1191 fn assert_generated_query_provenance_mutation_rejected(
1192 mutate: impl FnOnce(&mut GeneratedQueryRuleProvenance),
1193 ) {
1194 let mut plan = representative_plan();
1195 plan.sccs[0].predicates[0] = "__xlog_query_0".to_owned();
1196 plan.rules_by_scc[0][0].head = "__xlog_query_0".to_owned();
1197 plan.generated_query_rules = vec![GeneratedQueryRuleProvenance {
1198 query_index: 0,
1199 scc_index: 0,
1200 rule_index: 0,
1201 }];
1202 let certificate = ResidentGraphRouteCertificate::inspect(&plan, &catalog()).unwrap();
1203 mutate(&mut plan.generated_query_rules[0]);
1204 assert!(!certificate.matches_plan(&plan).unwrap());
1205 }
1206
1207 #[test]
1208 fn certified_plan_seals_the_exact_arc_and_certificate() {
1209 let plan = Arc::new(representative_plan());
1210 let certified = ResidentGraphCertifiedPlan::inspect(Arc::clone(&plan), &catalog()).unwrap();
1211
1212 assert!(std::ptr::eq(Arc::as_ptr(&plan), certified.plan()));
1213 assert!(certified
1214 .certificate()
1215 .matches_plan(certified.plan())
1216 .unwrap());
1217
1218 let mut independent_mutation = (*plan).clone();
1219 independent_mutation.est_memory_peak += 1;
1220 assert!(!certified
1221 .certificate()
1222 .matches_plan(&independent_mutation)
1223 .unwrap());
1224 }
1225
1226 #[test]
1227 fn prepare_latency_diagnostics_are_explicit_and_sample_bound() {
1228 let disabled = ResidentGraphPrepareOptions::default();
1229 assert_eq!(disabled.latency_diagnostic_sample, None);
1230
1231 let enabled = disabled.with_latency_diagnostic_sample(7);
1232 assert_eq!(enabled.latency_diagnostic_sample, Some(7));
1233 }
1234
1235 #[test]
1236 fn prepare_latency_diagnostic_format_is_aligned_and_default_off_is_silent() {
1237 let mut snapshot = ResidentGraphPrepareDiagnosticSnapshot {
1238 sample: 17,
1239 total_ns: 101,
1240 admission_and_source_snapshot_ns: 102,
1241 execution_domain_and_build_setup_ns: 109,
1242 metadata_binding_construction_ns: 110,
1243 reservation_validation_and_release_ns: 111,
1244 relation_device_allocation_calls: 103,
1245 workspace_provider_calls: 104,
1246 metadata_initial_htod_calls: 105,
1247 metadata_initial_htod_bytes: 106,
1248 graph_instantiate_ns: 107,
1249 unattributed_ns: 108,
1250 ..ResidentGraphPrepareDiagnosticSnapshot::default()
1251 };
1252 snapshot.schema_winners_initial_htod_calls = 1;
1253 snapshot.schema_winners_initial_htod_bytes = 12;
1254
1255 let line = resident_graph_prepare_diagnostic_line(Some(snapshot))
1256 .expect("enabled diagnostics produce one line");
1257 for expected in [
1258 "sample=17",
1259 "total_ns=101",
1260 "admission_and_source_snapshot_ns=102",
1261 "execution_domain_and_build_setup_ns=109",
1262 "metadata_binding_construction_ns=110",
1263 "reservation_validation_and_release_ns=111",
1264 "relation_device_allocation_calls=103",
1265 "workspace_provider_calls=104",
1266 "metadata_initial_htod_calls=105",
1267 "metadata_initial_htod_bytes=106",
1268 "schema_winners_initial_htod_calls=1",
1269 "schema_winners_initial_htod_bytes=12",
1270 "graph_instantiate_ns=107",
1271 "unattributed_ns=108",
1272 ] {
1273 assert!(line.split_ascii_whitespace().any(|field| field == expected));
1274 }
1275 assert_eq!(line.matches("sample=17").count(), 1);
1276 assert!(!line.contains("admission_source_validation_ns="));
1277 assert_eq!(resident_graph_prepare_diagnostic_line(None), None);
1278 }
1279
1280 #[test]
1281 fn certificate_binds_strata_vector_length() {
1282 assert_mutation_rejected(|plan| {
1283 plan.strata.push(Stratum {
1284 id: 3,
1285 sccs: Vec::new(),
1286 });
1287 });
1288 }
1289
1290 #[test]
1291 fn certificate_binds_empty_stratum_position() {
1292 assert_mutation_rejected(|plan| plan.strata.swap(1, 2));
1293 }
1294
1295 #[test]
1296 fn certificate_binds_stratum_id() {
1297 assert_mutation_rejected(|plan| plan.strata[1].id = 99);
1298 }
1299
1300 #[test]
1301 fn certificate_binds_ordered_stratum_scc_membership() {
1302 assert_mutation_rejected(|plan| plan.strata[2].sccs.swap(0, 1));
1303 }
1304
1305 #[test]
1306 fn certificate_binds_scc_vector_length_including_empty_sccs() {
1307 assert_mutation_rejected(|plan| {
1308 plan.sccs.pop();
1309 });
1310 }
1311
1312 #[test]
1313 fn certificate_binds_scc_id() {
1314 assert_mutation_rejected(|plan| plan.sccs[1].id = 99);
1315 }
1316
1317 #[test]
1318 fn certificate_binds_ordered_scc_predicate_membership() {
1319 assert_mutation_rejected(|plan| plan.sccs[0].predicates.swap(0, 1));
1320 }
1321
1322 #[test]
1323 fn certificate_binds_scc_recursive_flag() {
1324 assert_mutation_rejected(|plan| plan.sccs[1].is_recursive = true);
1325 }
1326
1327 #[test]
1328 fn certificate_binds_rules_by_scc_vector_length() {
1329 assert_mutation_rejected(|plan| plan.rules_by_scc.push(Vec::new()));
1330 }
1331
1332 #[test]
1333 fn certificate_binds_rule_occurrence_count() {
1334 assert_mutation_rejected(|plan| {
1335 plan.rules_by_scc[0].pop();
1336 });
1337 }
1338
1339 #[test]
1340 fn certificate_binds_rule_order() {
1341 assert_mutation_rejected(|plan| plan.rules_by_scc[0].swap(0, 1));
1342 }
1343
1344 #[test]
1345 fn certificate_binds_generated_query_provenance_omission() {
1346 let mut plan = representative_plan();
1347 plan.generated_query_rules = vec![GeneratedQueryRuleProvenance {
1348 query_index: 0,
1349 scc_index: 0,
1350 rule_index: 0,
1351 }];
1352 let certificate = ResidentGraphRouteCertificate::inspect(&plan, &catalog()).unwrap();
1353 plan.generated_query_rules.clear();
1354 assert!(!certificate.matches_plan(&plan).unwrap());
1355 }
1356
1357 #[test]
1358 fn certificate_binds_generated_query_index() {
1359 assert_generated_query_provenance_mutation_rejected(|provenance| {
1360 provenance.query_index = 1;
1361 });
1362 }
1363
1364 #[test]
1365 fn certificate_binds_generated_query_rule_position() {
1366 assert_generated_query_provenance_mutation_rejected(|provenance| {
1367 provenance.rule_index = 1;
1368 });
1369 }
1370
1371 #[test]
1372 fn certificate_binds_rule_schema_type() {
1373 assert_mutation_rejected(|plan| {
1374 plan.rules_by_scc[0][0].meta.schema =
1375 schema(&[("left", ScalarType::U64), ("right", ScalarType::Symbol)]);
1376 });
1377 }
1378
1379 #[test]
1380 fn certificate_binds_rule_schema_arity() {
1381 assert_mutation_rejected(|plan| {
1382 plan.rules_by_scc[0][0].meta.schema = schema(&[("left", ScalarType::U32)]);
1383 });
1384 }
1385
1386 #[test]
1387 fn certificate_binds_primary_rir() {
1388 assert_mutation_rejected(|plan| plan.rules_by_scc[0][0].body = RirNode::Unit);
1389 }
1390
1391 #[test]
1392 fn certificate_binds_captured_alternative_rir() {
1393 assert_mutation_rejected(|plan| {
1394 let RirNode::ChainJoin { fallback, .. } = &mut plan.rules_by_scc[0][1].body else {
1395 panic!("representative rule must carry a captured alternative");
1396 };
1397 **fallback = RirNode::Unit;
1398 });
1399 }
1400
1401 #[test]
1402 fn certificate_binds_implicit_device_convergence_route() {
1403 let plan = representative_plan();
1404 let mut certificate = ResidentGraphRouteCertificate::inspect(&plan, &catalog()).unwrap();
1405 let convergence = certificate
1406 .covered_route_descriptors
1407 .iter()
1408 .find(|descriptor| {
1409 descriptor.contains("scc=0;rule=0;head=alpha;")
1410 && descriptor.ends_with(";implicit=device_convergence")
1411 })
1412 .cloned()
1413 .expect("recursive rule must certify implicit device convergence");
1414 assert!(certificate.covered_route_descriptors.remove(&convergence));
1415 certificate
1416 .covered_route_descriptors
1417 .insert(format!("{convergence};mutated"));
1418 certificate.plan_fingerprint = stable_descriptor_fingerprint(
1419 certificate
1420 .covered_route_descriptors
1421 .iter()
1422 .map(String::as_bytes),
1423 );
1424 assert!(!certificate.matches_plan(&plan).unwrap());
1425 }
1426
1427 #[test]
1428 fn certificate_binds_rel_arities_vector_length() {
1429 assert_mutation_rejected(|plan| {
1430 plan.rel_arities.insert(RelId(7), 3);
1431 });
1432 }
1433
1434 #[test]
1435 fn certificate_binds_relation_arity() {
1436 assert_mutation_rejected(|plan| {
1437 plan.rel_arities.insert(RelId(1), 3);
1438 });
1439 }
1440
1441 #[test]
1442 fn certificate_rel_arities_are_order_independent() {
1443 let plan = representative_plan();
1444 let certificate = ResidentGraphRouteCertificate::inspect(&plan, &catalog()).unwrap();
1445 let mut reordered = plan.clone();
1446 reordered.rel_arities.clear();
1447 reordered.rel_arities.insert(RelId(1), 2);
1448 reordered.rel_arities.insert(RelId(9), 0);
1449 assert!(certificate.matches_plan(&reordered).unwrap());
1450 }
1451
1452 #[test]
1453 fn certificate_binds_estimated_memory_peak() {
1454 assert_mutation_rejected(|plan| plan.est_memory_peak += 1);
1455 }
1456
1457 #[test]
1458 fn certificate_rejects_intermediate_arity_above_fixed_metadata_envelope() {
1459 let columns = (0..18)
1460 .map(|index| (format!("column_{index}"), ScalarType::U32))
1461 .collect::<Vec<_>>();
1462 let wide_schema = Schema::new(columns);
1463 let plan = ExecutionPlan {
1464 sccs: vec![Scc {
1465 id: 1,
1466 predicates: vec!["wide".to_owned()],
1467 is_recursive: false,
1468 }],
1469 strata: vec![Stratum {
1470 id: 0,
1471 sccs: vec![1],
1472 }],
1473 rules_by_scc: vec![vec![CompiledRule {
1474 head: "wide".to_owned(),
1475 body: RirNode::Scan { rel: RelId(1) },
1476 meta: RirMeta {
1477 schema: wide_schema.clone(),
1478 ..RirMeta::default()
1479 },
1480 }]],
1481 generated_query_rules: vec![],
1482 est_memory_peak: 0,
1483 rel_arities: HashMap::from([(RelId(1), 18)]),
1484 };
1485 let catalog = ResidentGraphSchemaCatalog::from_named_schemas([(
1486 "wide".to_owned(),
1487 RelId(1),
1488 wide_schema,
1489 )]);
1490 let certificate = ResidentGraphRouteCertificate::inspect(&plan, &catalog).unwrap();
1491 assert!(!certificate.is_supported());
1492 assert!(certificate.declines().iter().any(|decline| matches!(
1493 decline,
1494 ResidentGraphDeclineReason::UnsupportedNode {
1495 node: "intermediate_arity",
1496 ..
1497 }
1498 )));
1499 }
1500
1501 #[test]
1502 fn certificate_accepts_intermediate_arity_at_fixed_metadata_envelope() {
1503 let columns = (0..17)
1504 .map(|index| (format!("column_{index}"), ScalarType::U32))
1505 .collect::<Vec<_>>();
1506 let wide_schema = Schema::new(columns);
1507 let plan = ExecutionPlan {
1508 sccs: vec![Scc {
1509 id: 1,
1510 predicates: vec!["wide".to_owned()],
1511 is_recursive: false,
1512 }],
1513 strata: vec![Stratum {
1514 id: 0,
1515 sccs: vec![1],
1516 }],
1517 rules_by_scc: vec![vec![CompiledRule {
1518 head: "wide".to_owned(),
1519 body: RirNode::Scan { rel: RelId(1) },
1520 meta: RirMeta {
1521 schema: wide_schema.clone(),
1522 ..RirMeta::default()
1523 },
1524 }]],
1525 generated_query_rules: vec![],
1526 est_memory_peak: 0,
1527 rel_arities: HashMap::from([(RelId(1), 17)]),
1528 };
1529 let catalog = ResidentGraphSchemaCatalog::from_named_schemas([(
1530 "wide".to_owned(),
1531 RelId(1),
1532 wide_schema,
1533 )]);
1534 let certificate = ResidentGraphRouteCertificate::inspect(&plan, &catalog).unwrap();
1535 assert!(certificate.is_supported(), "{:#?}", certificate.declines());
1536 }
1537}