1use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
2use std::marker::PhantomData;
3use std::sync::Arc;
4use std::time::Instant;
5
6use cudarc::driver::{sys::CUevent_flags, CudaEvent};
7use xlog_core::{symbol, RelId, Result, ScalarType, Schema, XlogError};
8use xlog_cuda::cuda_graph::{
9 CapturedCudaGraph, ConditionalCudaGraphSequenceBuilder, CudaConditionalGraphUnavailable,
10 CudaGraphNodeKind,
11};
12use xlog_cuda::device_runtime::{BlockId, ResidentCompletionEvent, XlogDeviceRuntime};
13use xlog_cuda::launch::LaunchRecorder;
14use xlog_cuda::memory::GpuMemoryReservation;
15use xlog_cuda::provider::resident_filter_project::{
16 resident_filter_scratch_device_bytes, ResidentFilterScratch,
17};
18use xlog_cuda::provider::resident_relational::{
19 resident_control_device_bytes, resident_device_trace_bytes,
20 resident_join_workspace_device_bytes, resident_packed_receipt_with_schema_winners_device_bytes,
21 resident_relation_device_bytes, resident_schema_winners_device_bytes,
22 resident_set_workspace_device_bytes, ResidentConvergenceControl, ResidentDeviceTrace,
23 ResidentJoinKind, ResidentJoinWorkspace, ResidentPackedReceipt, ResidentPinnedReceipt,
24 ResidentRelation, ResidentResourceCode, ResidentSchemaWinners, ResidentSetWorkspace,
25 ResidentTerminalCode, ResidentTerminalStatus,
26};
27use xlog_cuda::provider::resident_schedule::{
28 resident_schedule_metadata_device_bytes, ResidentExecutionDomain,
29 ResidentFilterComparisonDescriptor, ResidentOpDescriptor, ResidentProjectExpressionDescriptor,
30 ResidentRegionDescriptor, ResidentScheduleDeviceProgram, ResidentScheduleExternalBindings,
31 ResidentScheduleOpKind, ResidentScheduleSlotBinding, ResidentWaveDescriptor,
32 RESIDENT_SCHEDULE_OP_MARK_NOVELTY, RESIDENT_SCHEDULE_OP_MARK_SCHEMA_WINNER,
33 RESIDENT_SCHEDULE_REGION_FINALIZE, RESIDENT_SCHEDULE_REGION_INITIALIZE,
34 RESIDENT_SCHEDULE_REGION_RECURSIVE, RESIDENT_SCHEDULE_REGION_SCC_BEGIN,
35};
36use xlog_cuda::{
37 CompareOp as CudaCompareOp, CudaBuffer, CudaColumn, CudaKernelProvider, CudaStream,
38};
39use xlog_ir::{
40 CompareOp as RirCompareOp, CompiledRule, ConstValue, ExecutionPlan, Expr, JoinType,
41 ProjectExpr, RirNode,
42};
43
44use super::Executor;
45use crate::resident_graph::{
46 ResidentGraphCertifiedPlan, ResidentGraphDeclineReason, ResidentGraphDeviceStatus,
47 ResidentGraphExecutionError, ResidentGraphPrepareDiagnosticSnapshot,
48 ResidentGraphPrepareOptions, ResidentGraphRouteCertificate,
49};
50
51const MAX_RESIDENT_CAPACITY: u32 = 65_536;
52const RESIDENT_DYNAMIC_SCHEMA_ID: u32 = u32::MAX;
53
54fn resident_schemas_type_compatible(left: &Schema, right: &Schema) -> bool {
55 left.arity() == right.arity()
56 && (0..left.arity()).all(|column| left.column_type(column) == right.column_type(column))
57}
58
59fn resident_intern_schema(candidates: &mut Vec<Schema>, schema: Schema) -> Result<u32> {
60 let index = candidates
61 .iter()
62 .position(|candidate| candidate == &schema)
63 .unwrap_or_else(|| {
64 candidates.push(schema);
65 candidates.len() - 1
66 });
67 u32::try_from(index)
68 .map_err(|_| XlogError::Execution("resident schema candidate count exceeds u32".into()))
69}
70#[derive(Debug, Clone)]
71pub(super) struct ResidentWorkspaceAdmission {
72 pub(super) relation_capacity: u32,
73 pub(super) head_schemas: BTreeMap<String, Schema>,
74 head_schema_choices: BTreeMap<String, Vec<Schema>>,
75 rule_schema_ids: Vec<Vec<Option<u32>>>,
76 head_schema_selections: BTreeMap<String, ResidentHeadSchemaSelection>,
77}
78
79#[derive(Debug, Clone)]
80struct ResidentHeadSchemaSelection {
81 source_head: String,
82 output_schemas_by_source_winner: Vec<Schema>,
83}
84
85#[derive(Debug, Clone)]
86enum ResidentSchemaVariants {
87 Fixed(Schema),
88 Dynamic {
89 source_head: String,
90 schemas: Vec<Schema>,
91 },
92}
93
94fn resident_schema_variants_from_source(
95 source_head: String,
96 schemas: Vec<Schema>,
97) -> std::result::Result<ResidentSchemaVariants, ResidentGraphDeclineReason> {
98 let Some(first) = schemas.first() else {
99 return Err(resident_workspace_decline(format!(
100 "resident schema source {source_head} has no admitted candidate"
101 )));
102 };
103 if schemas.iter().all(|schema| schema == first) {
104 Ok(ResidentSchemaVariants::Fixed(first.clone()))
105 } else {
106 Ok(ResidentSchemaVariants::Dynamic {
107 source_head,
108 schemas,
109 })
110 }
111}
112
113fn resident_register_schema_selection(
114 target_head: &str,
115 selection: ResidentHeadSchemaSelection,
116 head_schema_choices: &BTreeMap<String, Vec<Schema>>,
117 head_schema_selections: &mut BTreeMap<String, ResidentHeadSchemaSelection>,
118) -> std::result::Result<(), ResidentGraphDeclineReason> {
119 if selection.source_head == target_head {
120 return Err(resident_workspace_decline(format!(
121 "resident schema lineage for {target_head} contains a cycle"
122 )));
123 }
124 let source_candidates = head_schema_choices
125 .get(&selection.source_head)
126 .filter(|candidates| !candidates.is_empty())
127 .ok_or_else(|| {
128 resident_workspace_decline(format!(
129 "resident schema source {} for {target_head} has no admitted candidate",
130 selection.source_head
131 ))
132 })?;
133 if source_candidates.len() != selection.output_schemas_by_source_winner.len() {
134 return Err(resident_workspace_decline(format!(
135 "resident schema source {} for {target_head} has {} candidates but {} mappings",
136 selection.source_head,
137 source_candidates.len(),
138 selection.output_schemas_by_source_winner.len()
139 )));
140 }
141
142 let mut source = selection.source_head.as_str();
143 let mut seen = HashSet::new();
144 while let Some(parent) = head_schema_selections.get(source) {
145 if source == target_head || !seen.insert(source) {
146 return Err(resident_workspace_decline(format!(
147 "resident schema lineage for {target_head} contains a cycle"
148 )));
149 }
150 source = parent.source_head.as_str();
151 }
152 if source == target_head {
153 return Err(resident_workspace_decline(format!(
154 "resident schema lineage for {target_head} contains a cycle"
155 )));
156 }
157
158 if let Some(existing) = head_schema_selections.get(target_head) {
159 if existing.source_head != selection.source_head
160 || existing.output_schemas_by_source_winner != selection.output_schemas_by_source_winner
161 {
162 return Err(resident_workspace_decline(format!(
163 "resident head {target_head} has multiple schema sources or mappings"
164 )));
165 }
166 return Ok(());
167 }
168 head_schema_selections.insert(target_head.to_string(), selection);
169 Ok(())
170}
171
172#[derive(Debug, Clone)]
173enum ResidentOutputSchemaSelection {
174 OwnWinner,
175 SourceWinner {
176 source_output: usize,
177 schemas: Vec<Schema>,
178 },
179}
180
181#[derive(Debug, Clone)]
182struct ResidentOutputSchemaPlan {
183 candidates: Vec<Schema>,
184 selection: ResidentOutputSchemaSelection,
185}
186
187fn resident_resolve_output_schemas(
188 plans: &[ResidentOutputSchemaPlan],
189 winner_ids: &[u32],
190) -> Result<Vec<Schema>> {
191 if plans.len() != winner_ids.len() {
192 return Err(XlogError::Execution(format!(
193 "resident schema plan count {} does not match winner count {}",
194 plans.len(),
195 winner_ids.len()
196 )));
197 }
198 (0..plans.len())
199 .map(|output| {
200 resident_resolve_output_schema(plans, winner_ids, output, &mut HashSet::new())
201 })
202 .collect()
203}
204
205fn resident_resolve_output_schema(
206 plans: &[ResidentOutputSchemaPlan],
207 winner_ids: &[u32],
208 output: usize,
209 resolving: &mut HashSet<usize>,
210) -> Result<Schema> {
211 if !resolving.insert(output) {
212 return Err(XlogError::Execution(
213 "resident schema source lineage contains a cycle".into(),
214 ));
215 }
216 let result = (|| {
217 let plan = plans.get(output).ok_or_else(|| {
218 XlogError::Execution(format!("resident schema output {output} is missing"))
219 })?;
220 let winner = *winner_ids.get(output).ok_or_else(|| {
221 XlogError::Execution(format!("resident schema winner {output} is missing"))
222 })?;
223 if winner != RESIDENT_DYNAMIC_SCHEMA_ID {
224 return plan
225 .candidates
226 .get(winner as usize)
227 .cloned()
228 .ok_or_else(|| {
229 XlogError::Execution(format!(
230 "resident schema winner {winner} exceeds {} candidates",
231 plan.candidates.len()
232 ))
233 });
234 }
235 let ResidentOutputSchemaSelection::SourceWinner {
236 source_output,
237 schemas,
238 } = &plan.selection
239 else {
240 return Err(XlogError::Execution(
241 "resident dynamic schema winner has no source lineage".into(),
242 ));
243 };
244 let source_schema =
245 resident_resolve_output_schema(plans, winner_ids, *source_output, resolving)?;
246 let source_plan = plans
247 .get(*source_output)
248 .ok_or_else(|| XlogError::Execution("resident schema source plan is missing".into()))?;
249 let source_winner = source_plan
250 .candidates
251 .iter()
252 .position(|candidate| candidate == &source_schema)
253 .ok_or_else(|| {
254 XlogError::Execution(
255 "resident resolved source schema has no candidate index".into(),
256 )
257 })?;
258 schemas.get(source_winner).cloned().ok_or_else(|| {
259 XlogError::Execution(format!(
260 "resident schema source winner {source_winner} exceeds {} mappings",
261 schemas.len()
262 ))
263 })
264 })();
265 resolving.remove(&output);
266 result
267}
268
269struct ResidentRunOwners {
270 provider: Arc<CudaKernelProvider>,
271 runtime: Arc<XlogDeviceRuntime>,
272 stream: Arc<CudaStream>,
273 graph: CapturedCudaGraph,
276 execution_domain: ResidentExecutionDomain,
277 schedule_program: ResidentScheduleDeviceProgram,
278 recorder: LaunchRecorder,
279 relations: Vec<Option<ResidentRelation>>,
280 output_indices: Vec<(String, usize)>,
281 filter_scratch: Option<ResidentFilterScratch>,
282 set_workspace: ResidentSetWorkspace,
283 join_workspace: ResidentJoinWorkspace,
284 control: ResidentConvergenceControl,
285 device_trace: ResidentDeviceTrace,
286 schema_winners: ResidentSchemaWinners,
287 receipt: ResidentPackedReceipt,
288 pinned_receipt: ResidentPinnedReceipt,
289 source_epoch: u64,
290 relation_registration: Vec<(RelId, String)>,
291 transaction_identity: Arc<()>,
292 output_schema_plans: Vec<ResidentOutputSchemaPlan>,
293}
294
295#[derive(Clone, Debug, PartialEq, Eq)]
296struct ResidentSourceSetSnapshot {
297 name: String,
298 version: u64,
299 schema: Schema,
300 row_capacity: u64,
301 column_blocks: Vec<Option<BlockId>>,
302 row_count_block: BlockId,
303}
304
305#[derive(Clone)]
306enum ResidentBufferRef {
307 Source(String),
308 Private(usize),
309}
310
311enum ResidentRecordedOp {
312 Unit {
313 output: usize,
314 op_id: u32,
315 },
316 Scan {
317 relation: ResidentBufferRef,
318 op_id: u32,
319 },
320 TraceDelta {
321 scan_delta: u32,
322 filter_delta: u32,
323 semantic_guard: Option<ResidentBufferRef>,
324 },
325 Clear {
326 output: usize,
327 },
328 Filter {
329 input: ResidentBufferRef,
330 output: usize,
331 workspace: usize,
332 op_id: u32,
333 },
334 Project {
335 input: ResidentBufferRef,
336 output: usize,
337 workspace: usize,
338 op_id: u32,
339 },
340 Union {
341 left: ResidentBufferRef,
342 right: ResidentBufferRef,
343 output: usize,
344 op_id: u32,
345 },
346 Diff {
347 left: ResidentBufferRef,
348 right: ResidentBufferRef,
349 output: usize,
350 op_id: u32,
351 },
352 Join {
353 kind: ResidentJoinKind,
354 left: ResidentBufferRef,
355 left_key: usize,
356 right: ResidentBufferRef,
357 right_key: usize,
358 output: usize,
359 op_id: u32,
360 },
361 ChangedReset,
362 ChangedMark {
363 relation: usize,
364 },
365 TestStatus(ResidentTerminalStatus),
366 SchemaWinnerMark {
367 contribution: ResidentBufferRef,
368 head_index: u32,
369 schema_id: u32,
370 },
371}
372
373fn resident_record_unit_leaf(
374 output: usize,
375 op_id: u32,
376 ops: &mut Vec<ResidentRecordedOp>,
377 push_physical: impl FnOnce(&mut Vec<ResidentRecordedOp>, ResidentRecordedOp, u32),
378) -> ResidentBufferRef {
379 push_physical(ops, ResidentRecordedOp::Unit { output, op_id }, op_id);
380 ResidentBufferRef::Private(output)
381}
382
383fn resident_new_phase_unit(
384 relations: &mut Vec<ResidentLogicalRelation>,
385 op_id: u32,
386) -> Result<(ResidentBufferRef, ResidentRecordedOp)> {
387 let output = relations.len();
388 relations.push(ResidentLogicalRelation {
389 schema: Schema::new(Vec::new()),
390 initial_count: 0,
391 permanent: false,
392 });
393 Ok((
394 ResidentBufferRef::Private(output),
395 ResidentRecordedOp::Unit { output, op_id },
396 ))
397}
398
399fn resident_record_scan_leaf(
400 relation: ResidentBufferRef,
401 op_id: u32,
402 semantic_guard: Option<ResidentBufferRef>,
403 ops: &mut Vec<ResidentRecordedOp>,
404 push_physical: impl FnOnce(&mut Vec<ResidentRecordedOp>, ResidentRecordedOp, u32),
405) -> ResidentBufferRef {
406 push_physical(
407 ops,
408 ResidentRecordedOp::Scan {
409 relation: relation.clone(),
410 op_id,
411 },
412 op_id,
413 );
414 ops.push(ResidentRecordedOp::TraceDelta {
415 scan_delta: 1,
416 filter_delta: 0,
417 semantic_guard,
418 });
419 relation
420}
421
422fn resident_semantic_trace_guard(
423 override_scan: Option<(RelId, usize, usize)>,
424) -> Option<ResidentBufferRef> {
425 override_scan.map(|(_, _, delta)| ResidentBufferRef::Private(delta))
426}
427
428fn resident_record_schema_winner_mark(
429 ops: &mut Vec<ResidentRecordedOp>,
430 contribution: ResidentBufferRef,
431 head_index: u32,
432 schema_id: u32,
433) {
434 ops.push(ResidentRecordedOp::SchemaWinnerMark {
435 contribution,
436 head_index,
437 schema_id,
438 });
439}
440
441#[derive(Debug, Clone, Copy, PartialEq, Eq)]
442enum ResidentUnionFoldMode {
443 SelfUnion,
444 LeftAssociated,
445}
446
447fn resident_union_fold_mode(input_count: usize) -> Result<ResidentUnionFoldMode> {
448 match input_count {
449 0 => Err(XlogError::Execution("resident union has no inputs".into())),
450 1 => Ok(ResidentUnionFoldMode::SelfUnion),
451 _ => Ok(ResidentUnionFoldMode::LeftAssociated),
452 }
453}
454
455enum ResidentPhaseMergeStep<T> {
456 Deduplicate(T),
457 Union(T, T),
458}
459
460fn resident_phase_merge<T, E>(
461 current: Option<T>,
462 contribution: T,
463 mut execute: impl FnMut(ResidentPhaseMergeStep<T>) -> std::result::Result<T, E>,
464) -> std::result::Result<T, E> {
465 let contribution = execute(ResidentPhaseMergeStep::Deduplicate(contribution))?;
466 match current {
467 Some(current) => execute(ResidentPhaseMergeStep::Union(current, contribution)),
468 None => Ok(contribution),
469 }
470}
471
472enum ResidentCapturePhase {
473 Segment {
474 ops: Vec<ResidentRecordedOp>,
475 scc_begin: Option<(u32, u32)>,
476 },
477 ConditionalWhile {
478 ops: Vec<ResidentRecordedOp>,
479 iteration_limit: u32,
480 convergence_op_id: u32,
481 },
482}
483
484struct ResidentCompactLogicalRegion {
485 ops: Vec<ResidentRecordedOp>,
486 iteration_limit: u32,
487 op_id: u32,
488 flags: u32,
489}
490
491struct ResidentCompactSchedulePlan {
492 source_slots: BTreeMap<String, u32>,
493 ops: Vec<ResidentOpDescriptor>,
494 waves: Vec<ResidentWaveDescriptor>,
495 regions: Vec<ResidentRegionDescriptor>,
496 generation_bases: Vec<u32>,
497 filter_comparisons: Vec<ResidentFilterComparisonDescriptor>,
498 project_expressions: Vec<ResidentProjectExpressionDescriptor>,
499}
500
501#[derive(Debug, Clone, Copy, PartialEq, Eq)]
502enum ResidentCaptureParentKind {
503 Kernel,
504 Conditional,
505}
506
507struct ResidentCompactTopology {
508 parent_kinds: Vec<ResidentCaptureParentKind>,
509 conditional_body_kernel_counts: Vec<usize>,
510 hierarchical_node_count: usize,
511}
512
513trait ResidentRegionFlags {
514 fn region_flags(&self) -> u32;
515}
516
517impl ResidentRegionFlags for ResidentCompactLogicalRegion {
518 fn region_flags(&self) -> u32 {
519 self.flags
520 }
521}
522
523impl ResidentRegionFlags for ResidentRegionDescriptor {
524 fn region_flags(&self) -> u32 {
525 self.flags
526 }
527}
528
529fn resident_compact_topology<R: ResidentRegionFlags>(
530 regions: &[R],
531) -> Result<ResidentCompactTopology> {
532 let parent_kinds = regions
533 .iter()
534 .map(|region| {
535 if region.region_flags() == RESIDENT_SCHEDULE_REGION_RECURSIVE {
536 ResidentCaptureParentKind::Conditional
537 } else {
538 ResidentCaptureParentKind::Kernel
539 }
540 })
541 .collect::<Vec<_>>();
542 let conditional_count = parent_kinds
543 .iter()
544 .filter(|kind| **kind == ResidentCaptureParentKind::Conditional)
545 .count();
546 let hierarchical_node_count = parent_kinds
547 .len()
548 .checked_add(conditional_count)
549 .ok_or_else(|| XlogError::Execution("resident topology node count overflow".into()))?;
550 Ok(ResidentCompactTopology {
551 parent_kinds,
552 conditional_body_kernel_counts: vec![1; conditional_count],
553 hierarchical_node_count,
554 })
555}
556
557fn resident_validate_parent_graph_kinds(
558 actual: &[CudaGraphNodeKind],
559 expected: &[ResidentCaptureParentKind],
560) -> Result<()> {
561 if actual.len() != expected.len()
562 || actual.iter().zip(expected).any(|(actual, expected)| {
563 !matches!(
564 (actual, expected),
565 (CudaGraphNodeKind::Kernel, ResidentCaptureParentKind::Kernel)
566 | (
567 CudaGraphNodeKind::Conditional,
568 ResidentCaptureParentKind::Conditional
569 )
570 )
571 })
572 {
573 return Err(XlogError::Execution(
574 "resident parent graph topology differs from the compact schedule".into(),
575 ));
576 }
577 Ok(())
578}
579
580fn resident_validate_conditional_body_node_kinds(
581 actual: &[Vec<CudaGraphNodeKind>],
582 expected_conditional_count: usize,
583) -> Result<Vec<usize>> {
584 if actual.len() != expected_conditional_count {
585 return Err(XlogError::Execution(format!(
586 "resident conditional body inventory has {} bodies, expected {expected_conditional_count}",
587 actual.len()
588 )));
589 }
590 let mut kernel_counts = Vec::with_capacity(actual.len());
591 for (index, kinds) in actual.iter().enumerate() {
592 if kinds.as_slice() != [CudaGraphNodeKind::Kernel] {
593 return Err(XlogError::Execution(format!(
594 "resident conditional body {index} must contain exactly one kernel node, found {kinds:?}"
595 )));
596 }
597 kernel_counts.push(1);
598 }
599 Ok(kernel_counts)
600}
601
602#[derive(Default)]
603struct ResidentCompactDescriptorTables {
604 filter_comparisons: Vec<ResidentFilterComparisonDescriptor>,
605 filter_ranges: Vec<(u32, u32)>,
606 project_expressions: Vec<ResidentProjectExpressionDescriptor>,
607 project_ranges: Vec<(u32, u32)>,
608}
609
610#[allow(clippy::too_many_arguments)]
611fn resident_compact_schedule_metadata_bytes(
612 slot_count: usize,
613 op_count: usize,
614 wave_count: usize,
615 region_count: usize,
616 generation_base_count: usize,
617 head_count: usize,
618 filter_comparison_count: usize,
619 project_expression_count: usize,
620) -> Result<u64> {
621 let generation_metadata_count =
622 generation_base_count
623 .checked_add(head_count)
624 .ok_or_else(|| {
625 XlogError::Execution("resident generation metadata count overflow".into())
626 })?;
627 resident_schedule_metadata_device_bytes(
628 slot_count,
629 op_count,
630 wave_count,
631 region_count,
632 generation_metadata_count,
633 filter_comparison_count,
634 project_expression_count,
635 )
636}
637
638fn resident_compact_schema_defaults(
639 ops: &[ResidentOpDescriptor],
640 head_count: usize,
641) -> Result<Vec<u32>> {
642 let mut defaults = vec![None; head_count];
643 for op in ops {
644 if op.flags & RESIDENT_SCHEDULE_OP_MARK_SCHEMA_WINNER == 0 {
645 continue;
646 }
647 let head = usize::try_from(op.schema_winner_head).map_err(|_| {
648 XlogError::Execution("resident schema winner head index overflow".into())
649 })?;
650 let default = defaults.get_mut(head).ok_or_else(|| {
651 XlogError::Execution("resident schema winner head is out of range".into())
652 })?;
653 if default.is_none() {
654 *default = Some(op.schema_winner_id);
655 }
656 }
657 defaults
658 .into_iter()
659 .enumerate()
660 .map(|(head, default)| {
661 default.ok_or_else(|| {
662 XlogError::Execution(format!(
663 "resident head {head} has no schema winner candidate"
664 ))
665 })
666 })
667 .collect()
668}
669
670fn resident_compact_allocation_bytes(
671 relation_bytes: u64,
672 filter_scratch_bytes: u64,
673 fixed_workspace_bytes: u64,
674 private_slot_count: usize,
675 head_count: usize,
676 plan: &ResidentCompactSchedulePlan,
677) -> Result<(u64, u64)> {
678 let slot_count = private_slot_count
679 .checked_add(plan.source_slots.len())
680 .ok_or_else(|| XlogError::Execution("resident compact slot count overflow".into()))?;
681 let metadata_bytes = resident_compact_schedule_metadata_bytes(
682 slot_count,
683 plan.ops.len(),
684 plan.waves.len(),
685 plan.regions.len(),
686 plan.generation_bases.len(),
687 head_count,
688 plan.filter_comparisons.len(),
689 plan.project_expressions.len(),
690 )?;
691 let required_bytes = relation_bytes
692 .checked_add(filter_scratch_bytes)
693 .and_then(|bytes| bytes.checked_add(fixed_workspace_bytes))
694 .and_then(|bytes| bytes.checked_add(metadata_bytes))
695 .ok_or_else(|| XlogError::Execution("resident compact manifest overflow".into()))?;
696 Ok((required_bytes, metadata_bytes))
697}
698
699fn resident_compact_preflight_device_bytes(
700 manifest: &ResidentAllocationManifest,
701 plan: &ResidentCompactSchedulePlan,
702) -> Result<(u64, u64, u64)> {
703 let filter_count = u64::try_from(plan.filter_comparisons.len().max(1)).map_err(|_| {
704 XlogError::Execution("resident compact filter descriptor count exceeds u64".into())
705 })?;
706 let project_count = u64::try_from(plan.project_expressions.len().max(1)).map_err(|_| {
707 XlogError::Execution("resident compact project descriptor count exceeds u64".into())
708 })?;
709 let filter_bytes = u64::try_from(std::mem::size_of::<ResidentFilterComparisonDescriptor>())
710 .ok()
711 .and_then(|size| size.checked_mul(filter_count))
712 .ok_or_else(|| {
713 XlogError::Execution("resident compact filter descriptor bytes overflow".into())
714 })?;
715 let project_bytes = u64::try_from(std::mem::size_of::<ResidentProjectExpressionDescriptor>())
716 .ok()
717 .and_then(|size| size.checked_mul(project_count))
718 .ok_or_else(|| {
719 XlogError::Execution("resident compact project descriptor bytes overflow".into())
720 })?;
721 let descriptor_bytes = filter_bytes
722 .checked_add(project_bytes)
723 .ok_or_else(|| XlogError::Execution("resident compact descriptor bytes overflow".into()))?;
724 let remaining_metadata_bytes = manifest
725 .schedule_metadata_bytes
726 .checked_sub(descriptor_bytes)
727 .ok_or_else(|| {
728 XlogError::Execution(
729 "resident compact descriptor bytes exceed schedule metadata".into(),
730 )
731 })?;
732 let fixed_bytes = manifest
733 .fixed_workspace_bytes
734 .checked_add(remaining_metadata_bytes)
735 .ok_or_else(|| XlogError::Execution("resident compact fixed bytes overflow".into()))?;
736 let reported_total = manifest
737 .relation_bytes
738 .checked_add(manifest.filter_scratch_bytes)
739 .and_then(|bytes| bytes.checked_add(filter_bytes))
740 .and_then(|bytes| bytes.checked_add(project_bytes))
741 .and_then(|bytes| bytes.checked_add(fixed_bytes))
742 .ok_or_else(|| XlogError::Execution("resident compact preflight bytes overflow".into()))?;
743 if reported_total != manifest.required_bytes {
744 return Err(XlogError::Execution(format!(
745 "resident compact preflight component mismatch: reported {reported_total}, required {}",
746 manifest.required_bytes
747 )));
748 }
749 Ok((filter_bytes, project_bytes, fixed_bytes))
750}
751
752fn resident_compact_tables(
753 filter_workspaces: &[ResidentFilterPlan],
754 project_workspaces: &[ResidentProjectPlan],
755) -> Result<ResidentCompactDescriptorTables> {
756 let mut tables = ResidentCompactDescriptorTables::default();
757 for workspace in filter_workspaces {
758 let offset = u32::try_from(tables.filter_comparisons.len()).map_err(|_| {
759 XlogError::Execution("resident filter comparison offset exceeds u32".into())
760 })?;
761 let count = u32::try_from(workspace.compact_comparisons.len()).map_err(|_| {
762 XlogError::Execution("resident filter comparison count exceeds u32".into())
763 })?;
764 tables
765 .filter_comparisons
766 .extend_from_slice(&workspace.compact_comparisons);
767 tables.filter_ranges.push((offset, count));
768 }
769 for workspace in project_workspaces {
770 let offset = u32::try_from(tables.project_expressions.len()).map_err(|_| {
771 XlogError::Execution("resident project expression offset exceeds u32".into())
772 })?;
773 let count = u32::try_from(workspace.compact_expressions.len()).map_err(|_| {
774 XlogError::Execution("resident project expression count exceeds u32".into())
775 })?;
776 tables
777 .project_expressions
778 .extend_from_slice(&workspace.compact_expressions);
779 tables.project_ranges.push((offset, count));
780 }
781 Ok(tables)
782}
783
784#[derive(Debug, Clone, Copy, PartialEq, Eq)]
785struct ResidentCompactSlotRef {
786 slot: u32,
787 generation: u32,
788}
789
790fn resident_compact_slot_ref(
791 reference: &ResidentBufferRef,
792 assignments: &[ResidentSlotAssignment],
793 source_slots: &BTreeMap<String, u32>,
794) -> Result<ResidentCompactSlotRef> {
795 match reference {
796 ResidentBufferRef::Private(logical) => {
797 let assignment = assignments.get(*logical).ok_or_else(|| {
798 XlogError::Execution(format!(
799 "resident logical relation {logical} has no compact slot assignment"
800 ))
801 })?;
802 Ok(ResidentCompactSlotRef {
803 slot: u32::try_from(assignment.slot).map_err(|_| {
804 XlogError::Execution("resident compact slot index exceeds u32".into())
805 })?,
806 generation: assignment.generation,
807 })
808 }
809 ResidentBufferRef::Source(name) => Ok(ResidentCompactSlotRef {
810 slot: *source_slots.get(name).ok_or_else(|| {
811 XlogError::Execution(format!(
812 "resident source {name} has no compact slot assignment"
813 ))
814 })?,
815 generation: 0,
816 }),
817 }
818}
819
820#[allow(clippy::too_many_arguments)]
821fn resident_lower_compact_regions<'a>(
822 logical_regions: Vec<ResidentCompactLogicalRegion>,
823 physical_slots: &[ResidentPhysicalSlotPlan],
824 assignments: &[ResidentSlotAssignment],
825 source_names: impl IntoIterator<Item = &'a str>,
826 tables: ResidentCompactDescriptorTables,
827) -> Result<ResidentCompactSchedulePlan> {
828 let source_slots = resident_source_slot_map(physical_slots.len(), source_names)?;
829 let slot_count = physical_slots
830 .len()
831 .checked_add(source_slots.len())
832 .ok_or_else(|| XlogError::Execution("resident compact slot count overflow".into()))?;
833 let slot_count_u32 = u32::try_from(slot_count)
834 .map_err(|_| XlogError::Execution("resident compact slot count exceeds u32".into()))?;
835 let mut ops = Vec::new();
836 let mut waves = Vec::new();
837 let mut regions = Vec::with_capacity(logical_regions.len());
838 let filter_total = u32::try_from(tables.filter_comparisons.len())
839 .map_err(|_| XlogError::Execution("resident filter comparison count exceeds u32".into()))?;
840 let project_total = u32::try_from(tables.project_expressions.len()).map_err(|_| {
841 XlogError::Execution("resident project expression count exceeds u32".into())
842 })?;
843 let mut filter_cursor = 0_u32;
844 let mut project_cursor = 0_u32;
845 let mut generation_bases = Vec::with_capacity(
846 logical_regions
847 .len()
848 .checked_mul(slot_count)
849 .ok_or_else(|| {
850 XlogError::Execution("resident generation baseline count overflow".into())
851 })?,
852 );
853
854 for logical_region in logical_regions {
855 let region_flags = logical_region.flags;
856 let first_wave = u32::try_from(waves.len())
857 .map_err(|_| XlogError::Execution("resident wave count exceeds u32".into()))?;
858 let generation_offset = u32::try_from(generation_bases.len()).map_err(|_| {
859 XlogError::Execution("resident generation baseline offset exceeds u32".into())
860 })?;
861 let mut first_generations = vec![None; slot_count];
862 let mut last_relation_descriptor = None::<(usize, ResidentCompactSlotRef)>;
863
864 for (logical_op_index, logical_op) in logical_region.ops.into_iter().enumerate() {
865 let mut emit = |descriptor: ResidentOpDescriptor,
866 output: ResidentCompactSlotRef,
867 references: Vec<ResidentCompactSlotRef>|
868 -> Result<()> {
869 for reference in &references {
870 let generation = first_generations
871 .get_mut(reference.slot as usize)
872 .ok_or_else(|| {
873 XlogError::Execution(
874 "resident compact operation slot is out of range".into(),
875 )
876 })?;
877 if generation.is_none() {
878 *generation = Some(reference.generation);
879 }
880 }
881 let op_index = ops.len();
882 let first_op = u32::try_from(op_index).map_err(|_| {
883 XlogError::Execution("resident compact op count exceeds u32".into())
884 })?;
885 ops.push(descriptor);
886 waves.push(ResidentWaveDescriptor {
887 first_op,
888 op_count: 1,
889 flags: 0,
890 reserved: 0,
891 });
892 last_relation_descriptor = Some((op_index, output));
893 Ok(())
894 };
895
896 match logical_op {
897 ResidentRecordedOp::Unit { output, op_id } => {
898 let output = resident_compact_slot_ref(
899 &ResidentBufferRef::Private(output),
900 assignments,
901 &source_slots,
902 )?;
903 emit(
904 ResidentOpDescriptor::unit(op_id, output.slot, output.generation),
905 output,
906 vec![output],
907 )?;
908 }
909 ResidentRecordedOp::Scan { relation, op_id } => {
910 let source = resident_compact_slot_ref(&relation, assignments, &source_slots)?;
911 emit(
912 ResidentOpDescriptor::scan(op_id, source.slot, source.generation),
913 source,
914 vec![source],
915 )?;
916 }
917 op @ (ResidentRecordedOp::Filter { .. } | ResidentRecordedOp::Project { .. }) => {
918 let (is_filter, input, output, workspace, op_id) = match op {
919 ResidentRecordedOp::Filter {
920 input,
921 output,
922 workspace,
923 op_id,
924 } => (true, input, output, workspace, op_id),
925 ResidentRecordedOp::Project {
926 input,
927 output,
928 workspace,
929 op_id,
930 } => (false, input, output, workspace, op_id),
931 _ => unreachable!("compact filter/project arm"),
932 };
933 let input = resident_compact_slot_ref(&input, assignments, &source_slots)?;
934 let output = resident_compact_slot_ref(
935 &ResidentBufferRef::Private(output),
936 assignments,
937 &source_slots,
938 )?;
939 let (aux_offset, aux_count) = if is_filter {
940 *tables.filter_ranges.get(workspace).ok_or_else(|| {
941 XlogError::Execution(
942 "resident compact filter workspace is out of range".into(),
943 )
944 })?
945 } else {
946 *tables.project_ranges.get(workspace).ok_or_else(|| {
947 XlogError::Execution(
948 "resident compact project workspace is out of range".into(),
949 )
950 })?
951 };
952 let (cursor, total, table_name) = if is_filter {
953 (&mut filter_cursor, filter_total, "filter")
954 } else {
955 (&mut project_cursor, project_total, "project")
956 };
957 if aux_offset != *cursor || aux_offset > total || aux_count > total - aux_offset
958 {
959 return Err(XlogError::Execution(format!(
960 "resident compact {table_name} descriptor range is not contiguous"
961 )));
962 }
963 *cursor = aux_offset + aux_count;
964 emit(
965 ResidentOpDescriptor {
966 kind: if is_filter {
967 ResidentScheduleOpKind::Filter
968 } else {
969 ResidentScheduleOpKind::Project
970 },
971 op_id,
972 out: output.slot,
973 in0: input.slot,
974 in0_generation: input.generation,
975 out_generation: output.generation,
976 aux_offset,
977 aux_count,
978 ..Default::default()
979 },
980 output,
981 vec![input, output],
982 )?;
983 }
984 op @ (ResidentRecordedOp::Union { .. } | ResidentRecordedOp::Diff { .. }) => {
985 let (is_union, left, right, output, op_id) = match op {
986 ResidentRecordedOp::Union {
987 left,
988 right,
989 output,
990 op_id,
991 } => (true, left, right, output, op_id),
992 ResidentRecordedOp::Diff {
993 left,
994 right,
995 output,
996 op_id,
997 } => (false, left, right, output, op_id),
998 _ => unreachable!("compact set arm"),
999 };
1000 let left = resident_compact_slot_ref(&left, assignments, &source_slots)?;
1001 let right = resident_compact_slot_ref(&right, assignments, &source_slots)?;
1002 let output = resident_compact_slot_ref(
1003 &ResidentBufferRef::Private(output),
1004 assignments,
1005 &source_slots,
1006 )?;
1007 emit(
1008 ResidentOpDescriptor {
1009 kind: if is_union {
1010 ResidentScheduleOpKind::Union
1011 } else {
1012 ResidentScheduleOpKind::Diff
1013 },
1014 op_id,
1015 out: output.slot,
1016 in0: left.slot,
1017 in1: right.slot,
1018 in0_generation: left.generation,
1019 in1_generation: right.generation,
1020 out_generation: output.generation,
1021 ..Default::default()
1022 },
1023 output,
1024 vec![left, right, output],
1025 )?;
1026 }
1027 ResidentRecordedOp::Join {
1028 kind,
1029 left,
1030 left_key,
1031 right,
1032 right_key,
1033 output,
1034 op_id,
1035 } => {
1036 let left = resident_compact_slot_ref(&left, assignments, &source_slots)?;
1037 let right = resident_compact_slot_ref(&right, assignments, &source_slots)?;
1038 let output = resident_compact_slot_ref(
1039 &ResidentBufferRef::Private(output),
1040 assignments,
1041 &source_slots,
1042 )?;
1043 emit(
1044 ResidentOpDescriptor {
1045 kind: match kind {
1046 ResidentJoinKind::Inner => ResidentScheduleOpKind::JoinInner,
1047 ResidentJoinKind::Semi => ResidentScheduleOpKind::JoinSemi,
1048 },
1049 op_id,
1050 out: output.slot,
1051 in0: left.slot,
1052 in1: right.slot,
1053 in0_generation: left.generation,
1054 in1_generation: right.generation,
1055 out_generation: output.generation,
1056 left_key: u32::try_from(left_key).map_err(|_| {
1057 XlogError::Execution(
1058 "resident compact left join key exceeds u32".into(),
1059 )
1060 })?,
1061 right_key: u32::try_from(right_key).map_err(|_| {
1062 XlogError::Execution(
1063 "resident compact right join key exceeds u32".into(),
1064 )
1065 })?,
1066 ..Default::default()
1067 },
1068 output,
1069 vec![left, right, output],
1070 )?;
1071 }
1072 ResidentRecordedOp::TraceDelta {
1073 scan_delta,
1074 filter_delta,
1075 semantic_guard,
1076 } => {
1077 let semantic_guard = semantic_guard
1078 .as_ref()
1079 .map(|reference| {
1080 resident_compact_slot_ref(reference, assignments, &source_slots)
1081 })
1082 .transpose()?;
1083 if let Some(reference) = semantic_guard {
1084 let generation = first_generations
1085 .get_mut(reference.slot as usize)
1086 .ok_or_else(|| {
1087 XlogError::Execution(
1088 "resident compact trace guard slot is out of range".into(),
1089 )
1090 })?;
1091 if generation.is_none() {
1092 *generation = Some(reference.generation);
1093 }
1094 }
1095 let first_op = u32::try_from(ops.len()).map_err(|_| {
1096 XlogError::Execution("resident compact op count exceeds u32".into())
1097 })?;
1098 ops.push(ResidentOpDescriptor::trace_delta(
1099 scan_delta,
1100 filter_delta,
1101 semantic_guard.map(|reference| (reference.slot, reference.generation)),
1102 ));
1103 waves.push(ResidentWaveDescriptor {
1104 first_op,
1105 op_count: 1,
1106 flags: 0,
1107 reserved: 0,
1108 });
1109 }
1110 ResidentRecordedOp::TestStatus(status) => {
1111 let first_op = u32::try_from(ops.len()).map_err(|_| {
1112 XlogError::Execution("resident compact op count exceeds u32".into())
1113 })?;
1114 ops.push(ResidentOpDescriptor::test_status(status)?);
1115 waves.push(ResidentWaveDescriptor {
1116 first_op,
1117 op_count: 1,
1118 flags: 0,
1119 reserved: 0,
1120 });
1121 }
1122 ResidentRecordedOp::SchemaWinnerMark {
1123 contribution,
1124 head_index,
1125 schema_id,
1126 } => {
1127 let contribution =
1128 resident_compact_slot_ref(&contribution, assignments, &source_slots)?;
1129 let (index, output) = last_relation_descriptor.as_ref().ok_or_else(|| {
1130 XlogError::Execution(
1131 "resident schema marker has no contribution operation".into(),
1132 )
1133 })?;
1134 if *output != contribution {
1135 return Err(XlogError::Execution(
1136 "resident schema marker does not match its contribution operation"
1137 .into(),
1138 ));
1139 }
1140 ops[*index] = ops[*index].with_schema_winner(head_index, schema_id);
1141 }
1142 ResidentRecordedOp::ChangedMark { relation } => {
1143 let relation = resident_compact_slot_ref(
1144 &ResidentBufferRef::Private(relation),
1145 assignments,
1146 &source_slots,
1147 )?;
1148 let (index, output) = last_relation_descriptor.as_ref().ok_or_else(|| {
1149 XlogError::Execution(
1150 "resident novelty marker has no completed delta copy".into(),
1151 )
1152 })?;
1153 if *output != relation || ops[*index].kind != ResidentScheduleOpKind::Project {
1154 return Err(XlogError::Execution(
1155 "resident novelty marker does not follow its completed delta copy"
1156 .into(),
1157 ));
1158 }
1159 ops[*index].flags |= RESIDENT_SCHEDULE_OP_MARK_NOVELTY;
1160 }
1161 ResidentRecordedOp::ChangedReset => {
1162 if region_flags != RESIDENT_SCHEDULE_REGION_RECURSIVE || logical_op_index != 0 {
1163 return Err(XlogError::Execution(
1164 "resident changed reset is not the recursive region entry".into(),
1165 ));
1166 }
1167 }
1168 ResidentRecordedOp::Clear { .. } => {
1169 return Err(XlogError::Execution(
1170 "resident compact SSA still contains a Clear operation".into(),
1171 ));
1172 }
1173 }
1174 }
1175
1176 for (slot, generation) in first_generations.into_iter().enumerate() {
1177 let baseline = if slot < physical_slots.len() && physical_slots[slot].permanent {
1178 0
1179 } else {
1180 generation.unwrap_or(0)
1181 };
1182 generation_bases.push(baseline);
1183 }
1184 let wave_count = u32::try_from(waves.len())
1185 .map_err(|_| XlogError::Execution("resident wave count exceeds u32".into()))?
1186 .checked_sub(first_wave)
1187 .ok_or_else(|| XlogError::Execution("resident wave range underflow".into()))?;
1188 regions.push(ResidentRegionDescriptor {
1189 first_wave,
1190 wave_count,
1191 iteration_limit: logical_region.iteration_limit,
1192 op_id: logical_region.op_id,
1193 flags: region_flags,
1194 first_slot: 0,
1195 slot_count: slot_count_u32,
1196 generation_offset,
1197 });
1198 }
1199
1200 if filter_cursor != filter_total || project_cursor != project_total {
1201 return Err(XlogError::Execution(
1202 "resident compact descriptor tables are not exactly covered".into(),
1203 ));
1204 }
1205
1206 Ok(ResidentCompactSchedulePlan {
1207 source_slots,
1208 ops,
1209 waves,
1210 regions,
1211 generation_bases,
1212 filter_comparisons: tables.filter_comparisons,
1213 project_expressions: tables.project_expressions,
1214 })
1215}
1216
1217impl ResidentCompactLogicalRegion {
1218 fn initializes(&self) -> bool {
1219 self.flags & RESIDENT_SCHEDULE_REGION_INITIALIZE != 0
1220 }
1221
1222 fn begins_scc(&self) -> bool {
1223 self.flags & RESIDENT_SCHEDULE_REGION_SCC_BEGIN != 0
1224 }
1225
1226 fn recursive(&self) -> bool {
1227 self.flags == RESIDENT_SCHEDULE_REGION_RECURSIVE
1228 }
1229
1230 fn finalizes(&self) -> bool {
1231 self.flags & RESIDENT_SCHEDULE_REGION_FINALIZE != 0
1232 }
1233}
1234
1235fn resident_compact_regions(
1236 initial_ops: Vec<ResidentRecordedOp>,
1237 phases: Vec<ResidentCapturePhase>,
1238 success_op_id: u32,
1239) -> Result<Vec<ResidentCompactLogicalRegion>> {
1240 let mut regions = Vec::new();
1241 let mut pending = initial_ops;
1242 let mut pending_seed_region = None;
1243
1244 for phase in phases {
1245 match phase {
1246 ResidentCapturePhase::Segment {
1247 mut ops,
1248 scc_begin: None,
1249 } => {
1250 if pending_seed_region.is_some() {
1251 return Err(XlogError::Execution(
1252 "resident SCC seed is not followed by its recursive body".into(),
1253 ));
1254 }
1255 pending.append(&mut ops);
1256 }
1257 ResidentCapturePhase::Segment {
1258 mut ops,
1259 scc_begin: Some((iteration_limit, op_id)),
1260 } => {
1261 if pending_seed_region.is_some() {
1262 return Err(XlogError::Execution(
1263 "resident SCC seed is not followed by its recursive body".into(),
1264 ));
1265 }
1266 pending.append(&mut ops);
1267 let mut flags = RESIDENT_SCHEDULE_REGION_SCC_BEGIN;
1268 if regions.is_empty() {
1269 flags |= RESIDENT_SCHEDULE_REGION_INITIALIZE;
1270 }
1271 regions.push(ResidentCompactLogicalRegion {
1272 ops: std::mem::take(&mut pending),
1273 iteration_limit,
1274 op_id,
1275 flags,
1276 });
1277 pending_seed_region = Some(regions.len() - 1);
1278 }
1279 ResidentCapturePhase::ConditionalWhile {
1280 ops,
1281 iteration_limit,
1282 convergence_op_id,
1283 } => {
1284 let seed_region = pending_seed_region.take().ok_or_else(|| {
1285 XlogError::Execution("resident recursive body has no preceding SCC seed".into())
1286 })?;
1287 if regions[seed_region].iteration_limit != iteration_limit {
1288 return Err(XlogError::Execution(
1289 "resident SCC seed and recursive body iteration limits differ".into(),
1290 ));
1291 }
1292 regions[seed_region].op_id = convergence_op_id;
1293 regions.push(ResidentCompactLogicalRegion {
1294 ops,
1295 iteration_limit,
1296 op_id: convergence_op_id,
1297 flags: RESIDENT_SCHEDULE_REGION_RECURSIVE,
1298 });
1299 }
1300 }
1301 }
1302 if pending_seed_region.is_some() {
1303 return Err(XlogError::Execution(
1304 "resident SCC seed is missing its recursive body".into(),
1305 ));
1306 }
1307
1308 let mut flags = RESIDENT_SCHEDULE_REGION_FINALIZE;
1309 if regions.is_empty() {
1310 flags |= RESIDENT_SCHEDULE_REGION_INITIALIZE;
1311 }
1312 regions.push(ResidentCompactLogicalRegion {
1313 ops: pending,
1314 iteration_limit: 1,
1315 op_id: success_op_id,
1316 flags,
1317 });
1318 Ok(regions)
1319}
1320
1321fn coalesce_resident_capture_phases(
1322 phases: Vec<ResidentCapturePhase>,
1323) -> Vec<ResidentCapturePhase> {
1324 let mut coalesced = Vec::new();
1325 let mut ordinary_ops = Vec::new();
1326 let flush_ordinary = |coalesced: &mut Vec<ResidentCapturePhase>,
1327 ordinary_ops: &mut Vec<ResidentRecordedOp>| {
1328 if !ordinary_ops.is_empty() {
1329 coalesced.push(ResidentCapturePhase::Segment {
1330 ops: std::mem::take(ordinary_ops),
1331 scc_begin: None,
1332 });
1333 }
1334 };
1335
1336 for phase in phases {
1337 match phase {
1338 ResidentCapturePhase::Segment {
1339 mut ops,
1340 scc_begin: None,
1341 } => ordinary_ops.append(&mut ops),
1342 boundary => {
1343 flush_ordinary(&mut coalesced, &mut ordinary_ops);
1344 coalesced.push(boundary);
1345 }
1346 }
1347 }
1348 flush_ordinary(&mut coalesced, &mut ordinary_ops);
1349 coalesced
1350}
1351
1352#[cfg(test)]
1353mod capture_phase_tests {
1354 use super::{coalesce_resident_capture_phases, ResidentCapturePhase, ResidentRecordedOp};
1355
1356 fn segment(output: usize, scc_begin: Option<(u32, u32)>) -> ResidentCapturePhase {
1357 ResidentCapturePhase::Segment {
1358 ops: vec![ResidentRecordedOp::Clear { output }],
1359 scc_begin,
1360 }
1361 }
1362
1363 fn assert_clear_range(phase: &ResidentCapturePhase, expected: std::ops::Range<usize>) {
1364 let ResidentCapturePhase::Segment {
1365 ops,
1366 scc_begin: None,
1367 } = phase
1368 else {
1369 panic!("expected an ordinary capture segment");
1370 };
1371 assert_eq!(ops.len(), expected.len());
1372 for (op, output) in ops.iter().zip(expected) {
1373 assert!(
1374 matches!(op, ResidentRecordedOp::Clear { output: actual } if *actual == output)
1375 );
1376 }
1377 }
1378
1379 #[test]
1380 fn coalesces_only_maximal_ordinary_phase_runs() {
1381 let mut phases = (0..1_000)
1382 .map(|output| segment(output, None))
1383 .collect::<Vec<_>>();
1384 phases.push(segment(10_000, Some((64, 41))));
1385 phases.extend((1_000..1_500).map(|output| segment(output, None)));
1386 phases.push(ResidentCapturePhase::ConditionalWhile {
1387 ops: vec![ResidentRecordedOp::Clear { output: 20_000 }],
1388 iteration_limit: 64,
1389 convergence_op_id: 42,
1390 });
1391 phases.extend((1_500..1_800).map(|output| segment(output, None)));
1392
1393 let coalesced = coalesce_resident_capture_phases(phases);
1394 assert_eq!(coalesced.len(), 5);
1395 assert_clear_range(&coalesced[0], 0..1_000);
1396 assert!(matches!(
1397 &coalesced[1],
1398 ResidentCapturePhase::Segment {
1399 ops,
1400 scc_begin: Some((64, 41)),
1401 } if matches!(ops.as_slice(), [ResidentRecordedOp::Clear { output: 10_000 }])
1402 ));
1403 assert_clear_range(&coalesced[2], 1_000..1_500);
1404 assert!(matches!(
1405 &coalesced[3],
1406 ResidentCapturePhase::ConditionalWhile {
1407 ops,
1408 iteration_limit: 64,
1409 convergence_op_id: 42,
1410 } if matches!(ops.as_slice(), [ResidentRecordedOp::Clear { output: 20_000 }])
1411 ));
1412 assert_clear_range(&coalesced[4], 1_500..1_800);
1413 }
1414}
1415
1416struct ResidentBuild<'executor> {
1417 executor: &'executor Executor,
1418 certificate: &'executor ResidentGraphRouteCertificate,
1419 capacity: u32,
1420 relations: Vec<ResidentLogicalRelation>,
1421 filter_workspaces: Vec<ResidentFilterPlan>,
1422 project_workspaces: Vec<ResidentProjectPlan>,
1423 heads: BTreeMap<String, usize>,
1424 head_winner_indices: BTreeMap<String, u32>,
1425 source_names: HashSet<String>,
1426 source_aliases: HashMap<String, usize>,
1427 next_op_id: u32,
1428 injection: Option<crate::resident_graph::ResidentGraphDeviceStatusTestInjection>,
1429 injection_recorded: bool,
1430}
1431
1432#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1433enum ResidentSourceBindingRoute {
1434 Direct,
1435 NormalizeEmpty,
1436}
1437
1438fn resident_source_logical_count(cached: Option<u32>) -> Result<u64> {
1439 cached.map(u64::from).ok_or_else(|| {
1440 XlogError::Execution("resident source requires a cold-path cached logical row count".into())
1441 })
1442}
1443
1444fn resident_source_binding_route(
1445 row_count: u64,
1446 count_tracked: bool,
1447 columns_tracked: bool,
1448) -> Result<ResidentSourceBindingRoute> {
1449 if count_tracked && columns_tracked {
1450 return Ok(ResidentSourceBindingRoute::Direct);
1451 }
1452 if row_count == 0 {
1453 return Ok(ResidentSourceBindingRoute::NormalizeEmpty);
1454 }
1455 Err(XlogError::Execution(
1456 "nonempty resident source is not fully runtime tracked".into(),
1457 ))
1458}
1459
1460#[derive(Debug, Clone)]
1461struct ResidentLogicalRelation {
1462 schema: Schema,
1463 initial_count: u32,
1464 permanent: bool,
1465}
1466
1467#[derive(Debug, Clone)]
1468struct ResidentFilterPlan {
1469 compact_comparisons: Vec<ResidentFilterComparisonDescriptor>,
1470}
1471
1472#[derive(Debug, Clone)]
1473struct ResidentProjectPlan {
1474 compact_expressions: Vec<ResidentProjectExpressionDescriptor>,
1475}
1476
1477#[derive(Debug, Clone, Copy)]
1478struct ResidentSlotAssignment {
1479 slot: usize,
1480 generation: u32,
1481}
1482
1483#[derive(Debug, Clone)]
1484struct ResidentPhysicalSlotPlan {
1485 schema: Schema,
1486 initial_count: u32,
1487 permanent: bool,
1488}
1489
1490#[derive(Debug, Clone)]
1491struct ResidentAllocationManifest {
1492 slots: Vec<ResidentPhysicalSlotPlan>,
1493 logical_to_slot: Vec<ResidentSlotAssignment>,
1494 required_bytes: u64,
1495 relation_bytes: u64,
1496 filter_scratch_bytes: u64,
1497 schedule_metadata_bytes: u64,
1498 fixed_workspace_bytes: u64,
1499 logical_relation_values: usize,
1500 permanent_slots: u32,
1501 scratch_slots: u32,
1502 filter_scratch_allocations: u32,
1503 max_row_bytes: u64,
1504}
1505
1506struct ResidentPhysicalBuild {
1507 relations: Vec<Option<ResidentRelation>>,
1508 filter_scratch: Option<ResidentFilterScratch>,
1509 set_workspace: ResidentSetWorkspace,
1510 join_workspace: ResidentJoinWorkspace,
1511 control: ResidentConvergenceControl,
1512}
1513
1514#[doc(hidden)]
1516#[derive(Default)]
1517pub struct ResidentPrepareDiagnostics {
1518 sample: u64,
1519 total_ns: u64,
1520 required_reservation_bytes: u64,
1521 logical_relation_values: usize,
1522 physical_relation_slots: usize,
1523 relation_device_allocation_calls: u64,
1524 compact_ops: usize,
1525 compact_waves: usize,
1526 compact_regions: usize,
1527 conditional_regions: usize,
1528 parent_graph_nodes: usize,
1529 conditional_body_nodes: usize,
1530 admission_and_source_snapshot_ns: u64,
1531 execution_domain_and_build_setup_ns: u64,
1532 logical_schedule_planning_ns: u64,
1533 manifest_compact_construction_ns: u64,
1534 schedule_lowering_ns: u64,
1535 reservation_ns: u64,
1536 relation_slot_allocation_ns: u64,
1537 relation_slot_allocation_ns_max: u64,
1538 relation_slot_allocation_bytes: u64,
1539 count_initialization_ns: u64,
1540 count_initialization_ns_max: u64,
1541 count_memset_calls: u64,
1542 workspace_provider_calls: u64,
1543 filter_scratch_allocation_ns: u64,
1544 filter_scratch_allocation_bytes: u64,
1545 set_workspace_allocation_ns: u64,
1546 set_workspace_allocation_bytes: u64,
1547 join_workspace_allocation_ns: u64,
1548 join_workspace_allocation_bytes: u64,
1549 control_allocation_ns: u64,
1550 control_allocation_bytes: u64,
1551 metadata_binding_construction_ns: u64,
1552 metadata_provider_calls: u64,
1553 device_trace_preparation_ns: u64,
1554 device_trace_reserved_bytes: u64,
1555 device_trace_initial_htod_calls: u64,
1556 device_trace_initial_htod_bytes: u64,
1557 schema_winners_preparation_ns: u64,
1558 schema_winners_reserved_bytes: u64,
1559 schema_winners_initial_htod_calls: u64,
1560 schema_winners_initial_htod_bytes: u64,
1561 receipt_preparation_ns: u64,
1562 receipt_reserved_bytes: u64,
1563 receipt_initial_htod_calls: u64,
1564 receipt_initial_htod_bytes: u64,
1565 schedule_program_preparation_ns: u64,
1566 schedule_program_reserved_bytes: u64,
1567 schedule_program_initial_htod_calls: u64,
1568 schedule_program_initial_htod_bytes: u64,
1569 reservation_validation_and_release_ns: u64,
1570 pinned_receipt_ns: u64,
1571 graph_body_capture_ns: u64,
1572 graph_instantiate_ns: u64,
1573 validation_owner_assembly_ns: u64,
1574}
1575
1576fn resident_prepare_diagnostics_for_sample(
1577 sample: Option<u64>,
1578) -> Option<ResidentPrepareDiagnostics> {
1579 sample.map(|sample| ResidentPrepareDiagnostics {
1580 sample,
1581 ..ResidentPrepareDiagnostics::default()
1582 })
1583}
1584
1585fn resident_prepare_elapsed_ns(started: std::time::Instant) -> u64 {
1586 u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX)
1587}
1588
1589fn resident_schema_winners_initial_htod(default_count: usize) -> (u64, u64) {
1590 if default_count == 0 {
1591 return (0, 0);
1592 }
1593 (
1594 1,
1595 u64::try_from(default_count)
1596 .unwrap_or(u64::MAX)
1597 .saturating_mul(std::mem::size_of::<u32>() as u64),
1598 )
1599}
1600
1601fn resident_receipt_initial_htod(output_count: usize) -> (u64, u64) {
1602 let pointee_count = output_count.saturating_mul(2).saturating_add(4);
1603 if pointee_count == 0 {
1604 return (0, 0);
1605 }
1606 (
1607 1,
1608 u64::try_from(pointee_count)
1609 .unwrap_or(u64::MAX)
1610 .saturating_mul(std::mem::size_of::<u64>() as u64),
1611 )
1612}
1613
1614fn resident_schedule_initial_htod(reserved_bytes: u64) -> (u64, u64) {
1615 (8, reserved_bytes)
1616}
1617
1618impl ResidentPrepareDiagnostics {
1619 pub fn into_snapshot(self) -> ResidentGraphPrepareDiagnosticSnapshot {
1620 let workspace_allocation_ns = self
1621 .filter_scratch_allocation_ns
1622 .saturating_add(self.set_workspace_allocation_ns)
1623 .saturating_add(self.join_workspace_allocation_ns)
1624 .saturating_add(self.control_allocation_ns);
1625 let workspace_allocation_bytes = self
1626 .filter_scratch_allocation_bytes
1627 .saturating_add(self.set_workspace_allocation_bytes)
1628 .saturating_add(self.join_workspace_allocation_bytes)
1629 .saturating_add(self.control_allocation_bytes);
1630 let metadata_preparation_ns = self
1631 .device_trace_preparation_ns
1632 .saturating_add(self.schema_winners_preparation_ns)
1633 .saturating_add(self.receipt_preparation_ns)
1634 .saturating_add(self.schedule_program_preparation_ns);
1635 let metadata_reserved_bytes = self
1636 .device_trace_reserved_bytes
1637 .saturating_add(self.schema_winners_reserved_bytes)
1638 .saturating_add(self.receipt_reserved_bytes)
1639 .saturating_add(self.schedule_program_reserved_bytes);
1640 let additive = [
1641 self.admission_and_source_snapshot_ns,
1642 self.execution_domain_and_build_setup_ns,
1643 self.logical_schedule_planning_ns,
1644 self.manifest_compact_construction_ns,
1645 self.schedule_lowering_ns,
1646 self.reservation_ns,
1647 self.relation_slot_allocation_ns,
1648 self.count_initialization_ns,
1649 workspace_allocation_ns,
1650 self.metadata_binding_construction_ns,
1651 metadata_preparation_ns,
1652 self.reservation_validation_and_release_ns,
1653 self.pinned_receipt_ns,
1654 self.graph_body_capture_ns,
1655 self.graph_instantiate_ns,
1656 self.validation_owner_assembly_ns,
1657 ];
1658 let attributed_ns = additive.iter().copied().fold(0u64, u64::saturating_add);
1659 let unattributed_ns = self.total_ns.saturating_sub(attributed_ns);
1660 let metadata_initial_htod_calls = self
1661 .device_trace_initial_htod_calls
1662 .saturating_add(self.schema_winners_initial_htod_calls)
1663 .saturating_add(self.receipt_initial_htod_calls)
1664 .saturating_add(self.schedule_program_initial_htod_calls);
1665 let metadata_initial_htod_bytes = self
1666 .device_trace_initial_htod_bytes
1667 .saturating_add(self.schema_winners_initial_htod_bytes)
1668 .saturating_add(self.receipt_initial_htod_bytes)
1669 .saturating_add(self.schedule_program_initial_htod_bytes);
1670 let to_u64 = |value: usize| u64::try_from(value).unwrap_or(u64::MAX);
1671 ResidentGraphPrepareDiagnosticSnapshot {
1672 sample: self.sample,
1673 total_ns: self.total_ns,
1674 admission_and_source_snapshot_ns: self.admission_and_source_snapshot_ns,
1675 execution_domain_and_build_setup_ns: self.execution_domain_and_build_setup_ns,
1676 logical_schedule_planning_ns: self.logical_schedule_planning_ns,
1677 manifest_compact_construction_ns: self.manifest_compact_construction_ns,
1678 schedule_lowering_ns: self.schedule_lowering_ns,
1679 reservation_ns: self.reservation_ns,
1680 relation_preparation_ns: self.relation_slot_allocation_ns,
1681 count_initialization_ns: self.count_initialization_ns,
1682 workspace_preparation_ns: workspace_allocation_ns,
1683 metadata_binding_construction_ns: self.metadata_binding_construction_ns,
1684 metadata_preparation_ns,
1685 reservation_validation_and_release_ns: self.reservation_validation_and_release_ns,
1686 pinned_receipt_ns: self.pinned_receipt_ns,
1687 graph_body_capture_ns: self.graph_body_capture_ns,
1688 graph_instantiate_ns: self.graph_instantiate_ns,
1689 validation_owner_assembly_ns: self.validation_owner_assembly_ns,
1690 unattributed_ns,
1691 required_reservation_bytes: self.required_reservation_bytes,
1692 logical_relation_values: to_u64(self.logical_relation_values),
1693 physical_relation_slots: to_u64(self.physical_relation_slots),
1694 relation_device_allocation_calls: self.relation_device_allocation_calls,
1695 relation_reserved_bytes: self.relation_slot_allocation_bytes,
1696 relation_slot_preparation_ns_max: self.relation_slot_allocation_ns_max,
1697 count_memset_calls: self.count_memset_calls,
1698 count_memset_bytes: self
1699 .count_memset_calls
1700 .saturating_mul(std::mem::size_of::<u32>() as u64),
1701 count_initialization_ns_max: self.count_initialization_ns_max,
1702 workspace_provider_calls: self.workspace_provider_calls,
1703 workspace_reserved_bytes: workspace_allocation_bytes,
1704 filter_scratch_preparation_ns: self.filter_scratch_allocation_ns,
1705 filter_scratch_reserved_bytes: self.filter_scratch_allocation_bytes,
1706 set_workspace_preparation_ns: self.set_workspace_allocation_ns,
1707 set_workspace_reserved_bytes: self.set_workspace_allocation_bytes,
1708 join_workspace_preparation_ns: self.join_workspace_allocation_ns,
1709 join_workspace_reserved_bytes: self.join_workspace_allocation_bytes,
1710 control_preparation_ns: self.control_allocation_ns,
1711 control_reserved_bytes: self.control_allocation_bytes,
1712 metadata_provider_calls: self.metadata_provider_calls,
1713 metadata_reserved_bytes,
1714 metadata_initial_htod_calls,
1715 metadata_initial_htod_bytes,
1716 device_trace_preparation_ns: self.device_trace_preparation_ns,
1717 device_trace_reserved_bytes: self.device_trace_reserved_bytes,
1718 device_trace_initial_htod_calls: self.device_trace_initial_htod_calls,
1719 device_trace_initial_htod_bytes: self.device_trace_initial_htod_bytes,
1720 schema_winners_preparation_ns: self.schema_winners_preparation_ns,
1721 schema_winners_reserved_bytes: self.schema_winners_reserved_bytes,
1722 schema_winners_initial_htod_calls: self.schema_winners_initial_htod_calls,
1723 schema_winners_initial_htod_bytes: self.schema_winners_initial_htod_bytes,
1724 receipt_preparation_ns: self.receipt_preparation_ns,
1725 receipt_reserved_bytes: self.receipt_reserved_bytes,
1726 receipt_initial_htod_calls: self.receipt_initial_htod_calls,
1727 receipt_initial_htod_bytes: self.receipt_initial_htod_bytes,
1728 schedule_program_preparation_ns: self.schedule_program_preparation_ns,
1729 schedule_program_reserved_bytes: self.schedule_program_reserved_bytes,
1730 schedule_program_initial_htod_calls: self.schedule_program_initial_htod_calls,
1731 schedule_program_initial_htod_bytes: self.schedule_program_initial_htod_bytes,
1732 compact_ops: to_u64(self.compact_ops),
1733 compact_waves: to_u64(self.compact_waves),
1734 compact_regions: to_u64(self.compact_regions),
1735 conditional_regions: to_u64(self.conditional_regions),
1736 parent_graph_nodes: to_u64(self.parent_graph_nodes),
1737 conditional_body_nodes: to_u64(self.conditional_body_nodes),
1738 }
1739 }
1740}
1741
1742#[cfg(test)]
1743mod prepare_diagnostic_tests {
1744 use super::{
1745 resident_prepare_diagnostics_for_sample, resident_receipt_initial_htod,
1746 resident_schedule_initial_htod, resident_schema_winners_initial_htod,
1747 ResidentPrepareDiagnostics,
1748 };
1749
1750 #[test]
1751 fn prepare_diagnostic_counts_only_source_proven_initial_transfers() {
1752 assert_eq!(resident_schema_winners_initial_htod(0), (0, 0));
1753 assert_eq!(resident_schema_winners_initial_htod(3), (1, 12));
1754 assert_eq!(resident_receipt_initial_htod(0), (1, 32));
1755 assert_eq!(resident_receipt_initial_htod(3), (1, 80));
1756 assert_eq!(resident_schedule_initial_htod(4096), (8, 4096));
1757
1758 let (schema_calls, schema_bytes) = resident_schema_winners_initial_htod(3);
1759 let (receipt_calls, receipt_bytes) = resident_receipt_initial_htod(0);
1760 let (schedule_calls, schedule_bytes) = resident_schedule_initial_htod(4096);
1761 let diagnostics = ResidentPrepareDiagnostics {
1762 sample: 9,
1763 total_ns: 100,
1764 relation_slot_allocation_ns: 11,
1765 relation_slot_allocation_ns_max: 7,
1766 relation_slot_allocation_bytes: 2048,
1767 workspace_provider_calls: 4,
1768 count_initialization_ns: 10,
1769 count_initialization_ns_max: 5,
1770 count_memset_calls: 3,
1771 metadata_provider_calls: 4,
1772 schema_winners_initial_htod_calls: schema_calls,
1773 schema_winners_initial_htod_bytes: schema_bytes,
1774 receipt_initial_htod_calls: receipt_calls,
1775 receipt_initial_htod_bytes: receipt_bytes,
1776 schedule_program_initial_htod_calls: schedule_calls,
1777 schedule_program_initial_htod_bytes: schedule_bytes,
1778 ..ResidentPrepareDiagnostics::default()
1779 };
1780 let snapshot = diagnostics.into_snapshot();
1781 assert_eq!(snapshot.sample, 9);
1782 assert_eq!(snapshot.total_ns, 100);
1783 assert_eq!(snapshot.relation_preparation_ns, 11);
1784 assert_eq!(snapshot.relation_reserved_bytes, 2048);
1785 assert_eq!(snapshot.relation_slot_preparation_ns_max, 7);
1786 assert_eq!(snapshot.workspace_provider_calls, 4);
1787 assert_eq!(snapshot.count_memset_calls, 3);
1788 assert_eq!(snapshot.count_initialization_ns, 10);
1789 assert_eq!(snapshot.count_initialization_ns_max, 5);
1790 assert_eq!(
1791 snapshot.count_memset_bytes,
1792 3 * std::mem::size_of::<u32>() as u64
1793 );
1794 assert_eq!(snapshot.metadata_provider_calls, 4);
1795 assert_eq!(snapshot.device_trace_initial_htod_calls, 0);
1796 assert_eq!(snapshot.device_trace_initial_htod_bytes, 0);
1797 assert_eq!(snapshot.schema_winners_initial_htod_calls, 1);
1798 assert_eq!(snapshot.schema_winners_initial_htod_bytes, 12);
1799 assert_eq!(snapshot.receipt_initial_htod_calls, 1);
1800 assert_eq!(snapshot.receipt_initial_htod_bytes, 32);
1801 assert_eq!(snapshot.schedule_program_initial_htod_calls, 8);
1802 assert_eq!(snapshot.schedule_program_initial_htod_bytes, 4096);
1803 assert_eq!(snapshot.metadata_initial_htod_calls, 10);
1804 assert_eq!(snapshot.metadata_initial_htod_bytes, 4140);
1805 }
1806
1807 #[test]
1808 fn prepare_diagnostic_is_absent_when_sampling_is_disabled() {
1809 assert!(resident_prepare_diagnostics_for_sample(None).is_none());
1810 assert_eq!(
1811 resident_prepare_diagnostics_for_sample(Some(23))
1812 .expect("enabled diagnostics create raw timing state")
1813 .sample,
1814 23
1815 );
1816 }
1817
1818 #[test]
1819 fn prepare_diagnostic_keeps_setup_and_reservation_validation_separate() {
1820 let snapshot = ResidentPrepareDiagnostics {
1821 sample: 5,
1822 total_ns: 100,
1823 admission_and_source_snapshot_ns: 11,
1824 execution_domain_and_build_setup_ns: 13,
1825 metadata_binding_construction_ns: 17,
1826 reservation_validation_and_release_ns: 19,
1827 ..ResidentPrepareDiagnostics::default()
1828 }
1829 .into_snapshot();
1830
1831 assert_eq!(snapshot.admission_and_source_snapshot_ns, 11);
1832 assert_eq!(snapshot.execution_domain_and_build_setup_ns, 13);
1833 assert_eq!(snapshot.metadata_binding_construction_ns, 17);
1834 assert_eq!(snapshot.reservation_validation_and_release_ns, 19);
1835 assert_eq!(snapshot.unattributed_ns, 40);
1836 }
1837}
1838
1839impl ResidentAllocationManifest {
1840 fn finalize_compact_schedule(
1841 &mut self,
1842 schedule: &ResidentCompactSchedulePlan,
1843 head_count: usize,
1844 ) -> Result<()> {
1845 let (required_bytes, schedule_metadata_bytes) = resident_compact_allocation_bytes(
1846 self.relation_bytes,
1847 self.filter_scratch_bytes,
1848 self.fixed_workspace_bytes,
1849 self.slots.len(),
1850 head_count,
1851 schedule,
1852 )?;
1853 self.required_bytes = required_bytes;
1854 self.schedule_metadata_bytes = schedule_metadata_bytes;
1855 Ok(())
1856 }
1857}
1858
1859fn resident_output_indices(
1860 heads: &BTreeMap<String, usize>,
1861 assignments: &[ResidentSlotAssignment],
1862 slots: &[ResidentPhysicalSlotPlan],
1863) -> Result<Vec<(String, usize)>> {
1864 heads
1865 .iter()
1866 .map(|(name, logical)| {
1867 let assignment = assignments.get(*logical).ok_or_else(|| {
1868 XlogError::Execution("resident head has no physical slot assignment".into())
1869 })?;
1870 if !slots
1871 .get(assignment.slot)
1872 .is_some_and(|slot| slot.permanent)
1873 {
1874 return Err(XlogError::Execution(
1875 "resident head is not assigned to a permanent physical slot".into(),
1876 ));
1877 }
1878 Ok((name.clone(), assignment.slot))
1879 })
1880 .collect()
1881}
1882
1883fn resident_validate_exact_reservation(required: u64, used: u64, remaining: u64) -> Result<()> {
1884 if used == required && remaining == 0 {
1885 return Ok(());
1886 }
1887 Err(XlogError::Execution(format!(
1888 "resident allocation manifest accounted for {required} manager-tracked bytes but materialization consumed {used} and left {remaining} reserved bytes"
1889 )))
1890}
1891
1892#[derive(Debug)]
1893struct ResidentScratchSlotState {
1894 slot: usize,
1895 layout: Vec<ScalarType>,
1896 last_use: usize,
1897 generation: u32,
1898}
1899
1900fn resident_schema_layout(schema: &Schema) -> Vec<ScalarType> {
1901 schema
1902 .columns
1903 .iter()
1904 .map(|(_, scalar)| scalar.clone())
1905 .collect()
1906}
1907
1908fn resident_private_inputs(op: &ResidentRecordedOp, mut visit: impl FnMut(usize)) {
1909 let mut visit_ref = |reference: &ResidentBufferRef| {
1910 if let ResidentBufferRef::Private(index) = reference {
1911 visit(*index);
1912 }
1913 };
1914 match op {
1915 ResidentRecordedOp::Scan {
1916 relation: input, ..
1917 }
1918 | ResidentRecordedOp::Filter { input, .. }
1919 | ResidentRecordedOp::Project { input, .. } => visit_ref(input),
1920 ResidentRecordedOp::Union { left, right, .. }
1921 | ResidentRecordedOp::Diff { left, right, .. }
1922 | ResidentRecordedOp::Join { left, right, .. } => {
1923 visit_ref(left);
1924 visit_ref(right);
1925 }
1926 ResidentRecordedOp::ChangedMark { relation } => visit(*relation),
1927 ResidentRecordedOp::TraceDelta { semantic_guard, .. } => {
1928 if let Some(semantic_guard) = semantic_guard {
1929 visit_ref(semantic_guard);
1930 }
1931 }
1932 ResidentRecordedOp::SchemaWinnerMark { contribution, .. } => visit_ref(contribution),
1933 ResidentRecordedOp::Unit { .. }
1934 | ResidentRecordedOp::Clear { .. }
1935 | ResidentRecordedOp::ChangedReset
1936 | ResidentRecordedOp::TestStatus(_) => {}
1937 }
1938}
1939
1940fn resident_private_output(op: &ResidentRecordedOp) -> Option<usize> {
1941 match op {
1942 ResidentRecordedOp::Unit { output, .. }
1943 | ResidentRecordedOp::Clear { output }
1944 | ResidentRecordedOp::Filter { output, .. }
1945 | ResidentRecordedOp::Project { output, .. }
1946 | ResidentRecordedOp::Union { output, .. }
1947 | ResidentRecordedOp::Diff { output, .. }
1948 | ResidentRecordedOp::Join { output, .. } => Some(*output),
1949 ResidentRecordedOp::Scan { .. }
1950 | ResidentRecordedOp::TraceDelta { .. }
1951 | ResidentRecordedOp::ChangedReset
1952 | ResidentRecordedOp::ChangedMark { .. }
1953 | ResidentRecordedOp::SchemaWinnerMark { .. }
1954 | ResidentRecordedOp::TestStatus(_) => None,
1955 }
1956}
1957
1958fn resident_source_slot_map<'a>(
1959 private_slot_count: usize,
1960 sources: impl IntoIterator<Item = &'a str>,
1961) -> Result<BTreeMap<String, u32>> {
1962 let first_source_slot = u32::try_from(private_slot_count)
1963 .map_err(|_| XlogError::Execution("resident private slot count exceeds u32".into()))?;
1964 let names = sources
1965 .into_iter()
1966 .map(str::to_owned)
1967 .collect::<BTreeSet<_>>();
1968 names
1969 .into_iter()
1970 .enumerate()
1971 .map(|(offset, name)| {
1972 let offset = u32::try_from(offset).map_err(|_| {
1973 XlogError::Execution("resident source slot count exceeds u32".into())
1974 })?;
1975 let slot = first_source_slot.checked_add(offset).ok_or_else(|| {
1976 XlogError::Execution("resident source slot index overflow".into())
1977 })?;
1978 Ok((name, slot))
1979 })
1980 .collect()
1981}
1982
1983fn resident_record_lifetimes(
1984 ops: &[ResidentRecordedOp],
1985 relations: &[ResidentLogicalRelation],
1986 definitions: &mut [Option<usize>],
1987 last_uses: &mut [Option<usize>],
1988 ordinal: &mut usize,
1989) -> Result<(usize, usize)> {
1990 let start = *ordinal;
1991 for op in ops {
1992 let current = *ordinal;
1993 let mut invalid_input = None;
1994 let mut input_before_definition = None;
1995 resident_private_inputs(op, |logical| {
1996 let Some(relation) = relations.get(logical) else {
1997 invalid_input.get_or_insert(logical);
1998 return;
1999 };
2000 if !relation.permanent {
2001 if definitions[logical].is_none() {
2002 input_before_definition.get_or_insert(logical);
2003 } else {
2004 last_uses[logical] = Some(last_uses[logical].unwrap_or(current).max(current));
2005 }
2006 }
2007 });
2008 if let Some(logical) = invalid_input {
2009 return Err(XlogError::Execution(format!(
2010 "resident logical input relation {logical} is missing"
2011 )));
2012 }
2013 if let Some(logical) = input_before_definition {
2014 return Err(XlogError::Execution(format!(
2015 "resident scratch relation {logical} is used before its definition"
2016 )));
2017 }
2018 if let Some(logical) = resident_private_output(op) {
2019 let relation = relations.get(logical).ok_or_else(|| {
2020 XlogError::Execution(format!(
2021 "resident logical output relation {logical} is missing"
2022 ))
2023 })?;
2024 if !relation.permanent {
2025 if definitions[logical].replace(current).is_some() {
2026 return Err(XlogError::Execution(format!(
2027 "resident scratch relation {logical} has multiple definitions"
2028 )));
2029 }
2030 last_uses[logical] = Some(last_uses[logical].unwrap_or(current).max(current));
2031 }
2032 }
2033 *ordinal = ordinal
2034 .checked_add(1)
2035 .ok_or_else(|| XlogError::Execution("resident operation ordinal overflow".into()))?;
2036 }
2037 Ok((start, *ordinal))
2038}
2039
2040fn resident_validate_slot_assignments(
2041 relations: &[ResidentLogicalRelation],
2042 definitions: &[Option<usize>],
2043 last_uses: &[Option<usize>],
2044 slots: &[ResidentPhysicalSlotPlan],
2045 assignments: &[ResidentSlotAssignment],
2046) -> Result<()> {
2047 if definitions.len() != relations.len()
2048 || last_uses.len() != relations.len()
2049 || assignments.len() != relations.len()
2050 {
2051 return Err(XlogError::Execution(
2052 "resident slot validation vector length mismatch".into(),
2053 ));
2054 }
2055
2056 let mut occupied_permanent_slots = HashSet::new();
2057 let mut scratch_by_slot = BTreeMap::<usize, Vec<(usize, usize, usize, u32)>>::new();
2058 let mut assigned_slots = HashSet::new();
2059 for (logical, relation) in relations.iter().enumerate() {
2060 let assignment = assignments[logical];
2061 let slot = slots.get(assignment.slot).ok_or_else(|| {
2062 XlogError::Execution(format!(
2063 "resident logical relation {logical} maps to missing physical slot {}",
2064 assignment.slot
2065 ))
2066 })?;
2067 assigned_slots.insert(assignment.slot);
2068 if relation.permanent {
2069 if assignment.generation != 0 {
2070 return Err(XlogError::Execution(format!(
2071 "resident permanent relation {logical} has nonzero generation {}",
2072 assignment.generation
2073 )));
2074 }
2075 if !slot.permanent
2076 || slot.schema != relation.schema
2077 || slot.initial_count != relation.initial_count
2078 {
2079 return Err(XlogError::Execution(format!(
2080 "resident permanent relation {logical} has an incompatible physical slot"
2081 )));
2082 }
2083 if !occupied_permanent_slots.insert(assignment.slot) {
2084 return Err(XlogError::Execution(format!(
2085 "resident permanent physical slot {} has multiple owners",
2086 assignment.slot
2087 )));
2088 }
2089 continue;
2090 }
2091
2092 if slot.permanent
2093 || resident_schema_layout(&slot.schema) != resident_schema_layout(&relation.schema)
2094 {
2095 return Err(XlogError::Execution(format!(
2096 "resident scratch relation {logical} has an incompatible physical slot"
2097 )));
2098 }
2099 let definition = definitions[logical].ok_or_else(|| {
2100 XlogError::Execution(format!(
2101 "resident scratch relation {logical} is never defined"
2102 ))
2103 })?;
2104 let last_use = last_uses[logical].unwrap_or(definition);
2105 if last_use < definition {
2106 return Err(XlogError::Execution(format!(
2107 "resident scratch relation {logical} ends before its definition"
2108 )));
2109 }
2110 scratch_by_slot.entry(assignment.slot).or_default().push((
2111 definition,
2112 last_use,
2113 logical,
2114 assignment.generation,
2115 ));
2116 }
2117
2118 if assigned_slots.len() != slots.len() {
2119 return Err(XlogError::Execution(
2120 "resident allocation manifest contains an unassigned physical slot".into(),
2121 ));
2122 }
2123 for (slot, mut generations) in scratch_by_slot {
2124 generations.sort_by_key(|(definition, _, logical, _)| (*definition, *logical));
2125 let mut previous_last_use = None;
2126 for (expected_generation, (definition, last_use, logical, generation)) in
2127 generations.into_iter().enumerate()
2128 {
2129 let expected_generation = u32::try_from(expected_generation).map_err(|_| {
2130 XlogError::Execution("resident scratch generation exceeds u32".into())
2131 })?;
2132 if generation != expected_generation {
2133 return Err(XlogError::Execution(format!(
2134 "resident scratch relation {logical} in slot {slot} has generation {generation} but expected {expected_generation}"
2135 )));
2136 }
2137 if previous_last_use.is_some_and(|previous| previous >= definition) {
2138 return Err(XlogError::Execution(format!(
2139 "resident scratch generations overlap in physical slot {slot}"
2140 )));
2141 }
2142 previous_last_use = Some(last_use);
2143 }
2144 }
2145 Ok(())
2146}
2147
2148fn resident_checked_add(total: &mut u64, bytes: u64, label: &str) -> Result<()> {
2149 *total = total.checked_add(bytes).ok_or_else(|| {
2150 XlogError::Execution(format!("resident {label} allocation byte overflow"))
2151 })?;
2152 Ok(())
2153}
2154
2155pub struct PreparedResidentGraph<'executor> {
2157 owners: ResidentRunOwners,
2158 preflight_report: ResidentGraphPreflightReport,
2159 has_device_status_writer: bool,
2160 source_guard: &'executor Executor,
2161 source_set_snapshots: Vec<ResidentSourceSetSnapshot>,
2162 prepare_diagnostic: Option<ResidentPrepareDiagnostics>,
2163}
2164
2165#[derive(Debug, Clone, PartialEq, Eq)]
2167pub struct ResidentGraphPreflightReport {
2168 pub relation_capacity: u32,
2170 pub estimated_required_bytes: u64,
2172 pub available_bytes_at_admission: u64,
2174 pub tracked_device_allocation_bytes: u64,
2176 pub relation_device_bytes: u64,
2178 pub filter_descriptor_device_bytes: u64,
2180 pub filter_scratch_device_bytes: u64,
2182 pub project_descriptor_device_bytes: u64,
2184 pub fixed_workspace_device_bytes: u64,
2186 pub parent_graph_nodes: usize,
2188 pub conditional_while_nodes: usize,
2190 pub parent_graph_node_kinds: Vec<CudaGraphNodeKind>,
2192 pub conditional_body_node_kinds: Vec<Vec<CudaGraphNodeKind>>,
2194 pub conditional_body_kernel_counts: Vec<usize>,
2196 pub hierarchical_graph_nodes: usize,
2198 pub private_relation_slots: usize,
2200 pub logical_relation_values: usize,
2202 pub permanent_relation_slots: u32,
2204 pub staged_output_relations: usize,
2206 pub scratch_slots: u32,
2208 pub filter_scratch_allocations: u32,
2210 pub max_row_bytes: u64,
2212}
2213
2214pub struct ResidentGraphInFlight<'executor> {
2216 completion: ResidentCompletionEvent,
2219 timing_start: CudaEvent,
2220 timing_end: CudaEvent,
2221 owners: ResidentRunOwners,
2222 _executor: PhantomData<&'executor Executor>,
2223}
2224
2225pub struct ResidentGraphSynchronized<'executor> {
2227 device_elapsed_ns: u64,
2228 owners: ResidentRunOwners,
2229 _executor: PhantomData<&'executor Executor>,
2230}
2231
2232struct StagedResidentOutput {
2233 name: String,
2234 buffer: CudaBuffer,
2235}
2236
2237pub struct ObservedResidentGraphReceipt {
2239 encoded_len: usize,
2240 device_elapsed_ns: u64,
2241 device_scan_invocations: u64,
2242 device_filter_invocations: u64,
2243 semantic_scan_invocations: u64,
2244 semantic_filter_invocations: u64,
2245 iterations: u32,
2246 terminal: std::result::Result<(), ResidentGraphExecutionError>,
2247 outputs: Vec<StagedResidentOutput>,
2248 source_epoch: u64,
2249 relation_registration: Vec<(RelId, String)>,
2250 transaction_identity: Arc<()>,
2251 provider: Arc<CudaKernelProvider>,
2252 phase_timings: Option<ResidentFinalObservationPhaseTimings>,
2253}
2254
2255#[doc(hidden)]
2257#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2258pub struct ResidentFinalObservationPhaseTimings {
2259 pub receipt_d2h_ns: u64,
2261 pub decode_schema_staging_ns: u64,
2263}
2264
2265fn resident_source_set_snapshot(
2266 provider: &CudaKernelProvider,
2267 name: &str,
2268 version: u64,
2269 buffer: &CudaBuffer,
2270) -> std::result::Result<ResidentSourceSetSnapshot, ResidentGraphDeclineReason> {
2271 if !buffer.canonical_full_row_set_certified() {
2272 return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2273 relation: name.to_owned(),
2274 });
2275 }
2276 let manager_ptr = Arc::as_ptr(provider.memory()) as usize;
2277 let runtime = provider.memory().runtime().ok_or_else(|| {
2278 ResidentGraphDeclineReason::SourceSetUncertified {
2279 relation: format!("{name} memory manager has no device runtime"),
2280 }
2281 })?;
2282 if !Arc::ptr_eq(provider.device(), provider.memory().device())
2283 || !Arc::ptr_eq(provider.device(), runtime.device())
2284 || u32::try_from(provider.device().ordinal()).ok() != Some(runtime.device_ordinal())
2285 {
2286 return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2287 relation: format!("{name} memory manager does not match the resident provider"),
2288 });
2289 }
2290 let column_blocks = buffer
2291 .columns()
2292 .iter()
2293 .enumerate()
2294 .map(|(column_index, column)| {
2295 let CudaColumn::Owned(column) = column else {
2296 return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2297 relation: format!("{name} column {column_index} is externally owned"),
2298 });
2299 };
2300 if column.memory_manager_ptr_value() != manager_ptr {
2301 return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2302 relation: format!(
2303 "{name} column {column_index} belongs to another memory manager"
2304 ),
2305 });
2306 }
2307 let block = column.runtime_block().map(BlockId::from_block);
2308 if block.is_none() && buffer.num_rows() != 0 {
2309 return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2310 relation: format!("{name} column {column_index} is not runtime tracked"),
2311 });
2312 }
2313 Ok(block)
2314 })
2315 .collect::<std::result::Result<Vec<_>, _>>()?;
2316 let row_count = buffer.num_rows_device();
2317 if row_count.memory_manager_ptr_value() != manager_ptr {
2318 return Err(ResidentGraphDeclineReason::SourceSetUncertified {
2319 relation: format!("{name} logical count belongs to another memory manager"),
2320 });
2321 }
2322 let row_count_block = row_count
2323 .runtime_block()
2324 .map(BlockId::from_block)
2325 .ok_or_else(|| ResidentGraphDeclineReason::SourceSetUncertified {
2326 relation: format!("{name} logical count is not runtime tracked"),
2327 })?;
2328 Ok(ResidentSourceSetSnapshot {
2329 name: name.to_owned(),
2330 version,
2331 schema: buffer.schema().clone(),
2332 row_capacity: buffer.num_rows(),
2333 column_blocks,
2334 row_count_block,
2335 })
2336}
2337
2338fn validate_resident_source_set_snapshots(
2339 executor: &Executor,
2340 source_epoch: u64,
2341 snapshots: &[ResidentSourceSetSnapshot],
2342) -> std::result::Result<(), ResidentGraphExecutionError> {
2343 if executor.store.mutation_epoch() != source_epoch {
2344 return Err(resident_decline_error(
2345 ResidentGraphDeclineReason::SourceSetUncertified {
2346 relation: "relation store changed after resident preparation".to_owned(),
2347 },
2348 ));
2349 }
2350 for snapshot in snapshots {
2351 let Some((buffer, version)) = executor.store.get_with_version(&snapshot.name) else {
2352 return Err(resident_decline_error(
2353 ResidentGraphDeclineReason::SourceSetUncertified {
2354 relation: snapshot.name.clone(),
2355 },
2356 ));
2357 };
2358 let current =
2359 resident_source_set_snapshot(&executor.provider, &snapshot.name, version, buffer)
2360 .map_err(resident_decline_error)?;
2361 if ¤t != snapshot {
2362 return Err(resident_decline_error(
2363 ResidentGraphDeclineReason::SourceSetUncertified {
2364 relation: snapshot.name.clone(),
2365 },
2366 ));
2367 }
2368 }
2369 Ok(())
2370}
2371
2372impl<'executor> PreparedResidentGraph<'executor> {
2373 pub fn preflight_report(&self) -> &ResidentGraphPreflightReport {
2375 &self.preflight_report
2376 }
2377
2378 #[doc(hidden)]
2380 pub fn take_prepare_diagnostic(&mut self) -> Option<ResidentPrepareDiagnostics> {
2381 self.prepare_diagnostic.take()
2382 }
2383
2384 #[cfg(feature = "resident-graph-tests")]
2385 pub(crate) fn invalidate_expected_source_epoch(&mut self) {
2386 self.owners.source_epoch = self.owners.source_epoch.wrapping_add(1);
2387 }
2388
2389 pub fn launch(
2391 mut self,
2392 ) -> std::result::Result<ResidentGraphInFlight<'executor>, ResidentGraphExecutionError> {
2393 validate_resident_source_set_snapshots(
2394 self.source_guard,
2395 self.owners.source_epoch,
2396 &self.source_set_snapshots,
2397 )?;
2398 self.owners
2399 .execution_domain
2400 .preflight(&mut self.owners.recorder)
2401 .map_err(runtime_error)?;
2402 let timing_start = self
2403 .owners
2404 .stream
2405 .record_event(Some(CUevent_flags::CU_EVENT_DEFAULT))
2406 .map_err(|error| {
2407 runtime_error(format!("resident timing-start event failed: {error}"))
2408 })?;
2409 self.owners
2410 .graph
2411 .launch(&self.owners.stream)
2412 .map_err(runtime_error)?;
2413 let timing_end = match self
2414 .owners
2415 .stream
2416 .record_event(Some(CUevent_flags::CU_EVENT_DEFAULT))
2417 {
2418 Ok(event) => event,
2419 Err(error) => {
2420 let _ = self.owners.stream.synchronize();
2421 return Err(runtime_error(format!(
2422 "resident timing-end event failed after graph launch: {error}"
2423 )));
2424 }
2425 };
2426 self.owners
2427 .runtime
2428 .record_conditional_graph_launch(self.has_device_status_writer);
2429 let recorder = std::mem::replace(
2430 &mut self.owners.recorder,
2431 self.owners.execution_domain.new_strict_recorder(),
2432 );
2433 if let Err(error) = self.owners.execution_domain.commit(recorder) {
2434 let _ = self.owners.stream.synchronize();
2435 return Err(runtime_error(error));
2436 }
2437 let completion = match self
2438 .owners
2439 .runtime
2440 .record_resident_completion_event(&self.owners.stream)
2441 {
2442 Ok(completion) => completion,
2443 Err(error) => {
2444 let _ = self.owners.stream.synchronize();
2445 return Err(runtime_error(error));
2446 }
2447 };
2448 Ok(ResidentGraphInFlight {
2449 completion,
2450 timing_start,
2451 timing_end,
2452 owners: self.owners,
2453 _executor: PhantomData,
2454 })
2455 }
2456}
2457
2458impl<'executor> ResidentGraphInFlight<'executor> {
2459 pub fn synchronize_core(
2461 mut self,
2462 ) -> std::result::Result<ResidentGraphSynchronized<'executor>, ResidentGraphExecutionError>
2463 {
2464 self.completion.synchronize().map_err(runtime_error)?;
2465 let elapsed_ms = self
2466 .timing_start
2467 .elapsed_ms(&self.timing_end)
2468 .map_err(|error| {
2469 runtime_error(format!("resident CUDA-event timing failed: {error}"))
2470 })?;
2471 if !elapsed_ms.is_finite() || elapsed_ms < 0.0 {
2472 return Err(runtime_error("resident CUDA-event timing was not finite"));
2473 }
2474 let device_elapsed_ns = (f64::from(elapsed_ms) * 1_000_000.0)
2475 .round()
2476 .clamp(0.0, u64::MAX as f64) as u64;
2477 Ok(ResidentGraphSynchronized {
2478 device_elapsed_ns,
2479 owners: self.owners,
2480 _executor: PhantomData,
2481 })
2482 }
2483}
2484
2485impl<'executor> ResidentGraphSynchronized<'executor> {
2486 pub fn observe_final_receipt(
2488 mut self,
2489 ) -> std::result::Result<ObservedResidentGraphReceipt, ResidentGraphExecutionError> {
2490 let phase_diagnostics =
2491 std::env::var("XLOG_RESIDENT_LATENCY_DIAGNOSTICS").as_deref() == Ok("1");
2492 let receipt_d2h_started = phase_diagnostics.then(Instant::now);
2493 let encoded_len = self.owners.receipt.len_bytes();
2494 let bytes = self
2495 .owners
2496 .provider
2497 .observe_resident_packed_receipt(
2498 &self.owners.receipt,
2499 &mut self.owners.pinned_receipt,
2500 &self.owners.stream,
2501 )
2502 .map_err(runtime_error)?;
2503 let receipt_d2h_ns = receipt_d2h_started
2504 .map(|started| u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX))
2505 .unwrap_or(0);
2506 let decode_started = phase_diagnostics.then(Instant::now);
2507 let relation_count_len = self.owners.output_indices.len();
2508 let schema_winner_count = self.owners.output_schema_plans.len();
2509 let expected_count_fields = relation_count_len + 4 + schema_winner_count;
2510 if self.owners.receipt.relation_count_len() as usize != relation_count_len
2511 || self.owners.receipt.device_trace_field_count() != 4
2512 || self.owners.receipt.schema_winner_count() as usize != schema_winner_count
2513 || self.owners.receipt.total_count_field_len() as usize != expected_count_fields
2514 || bytes.len() != encoded_len
2515 || encoded_len != 44 + 4 * expected_count_fields
2516 {
2517 return Err(runtime_error(format!(
2518 "malformed resident receipt length: got {}, expected {}",
2519 bytes.len(),
2520 44 + 4 * expected_count_fields
2521 )));
2522 }
2523 let field_u32 = |offset| {
2524 read_receipt_u32(&bytes, offset)
2525 .ok_or_else(|| runtime_error("truncated resident receipt u32 field"))
2526 };
2527 let field_u64 = |offset| {
2528 read_receipt_u64(&bytes, offset)
2529 .ok_or_else(|| runtime_error("truncated resident receipt u64 field"))
2530 };
2531 let code = field_u32(0)?;
2532 let op_id = field_u32(4)?;
2533 let resource_code = field_u32(8)?;
2534 let iterations = field_u32(12)?;
2535 let limit = field_u32(16)?;
2536 let reserved = field_u32(20)?;
2537 let required = field_u64(24)?;
2538 let capacity = field_u64(32)?;
2539 let changed = field_u32(40)?;
2540 if reserved != 0 || changed > 1 {
2541 return Err(runtime_error(
2542 "malformed resident terminal status reserved/changed field",
2543 ));
2544 }
2545 let terminal = match code {
2546 value if value == ResidentTerminalCode::Success as u32 => {
2547 if resource_code != ResidentResourceCode::None as u32
2548 || limit != 0
2549 || required != 0
2550 || capacity != 0
2551 {
2552 return Err(runtime_error("malformed resident success payload"));
2553 }
2554 Ok(())
2555 }
2556 value if value == ResidentTerminalCode::IterationLimit as u32 => {
2557 if resource_code != ResidentResourceCode::None as u32
2558 || required != 0
2559 || capacity != 0
2560 || iterations > limit
2561 {
2562 return Err(runtime_error("malformed resident iteration-limit payload"));
2563 }
2564 Err(ResidentGraphExecutionError::IterationLimit {
2565 limit,
2566 completed: iterations,
2567 })
2568 }
2569 value if value == ResidentTerminalCode::CapacityOverflow as u32 => {
2570 if resource_code != ResidentResourceCode::OutputRows as u32 || required <= capacity
2571 {
2572 return Err(runtime_error(
2573 "malformed resident capacity-overflow payload",
2574 ));
2575 }
2576 Err(ResidentGraphExecutionError::CapacityOverflow {
2577 op_id,
2578 required,
2579 capacity,
2580 })
2581 }
2582 value if value == ResidentTerminalCode::ResourceExhausted as u32 => {
2583 let resource = match resource_code {
2584 value
2585 if value == ResidentResourceCode::SetHashSlots as u32
2586 || value == ResidentResourceCode::JoinBuckets as u32
2587 || value == ResidentResourceCode::JoinChains as u32 =>
2588 {
2589 "workspace_slots"
2590 }
2591 value if value == ResidentResourceCode::InputRows as u32 => "input_rows",
2592 value if value == ResidentResourceCode::OutputRows as u32 => "output_rows",
2593 _ => return Err(runtime_error("unknown resident resource code")),
2594 };
2595 if required <= capacity {
2596 return Err(runtime_error("malformed resident resource payload"));
2597 }
2598 Err(ResidentGraphExecutionError::ResourceExhausted {
2599 op_id,
2600 resource,
2601 required,
2602 capacity,
2603 })
2604 }
2605 _ => {
2606 return Err(runtime_error(format!(
2607 "unknown resident terminal code {code}"
2608 )))
2609 }
2610 };
2611
2612 let mut counts = Vec::with_capacity(self.owners.output_indices.len());
2613 for index in 0..self.owners.output_indices.len() {
2614 counts.push(field_u32(44 + index * 4)?);
2615 }
2616 let device_scan_invocations = u64::from(field_u32(44 + relation_count_len * 4)?);
2617 let device_filter_invocations = u64::from(field_u32(48 + relation_count_len * 4)?);
2618 let semantic_scan_invocations = u64::from(field_u32(52 + relation_count_len * 4)?);
2619 let semantic_filter_invocations = u64::from(field_u32(56 + relation_count_len * 4)?);
2620 let schema_winner_offset = 60 + relation_count_len * 4;
2621 let mut schema_winner_ids = Vec::with_capacity(schema_winner_count);
2622 for index in 0..schema_winner_count {
2623 schema_winner_ids.push(field_u32(schema_winner_offset + index * 4)?);
2624 }
2625 let mut outputs = Vec::new();
2626 if terminal.is_ok() {
2627 let selected_schemas = resident_resolve_output_schemas(
2628 &self.owners.output_schema_plans,
2629 &schema_winner_ids,
2630 )
2631 .map_err(runtime_error)?;
2632 let mut cache_entries = Vec::with_capacity(counts.len());
2633 for ((_, relation_index), count) in self
2634 .owners
2635 .output_indices
2636 .iter()
2637 .zip(counts.iter().copied())
2638 {
2639 let relation = private_relation(&self.owners.relations, *relation_index)
2640 .map_err(runtime_error)?;
2641 if count > relation.capacity() {
2642 return Err(runtime_error(format!(
2643 "resident receipt count {count} exceeds output capacity {}",
2644 relation.capacity()
2645 )));
2646 }
2647 cache_entries.push((relation.buffer(), count));
2648 }
2649 self.owners
2650 .provider
2651 .finalize_resident_logical_counts(&cache_entries)
2652 .map_err(runtime_error)?;
2653 for ((name, relation_index), schema) in
2654 self.owners.output_indices.iter().zip(selected_schemas)
2655 {
2656 let relation = self
2657 .owners
2658 .relations
2659 .get_mut(*relation_index)
2660 .and_then(Option::take)
2661 .ok_or_else(|| runtime_error("resident output relation owner missing"))?;
2662 outputs.push(StagedResidentOutput {
2663 name: name.clone(),
2664 buffer: relation
2665 .into_buffer_with_observed_schema(schema)
2666 .map_err(runtime_error)?,
2667 });
2668 }
2669 }
2670
2671 let decode_schema_staging_ns = decode_started
2672 .map(|started| u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX))
2673 .unwrap_or(0);
2674 Ok(ObservedResidentGraphReceipt {
2675 encoded_len,
2676 device_elapsed_ns: self.device_elapsed_ns,
2677 device_scan_invocations,
2678 device_filter_invocations,
2679 semantic_scan_invocations,
2680 semantic_filter_invocations,
2681 iterations,
2682 terminal,
2683 outputs,
2684 source_epoch: self.owners.source_epoch,
2685 relation_registration: self.owners.relation_registration.clone(),
2686 transaction_identity: Arc::clone(&self.owners.transaction_identity),
2687 provider: Arc::clone(&self.owners.provider),
2688 phase_timings: phase_diagnostics.then_some(ResidentFinalObservationPhaseTimings {
2689 receipt_d2h_ns,
2690 decode_schema_staging_ns,
2691 }),
2692 })
2693 }
2694}
2695
2696impl ObservedResidentGraphReceipt {
2697 pub fn encoded_len(&self) -> usize {
2699 self.encoded_len
2700 }
2701
2702 #[doc(hidden)]
2704 pub fn phase_timings(&self) -> Option<ResidentFinalObservationPhaseTimings> {
2705 self.phase_timings
2706 }
2707
2708 #[doc(hidden)]
2710 pub fn relation_registration_count(&self) -> usize {
2711 self.relation_registration.len()
2712 }
2713
2714 pub fn device_elapsed_ns(&self) -> u64 {
2716 self.device_elapsed_ns
2717 }
2718
2719 pub fn device_scan_invocations(&self) -> u64 {
2721 self.device_scan_invocations
2722 }
2723
2724 pub fn device_filter_invocations(&self) -> u64 {
2726 self.device_filter_invocations
2727 }
2728
2729 pub fn semantic_scan_invocations(&self) -> u64 {
2731 self.semantic_scan_invocations
2732 }
2733
2734 pub fn semantic_filter_invocations(&self) -> u64 {
2736 self.semantic_filter_invocations
2737 }
2738
2739 pub fn staged_output_count(&self) -> u64 {
2741 self.outputs.len() as u64
2742 }
2743
2744 pub fn iterations(&self) -> u32 {
2746 self.iterations
2747 }
2748
2749 pub fn commit(
2751 mut self,
2752 executor: &mut Executor,
2753 ) -> std::result::Result<(), ResidentGraphExecutionError> {
2754 self.terminal?;
2755 if !Arc::ptr_eq(&self.transaction_identity, &executor.transaction_identity)
2756 || !Arc::ptr_eq(&self.provider, &executor.provider)
2757 || self.source_epoch != executor.store.mutation_epoch()
2758 {
2759 return Err(ResidentGraphExecutionError::Runtime(
2760 "resident transaction became stale before commit".into(),
2761 ));
2762 }
2763 let mut current_registration = executor
2764 .rel_names
2765 .iter()
2766 .map(|(rel, name)| (*rel, name.clone()))
2767 .collect::<Vec<_>>();
2768 current_registration.sort_by_key(|(rel, name)| (rel.0, name.clone()));
2769 if current_registration != self.relation_registration {
2770 return Err(ResidentGraphExecutionError::Runtime(
2771 "resident relation registration changed before commit".into(),
2772 ));
2773 }
2774 let additional = self
2775 .outputs
2776 .iter()
2777 .filter(|output| !executor.store.contains(&output.name))
2778 .count();
2779 executor
2780 .store
2781 .try_reserve_relations(additional)
2782 .map_err(runtime_error)?;
2783 executor.common_subexpression_cache.clear();
2784 for output in self.outputs.drain(..) {
2785 if let Some(&rel) = executor.name_to_rel.get(&output.name) {
2786 executor.join_index_cache.invalidate_rel(rel);
2787 }
2788 executor.store.put_owned(output.name, output.buffer);
2789 }
2790 Ok(())
2791 }
2792}
2793
2794fn runtime_error(error: impl std::fmt::Display) -> ResidentGraphExecutionError {
2795 ResidentGraphExecutionError::Runtime(error.to_string())
2796}
2797
2798impl ResidentBuild<'_> {
2799 fn source_reference(&mut self, name: &str) -> Result<ResidentBufferRef> {
2800 if let Some(index) = self.source_aliases.get(name) {
2801 return Ok(ResidentBufferRef::Private(*index));
2802 }
2803 let (row_count, count_tracked, columns_tracked, schema) = {
2804 let source = self.executor.store.get(name).ok_or_else(|| {
2805 XlogError::Execution(format!(
2806 "resident source {name} disappeared during planning"
2807 ))
2808 })?;
2809 (
2810 resident_source_logical_count(source.cached_row_count())?,
2811 source.num_rows_device().runtime_block().is_some(),
2812 source
2813 .columns()
2814 .iter()
2815 .all(|column| column.runtime_block().is_some()),
2816 source.schema().clone(),
2817 )
2818 };
2819 match resident_source_binding_route(row_count, count_tracked, columns_tracked)? {
2820 ResidentSourceBindingRoute::Direct => {
2821 self.source_names.insert(name.to_owned());
2822 Ok(ResidentBufferRef::Source(name.to_owned()))
2823 }
2824 ResidentSourceBindingRoute::NormalizeEmpty => {
2825 let index = self.allocate_permanent_relation(schema, 0)?;
2826 self.source_aliases.insert(name.to_owned(), index);
2827 Ok(ResidentBufferRef::Private(index))
2828 }
2829 }
2830 }
2831
2832 fn private(&self, index: usize) -> &ResidentLogicalRelation {
2833 &self.relations[index]
2834 }
2835
2836 fn schema(&self, reference: &ResidentBufferRef) -> Result<&Schema> {
2837 match reference {
2838 ResidentBufferRef::Source(name) => self
2839 .executor
2840 .store
2841 .get(name)
2842 .map(CudaBuffer::schema)
2843 .ok_or_else(|| {
2844 XlogError::Execution(format!("missing resident source relation {name}"))
2845 }),
2846 ResidentBufferRef::Private(index) => Ok(&self.private(*index).schema),
2847 }
2848 }
2849
2850 fn allocate_relation(
2851 &mut self,
2852 schema: Schema,
2853 initial_count: u32,
2854 permanent: bool,
2855 ) -> Result<usize> {
2856 if initial_count > 1 {
2857 return Err(XlogError::Execution(
2858 "resident logical initial count must be zero or one".into(),
2859 ));
2860 }
2861 let index = self.relations.len();
2862 self.relations.push(ResidentLogicalRelation {
2863 schema,
2864 initial_count,
2865 permanent,
2866 });
2867 Ok(index)
2868 }
2869
2870 fn allocate_permanent_relation(&mut self, schema: Schema, initial_count: u32) -> Result<usize> {
2871 self.allocate_relation(schema, initial_count, true)
2872 }
2873
2874 fn allocate_scratch_relation(&mut self, schema: Schema) -> Result<usize> {
2875 self.allocate_relation(schema, 0, false)
2876 }
2877
2878 fn allocation_manifest(
2879 &self,
2880 initial_ops: &[ResidentRecordedOp],
2881 phases: &[ResidentCapturePhase],
2882 ) -> Result<ResidentAllocationManifest> {
2883 let mut definitions = vec![None; self.relations.len()];
2884 let mut last_uses = vec![None; self.relations.len()];
2885 let mut ordinal = 0usize;
2886 let mut phase_ranges = Vec::with_capacity(phases.len() + 1);
2887 phase_ranges.push(resident_record_lifetimes(
2888 initial_ops,
2889 &self.relations,
2890 &mut definitions,
2891 &mut last_uses,
2892 &mut ordinal,
2893 )?);
2894 for phase in phases {
2895 let ops = match phase {
2896 ResidentCapturePhase::Segment { ops, .. }
2897 | ResidentCapturePhase::ConditionalWhile { ops, .. } => ops,
2898 };
2899 phase_ranges.push(resident_record_lifetimes(
2900 ops,
2901 &self.relations,
2902 &mut definitions,
2903 &mut last_uses,
2904 &mut ordinal,
2905 )?);
2906 }
2907
2908 let mut slots = Vec::<ResidentPhysicalSlotPlan>::new();
2909 let mut logical_to_slot = vec![
2910 ResidentSlotAssignment {
2911 slot: usize::MAX,
2912 generation: 0,
2913 };
2914 self.relations.len()
2915 ];
2916 for (logical, relation) in self.relations.iter().enumerate() {
2917 if relation.permanent {
2918 let slot = slots.len();
2919 slots.push(ResidentPhysicalSlotPlan {
2920 schema: relation.schema.clone(),
2921 initial_count: relation.initial_count,
2922 permanent: true,
2923 });
2924 logical_to_slot[logical] = ResidentSlotAssignment {
2925 slot,
2926 generation: 0,
2927 };
2928 }
2929 }
2930
2931 let mut scratch_relations = self
2932 .relations
2933 .iter()
2934 .enumerate()
2935 .filter(|(_, relation)| !relation.permanent)
2936 .map(|(logical, relation)| {
2937 let definition = definitions[logical].ok_or_else(|| {
2938 XlogError::Execution(format!(
2939 "resident scratch relation {logical} is never defined"
2940 ))
2941 })?;
2942 let last_use = last_uses[logical].unwrap_or(definition);
2943 if !phase_ranges
2944 .iter()
2945 .any(|(start, end)| *start <= definition && last_use < *end)
2946 {
2947 return Err(XlogError::Execution(format!(
2948 "resident scratch relation {logical} crosses a capture phase boundary"
2949 )));
2950 }
2951 Ok((
2952 logical,
2953 definition,
2954 last_use,
2955 resident_schema_layout(&relation.schema),
2956 ))
2957 })
2958 .collect::<Result<Vec<_>>>()?;
2959 scratch_relations.sort_by_key(|(logical, definition, _, _)| (*definition, *logical));
2960
2961 let mut scratch_slots = Vec::<ResidentScratchSlotState>::new();
2962 for (logical, definition, last_use, layout) in scratch_relations {
2963 let reusable = scratch_slots
2964 .iter_mut()
2965 .filter(|slot| slot.layout == layout && slot.last_use < definition)
2966 .min_by_key(|slot| slot.last_use);
2967 let assignment = if let Some(slot) = reusable {
2968 slot.generation = slot.generation.checked_add(1).ok_or_else(|| {
2969 XlogError::Execution("resident scratch generation overflow".into())
2970 })?;
2971 slot.last_use = last_use;
2972 ResidentSlotAssignment {
2973 slot: slot.slot,
2974 generation: slot.generation,
2975 }
2976 } else {
2977 let slot_index = slots.len();
2978 slots.push(ResidentPhysicalSlotPlan {
2979 schema: self.relations[logical].schema.clone(),
2980 initial_count: 0,
2981 permanent: false,
2982 });
2983 scratch_slots.push(ResidentScratchSlotState {
2984 slot: slot_index,
2985 layout,
2986 last_use,
2987 generation: 0,
2988 });
2989 ResidentSlotAssignment {
2990 slot: slot_index,
2991 generation: 0,
2992 }
2993 };
2994 logical_to_slot[logical] = assignment;
2995 }
2996 if logical_to_slot
2997 .iter()
2998 .any(|assignment| assignment.slot == usize::MAX)
2999 {
3000 return Err(XlogError::Execution(
3001 "resident logical relation has no physical slot assignment".into(),
3002 ));
3003 }
3004 resident_validate_slot_assignments(
3005 &self.relations,
3006 &definitions,
3007 &last_uses,
3008 &slots,
3009 &logical_to_slot,
3010 )?;
3011
3012 let capacity = u64::from(self.capacity);
3013 let mut relation_bytes = 0u64;
3014 let mut max_row_bytes = 1u64;
3015 for slot in &slots {
3016 max_row_bytes = max_row_bytes.max(slot.schema.row_size_bytes() as u64);
3017 resident_checked_add(
3018 &mut relation_bytes,
3019 resident_relation_device_bytes(&slot.schema, capacity)?,
3020 "relation",
3021 )?;
3022 }
3023 let filter_scratch_bytes = if self.filter_workspaces.is_empty() {
3024 0
3025 } else {
3026 resident_filter_scratch_device_bytes(capacity)?
3027 };
3028 let set_candidate_capacity = capacity.checked_mul(2).ok_or_else(|| {
3029 XlogError::Execution("resident set candidate capacity overflow".into())
3030 })?;
3031 let mut fixed_workspace_bytes = 0u64;
3032 for bytes in [
3033 resident_set_workspace_device_bytes(set_candidate_capacity)?,
3034 resident_join_workspace_device_bytes(capacity)?,
3035 resident_control_device_bytes(),
3036 resident_device_trace_bytes(),
3037 resident_schema_winners_device_bytes(self.heads.len())?,
3038 resident_packed_receipt_with_schema_winners_device_bytes(self.heads.len())?,
3039 ] {
3040 resident_checked_add(&mut fixed_workspace_bytes, bytes, "fixed workspace")?;
3041 }
3042 let mut required_bytes = 0u64;
3043 for bytes in [relation_bytes, filter_scratch_bytes, fixed_workspace_bytes] {
3044 resident_checked_add(&mut required_bytes, bytes, "manifest")?;
3045 }
3046
3047 let permanent_slots = u32::try_from(slots.iter().filter(|slot| slot.permanent).count())
3048 .map_err(|_| {
3049 XlogError::Execution("resident permanent slot count exceeds u32".into())
3050 })?;
3051 Ok(ResidentAllocationManifest {
3052 slots,
3053 logical_to_slot,
3054 required_bytes,
3055 relation_bytes,
3056 filter_scratch_bytes,
3057 schedule_metadata_bytes: 0,
3058 fixed_workspace_bytes,
3059 logical_relation_values: self.relations.len(),
3060 permanent_slots,
3061 scratch_slots: u32::try_from(scratch_slots.len()).map_err(|_| {
3062 XlogError::Execution("resident scratch slot count exceeds u32".into())
3063 })?,
3064 filter_scratch_allocations: u32::from(!self.filter_workspaces.is_empty()),
3065 max_row_bytes,
3066 })
3067 }
3068
3069 fn materialize(
3070 &self,
3071 manifest: &ResidentAllocationManifest,
3072 reservation: &mut GpuMemoryReservation,
3073 mut diagnostics: Option<&mut ResidentPrepareDiagnostics>,
3074 ) -> Result<ResidentPhysicalBuild> {
3075 let capacity = u64::from(self.capacity);
3076 let mut relations = Vec::with_capacity(manifest.slots.len());
3077 for slot in &manifest.slots {
3078 let allocation_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3079 let allocation_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3080 let mut relation = self
3081 .executor
3082 .provider
3083 .prepare_resident_relation_in_reservation(
3084 slot.schema.clone(),
3085 capacity,
3086 reservation,
3087 )?;
3088 if let Some(diagnostics) = diagnostics.as_deref_mut() {
3089 let allocation_ns = resident_prepare_elapsed_ns(
3090 allocation_started.expect("diagnostic timer exists when enabled"),
3091 );
3092 let allocation_bytes = reservation.used_bytes().saturating_sub(
3093 allocation_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3094 );
3095 diagnostics.relation_slot_allocation_ns = diagnostics
3096 .relation_slot_allocation_ns
3097 .saturating_add(allocation_ns);
3098 diagnostics.relation_slot_allocation_ns_max = diagnostics
3099 .relation_slot_allocation_ns_max
3100 .max(allocation_ns);
3101 diagnostics.relation_slot_allocation_bytes = diagnostics
3102 .relation_slot_allocation_bytes
3103 .saturating_add(allocation_bytes);
3104 diagnostics.relation_device_allocation_calls =
3105 diagnostics.relation_device_allocation_calls.saturating_add(
3106 u64::try_from(slot.schema.arity())
3107 .unwrap_or(u64::MAX)
3108 .saturating_add(1),
3109 );
3110 }
3111 let count_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3112 self.executor
3113 .provider
3114 .initialize_resident_relation_count(&mut relation, slot.initial_count)?;
3115 if let Some(diagnostics) = diagnostics.as_deref_mut() {
3116 let count_ns = resident_prepare_elapsed_ns(
3117 count_started.expect("diagnostic timer exists when enabled"),
3118 );
3119 diagnostics.count_initialization_ns =
3120 diagnostics.count_initialization_ns.saturating_add(count_ns);
3121 diagnostics.count_initialization_ns_max =
3122 diagnostics.count_initialization_ns_max.max(count_ns);
3123 diagnostics.count_memset_calls = diagnostics.count_memset_calls.saturating_add(1);
3124 }
3125 relations.push(Some(relation));
3126 }
3127 let filter_scratch = if self.filter_workspaces.is_empty() {
3128 None
3129 } else {
3130 let bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3131 let started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3132 let scratch = self
3133 .executor
3134 .provider
3135 .prepare_resident_filter_scratch_in_reservation(capacity, reservation)?;
3136 if let Some(diagnostics) = diagnostics.as_deref_mut() {
3137 let elapsed_ns = resident_prepare_elapsed_ns(
3138 started.expect("diagnostic timer exists when enabled"),
3139 );
3140 let reserved_bytes = reservation.used_bytes().saturating_sub(
3141 bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3142 );
3143 diagnostics.workspace_provider_calls =
3144 diagnostics.workspace_provider_calls.saturating_add(1);
3145 diagnostics.filter_scratch_allocation_ns = elapsed_ns;
3146 diagnostics.filter_scratch_allocation_bytes = reserved_bytes;
3147 }
3148 Some(scratch)
3149 };
3150 let set_candidate_capacity = capacity.checked_mul(2).ok_or_else(|| {
3151 XlogError::Execution("resident set candidate capacity overflow".into())
3152 })?;
3153 let set_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3154 let set_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3155 let set_workspace = self
3156 .executor
3157 .provider
3158 .prepare_resident_set_workspace_in_reservation(set_candidate_capacity, reservation)?;
3159 if let Some(diagnostics) = diagnostics.as_deref_mut() {
3160 let set_ns = resident_prepare_elapsed_ns(
3161 set_started.expect("diagnostic timer exists when enabled"),
3162 );
3163 let set_reserved_bytes = reservation.used_bytes().saturating_sub(
3164 set_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3165 );
3166 diagnostics.workspace_provider_calls =
3167 diagnostics.workspace_provider_calls.saturating_add(1);
3168 diagnostics.set_workspace_allocation_ns = set_ns;
3169 diagnostics.set_workspace_allocation_bytes = set_reserved_bytes;
3170 }
3171 let join_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3172 let join_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3173 let join_workspace = self
3174 .executor
3175 .provider
3176 .prepare_resident_join_workspace_in_reservation(capacity, reservation)?;
3177 if let Some(diagnostics) = diagnostics.as_deref_mut() {
3178 let join_ns = resident_prepare_elapsed_ns(
3179 join_started.expect("diagnostic timer exists when enabled"),
3180 );
3181 let join_reserved_bytes = reservation.used_bytes().saturating_sub(
3182 join_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3183 );
3184 diagnostics.workspace_provider_calls =
3185 diagnostics.workspace_provider_calls.saturating_add(1);
3186 diagnostics.join_workspace_allocation_ns = join_ns;
3187 diagnostics.join_workspace_allocation_bytes = join_reserved_bytes;
3188 }
3189 let control_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
3190 let control_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
3191 let control = self
3192 .executor
3193 .provider
3194 .prepare_resident_convergence_control_in_reservation(reservation)?;
3195 if let Some(diagnostics) = diagnostics.as_deref_mut() {
3196 let control_ns = resident_prepare_elapsed_ns(
3197 control_started.expect("diagnostic timer exists when enabled"),
3198 );
3199 let control_reserved_bytes = reservation.used_bytes().saturating_sub(
3200 control_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
3201 );
3202 diagnostics.workspace_provider_calls =
3203 diagnostics.workspace_provider_calls.saturating_add(1);
3204 diagnostics.control_allocation_ns = control_ns;
3205 diagnostics.control_allocation_bytes = control_reserved_bytes;
3206 }
3207 Ok(ResidentPhysicalBuild {
3208 relations,
3209 filter_scratch,
3210 set_workspace,
3211 join_workspace,
3212 control,
3213 })
3214 }
3215
3216 fn next_op_id(&mut self) -> Result<u32> {
3217 let op_id = self.next_op_id;
3218 self.next_op_id = self
3219 .next_op_id
3220 .checked_add(1)
3221 .ok_or_else(|| XlogError::Execution("resident physical op id overflow".into()))?;
3222 Ok(op_id)
3223 }
3224
3225 fn push_physical_op(
3226 &mut self,
3227 ops: &mut Vec<ResidentRecordedOp>,
3228 op: ResidentRecordedOp,
3229 op_id: u32,
3230 ) {
3231 ops.push(op);
3232 if !self.injection_recorded
3233 && self
3234 .injection
3235 .as_ref()
3236 .is_some_and(|injection| injection.after_op == op_id)
3237 {
3238 let status =
3239 terminal_status_for_injection(&self.injection.as_ref().expect("checked").status);
3240 ops.push(ResidentRecordedOp::TestStatus(status));
3241 self.injection_recorded = true;
3242 }
3243 }
3244
3245 fn scan_reference(
3246 &mut self,
3247 rel: RelId,
3248 override_scan: Option<(RelId, usize, usize)>,
3249 occurrences: &mut HashMap<RelId, usize>,
3250 ) -> Result<ResidentBufferRef> {
3251 let occurrence = occurrences.entry(rel).or_insert(0);
3252 let current = *occurrence;
3253 *occurrence += 1;
3254 if let Some((target, target_occurrence, delta)) = override_scan {
3255 if rel == target && current == target_occurrence {
3256 return Ok(ResidentBufferRef::Private(delta));
3257 }
3258 }
3259 let name = self
3260 .executor
3261 .rel_names
3262 .get(&rel)
3263 .ok_or_else(|| XlogError::Execution(format!("unknown resident relation id {rel:?}")))?
3264 .clone();
3265 if let Some(index) = self.heads.get(&name) {
3266 Ok(ResidentBufferRef::Private(*index))
3267 } else {
3268 self.source_reference(&name)
3269 }
3270 }
3271
3272 fn plan_node(
3273 &mut self,
3274 node: &RirNode,
3275 override_scan: Option<(RelId, usize, usize)>,
3276 occurrences: &mut HashMap<RelId, usize>,
3277 ops: &mut Vec<ResidentRecordedOp>,
3278 ) -> Result<ResidentBufferRef> {
3279 match node {
3280 RirNode::Unit => {
3281 let op_id = self.next_op_id()?;
3282 let (reference, op) = resident_new_phase_unit(&mut self.relations, op_id)?;
3283 self.push_physical_op(ops, op, op_id);
3284 Ok(reference)
3285 }
3286 RirNode::Scan { rel } => {
3287 let reference = self.scan_reference(*rel, override_scan, occurrences)?;
3288 let op_id = self.next_op_id()?;
3289 Ok(resident_record_scan_leaf(
3290 reference,
3291 op_id,
3292 resident_semantic_trace_guard(override_scan),
3293 ops,
3294 |ops, op, op_id| self.push_physical_op(ops, op, op_id),
3295 ))
3296 }
3297 RirNode::Filter { input, predicate } => {
3298 let input_ref = self.plan_node(input, override_scan, occurrences, ops)?;
3299 let input_schema = self.schema(&input_ref)?.clone();
3300 let compact_comparisons =
3301 resident_compact_filter_descriptors(predicate, &input_schema)?;
3302 let workspace = ResidentFilterPlan {
3303 compact_comparisons,
3304 };
3305 let workspace_index = self.filter_workspaces.len();
3306 self.filter_workspaces.push(workspace);
3307 let schema = self
3308 .certificate
3309 .node_schema(node)
3310 .ok_or_else(|| XlogError::Execution("resident filter schema missing".into()))?;
3311 let output = self.allocate_scratch_relation(schema)?;
3312 let op_id = self.next_op_id()?;
3313 self.push_physical_op(
3314 ops,
3315 ResidentRecordedOp::Filter {
3316 input: input_ref,
3317 output,
3318 workspace: workspace_index,
3319 op_id,
3320 },
3321 op_id,
3322 );
3323 ops.push(ResidentRecordedOp::TraceDelta {
3324 scan_delta: 0,
3325 filter_delta: 1,
3326 semantic_guard: resident_semantic_trace_guard(override_scan),
3327 });
3328 Ok(ResidentBufferRef::Private(output))
3329 }
3330 RirNode::Project { input, columns } => {
3331 let input_ref = self.plan_node(input, override_scan, occurrences, ops)?;
3332 let input_schema = self.schema(&input_ref)?.clone();
3333 let certified_schema = self.certificate.node_schema(node).ok_or_else(|| {
3334 XlogError::Execution("resident project schema missing".into())
3335 })?;
3336 let schema = self.executor.project_schema(&input_schema, columns)?;
3337 if schema.arity() != certified_schema.arity()
3338 || (0..schema.arity()).any(|column| {
3339 schema.column_type(column) != certified_schema.column_type(column)
3340 })
3341 {
3342 return Err(XlogError::Execution(
3343 "resident project physical schema differs from certificate".into(),
3344 ));
3345 }
3346 let compact_expressions =
3347 resident_compact_project_descriptors(columns, &input_schema, &schema)?;
3348 let workspace = ResidentProjectPlan {
3349 compact_expressions,
3350 };
3351 let workspace_index = self.project_workspaces.len();
3352 self.project_workspaces.push(workspace);
3353 let output = self.allocate_scratch_relation(schema)?;
3354 let op_id = self.next_op_id()?;
3355 self.push_physical_op(
3356 ops,
3357 ResidentRecordedOp::Project {
3358 input: input_ref,
3359 output,
3360 workspace: workspace_index,
3361 op_id,
3362 },
3363 op_id,
3364 );
3365 Ok(ResidentBufferRef::Private(output))
3366 }
3367 RirNode::Join {
3368 left,
3369 right,
3370 left_keys,
3371 right_keys,
3372 join_type,
3373 } => {
3374 let left_ref = self.plan_node(left, override_scan, occurrences, ops)?;
3375 let right_ref = self.plan_node(right, override_scan, occurrences, ops)?;
3376 let certified_schema = self
3377 .certificate
3378 .node_schema(node)
3379 .ok_or_else(|| XlogError::Execution("resident join schema missing".into()))?;
3380 let kind = match join_type {
3381 JoinType::Inner => ResidentJoinKind::Inner,
3382 JoinType::Semi => ResidentJoinKind::Semi,
3383 _ => return Err(XlogError::Execution("uncertified resident join".into())),
3384 };
3385 let left_schema = self.schema(&left_ref)?.clone();
3390 let right_schema = self.schema(&right_ref)?.clone();
3391 let schema = match kind {
3392 ResidentJoinKind::Semi => left_schema,
3393 ResidentJoinKind::Inner => {
3394 let mut columns = left_schema.columns;
3395 columns.extend(right_schema.columns);
3396 Schema::new(columns)
3397 }
3398 };
3399 if schema.arity() != certified_schema.arity()
3400 || (0..schema.arity()).any(|column| {
3401 schema.column_type(column) != certified_schema.column_type(column)
3402 })
3403 {
3404 return Err(XlogError::Execution(
3405 "resident join physical schema differs from certificate".into(),
3406 ));
3407 }
3408 let output = self.allocate_scratch_relation(schema).map_err(|error| {
3409 XlogError::Execution(format!(
3410 "resident join intermediate arity {} at {node:?} could not be allocated: {error}",
3411 certified_schema.arity()
3412 ))
3413 })?;
3414 let op_id = self.next_op_id()?;
3415 self.push_physical_op(
3416 ops,
3417 ResidentRecordedOp::Join {
3418 kind,
3419 left: left_ref,
3420 left_key: left_keys[0],
3421 right: right_ref,
3422 right_key: right_keys[0],
3423 output,
3424 op_id,
3425 },
3426 op_id,
3427 );
3428 Ok(ResidentBufferRef::Private(output))
3429 }
3430 RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
3431 self.plan_node(fallback, override_scan, occurrences, ops)
3432 }
3433 RirNode::Distinct { input, .. } => {
3434 let input_ref = self.plan_node(input, override_scan, occurrences, ops)?;
3435 self.plan_dedup(input_ref, node, ops)
3436 }
3437 RirNode::Diff { left, right } => {
3438 let left_ref = self.plan_node(left, override_scan, occurrences, ops)?;
3439 let right_ref = self.plan_node(right, override_scan, occurrences, ops)?;
3440 self.plan_diff(left_ref, right_ref, node, ops)
3441 }
3442 RirNode::Union { inputs } => {
3443 let fold_mode = resident_union_fold_mode(inputs.len())?;
3444 let mut iter = inputs.iter();
3445 let first = iter
3446 .next()
3447 .expect("union fold mode rejected the empty case");
3448 let mut result = self.plan_node(first, override_scan, occurrences, ops)?;
3449 if fold_mode == ResidentUnionFoldMode::SelfUnion {
3450 return self.plan_union(result.clone(), result, node, ops);
3451 }
3452 for input in iter {
3453 let right = self.plan_node(input, override_scan, occurrences, ops)?;
3454 result = self.plan_union(result, right, node, ops)?;
3455 }
3456 Ok(result)
3457 }
3458 RirNode::Fixpoint { .. }
3459 | RirNode::GroupBy { .. }
3460 | RirNode::TensorMaskedJoin { .. } => Err(XlogError::Execution(
3461 "uncertified node reached resident planner".into(),
3462 )),
3463 }
3464 }
3465
3466 fn plan_dedup(
3467 &mut self,
3468 input: ResidentBufferRef,
3469 node: &RirNode,
3470 ops: &mut Vec<ResidentRecordedOp>,
3471 ) -> Result<ResidentBufferRef> {
3472 let schema = self
3473 .certificate
3474 .node_schema(node)
3475 .ok_or_else(|| XlogError::Execution("resident dedup schema missing".into()))?;
3476 let output = self.allocate_scratch_relation(schema)?;
3477 let op_id = self.next_op_id()?;
3479 self.push_physical_op(
3480 ops,
3481 ResidentRecordedOp::Union {
3482 left: input.clone(),
3483 right: input,
3484 output,
3485 op_id,
3486 },
3487 op_id,
3488 );
3489 Ok(ResidentBufferRef::Private(output))
3490 }
3491
3492 fn plan_union(
3493 &mut self,
3494 left: ResidentBufferRef,
3495 right: ResidentBufferRef,
3496 schema_node: &RirNode,
3497 ops: &mut Vec<ResidentRecordedOp>,
3498 ) -> Result<ResidentBufferRef> {
3499 let schema = self
3500 .certificate
3501 .node_schema(schema_node)
3502 .ok_or_else(|| XlogError::Execution("resident union schema missing".into()))?;
3503 self.plan_union_schema(left, right, schema, ops)
3504 }
3505
3506 fn plan_union_schema(
3507 &mut self,
3508 left: ResidentBufferRef,
3509 right: ResidentBufferRef,
3510 schema: Schema,
3511 ops: &mut Vec<ResidentRecordedOp>,
3512 ) -> Result<ResidentBufferRef> {
3513 let output = self.allocate_scratch_relation(schema)?;
3514 let op_id = self.next_op_id()?;
3515 self.push_physical_op(
3516 ops,
3517 ResidentRecordedOp::Union {
3518 left,
3519 right,
3520 output,
3521 op_id,
3522 },
3523 op_id,
3524 );
3525 Ok(ResidentBufferRef::Private(output))
3526 }
3527
3528 fn plan_diff(
3529 &mut self,
3530 left: ResidentBufferRef,
3531 right: ResidentBufferRef,
3532 schema_node: &RirNode,
3533 ops: &mut Vec<ResidentRecordedOp>,
3534 ) -> Result<ResidentBufferRef> {
3535 let schema = self
3536 .certificate
3537 .node_schema(schema_node)
3538 .ok_or_else(|| XlogError::Execution("resident diff schema missing".into()))?;
3539 self.plan_diff_schema(left, right, schema, ops)
3540 }
3541
3542 fn plan_diff_schema(
3543 &mut self,
3544 left: ResidentBufferRef,
3545 right: ResidentBufferRef,
3546 schema: Schema,
3547 ops: &mut Vec<ResidentRecordedOp>,
3548 ) -> Result<ResidentBufferRef> {
3549 let output = self.allocate_scratch_relation(schema)?;
3550 let op_id = self.next_op_id()?;
3551 self.push_physical_op(
3552 ops,
3553 ResidentRecordedOp::Diff {
3554 left,
3555 right,
3556 output,
3557 op_id,
3558 },
3559 op_id,
3560 );
3561 Ok(ResidentBufferRef::Private(output))
3562 }
3563
3564 fn plan_copy(
3565 &mut self,
3566 input: ResidentBufferRef,
3567 output: usize,
3568 schema: &Schema,
3569 ops: &mut Vec<ResidentRecordedOp>,
3570 ) -> Result<()> {
3571 let compact_expressions = (0..schema.arity())
3572 .map(|column| {
3573 let scalar = schema.column_type(column).ok_or_else(|| {
3574 XlogError::Execution("resident identity project column is missing".into())
3575 })?;
3576 Ok(ResidentProjectExpressionDescriptor::column(
3577 u32::try_from(column).map_err(|_| {
3578 XlogError::Execution("resident project column exceeds u32".into())
3579 })?,
3580 u32::try_from(scalar.size_bytes()).map_err(|_| {
3581 XlogError::Execution("resident project scalar width exceeds u32".into())
3582 })?,
3583 ))
3584 })
3585 .collect::<Result<Vec<_>>>()?;
3586 let workspace = ResidentProjectPlan {
3587 compact_expressions,
3588 };
3589 let workspace_index = self.project_workspaces.len();
3590 self.project_workspaces.push(workspace);
3591 let op_id = self.next_op_id()?;
3592 self.push_physical_op(
3593 ops,
3594 ResidentRecordedOp::Project {
3595 input,
3596 output,
3597 workspace: workspace_index,
3598 op_id,
3599 },
3600 op_id,
3601 );
3602 Ok(())
3603 }
3604
3605 fn normalize_to_schema(
3606 &mut self,
3607 input: ResidentBufferRef,
3608 schema: &Schema,
3609 ops: &mut Vec<ResidentRecordedOp>,
3610 ) -> Result<ResidentBufferRef> {
3611 let input_schema = self.schema(&input)?;
3612 if input_schema == schema {
3613 return Ok(input);
3614 }
3615 if input_schema.arity() != schema.arity()
3616 || input_schema
3617 .columns
3618 .iter()
3619 .zip(&schema.columns)
3620 .any(|(left, right)| left.1 != right.1)
3621 {
3622 return Err(XlogError::Execution(format!(
3623 "resident rule output schema {:?} cannot be normalized to head schema {:?}",
3624 input_schema, schema
3625 )));
3626 }
3627 let output = self.allocate_scratch_relation(schema.clone())?;
3628 self.plan_copy(input, output, schema, ops)?;
3629 Ok(ResidentBufferRef::Private(output))
3630 }
3631
3632 fn merge_phase_contribution(
3633 &mut self,
3634 current: Option<ResidentBufferRef>,
3635 contribution: ResidentBufferRef,
3636 schema: &Schema,
3637 ops: &mut Vec<ResidentRecordedOp>,
3638 ) -> Result<ResidentBufferRef> {
3639 let contribution = self.normalize_to_schema(contribution, schema, ops)?;
3640 resident_phase_merge(current, contribution, |step| match step {
3641 ResidentPhaseMergeStep::Deduplicate(input) => {
3642 self.plan_union_schema(input.clone(), input, schema.clone(), ops)
3643 }
3644 ResidentPhaseMergeStep::Union(left, right) => {
3645 self.plan_union_schema(left, right, schema.clone(), ops)
3646 }
3647 })
3648 }
3649}
3650
3651fn private_relation(
3652 relations: &[Option<ResidentRelation>],
3653 index: usize,
3654) -> Result<&ResidentRelation> {
3655 relations
3656 .get(index)
3657 .and_then(Option::as_ref)
3658 .ok_or_else(|| XlogError::Execution(format!("resident relation slot {index} is missing")))
3659}
3660
3661fn collect_recorded_scans(node: &RirNode, output: &mut Vec<RelId>) {
3662 match node {
3663 RirNode::Unit | RirNode::TensorMaskedJoin { .. } => {}
3664 RirNode::Scan { rel } => output.push(*rel),
3665 RirNode::Filter { input, .. }
3666 | RirNode::Project { input, .. }
3667 | RirNode::GroupBy { input, .. }
3668 | RirNode::Distinct { input, .. } => collect_recorded_scans(input, output),
3669 RirNode::Join { left, right, .. } | RirNode::Diff { left, right } => {
3670 collect_recorded_scans(left, output);
3671 collect_recorded_scans(right, output);
3672 }
3673 RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
3674 collect_recorded_scans(fallback, output);
3675 }
3676 RirNode::Union { inputs } => {
3677 for input in inputs {
3678 collect_recorded_scans(input, output);
3679 }
3680 }
3681 RirNode::Fixpoint {
3682 base, recursive, ..
3683 } => {
3684 collect_recorded_scans(base, output);
3685 collect_recorded_scans(recursive, output);
3686 }
3687 }
3688}
3689
3690fn resident_decline_error(reason: ResidentGraphDeclineReason) -> ResidentGraphExecutionError {
3691 ResidentGraphExecutionError::Declined(reason)
3692}
3693
3694fn conditional_graph_error(error: CudaConditionalGraphUnavailable) -> ResidentGraphExecutionError {
3695 if error.is_unsupported() {
3696 resident_decline_error(ResidentGraphDeclineReason::ConditionalGraphUnavailable {
3697 detail: error.decline_detail(),
3698 })
3699 } else {
3700 runtime_error(error.decline_detail())
3701 }
3702}
3703
3704fn read_receipt_u32(bytes: &[u8], offset: usize) -> Option<u32> {
3705 bytes
3706 .get(offset..offset.checked_add(4)?)?
3707 .try_into()
3708 .ok()
3709 .map(u32::from_le_bytes)
3710}
3711
3712fn read_receipt_u64(bytes: &[u8], offset: usize) -> Option<u64> {
3713 bytes
3714 .get(offset..offset.checked_add(8)?)?
3715 .try_into()
3716 .ok()
3717 .map(u64::from_le_bytes)
3718}
3719
3720fn resident_compact_scalar(value: &ConstValue) -> Result<(ScalarType, u64)> {
3721 match value {
3722 ConstValue::Symbol(value) => Ok((ScalarType::Symbol, u64::from(symbol::intern(value)))),
3723 ConstValue::U32(value) => Ok((ScalarType::U32, u64::from(*value))),
3724 ConstValue::U64(value) => Ok((ScalarType::U64, *value)),
3725 _ => Err(XlogError::Execution(
3726 "uncertified resident constant type".into(),
3727 )),
3728 }
3729}
3730
3731fn resident_compact_filter_descriptors(
3732 expression: &Expr,
3733 input_schema: &Schema,
3734) -> Result<Vec<ResidentFilterComparisonDescriptor>> {
3735 fn operand(expression: &Expr, input_schema: &Schema) -> Result<(u32, u32, ScalarType, u64)> {
3736 match expression {
3737 Expr::Column(column) => {
3738 let scalar = input_schema.column_type(*column).ok_or_else(|| {
3739 XlogError::Execution("resident filter column is out of range".into())
3740 })?;
3741 Ok((
3742 0,
3743 u32::try_from(*column).map_err(|_| {
3744 XlogError::Execution("resident filter column exceeds u32".into())
3745 })?,
3746 scalar,
3747 0,
3748 ))
3749 }
3750 Expr::Const(value) => {
3751 let (scalar, bits) = resident_compact_scalar(value)?;
3752 Ok((1, 0, scalar, bits))
3753 }
3754 _ => Err(XlogError::Execution(
3755 "uncertified resident filter operand".into(),
3756 )),
3757 }
3758 }
3759
3760 fn append(
3761 expression: &Expr,
3762 input_schema: &Schema,
3763 output: &mut Vec<ResidentFilterComparisonDescriptor>,
3764 ) -> Result<()> {
3765 match expression {
3766 Expr::And(parts) => {
3767 for part in parts {
3768 append(part, input_schema, output)?;
3769 }
3770 Ok(())
3771 }
3772 Expr::Compare { left, op, right } => {
3773 let (left_kind, left_column, left_type, left_constant) =
3774 operand(left, input_schema)?;
3775 let (right_kind, right_column, right_type, right_constant) =
3776 operand(right, input_schema)?;
3777 if left_type != right_type {
3778 return Err(XlogError::Execution(
3779 "resident filter operand scalar types differ".into(),
3780 ));
3781 }
3782 let width = u32::try_from(left_type.size_bytes()).map_err(|_| {
3783 XlogError::Execution("resident filter scalar width exceeds u32".into())
3784 })?;
3785 if !matches!(width, 4 | 8) {
3786 return Err(XlogError::Execution(
3787 "resident filter scalar width is unsupported".into(),
3788 ));
3789 }
3790 output.push(ResidentFilterComparisonDescriptor {
3791 left_kind,
3792 left_column,
3793 right_kind,
3794 right_column,
3795 op: u32::from(resident_compare_op(*op) as u8),
3796 width,
3797 reserved_zero: 0,
3798 reserved_one: 0,
3799 left_constant,
3800 right_constant,
3801 });
3802 Ok(())
3803 }
3804 _ => Err(XlogError::Execution(
3805 "uncertified resident filter expression".into(),
3806 )),
3807 }
3808 }
3809
3810 let mut output = Vec::new();
3811 append(expression, input_schema, &mut output)?;
3812 Ok(output)
3813}
3814
3815fn resident_compact_project_descriptors(
3816 columns: &[ProjectExpr],
3817 input_schema: &Schema,
3818 output_schema: &Schema,
3819) -> Result<Vec<ResidentProjectExpressionDescriptor>> {
3820 if columns.len() != output_schema.arity() {
3821 return Err(XlogError::Execution(
3822 "resident project expression count differs from output arity".into(),
3823 ));
3824 }
3825 columns
3826 .iter()
3827 .enumerate()
3828 .map(|(output_column, expression)| {
3829 let output_type = output_schema.column_type(output_column).ok_or_else(|| {
3830 XlogError::Execution("resident project output column is out of range".into())
3831 })?;
3832 let width = u32::try_from(output_type.size_bytes()).map_err(|_| {
3833 XlogError::Execution("resident project scalar width exceeds u32".into())
3834 })?;
3835 if !matches!(width, 4 | 8) {
3836 return Err(XlogError::Execution(
3837 "resident project scalar width is unsupported".into(),
3838 ));
3839 }
3840 match expression {
3841 ProjectExpr::Column(input_column) => {
3842 if input_schema.column_type(*input_column) != Some(output_type) {
3843 return Err(XlogError::Execution(
3844 "resident project column scalar type differs from output".into(),
3845 ));
3846 }
3847 Ok(ResidentProjectExpressionDescriptor::column(
3848 u32::try_from(*input_column).map_err(|_| {
3849 XlogError::Execution("resident project column exceeds u32".into())
3850 })?,
3851 width,
3852 ))
3853 }
3854 ProjectExpr::Computed(Expr::Const(value), declared_type) => {
3855 let (constant_type, bits) = resident_compact_scalar(value)?;
3856 if constant_type != *declared_type || constant_type != output_type {
3857 return Err(XlogError::Execution(
3858 "resident project constant scalar type differs from output".into(),
3859 ));
3860 }
3861 Ok(ResidentProjectExpressionDescriptor::constant(width, bits))
3862 }
3863 _ => Err(XlogError::Execution(
3864 "uncertified resident projection expression".into(),
3865 )),
3866 }
3867 })
3868 .collect()
3869}
3870
3871fn resident_compare_op(op: RirCompareOp) -> CudaCompareOp {
3872 match op {
3873 RirCompareOp::Eq => CudaCompareOp::Eq,
3874 RirCompareOp::Ne => CudaCompareOp::Ne,
3875 RirCompareOp::Lt => CudaCompareOp::Lt,
3876 RirCompareOp::Le => CudaCompareOp::Le,
3877 RirCompareOp::Gt => CudaCompareOp::Gt,
3878 RirCompareOp::Ge => CudaCompareOp::Ge,
3879 }
3880}
3881
3882fn terminal_status_for_injection(status: &ResidentGraphDeviceStatus) -> ResidentTerminalStatus {
3883 match status {
3884 ResidentGraphDeviceStatus::Success { iterations } => ResidentTerminalStatus {
3885 code: ResidentTerminalCode::Success as u32,
3886 iterations: *iterations,
3887 ..ResidentTerminalStatus::default()
3888 },
3889 ResidentGraphDeviceStatus::IterationLimit { limit, completed } => ResidentTerminalStatus {
3890 code: ResidentTerminalCode::IterationLimit as u32,
3891 iterations: *completed,
3892 limit: *limit,
3893 ..ResidentTerminalStatus::default()
3894 },
3895 ResidentGraphDeviceStatus::CapacityOverflow {
3896 op_id,
3897 required,
3898 capacity,
3899 } => ResidentTerminalStatus {
3900 code: ResidentTerminalCode::CapacityOverflow as u32,
3901 op_id: *op_id,
3902 resource_code: ResidentResourceCode::OutputRows as u32,
3903 required: *required,
3904 capacity: *capacity,
3905 ..ResidentTerminalStatus::default()
3906 },
3907 ResidentGraphDeviceStatus::ResourceExhausted {
3908 op_id,
3909 required,
3910 capacity,
3911 ..
3912 } => ResidentTerminalStatus {
3913 code: ResidentTerminalCode::ResourceExhausted as u32,
3914 op_id: *op_id,
3915 resource_code: ResidentResourceCode::SetHashSlots as u32,
3916 required: *required,
3917 capacity: *capacity,
3918 ..ResidentTerminalStatus::default()
3919 },
3920 }
3921}
3922
3923fn validate_compact_resident_node_envelope(
3924 node: &RirNode,
3925 schema_for: &impl Fn(&RirNode) -> std::result::Result<Schema, ResidentGraphDeclineReason>,
3926) -> std::result::Result<(), ResidentGraphDeclineReason> {
3927 match node {
3928 RirNode::Unit | RirNode::Scan { .. } => Ok(()),
3929 RirNode::Filter { input, .. } | RirNode::Project { input, .. } => {
3930 validate_compact_resident_node_envelope(input, schema_for)
3931 }
3932 RirNode::Distinct { input, key_cols } => {
3933 validate_compact_resident_node_envelope(input, schema_for)?;
3934 let arity = schema_for(input)?.arity();
3935 if key_cols.len() != arity
3936 || key_cols
3937 .iter()
3938 .copied()
3939 .enumerate()
3940 .any(|(column, key)| column != key)
3941 {
3942 return Err(resident_workspace_decline(format!(
3943 "compact resident Distinct requires canonical full-row key columns 0..{arity}"
3944 )));
3945 }
3946 Ok(())
3947 }
3948 RirNode::Union { inputs } => {
3949 if inputs.is_empty() {
3950 return Err(resident_workspace_decline(
3951 "compact resident Union requires at least one input",
3952 ));
3953 }
3954 for input in inputs {
3955 validate_compact_resident_node_envelope(input, schema_for)?;
3956 }
3957 Ok(())
3958 }
3959 RirNode::Diff { left, right } => {
3960 validate_compact_resident_node_envelope(left, schema_for)?;
3961 validate_compact_resident_node_envelope(right, schema_for)
3962 }
3963 RirNode::Join {
3964 left,
3965 right,
3966 left_keys,
3967 right_keys,
3968 join_type,
3969 } => {
3970 if !matches!(join_type, JoinType::Inner | JoinType::Semi) {
3971 return Err(resident_workspace_decline(format!(
3972 "compact resident Join does not support {join_type:?}"
3973 )));
3974 }
3975 if left_keys.len() != 1 || right_keys.len() != 1 {
3976 return Err(resident_workspace_decline(
3977 "compact resident Join requires exactly one key per input",
3978 ));
3979 }
3980 validate_compact_resident_node_envelope(left, schema_for)?;
3981 validate_compact_resident_node_envelope(right, schema_for)?;
3982 let left_schema = schema_for(left)?;
3983 let right_schema = schema_for(right)?;
3984 let left_key = left_keys[0];
3985 let right_key = right_keys[0];
3986 if left_key >= left_schema.arity() || right_key >= right_schema.arity() {
3987 return Err(resident_workspace_decline(
3988 "compact resident Join key is outside its input schema",
3989 ));
3990 }
3991 let left_key_type = left_schema
3992 .column_type(left_key)
3993 .expect("left key bounds checked");
3994 let right_key_type = right_schema
3995 .column_type(right_key)
3996 .expect("right key bounds checked");
3997 if left_key_type != right_key_type
3998 || !matches!(
3999 left_key_type,
4000 ScalarType::U32 | ScalarType::U64 | ScalarType::Symbol
4001 )
4002 {
4003 return Err(resident_workspace_decline(
4004 "compact resident Join requires matching U32, U64, or Symbol key types",
4005 ));
4006 }
4007 if left_key_type.size_bytes() != right_key_type.size_bytes() {
4008 return Err(resident_workspace_decline(
4009 "compact resident Join key widths differ",
4010 ));
4011 }
4012 let output_arity = match join_type {
4013 JoinType::Inner => left_schema
4014 .arity()
4015 .checked_add(right_schema.arity())
4016 .ok_or_else(|| {
4017 resident_workspace_decline("compact resident Join arity overflow")
4018 })?,
4019 JoinType::Semi => left_schema.arity(),
4020 _ => unreachable!("join kind checked above"),
4021 };
4022 if output_arity > 17 {
4023 return Err(resident_workspace_decline(
4024 "compact resident Join output arity exceeds 17",
4025 ));
4026 }
4027 Ok(())
4028 }
4029 RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
4030 validate_compact_resident_node_envelope(fallback, schema_for)
4031 }
4032 RirNode::Fixpoint { .. } | RirNode::GroupBy { .. } | RirNode::TensorMaskedJoin { .. } => {
4033 Err(resident_workspace_decline(
4034 "compact resident route contains an unsupported node",
4035 ))
4036 }
4037 }
4038}
4039
4040fn resident_generated_query_heads(
4041 plan: &ExecutionPlan,
4042) -> std::result::Result<BTreeSet<String>, ResidentGraphDeclineReason> {
4043 let mut heads = BTreeSet::new();
4044 let mut rule_positions = HashSet::new();
4045 for (position, provenance) in plan.generated_query_rules.iter().enumerate() {
4046 if provenance.query_index != position {
4047 return Err(resident_workspace_decline(format!(
4048 "generated query provenance position {position} carries query index {}",
4049 provenance.query_index
4050 )));
4051 }
4052 if !rule_positions.insert((provenance.scc_index, provenance.rule_index)) {
4053 return Err(resident_workspace_decline(format!(
4054 "generated query provenance {} reuses compiled rule scc={} rule={}",
4055 provenance.query_index, provenance.scc_index, provenance.rule_index
4056 )));
4057 }
4058 let expected_head = format!("__xlog_query_{}", provenance.query_index);
4059 let rule = plan
4060 .rules_by_scc
4061 .get(provenance.scc_index)
4062 .and_then(|rules| rules.get(provenance.rule_index))
4063 .ok_or_else(|| {
4064 resident_workspace_decline(format!(
4065 "generated query provenance {} references missing compiled rule scc={} rule={}",
4066 provenance.query_index, provenance.scc_index, provenance.rule_index
4067 ))
4068 })?;
4069 if rule.head != expected_head {
4070 return Err(resident_workspace_decline(format!(
4071 "generated query provenance {} expects head {expected_head} but references authored head {}",
4072 provenance.query_index, rule.head
4073 )));
4074 }
4075 let occurrence_count = plan
4076 .rules_by_scc
4077 .iter()
4078 .flatten()
4079 .filter(|candidate| candidate.head == expected_head)
4080 .count();
4081 if occurrence_count != 1 {
4082 return Err(resident_workspace_decline(format!(
4083 "generated query head {expected_head} must have exactly one compiled rule, found {occurrence_count}"
4084 )));
4085 }
4086 heads.insert(expected_head);
4087 }
4088 Ok(heads)
4089}
4090
4091impl Executor {
4092 pub fn prepare_resident_graph<'executor>(
4094 &'executor self,
4095 plan: &ExecutionPlan,
4096 certificate: &ResidentGraphRouteCertificate,
4097 options: ResidentGraphPrepareOptions,
4098 ) -> std::result::Result<PreparedResidentGraph<'executor>, ResidentGraphExecutionError> {
4099 if !certificate.matches_plan(plan).map_err(runtime_error)? {
4100 return Err(resident_decline_error(
4101 ResidentGraphDeclineReason::WorkspaceUnbounded {
4102 detail: "route certificate does not match the plan being prepared".into(),
4103 },
4104 ));
4105 }
4106 if !certificate.is_supported() {
4107 return Err(resident_decline_error(
4108 certificate.declines().first().cloned().unwrap_or_else(|| {
4109 ResidentGraphDeclineReason::WorkspaceUnbounded {
4110 detail: "route certificate is not resident-capable".into(),
4111 }
4112 }),
4113 ));
4114 }
4115
4116 self.prepare_resident_graph_after_certification(plan, certificate, options)
4117 }
4118
4119 pub fn prepare_certified_resident_graph<'executor>(
4121 &'executor self,
4122 certified: &ResidentGraphCertifiedPlan,
4123 options: ResidentGraphPrepareOptions,
4124 ) -> std::result::Result<PreparedResidentGraph<'executor>, ResidentGraphExecutionError> {
4125 let certificate = certified.certificate();
4126 if !certificate.is_supported() {
4127 return Err(resident_decline_error(
4128 certificate.declines().first().cloned().unwrap_or_else(|| {
4129 ResidentGraphDeclineReason::WorkspaceUnbounded {
4130 detail: "route certificate is not resident-capable".into(),
4131 }
4132 }),
4133 ));
4134 }
4135 self.prepare_resident_graph_after_certification(certified.plan(), certificate, options)
4136 }
4137
4138 fn prepare_resident_graph_after_certification<'executor>(
4139 &'executor self,
4140 plan: &ExecutionPlan,
4141 certificate: &ResidentGraphRouteCertificate,
4142 options: ResidentGraphPrepareOptions,
4143 ) -> std::result::Result<PreparedResidentGraph<'executor>, ResidentGraphExecutionError> {
4144 let mut diagnostics =
4145 resident_prepare_diagnostics_for_sample(options.latency_diagnostic_sample);
4146 let total_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4147 let admission_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4148 let source_epoch = self.store.mutation_epoch();
4149 let resident_head_names = plan
4150 .rules_by_scc
4151 .iter()
4152 .flat_map(|rules| rules.iter().map(|rule| rule.head.clone()))
4153 .collect::<HashSet<_>>();
4154 let mut source_set_snapshots = Vec::new();
4155 let mut scans = Vec::new();
4156 for rules in &plan.rules_by_scc {
4157 for rule in rules {
4158 collect_recorded_scans(&rule.body, &mut scans);
4159 }
4160 }
4161 scans.sort_by_key(|rel| rel.0);
4162 scans.dedup();
4163 for rel in scans {
4164 let name = self
4165 .rel_names
4166 .get(&rel)
4167 .ok_or_else(|| runtime_error(format!("unknown resident relation id {rel:?}")))?;
4168 match self.store.get_with_version(name) {
4169 Some((buffer, version)) => {
4170 let schema = certificate.schema_for(rel).ok_or_else(|| {
4171 runtime_error(format!("resident scan {name} has no certified schema"))
4172 })?;
4173 if buffer.schema() != schema {
4174 return Err(runtime_error(format!(
4175 "resident scan {name} store schema differs from its certificate"
4176 )));
4177 }
4178 source_set_snapshots.push(
4179 resident_source_set_snapshot(&self.provider, name, version, buffer)
4180 .map_err(resident_decline_error)?,
4181 );
4182 }
4183 None if resident_head_names.contains(name) => {}
4184 None => {
4185 return Err(runtime_error(format!(
4186 "missing resident source relation {name}"
4187 )))
4188 }
4189 }
4190 }
4191 let admission = self
4192 .resident_workspace_admission_after_certification(plan, certificate)
4193 .map_err(resident_decline_error)?;
4194 for (name, schema) in &admission.head_schemas {
4195 if let Some(existing) = self.store.get(name) {
4196 if !resident_schemas_type_compatible(existing.schema(), schema) {
4197 return Err(runtime_error(format!(
4198 "existing resident head {name} has a physically incompatible schema: existing={:?}, staged={schema:?}",
4199 existing.schema()
4200 )));
4201 }
4202 }
4203 }
4204 if let Some(diagnostics) = diagnostics.as_mut() {
4205 diagnostics.admission_and_source_snapshot_ns = resident_prepare_elapsed_ns(
4206 admission_started.expect("diagnostic timer exists when enabled"),
4207 );
4208 }
4209
4210 let execution_setup_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4211 let runtime =
4212 Arc::clone(self.provider.memory().runtime().ok_or_else(|| {
4213 runtime_error("resident graph requires the async device runtime")
4214 })?);
4215 let stream_id = if let Some(stream_id) = self.resident_graph_stream.get() {
4216 *stream_id
4217 } else {
4218 let acquired = runtime.stream_pool().acquire().map_err(runtime_error)?;
4219 match self.resident_graph_stream.set(acquired) {
4220 Ok(()) => acquired,
4221 Err(_) => *self
4222 .resident_graph_stream
4223 .get()
4224 .ok_or_else(|| runtime_error("resident stream initialization raced"))?,
4225 }
4226 };
4227 let stream = runtime
4228 .stream_pool()
4229 .resolve(stream_id)
4230 .ok_or_else(|| runtime_error("resident graph stream is no longer live"))?;
4231 let execution_domain = self
4232 .provider
4233 .bind_resident_execution_domain(Arc::clone(&runtime), stream_id, Arc::clone(&stream))
4234 .map_err(runtime_error)?;
4235
4236 let mut relation_registration = self
4237 .rel_names
4238 .iter()
4239 .map(|(rel, name)| (*rel, name.clone()))
4240 .collect::<Vec<_>>();
4241 relation_registration.sort_by_key(|(rel, name)| (rel.0, name.clone()));
4242
4243 let mut build = ResidentBuild {
4244 executor: self,
4245 certificate,
4246 capacity: admission.relation_capacity,
4247 relations: Vec::new(),
4248 filter_workspaces: Vec::new(),
4249 project_workspaces: Vec::new(),
4250 heads: BTreeMap::new(),
4251 head_winner_indices: BTreeMap::new(),
4252 source_names: HashSet::new(),
4253 source_aliases: HashMap::new(),
4254 next_op_id: 0,
4255 injection: options.test_device_status,
4256 injection_recorded: false,
4257 };
4258 if let Some(diagnostics) = diagnostics.as_mut() {
4259 diagnostics.execution_domain_and_build_setup_ns = resident_prepare_elapsed_ns(
4260 execution_setup_started.expect("diagnostic timer exists when enabled"),
4261 );
4262 }
4263 let logical_planning_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4264 for (winner_index, (name, schema)) in admission.head_schemas.iter().enumerate() {
4265 let index = build
4266 .allocate_permanent_relation(schema.clone(), 0)
4267 .map_err(runtime_error)?;
4268 build.heads.insert(name.clone(), index);
4269 build.head_winner_indices.insert(
4270 name.clone(),
4271 u32::try_from(winner_index)
4272 .map_err(|_| runtime_error("resident head count exceeds u32"))?,
4273 );
4274 }
4275
4276 let mut initial_ops = Vec::new();
4277 for (name, schema) in &admission.head_schemas {
4278 if self.store.contains(name) {
4279 let source = build.source_reference(name).map_err(runtime_error)?;
4280 build
4281 .plan_copy(source.clone(), build.heads[name], schema, &mut initial_ops)
4282 .map_err(runtime_error)?;
4283 resident_record_schema_winner_mark(
4284 &mut initial_ops,
4285 ResidentBufferRef::Private(build.heads[name]),
4286 build.head_winner_indices[name],
4287 0,
4288 );
4289 }
4290 }
4291
4292 let mut scc_by_id = HashMap::new();
4293 for (index, scc) in plan.sccs.iter().enumerate() {
4294 if scc_by_id.insert(scc.id, index).is_some() {
4295 return Err(runtime_error(format!(
4296 "duplicate resident SCC id {}",
4297 scc.id
4298 )));
4299 }
4300 }
4301 let ordered_sccs = if plan.strata.is_empty() {
4302 (0..plan.sccs.len()).collect::<Vec<_>>()
4303 } else {
4304 let mut ordered = Vec::new();
4305 for stratum in &plan.strata {
4306 for scc_id in &stratum.sccs {
4307 ordered.push(*scc_by_id.get(scc_id).ok_or_else(|| {
4308 runtime_error(format!(
4309 "stratum {} references unknown SCC {scc_id}",
4310 stratum.id
4311 ))
4312 })?);
4313 }
4314 }
4315 ordered
4316 };
4317 let mut seen_sccs = HashSet::new();
4318 for index in &ordered_sccs {
4319 if !seen_sccs.insert(*index) {
4320 return Err(runtime_error(format!(
4321 "resident execution schedule repeats SCC {}",
4322 plan.sccs[*index].id
4323 )));
4324 }
4325 }
4326 if seen_sccs.len() != plan.sccs.len() {
4327 return Err(runtime_error(
4328 "resident execution schedule omits one or more SCCs",
4329 ));
4330 }
4331
4332 let mut phases = Vec::new();
4333 for scc_index in ordered_sccs {
4334 let scc = &plan.sccs[scc_index];
4335 let rules = plan.rules_by_scc.get(scc_index).ok_or_else(|| {
4336 runtime_error(format!("resident SCC {scc_index} has no rule vector"))
4337 })?;
4338 if rules.is_empty() {
4339 continue;
4340 }
4341 if !scc.is_recursive {
4342 let mut ops = Vec::new();
4343 let mut phase_heads = BTreeMap::<String, ResidentBufferRef>::new();
4344 for (rule_index, rule) in rules.iter().enumerate() {
4345 let mut occurrences = HashMap::new();
4346 let contribution = build
4347 .plan_node(&rule.body, None, &mut occurrences, &mut ops)
4348 .map_err(runtime_error)?;
4349 let schema_id =
4350 admission.rule_schema_ids[scc_index][rule_index].ok_or_else(|| {
4351 runtime_error(format!(
4352 "resident rule {} has no schema winner id",
4353 rule.head
4354 ))
4355 })?;
4356 ops.push(ResidentRecordedOp::SchemaWinnerMark {
4357 contribution: contribution.clone(),
4358 head_index: build.head_winner_indices[&rule.head],
4359 schema_id,
4360 });
4361 let target = *build.heads.get(&rule.head).ok_or_else(|| {
4362 runtime_error(format!("resident head {} was not staged", rule.head))
4363 })?;
4364 let target_schema = build
4365 .schema(&ResidentBufferRef::Private(target))
4366 .map_err(runtime_error)?
4367 .clone();
4368 let current = phase_heads
4369 .remove(&rule.head)
4370 .unwrap_or(ResidentBufferRef::Private(target));
4371 let merged = build
4372 .merge_phase_contribution(
4373 Some(current),
4374 contribution,
4375 &target_schema,
4376 &mut ops,
4377 )
4378 .map_err(runtime_error)?;
4379 phase_heads.insert(rule.head.clone(), merged);
4380 }
4381 for (head, value) in phase_heads {
4382 let target = build.heads[&head];
4383 let target_schema = build
4384 .schema(&ResidentBufferRef::Private(target))
4385 .map_err(runtime_error)?
4386 .clone();
4387 build
4388 .plan_copy(value, target, &target_schema, &mut ops)
4389 .map_err(runtime_error)?;
4390 }
4391 phases.push(ResidentCapturePhase::Segment {
4392 ops,
4393 scc_begin: None,
4394 });
4395 continue;
4396 }
4397
4398 let mut recursive_heads = BTreeMap::new();
4399 for rule in rules {
4400 recursive_heads
4401 .entry(rule.head.clone())
4402 .or_insert_with(|| rule.meta.schema.clone());
4403 }
4404 let mut delta_heads = BTreeMap::new();
4405 for (name, schema) in &recursive_heads {
4406 delta_heads.insert(
4407 name.clone(),
4408 build
4409 .allocate_permanent_relation(schema.clone(), 0)
4410 .map_err(runtime_error)?,
4411 );
4412 }
4413
4414 let mut seed_ops = Vec::new();
4415 let mut seed_values = BTreeMap::<String, ResidentBufferRef>::new();
4416 for (rule_index, rule) in rules.iter().enumerate() {
4417 let mut occurrences = HashMap::new();
4418 let contribution = build
4419 .plan_node(&rule.body, None, &mut occurrences, &mut seed_ops)
4420 .map_err(runtime_error)?;
4421 let schema_id =
4422 admission.rule_schema_ids[scc_index][rule_index].ok_or_else(|| {
4423 runtime_error(format!(
4424 "resident rule {} has no schema winner id",
4425 rule.head
4426 ))
4427 })?;
4428 seed_ops.push(ResidentRecordedOp::SchemaWinnerMark {
4429 contribution: contribution.clone(),
4430 head_index: build.head_winner_indices[&rule.head],
4431 schema_id,
4432 });
4433 let current = seed_values.remove(&rule.head);
4434 let merged = build
4435 .merge_phase_contribution(
4436 current,
4437 contribution,
4438 &rule.meta.schema,
4439 &mut seed_ops,
4440 )
4441 .map_err(runtime_error)?;
4442 seed_values.insert(rule.head.clone(), merged);
4443 }
4444 for (name, schema) in &recursive_heads {
4445 let full = build.heads[name];
4446 let seed = seed_values.remove(name).ok_or_else(|| {
4447 runtime_error(format!(
4448 "recursive resident seed has no contribution for {name}"
4449 ))
4450 })?;
4451 let candidate = build
4452 .plan_union_schema(
4453 ResidentBufferRef::Private(full),
4454 seed,
4455 schema.clone(),
4456 &mut seed_ops,
4457 )
4458 .map_err(runtime_error)?;
4459 let novel = build
4460 .plan_diff_schema(
4461 candidate.clone(),
4462 ResidentBufferRef::Private(full),
4463 schema.clone(),
4464 &mut seed_ops,
4465 )
4466 .map_err(runtime_error)?;
4467 build
4468 .plan_copy(candidate, full, schema, &mut seed_ops)
4469 .map_err(runtime_error)?;
4470 build
4471 .plan_copy(novel, delta_heads[name], schema, &mut seed_ops)
4472 .map_err(runtime_error)?;
4473 }
4474 let begin_op_id = build.next_op_id().map_err(runtime_error)?;
4475 phases.push(ResidentCapturePhase::Segment {
4476 ops: seed_ops,
4477 scc_begin: Some((self.config.max_iterations, begin_op_id)),
4478 });
4479
4480 let mut body_ops = vec![ResidentRecordedOp::ChangedReset];
4481 let mut body_values = BTreeMap::<String, ResidentBufferRef>::new();
4482 let recursive_names = recursive_heads.keys().cloned().collect::<HashSet<_>>();
4483 for (rule_index, rule) in rules.iter().enumerate() {
4484 let mut rule_scans = Vec::new();
4485 collect_recorded_scans(&rule.body, &mut rule_scans);
4486 let mut per_relation_occurrence = HashMap::<RelId, usize>::new();
4487 for rel in rule_scans {
4488 let occurrence = per_relation_occurrence.entry(rel).or_insert(0);
4489 let current = *occurrence;
4490 *occurrence += 1;
4491 let Some(name) = self.rel_names.get(&rel) else {
4492 return Err(runtime_error(format!(
4493 "recursive resident scan has unknown relation id {rel:?}"
4494 )));
4495 };
4496 if !recursive_names.contains(name) {
4497 continue;
4498 }
4499 let mut planning_occurrences = HashMap::new();
4500 let contribution = build
4501 .plan_node(
4502 &rule.body,
4503 Some((rel, current, delta_heads[name])),
4504 &mut planning_occurrences,
4505 &mut body_ops,
4506 )
4507 .map_err(runtime_error)?;
4508 let schema_id =
4509 admission.rule_schema_ids[scc_index][rule_index].ok_or_else(|| {
4510 runtime_error(format!(
4511 "resident rule {} has no schema winner id",
4512 rule.head
4513 ))
4514 })?;
4515 body_ops.push(ResidentRecordedOp::SchemaWinnerMark {
4516 contribution: contribution.clone(),
4517 head_index: build.head_winner_indices[&rule.head],
4518 schema_id,
4519 });
4520 let current = body_values.remove(&rule.head);
4521 let merged = build
4522 .merge_phase_contribution(
4523 current,
4524 contribution,
4525 &rule.meta.schema,
4526 &mut body_ops,
4527 )
4528 .map_err(runtime_error)?;
4529 body_values.insert(rule.head.clone(), merged);
4530 }
4531 }
4532 for (name, schema) in &recursive_heads {
4533 let full = build.heads[name];
4534 let accumulated = match body_values.remove(name) {
4535 Some(value) => value,
4536 None => build
4537 .plan_diff_schema(
4538 ResidentBufferRef::Private(full),
4539 ResidentBufferRef::Private(full),
4540 schema.clone(),
4541 &mut body_ops,
4542 )
4543 .map_err(runtime_error)?,
4544 };
4545 let novel = build
4546 .plan_diff_schema(
4547 accumulated,
4548 ResidentBufferRef::Private(full),
4549 schema.clone(),
4550 &mut body_ops,
4551 )
4552 .map_err(runtime_error)?;
4553 let next_full = build
4554 .plan_union_schema(
4555 ResidentBufferRef::Private(full),
4556 novel.clone(),
4557 schema.clone(),
4558 &mut body_ops,
4559 )
4560 .map_err(runtime_error)?;
4561 build
4562 .plan_copy(next_full, full, schema, &mut body_ops)
4563 .map_err(runtime_error)?;
4564 build
4565 .plan_copy(novel, delta_heads[name], schema, &mut body_ops)
4566 .map_err(runtime_error)?;
4567 body_ops.push(ResidentRecordedOp::ChangedMark {
4570 relation: delta_heads[name],
4571 });
4572 }
4573 let convergence_op_id = build.next_op_id().map_err(runtime_error)?;
4574 phases.push(ResidentCapturePhase::ConditionalWhile {
4575 ops: body_ops,
4576 iteration_limit: self.config.max_iterations,
4577 convergence_op_id,
4578 });
4579 }
4580
4581 let success_op_id = build.next_op_id().map_err(runtime_error)?;
4582 if let Some(diagnostics) = diagnostics.as_mut() {
4583 diagnostics.logical_schedule_planning_ns = resident_prepare_elapsed_ns(
4584 logical_planning_started.expect("diagnostic timer exists when enabled"),
4585 );
4586 }
4587 let manifest_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4588 let mut manifest = build
4589 .allocation_manifest(&initial_ops, &phases)
4590 .map_err(runtime_error)?;
4591 let compact_tables =
4592 resident_compact_tables(&build.filter_workspaces, &build.project_workspaces)
4593 .map_err(runtime_error)?;
4594 let compact_regions =
4595 resident_compact_regions(initial_ops, phases, success_op_id).map_err(runtime_error)?;
4596 if let Some(diagnostics) = diagnostics.as_mut() {
4597 diagnostics.manifest_compact_construction_ns = resident_prepare_elapsed_ns(
4598 manifest_started.expect("diagnostic timer exists when enabled"),
4599 );
4600 diagnostics.compact_regions = compact_regions.len();
4601 diagnostics.conditional_regions = compact_regions
4602 .iter()
4603 .filter(|region| region.iteration_limit != 0)
4604 .count();
4605 }
4606 let schedule_lowering_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4607 let compact_schedule = resident_lower_compact_regions(
4608 compact_regions,
4609 &manifest.slots,
4610 &manifest.logical_to_slot,
4611 build.source_names.iter().map(String::as_str),
4612 compact_tables,
4613 )
4614 .map_err(runtime_error)?;
4615 manifest
4616 .finalize_compact_schedule(&compact_schedule, build.heads.len())
4617 .map_err(runtime_error)?;
4618 if let Some(diagnostics) = diagnostics.as_mut() {
4619 diagnostics.schedule_lowering_ns = resident_prepare_elapsed_ns(
4620 schedule_lowering_started.expect("diagnostic timer exists when enabled"),
4621 );
4622 diagnostics.required_reservation_bytes = manifest.required_bytes;
4623 diagnostics.logical_relation_values = manifest.logical_relation_values;
4624 diagnostics.physical_relation_slots = manifest.slots.len();
4625 diagnostics.compact_ops = compact_schedule.ops.len();
4626 diagnostics.compact_waves = compact_schedule.waves.len();
4627 }
4628 let reservation_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4629 let manifest_available_bytes = self.provider.memory().remaining_bytes();
4630 let mut reservation = self
4631 .provider
4632 .memory()
4633 .reserve_bytes(manifest.required_bytes)
4634 .map_err(|error| {
4635 resident_decline_error(ResidentGraphDeclineReason::WorkspaceUnbounded {
4636 detail: format!(
4637 "resident allocation manifest reservation of {} bytes failed: {error}",
4638 manifest.required_bytes
4639 ),
4640 })
4641 })?;
4642 if let Some(diagnostics) = diagnostics.as_mut() {
4643 diagnostics.reservation_ns = resident_prepare_elapsed_ns(
4644 reservation_started.expect("diagnostic timer exists when enabled"),
4645 );
4646 }
4647 let physical = build
4648 .materialize(&manifest, &mut reservation, diagnostics.as_mut())
4649 .map_err(runtime_error)?;
4650
4651 if build.injection.is_some() && !build.injection_recorded {
4652 return Err(runtime_error(
4653 "test device status requested after a nonexistent physical op",
4654 ));
4655 }
4656 let metadata_binding_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4657 let output_indices =
4658 resident_output_indices(&build.heads, &manifest.logical_to_slot, &manifest.slots)
4659 .map_err(runtime_error)?;
4660 let output_relations = output_indices
4661 .iter()
4662 .map(|(_, index)| private_relation(&physical.relations, *index))
4663 .collect::<Result<Vec<_>>>()
4664 .map_err(runtime_error)?;
4665 let output_schema_plans = output_indices
4666 .iter()
4667 .map(|(name, _)| {
4668 let candidates = admission
4669 .head_schema_choices
4670 .get(name)
4671 .cloned()
4672 .ok_or_else(|| {
4673 runtime_error(format!("resident head {name} has no schema plan"))
4674 })?;
4675 let selection = if let Some(selection) = admission.head_schema_selections.get(name)
4676 {
4677 let source_output = usize::try_from(
4678 *build
4679 .head_winner_indices
4680 .get(&selection.source_head)
4681 .ok_or_else(|| {
4682 runtime_error(format!(
4683 "resident schema source {} is not staged",
4684 selection.source_head
4685 ))
4686 })?,
4687 )
4688 .map_err(|_| runtime_error("resident schema source index overflow"))?;
4689 let source_candidates = admission
4690 .head_schema_choices
4691 .get(&selection.source_head)
4692 .ok_or_else(|| runtime_error("resident schema source is missing"))?;
4693 if source_candidates.len() != selection.output_schemas_by_source_winner.len() {
4694 return Err(runtime_error(
4695 "resident schema source mapping has the wrong length",
4696 ));
4697 }
4698 ResidentOutputSchemaSelection::SourceWinner {
4699 source_output,
4700 schemas: selection.output_schemas_by_source_winner.clone(),
4701 }
4702 } else {
4703 ResidentOutputSchemaSelection::OwnWinner
4704 };
4705 Ok(ResidentOutputSchemaPlan {
4706 candidates,
4707 selection,
4708 })
4709 })
4710 .collect::<std::result::Result<Vec<_>, ResidentGraphExecutionError>>()?;
4711 if let Some(diagnostics) = diagnostics.as_mut() {
4712 let metadata_binding_ns = resident_prepare_elapsed_ns(
4713 metadata_binding_started.expect("diagnostic timer exists when enabled"),
4714 );
4715 diagnostics.metadata_binding_construction_ns = diagnostics
4716 .metadata_binding_construction_ns
4717 .saturating_add(metadata_binding_ns);
4718 }
4719 let device_trace_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4720 let device_trace_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4721 let device_trace = self
4722 .provider
4723 .prepare_resident_device_trace_in_reservation(&mut reservation)
4724 .map_err(runtime_error)?;
4725 if let Some(diagnostics) = diagnostics.as_mut() {
4726 let device_trace_ns = resident_prepare_elapsed_ns(
4727 device_trace_started.expect("diagnostic timer exists when enabled"),
4728 );
4729 let device_trace_reserved_bytes = reservation.used_bytes().saturating_sub(
4730 device_trace_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
4731 );
4732 diagnostics.metadata_provider_calls =
4733 diagnostics.metadata_provider_calls.saturating_add(1);
4734 diagnostics.device_trace_preparation_ns = device_trace_ns;
4735 diagnostics.device_trace_reserved_bytes = device_trace_reserved_bytes;
4736 }
4737 let schema_defaults_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4738 let default_schema_ids =
4739 resident_compact_schema_defaults(&compact_schedule.ops, output_relations.len())
4740 .map_err(runtime_error)?;
4741 if let Some(diagnostics) = diagnostics.as_mut() {
4742 let schema_defaults_ns = resident_prepare_elapsed_ns(
4743 schema_defaults_started.expect("diagnostic timer exists when enabled"),
4744 );
4745 diagnostics.metadata_binding_construction_ns = diagnostics
4746 .metadata_binding_construction_ns
4747 .saturating_add(schema_defaults_ns);
4748 }
4749 let schema_winners_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4750 let schema_winners_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4751 let schema_winners = self
4752 .provider
4753 .prepare_resident_schema_winners_in_reservation(&default_schema_ids, &mut reservation)
4754 .map_err(runtime_error)?;
4755 if let Some(diagnostics) = diagnostics.as_mut() {
4756 let schema_winners_ns = resident_prepare_elapsed_ns(
4757 schema_winners_started.expect("diagnostic timer exists when enabled"),
4758 );
4759 let schema_winners_reserved_bytes = reservation.used_bytes().saturating_sub(
4760 schema_winners_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
4761 );
4762 diagnostics.metadata_provider_calls =
4763 diagnostics.metadata_provider_calls.saturating_add(1);
4764 diagnostics.schema_winners_preparation_ns = schema_winners_ns;
4765 diagnostics.schema_winners_reserved_bytes = schema_winners_reserved_bytes;
4766 (
4767 diagnostics.schema_winners_initial_htod_calls,
4768 diagnostics.schema_winners_initial_htod_bytes,
4769 ) = resident_schema_winners_initial_htod(default_schema_ids.len());
4770 }
4771 let receipt_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4772 let receipt_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4773 let receipt = self
4774 .provider
4775 .prepare_resident_packed_receipt_with_trace_and_schema_winners_in_reservation(
4776 &output_relations,
4777 &device_trace,
4778 &schema_winners,
4779 &mut reservation,
4780 )
4781 .map_err(runtime_error)?;
4782 if let Some(diagnostics) = diagnostics.as_mut() {
4783 let receipt_ns = resident_prepare_elapsed_ns(
4784 receipt_started.expect("diagnostic timer exists when enabled"),
4785 );
4786 let receipt_reserved_bytes = reservation.used_bytes().saturating_sub(
4787 receipt_bytes_before.expect("diagnostic byte snapshot exists when enabled"),
4788 );
4789 diagnostics.metadata_provider_calls =
4790 diagnostics.metadata_provider_calls.saturating_add(1);
4791 diagnostics.receipt_preparation_ns = receipt_ns;
4792 diagnostics.receipt_reserved_bytes = receipt_reserved_bytes;
4793 (
4794 diagnostics.receipt_initial_htod_calls,
4795 diagnostics.receipt_initial_htod_bytes,
4796 ) = resident_receipt_initial_htod(output_relations.len());
4797 }
4798 let schedule_bindings_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4799 let mut schedule_bindings = Vec::with_capacity(
4800 manifest
4801 .slots
4802 .len()
4803 .checked_add(compact_schedule.source_slots.len())
4804 .ok_or_else(|| runtime_error("resident compact slot binding count overflow"))?,
4805 );
4806 for (slot_index, slot) in manifest.slots.iter().enumerate() {
4807 let relation =
4808 private_relation(&physical.relations, slot_index).map_err(runtime_error)?;
4809 schedule_bindings.push(if slot.permanent {
4810 ResidentScheduleSlotBinding::permanent(relation.buffer(), 0)
4811 } else {
4812 ResidentScheduleSlotBinding::scratch(relation.buffer(), 0)
4813 });
4814 }
4815 for (name, slot) in &compact_schedule.source_slots {
4816 let expected_slot = u32::try_from(schedule_bindings.len())
4817 .map_err(|_| runtime_error("resident compact slot binding count exceeds u32"))?;
4818 if *slot != expected_slot {
4819 return Err(runtime_error(
4820 "resident compact source bindings are not contiguous",
4821 ));
4822 }
4823 let source = self.store.get(name).ok_or_else(|| {
4824 runtime_error(format!("resident source {name} disappeared during prepare"))
4825 })?;
4826 schedule_bindings
4827 .push(ResidentScheduleSlotBinding::source(source, 0).map_err(runtime_error)?);
4828 }
4829 let receipt_slots = output_indices
4830 .iter()
4831 .map(|(_, slot)| {
4832 u32::try_from(*slot)
4833 .map_err(|_| runtime_error("resident receipt slot index exceeds u32"))
4834 })
4835 .collect::<std::result::Result<Vec<_>, _>>()?;
4836 if let Some(diagnostics) = diagnostics.as_mut() {
4837 let schedule_bindings_ns = resident_prepare_elapsed_ns(
4838 schedule_bindings_started.expect("diagnostic timer exists when enabled"),
4839 );
4840 diagnostics.metadata_binding_construction_ns = diagnostics
4841 .metadata_binding_construction_ns
4842 .saturating_add(schedule_bindings_ns);
4843 }
4844 let schedule_program_bytes_before = diagnostics.as_ref().map(|_| reservation.used_bytes());
4845 let schedule_program_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4846 let schedule_program = self
4847 .provider
4848 .prepare_resident_schedule_program_in_reservation(
4849 &execution_domain,
4850 &schedule_bindings,
4851 &compact_schedule.ops,
4852 &compact_schedule.waves,
4853 &compact_schedule.regions,
4854 &compact_schedule.generation_bases,
4855 &compact_schedule.filter_comparisons,
4856 &compact_schedule.project_expressions,
4857 &receipt_slots,
4858 ResidentScheduleExternalBindings::new(
4859 physical.filter_scratch.as_ref(),
4860 &physical.set_workspace,
4861 &physical.join_workspace,
4862 &physical.control,
4863 &device_trace,
4864 &schema_winners,
4865 &receipt,
4866 ),
4867 &mut reservation,
4868 )
4869 .map_err(runtime_error)?;
4870 if let Some(diagnostics) = diagnostics.as_mut() {
4871 let schedule_program_ns = resident_prepare_elapsed_ns(
4872 schedule_program_started.expect("diagnostic timer exists when enabled"),
4873 );
4874 let schedule_program_reserved_bytes = reservation.used_bytes().saturating_sub(
4875 schedule_program_bytes_before
4876 .expect("diagnostic byte snapshot exists when enabled"),
4877 );
4878 diagnostics.metadata_provider_calls =
4879 diagnostics.metadata_provider_calls.saturating_add(1);
4880 diagnostics.schedule_program_preparation_ns = schedule_program_ns;
4881 diagnostics.schedule_program_reserved_bytes = schedule_program_reserved_bytes;
4882 (
4883 diagnostics.schedule_program_initial_htod_calls,
4884 diagnostics.schedule_program_initial_htod_bytes,
4885 ) = resident_schedule_initial_htod(diagnostics.schedule_program_reserved_bytes);
4886 }
4887 let reservation_validation_started =
4888 diagnostics.as_ref().map(|_| std::time::Instant::now());
4889 let tracked_device_allocation_bytes = reservation.used_bytes();
4890 resident_validate_exact_reservation(
4891 manifest.required_bytes,
4892 tracked_device_allocation_bytes,
4893 reservation.remaining_bytes(),
4894 )
4895 .map_err(runtime_error)?;
4896 drop(reservation);
4897 if let Some(diagnostics) = diagnostics.as_mut() {
4898 diagnostics.reservation_validation_and_release_ns = resident_prepare_elapsed_ns(
4899 reservation_validation_started.expect("diagnostic timer exists when enabled"),
4900 );
4901 }
4902 let pinned_receipt_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4903 let pinned_receipt = self
4904 .provider
4905 .prepare_resident_pinned_receipt(&receipt)
4906 .map_err(runtime_error)?;
4907 if let Some(diagnostics) = diagnostics.as_mut() {
4908 diagnostics.pinned_receipt_ns = resident_prepare_elapsed_ns(
4909 pinned_receipt_started.expect("diagnostic timer exists when enabled"),
4910 );
4911 }
4912
4913 let graph_capture_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4914 let mut sequence =
4915 ConditionalCudaGraphSequenceBuilder::new(&stream).map_err(conditional_graph_error)?;
4916 let capture_topology =
4917 resident_compact_topology(&compact_schedule.regions).map_err(runtime_error)?;
4918 let mut conditional_body_node_kinds =
4919 Vec::with_capacity(capture_topology.conditional_body_kernel_counts.len());
4920 for ((region_index, region), parent_kind) in compact_schedule
4921 .regions
4922 .iter()
4923 .enumerate()
4924 .zip(capture_topology.parent_kinds.iter().copied())
4925 {
4926 let region_index = u32::try_from(region_index)
4927 .map_err(|_| runtime_error("resident compact region index exceeds u32"))?;
4928 if parent_kind == ResidentCaptureParentKind::Conditional {
4929 sequence
4930 .add_conditional_while(u32::from(region.iteration_limit != 0), true, |body| {
4931 body.capture_on_stream(&stream, || unsafe {
4932 self.provider.record_resident_schedule_region_on_stream(
4933 &schedule_program,
4934 region_index,
4935 Some(&body),
4936 &stream,
4937 )
4938 })?;
4939 conditional_body_node_kinds.push(body.linear_chain_node_kinds()?);
4940 Ok(())
4941 })
4942 .map_err(conditional_graph_error)?;
4943 } else {
4944 sequence
4945 .capture_segment_on_stream(&stream, || unsafe {
4946 self.provider.record_resident_schedule_region_on_stream(
4947 &schedule_program,
4948 region_index,
4949 None,
4950 &stream,
4951 )
4952 })
4953 .map_err(conditional_graph_error)?;
4954 }
4955 }
4956 if let Some(diagnostics) = diagnostics.as_mut() {
4957 diagnostics.graph_body_capture_ns = resident_prepare_elapsed_ns(
4958 graph_capture_started.expect("diagnostic timer exists when enabled"),
4959 );
4960 }
4961 let graph_instantiate_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4962 let graph = sequence
4963 .instantiate()
4964 .map_err(conditional_graph_error)?
4965 .bind_resident_lifecycle(&runtime);
4966 if let Some(diagnostics) = diagnostics.as_mut() {
4967 diagnostics.graph_instantiate_ns = resident_prepare_elapsed_ns(
4968 graph_instantiate_started.expect("diagnostic timer exists when enabled"),
4969 );
4970 }
4971
4972 let validation_started = diagnostics.as_ref().map(|_| std::time::Instant::now());
4973 let mut recorder = execution_domain.new_strict_recorder();
4974 for binding in &schedule_bindings {
4975 binding.record_uses(&mut recorder);
4976 }
4977 ResidentScheduleExternalBindings::new(
4978 physical.filter_scratch.as_ref(),
4979 &physical.set_workspace,
4980 &physical.join_workspace,
4981 &physical.control,
4982 &device_trace,
4983 &schema_winners,
4984 &receipt,
4985 )
4986 .record_uses(&mut recorder);
4987 schedule_program.record_uses(&mut recorder);
4988
4989 let graph_node_kinds = graph.linear_chain_node_kinds().map_err(runtime_error)?;
4990 resident_validate_parent_graph_kinds(&graph_node_kinds, &capture_topology.parent_kinds)
4991 .map_err(runtime_error)?;
4992 let conditional_body_kernel_counts = resident_validate_conditional_body_node_kinds(
4993 &conditional_body_node_kinds,
4994 capture_topology.conditional_body_kernel_counts.len(),
4995 )
4996 .map_err(runtime_error)?;
4997 if conditional_body_kernel_counts != capture_topology.conditional_body_kernel_counts {
4998 return Err(runtime_error(
4999 "resident conditional body topology differs from the compact schedule",
5000 ));
5001 }
5002 let body_node_count =
5003 conditional_body_node_kinds
5004 .iter()
5005 .try_fold(0usize, |total, kinds| {
5006 total.checked_add(kinds.len()).ok_or_else(|| {
5007 runtime_error("resident hierarchical graph node count overflow")
5008 })
5009 })?;
5010 let hierarchical_graph_nodes = graph_node_kinds
5011 .len()
5012 .checked_add(body_node_count)
5013 .ok_or_else(|| runtime_error("resident hierarchical graph node count overflow"))?;
5014 if hierarchical_graph_nodes != capture_topology.hierarchical_node_count {
5015 return Err(runtime_error(
5016 "resident hierarchical graph topology differs from the compact schedule",
5017 ));
5018 }
5019 let (
5020 filter_descriptor_device_bytes,
5021 project_descriptor_device_bytes,
5022 fixed_workspace_device_bytes,
5023 ) = resident_compact_preflight_device_bytes(&manifest, &compact_schedule)
5024 .map_err(runtime_error)?;
5025 let parent_graph_node_count = graph_node_kinds.len();
5026 let preflight_report = ResidentGraphPreflightReport {
5027 relation_capacity: admission.relation_capacity,
5028 estimated_required_bytes: manifest.required_bytes,
5029 available_bytes_at_admission: manifest_available_bytes,
5030 tracked_device_allocation_bytes,
5031 relation_device_bytes: manifest.relation_bytes,
5032 filter_descriptor_device_bytes,
5033 filter_scratch_device_bytes: manifest.filter_scratch_bytes,
5034 project_descriptor_device_bytes,
5035 fixed_workspace_device_bytes,
5036 parent_graph_nodes: graph_node_kinds.len(),
5037 conditional_while_nodes: graph_node_kinds
5038 .iter()
5039 .filter(|kind| **kind == CudaGraphNodeKind::Conditional)
5040 .count(),
5041 parent_graph_node_kinds: graph_node_kinds,
5042 conditional_body_node_kinds,
5043 conditional_body_kernel_counts,
5044 hierarchical_graph_nodes,
5045 private_relation_slots: physical.relations.len(),
5046 logical_relation_values: manifest.logical_relation_values,
5047 permanent_relation_slots: manifest.permanent_slots,
5048 staged_output_relations: output_indices.len(),
5049 scratch_slots: manifest.scratch_slots,
5050 filter_scratch_allocations: manifest.filter_scratch_allocations,
5051 max_row_bytes: manifest.max_row_bytes,
5052 };
5053 let has_device_status_writer = build.injection_recorded;
5054 let owners = ResidentRunOwners {
5055 provider: Arc::clone(&self.provider),
5056 runtime,
5057 stream,
5058 graph,
5059 execution_domain,
5060 schedule_program,
5061 recorder,
5062 relations: physical.relations,
5063 output_indices,
5064 filter_scratch: physical.filter_scratch,
5065 set_workspace: physical.set_workspace,
5066 join_workspace: physical.join_workspace,
5067 control: physical.control,
5068 device_trace,
5069 schema_winners,
5070 receipt,
5071 pinned_receipt,
5072 source_epoch,
5073 relation_registration,
5074 transaction_identity: Arc::clone(&self.transaction_identity),
5075 output_schema_plans,
5076 };
5077 let validation_owner_assembly_ns = diagnostics.as_ref().map(|_| {
5078 resident_prepare_elapsed_ns(
5079 validation_started.expect("diagnostic timer exists when enabled"),
5080 )
5081 });
5082 let prepare_total_ns = diagnostics.as_ref().map(|_| {
5083 resident_prepare_elapsed_ns(
5084 total_started.expect("diagnostic total timer exists when enabled"),
5085 )
5086 });
5087 let prepare_diagnostic = diagnostics.map(|mut diagnostics| {
5088 diagnostics.total_ns =
5089 prepare_total_ns.expect("diagnostic total sample exists when enabled");
5090 diagnostics.validation_owner_assembly_ns = validation_owner_assembly_ns
5091 .expect("diagnostic validation sample exists when enabled");
5092 diagnostics.parent_graph_nodes = parent_graph_node_count;
5093 diagnostics.conditional_body_nodes = body_node_count;
5094 diagnostics
5095 });
5096 Ok(PreparedResidentGraph {
5097 owners,
5098 preflight_report,
5099 has_device_status_writer,
5100 source_guard: self,
5101 source_set_snapshots,
5102 prepare_diagnostic,
5103 })
5104 }
5105
5106 fn resident_schema_variants(
5107 &self,
5108 node: &RirNode,
5109 certificate: &ResidentGraphRouteCertificate,
5110 head_schema_choices: &BTreeMap<String, Vec<Schema>>,
5111 ) -> std::result::Result<ResidentSchemaVariants, ResidentGraphDeclineReason> {
5112 match node {
5113 RirNode::Unit => certificate
5114 .node_schema(node)
5115 .map(ResidentSchemaVariants::Fixed)
5116 .ok_or_else(|| resident_workspace_decline("resident node schema missing")),
5117 RirNode::Scan { rel } => {
5118 let name = self.rel_names.get(rel).ok_or_else(|| {
5119 resident_workspace_decline(format!(
5120 "resident scan relation {rel:?} is not registered"
5121 ))
5122 })?;
5123 if let Some(schemas) = head_schema_choices.get(name) {
5124 return resident_schema_variants_from_source(name.clone(), schemas.clone());
5125 }
5126 let schema = self
5127 .store
5128 .get(name)
5129 .map(|buffer| buffer.schema().clone())
5130 .or_else(|| certificate.schema_for(*rel).cloned())
5131 .ok_or_else(|| {
5132 resident_workspace_decline(format!("resident scan {name} has no schema"))
5133 })?;
5134 Ok(ResidentSchemaVariants::Fixed(schema))
5135 }
5136 RirNode::Filter { input, .. } | RirNode::Distinct { input, .. } => {
5137 self.resident_schema_variants(input, certificate, head_schema_choices)
5138 }
5139 RirNode::Project { input, columns } => {
5140 match self.resident_schema_variants(input, certificate, head_schema_choices)? {
5141 ResidentSchemaVariants::Fixed(schema) => self
5142 .project_schema(&schema, columns)
5143 .map(ResidentSchemaVariants::Fixed)
5144 .map_err(|error| resident_workspace_decline(error.to_string())),
5145 ResidentSchemaVariants::Dynamic {
5146 source_head,
5147 schemas,
5148 } => {
5149 let schemas = schemas
5150 .iter()
5151 .map(|schema| self.project_schema(schema, columns))
5152 .collect::<Result<Vec<_>>>()
5153 .map_err(|error| resident_workspace_decline(error.to_string()))?;
5154 resident_schema_variants_from_source(source_head, schemas)
5155 }
5156 }
5157 }
5158 RirNode::Join {
5159 left,
5160 right,
5161 join_type,
5162 ..
5163 } => {
5164 let left = self.resident_schema_variants(left, certificate, head_schema_choices)?;
5165 match join_type {
5166 JoinType::Semi => Ok(left),
5167 JoinType::Inner => {
5168 let right =
5169 self.resident_schema_variants(right, certificate, head_schema_choices)?;
5170 match (left, right) {
5171 (
5172 ResidentSchemaVariants::Fixed(left),
5173 ResidentSchemaVariants::Fixed(right),
5174 ) => {
5175 let mut columns = left.columns;
5176 columns.extend(right.columns);
5177 Ok(ResidentSchemaVariants::Fixed(Schema::new(columns)))
5178 }
5179 (
5180 ResidentSchemaVariants::Dynamic {
5181 source_head,
5182 schemas,
5183 },
5184 ResidentSchemaVariants::Fixed(right),
5185 ) => {
5186 let schemas = schemas
5187 .into_iter()
5188 .map(|left| {
5189 let mut columns = left.columns;
5190 columns.extend(right.columns.iter().cloned());
5191 Schema::new(columns)
5192 })
5193 .collect();
5194 resident_schema_variants_from_source(source_head, schemas)
5195 }
5196 (
5197 ResidentSchemaVariants::Fixed(left),
5198 ResidentSchemaVariants::Dynamic {
5199 source_head,
5200 schemas,
5201 },
5202 ) => {
5203 let schemas = schemas
5204 .into_iter()
5205 .map(|right| {
5206 let mut columns = left.columns.clone();
5207 columns.extend(right.columns);
5208 Schema::new(columns)
5209 })
5210 .collect();
5211 resident_schema_variants_from_source(source_head, schemas)
5212 }
5213 (
5214 ResidentSchemaVariants::Dynamic {
5215 source_head: left_source,
5216 schemas: left_schemas,
5217 },
5218 ResidentSchemaVariants::Dynamic {
5219 source_head: right_source,
5220 schemas: right_schemas,
5221 },
5222 ) => {
5223 if left_source != right_source
5224 || left_schemas.len() != right_schemas.len()
5225 {
5226 return Err(resident_workspace_decline(
5227 "resident join has multiple schema sources",
5228 ));
5229 }
5230 let schemas = left_schemas
5231 .into_iter()
5232 .zip(right_schemas)
5233 .map(|(left, right)| {
5234 let mut columns = left.columns;
5235 columns.extend(right.columns);
5236 Schema::new(columns)
5237 })
5238 .collect();
5239 resident_schema_variants_from_source(left_source, schemas)
5240 }
5241 }
5242 }
5243 _ => Err(resident_workspace_decline(
5244 "uncertified join reached resident schema planning",
5245 )),
5246 }
5247 }
5248 RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
5249 self.resident_schema_variants(fallback, certificate, head_schema_choices)
5250 }
5251 RirNode::Diff { left, .. } => {
5252 self.resident_schema_variants(left, certificate, head_schema_choices)
5253 }
5254 RirNode::Union { inputs } => {
5255 if inputs.is_empty() {
5256 return Err(resident_workspace_decline(
5257 "resident schema union has no inputs",
5258 ));
5259 }
5260 let mut fixed = None;
5261 for input in inputs {
5262 match self.resident_schema_variants(input, certificate, head_schema_choices)? {
5263 ResidentSchemaVariants::Fixed(schema) => match &fixed {
5264 Some(existing) if existing != &schema => {
5265 return Err(resident_workspace_decline(
5266 "resident schema union has ambiguous input schemas",
5267 ));
5268 }
5269 Some(_) => {}
5270 None => fixed = Some(schema),
5271 },
5272 ResidentSchemaVariants::Dynamic { .. } => {
5273 return Err(resident_workspace_decline(
5274 "resident schema union has ambiguous dynamic lineage",
5275 ));
5276 }
5277 }
5278 }
5279 fixed.map(ResidentSchemaVariants::Fixed).ok_or_else(|| {
5280 resident_workspace_decline("resident schema union has no inputs")
5281 })
5282 }
5283 RirNode::Fixpoint { .. }
5284 | RirNode::GroupBy { .. }
5285 | RirNode::TensorMaskedJoin { .. } => Err(resident_workspace_decline(
5286 "unsupported node reached resident schema planning",
5287 )),
5288 }
5289 }
5290
5291 fn resident_workspace_admission_after_certification(
5292 &self,
5293 plan: &ExecutionPlan,
5294 certificate: &ResidentGraphRouteCertificate,
5295 ) -> std::result::Result<ResidentWorkspaceAdmission, ResidentGraphDeclineReason> {
5296 for rule in plan.rules_by_scc.iter().flatten() {
5297 validate_compact_resident_node_envelope(&rule.body, &|node| {
5298 certificate.node_schema(node).ok_or_else(|| {
5299 resident_workspace_decline("compact resident node schema is missing")
5300 })
5301 })?;
5302 }
5303
5304 let generated_query_heads = resident_generated_query_heads(plan)?;
5305
5306 let mut generated_query_occurrences = generated_query_heads
5307 .iter()
5308 .map(|head| (head.as_str(), 0usize))
5309 .collect::<BTreeMap<_, _>>();
5310 for rule in plan.rules_by_scc.iter().flatten() {
5311 if let Some(count) = generated_query_occurrences.get_mut(rule.head.as_str()) {
5312 *count = count.checked_add(1).ok_or_else(|| {
5313 resident_workspace_decline(format!(
5314 "generated query head {} rule count overflowed",
5315 rule.head
5316 ))
5317 })?;
5318 }
5319 }
5320 for (head, count) in generated_query_occurrences {
5321 if count != 1 {
5322 return Err(resident_workspace_decline(format!(
5323 "generated query head {head} must have exactly one compiled rule, found {count}"
5324 )));
5325 }
5326 }
5327
5328 let (required_row_capacity, capacity_witness) =
5329 self.resident_required_row_capacity(plan, certificate)?;
5330 let required_row_capacity = required_row_capacity.max(1);
5331 let required_row_capacity = u32::try_from(required_row_capacity).map_err(|_| {
5332 ResidentGraphDeclineReason::WorkspaceUnbounded {
5333 detail: format!(
5334 "certified resident row bound {required_row_capacity} exceeds the u32 count ABI; witness: {capacity_witness}"
5335 ),
5336 }
5337 })?;
5338 let relation_capacity = checked_capacity_class(required_row_capacity).map_err(|error| {
5339 ResidentGraphDeclineReason::WorkspaceUnbounded {
5340 detail: format!(
5341 "{error}; certified_required_rows={required_row_capacity}; witness: {capacity_witness}"
5342 ),
5343 }
5344 })?;
5345
5346 let mut head_schemas = BTreeMap::<String, Schema>::new();
5347 for rules in &plan.rules_by_scc {
5348 for rule in rules {
5349 match head_schemas.get(&rule.head) {
5350 Some(existing) if existing != &rule.meta.schema => {
5351 return Err(ResidentGraphDeclineReason::WorkspaceUnbounded {
5352 detail: format!(
5353 "relation {} has inconsistent compiled head schemas",
5354 rule.head
5355 ),
5356 });
5357 }
5358 Some(_) => {}
5359 None => {
5360 head_schemas.insert(rule.head.clone(), rule.meta.schema.clone());
5361 }
5362 }
5363 }
5364 }
5365
5366 let mut head_schema_choices = head_schemas
5367 .keys()
5368 .map(|name| {
5369 let mut candidates = Vec::new();
5370 if let Some(existing) = self.store.get(name) {
5371 let physical = &head_schemas[name];
5372 if !resident_schemas_type_compatible(existing.schema(), physical) {
5373 return Err(resident_workspace_decline(format!(
5374 "existing resident head {name} is physically incompatible with its compiled schema"
5375 )));
5376 }
5377 if !(generated_query_heads.contains(name)
5378 && existing.cached_row_count() == Some(0))
5379 {
5380 candidates.push(existing.schema().clone());
5381 }
5382 }
5383 Ok((name.clone(), candidates))
5384 })
5385 .collect::<std::result::Result<BTreeMap<_, _>, _>>()?;
5386 let mut rule_schema_ids = plan
5387 .rules_by_scc
5388 .iter()
5389 .map(|rules| vec![None; rules.len()])
5390 .collect::<Vec<_>>();
5391 let mut head_schema_selections = BTreeMap::new();
5392
5393 for (scc_index, rules) in plan.rules_by_scc.iter().enumerate() {
5394 let recursive = plan.sccs.get(scc_index).is_some_and(|scc| scc.is_recursive);
5395 for (rule_index, rule) in rules.iter().enumerate() {
5396 if generated_query_heads.contains(&rule.head) {
5397 continue;
5398 }
5399 let variants =
5400 self.resident_schema_variants(&rule.body, certificate, &head_schema_choices)?;
5401 let physical = &head_schemas[&rule.head];
5402 let validate_schema = |schema: &Schema| {
5403 if !resident_schemas_type_compatible(schema, physical) {
5404 return Err(resident_workspace_decline(format!(
5405 "resident rule contribution for {} is physically incompatible with its compiled head",
5406 rule.head
5407 )));
5408 }
5409 if recursive {
5410 let rel = self.name_to_rel.get(&rule.head).ok_or_else(|| {
5411 resident_workspace_decline(format!(
5412 "recursive resident head {} is not registered",
5413 rule.head
5414 ))
5415 })?;
5416 let candidate = certificate.schema_for(*rel).ok_or_else(|| {
5417 resident_workspace_decline(format!(
5418 "recursive resident head {} has no catalog schema candidate",
5419 rule.head
5420 ))
5421 })?;
5422 if !resident_schemas_type_compatible(schema, candidate) {
5423 return Err(resident_workspace_decline(format!(
5424 "recursive resident head {} contribution schema is physically incompatible with its catalog candidate: contribution={schema:?}, catalog={candidate:?}",
5425 rule.head
5426 )));
5427 }
5428 if schema.key_columns != candidate.key_columns {
5429 return Err(resident_workspace_decline(format!(
5430 "recursive resident head {} contribution key columns do not equal its catalog candidate: contribution={:?}, catalog={:?}",
5431 rule.head, schema.key_columns, candidate.key_columns
5432 )));
5433 }
5434 }
5435 Ok(())
5436 };
5437 match &variants {
5438 ResidentSchemaVariants::Fixed(schema) => validate_schema(schema)?,
5439 ResidentSchemaVariants::Dynamic { schemas, .. } => {
5440 for schema in schemas {
5441 validate_schema(schema)?;
5442 }
5443 }
5444 }
5445 match variants {
5446 ResidentSchemaVariants::Fixed(schema) => {
5447 let schema_id = resident_intern_schema(
5448 head_schema_choices
5449 .get_mut(&rule.head)
5450 .expect("compiled head choice exists"),
5451 schema,
5452 )
5453 .map_err(|error| resident_workspace_decline(error.to_string()))?;
5454 rule_schema_ids[scc_index][rule_index] = Some(schema_id);
5455 }
5456 ResidentSchemaVariants::Dynamic {
5457 source_head,
5458 schemas,
5459 } => {
5460 resident_register_schema_selection(
5461 &rule.head,
5462 ResidentHeadSchemaSelection {
5463 source_head,
5464 output_schemas_by_source_winner: schemas.clone(),
5465 },
5466 &head_schema_choices,
5467 &mut head_schema_selections,
5468 )?;
5469 for schema in schemas {
5470 resident_intern_schema(
5471 head_schema_choices
5472 .get_mut(&rule.head)
5473 .expect("compiled head choice exists"),
5474 schema,
5475 )
5476 .map_err(|error| resident_workspace_decline(error.to_string()))?;
5477 }
5478 rule_schema_ids[scc_index][rule_index] = Some(RESIDENT_DYNAMIC_SCHEMA_ID);
5479 }
5480 }
5481 }
5482 }
5483
5484 let mut seen_query_heads = HashSet::new();
5485 for (scc_index, rules) in plan.rules_by_scc.iter().enumerate() {
5486 for (rule_index, rule) in rules.iter().enumerate() {
5487 if !generated_query_heads.contains(&rule.head) {
5488 continue;
5489 }
5490 if !seen_query_heads.insert(rule.head.clone()) {
5491 return Err(resident_workspace_decline(format!(
5492 "synthetic query head {} has more than one compiled rule",
5493 rule.head
5494 )));
5495 }
5496 let variants =
5497 self.resident_schema_variants(&rule.body, certificate, &head_schema_choices)?;
5498 let physical = &head_schemas[&rule.head];
5499 match variants {
5500 ResidentSchemaVariants::Fixed(schema) => {
5501 if !resident_schemas_type_compatible(&schema, physical) {
5502 return Err(resident_workspace_decline(format!(
5503 "synthetic query head {} is physically incompatible with its compiled head",
5504 rule.head
5505 )));
5506 }
5507 let schema_id = resident_intern_schema(
5508 head_schema_choices
5509 .get_mut(&rule.head)
5510 .expect("compiled query head choice exists"),
5511 schema,
5512 )
5513 .map_err(|error| resident_workspace_decline(error.to_string()))?;
5514 rule_schema_ids[scc_index][rule_index] = Some(schema_id);
5515 }
5516 ResidentSchemaVariants::Dynamic {
5517 source_head,
5518 schemas,
5519 } => {
5520 if schemas
5521 .iter()
5522 .any(|schema| !resident_schemas_type_compatible(schema, physical))
5523 {
5524 return Err(resident_workspace_decline(format!(
5525 "synthetic query head {} has a physically incompatible dynamic schema",
5526 rule.head
5527 )));
5528 }
5529 for schema in &schemas {
5530 resident_intern_schema(
5531 head_schema_choices
5532 .get_mut(&rule.head)
5533 .expect("compiled query head choice exists"),
5534 schema.clone(),
5535 )
5536 .map_err(|error| resident_workspace_decline(error.to_string()))?;
5537 }
5538 rule_schema_ids[scc_index][rule_index] = Some(RESIDENT_DYNAMIC_SCHEMA_ID);
5539 resident_register_schema_selection(
5540 &rule.head,
5541 ResidentHeadSchemaSelection {
5542 source_head,
5543 output_schemas_by_source_winner: schemas,
5544 },
5545 &head_schema_choices,
5546 &mut head_schema_selections,
5547 )?;
5548 }
5549 }
5550 }
5551 }
5552
5553 for (name, choices) in &head_schema_choices {
5554 if choices.is_empty() {
5555 return Err(resident_workspace_decline(format!(
5556 "resident head {name} has no schema candidate"
5557 )));
5558 }
5559 }
5560 Ok(ResidentWorkspaceAdmission {
5561 relation_capacity,
5562 head_schemas,
5563 head_schema_choices,
5564 rule_schema_ids,
5565 head_schema_selections,
5566 })
5567 }
5568
5569 fn resident_required_row_capacity(
5570 &self,
5571 plan: &ExecutionPlan,
5572 certificate: &ResidentGraphRouteCertificate,
5573 ) -> std::result::Result<(u64, String), ResidentGraphDeclineReason> {
5574 let mut relation_domains = HashMap::<RelId, ResidentColumnDomains>::new();
5575 let mut synthetic_domains = HashMap::<String, ResidentColumnDomains>::new();
5576 let mut source_rows = HashMap::<RelId, u64>::new();
5577 let plan_relations =
5578 resident_plan_relation_ids(plan.rules_by_scc.iter().flatten(), &self.name_to_rel);
5579 for (relation, name) in &self.rel_names {
5580 if !plan_relations.contains(relation) {
5581 continue;
5582 }
5583 let schema = certificate.schema_for(*relation).ok_or_else(|| {
5584 resident_workspace_decline(format!(
5585 "relation {name} has no schema in the resident certificate"
5586 ))
5587 })?;
5588 let mut columns = vec![BTreeMap::new(); schema.arity()];
5589 let rows = self.store.get(name).map(CudaBuffer::num_rows).unwrap_or(0);
5590 if rows != 0 {
5591 for (column, domain) in columns.iter_mut().enumerate() {
5592 domain.insert(format!("source:{}:{column}", relation.0), rows);
5593 }
5594 }
5595 relation_domains.insert(*relation, columns);
5596 source_rows.insert(*relation, rows);
5597 }
5598
5599 loop {
5600 let mut changed = false;
5601 for rules in &plan.rules_by_scc {
5602 for rule in rules {
5603 let contribution = resident_node_domains(&rule.body, &relation_domains)?;
5604 if contribution.len() != rule.meta.schema.arity() {
5605 return Err(resident_workspace_decline(format!(
5606 "resident rule head {} has {} certified domains for schema arity {}",
5607 rule.head,
5608 contribution.len(),
5609 rule.meta.schema.arity()
5610 )));
5611 }
5612 let target = if let Some(head) = self.name_to_rel.get(&rule.head) {
5613 relation_domains.get_mut(head).ok_or_else(|| {
5614 resident_workspace_decline(format!(
5615 "resident rule head {} has no domain slot",
5616 rule.head
5617 ))
5618 })?
5619 } else {
5620 synthetic_domains
5621 .entry(rule.head.clone())
5622 .or_insert_with(|| vec![BTreeMap::new(); rule.meta.schema.arity()])
5623 };
5624 changed |= merge_resident_domains(&rule.head, target, contribution)?;
5625 }
5626 }
5627 if !changed {
5628 break;
5629 }
5630 }
5631
5632 let mut relation_set_bounds = source_rows.clone();
5633 let mut synthetic_set_bounds = HashMap::<String, u64>::new();
5634 let mut required = source_rows.values().copied().max().unwrap_or(1).max(1);
5635 let mut witness = source_rows
5636 .iter()
5637 .max_by_key(|(_, rows)| *rows)
5638 .map(|(relation, rows)| {
5639 let name = self
5640 .rel_names
5641 .get(relation)
5642 .map(String::as_str)
5643 .unwrap_or("<unregistered>");
5644 format!("source relation {name} ({relation:?}) rows={rows}")
5645 })
5646 .unwrap_or_else(|| "unit relation bound=1".to_string());
5647
5648 let mut scc_by_id = HashMap::new();
5649 for (index, scc) in plan.sccs.iter().enumerate() {
5650 if scc_by_id.insert(scc.id, index).is_some() {
5651 return Err(resident_workspace_decline(format!(
5652 "resident row proof found duplicate SCC id {}",
5653 scc.id
5654 )));
5655 }
5656 }
5657 let ordered_sccs = if plan.strata.is_empty() {
5658 (0..plan.sccs.len()).collect::<Vec<_>>()
5659 } else {
5660 let mut ordered = Vec::new();
5661 for stratum in &plan.strata {
5662 for scc_id in &stratum.sccs {
5663 ordered.push(*scc_by_id.get(scc_id).ok_or_else(|| {
5664 resident_workspace_decline(format!(
5665 "resident row proof stratum {} references unknown SCC {scc_id}",
5666 stratum.id
5667 ))
5668 })?);
5669 }
5670 }
5671 ordered
5672 };
5673 let mut seen_sccs = HashSet::new();
5674 for scc_index in ordered_sccs {
5675 if !seen_sccs.insert(scc_index) {
5676 return Err(resident_workspace_decline(format!(
5677 "resident row proof schedule repeats SCC {}",
5678 plan.sccs[scc_index].id
5679 )));
5680 }
5681 let scc = &plan.sccs[scc_index];
5682 let rules = plan.rules_by_scc.get(scc_index).ok_or_else(|| {
5683 resident_workspace_decline(format!(
5684 "resident row proof SCC {scc_index} has no rule vector"
5685 ))
5686 })?;
5687
5688 if scc.is_recursive {
5689 for name in &scc.predicates {
5690 let relation = self.name_to_rel.get(name).ok_or_else(|| {
5691 resident_workspace_decline(format!(
5692 "recursive relation {name} has no registered relation id"
5693 ))
5694 })?;
5695 let columns = relation_domains.get(relation).ok_or_else(|| {
5696 resident_workspace_decline(format!(
5697 "recursive relation {relation:?} has no active-domain proof"
5698 ))
5699 })?;
5700 let context = format!("recursive relation {name} ({relation:?})");
5701 let finite_cap = resident_domain_product(columns, &context)?;
5702 relation_set_bounds.insert(*relation, finite_cap);
5703 if finite_cap > required {
5704 required = finite_cap;
5705 witness = format!(
5706 "{context} finite_domain_cap={finite_cap} domains={}",
5707 resident_domain_description(columns)
5708 );
5709 }
5710 }
5711 }
5712
5713 let mut additions = BTreeMap::<String, u64>::new();
5714 let mut raw_additions = BTreeMap::<String, u64>::new();
5715 let mut addition_witnesses = BTreeMap::<String, String>::new();
5716 for (rule_index, rule) in rules.iter().enumerate() {
5717 let path = format!("scc[{scc_index}].rule[{rule_index}] head={}", rule.head);
5718 let proof = resident_node_row_bound(
5719 &rule.body,
5720 &source_rows,
5721 &relation_set_bounds,
5722 &relation_domains,
5723 &path,
5724 )?;
5725 let rows = proof.rows;
5726 let transient = proof.peak;
5727 let rule_domains = resident_node_domains(&rule.body, &relation_domains)?;
5728 let set_rows = resident_domain_product_capped(
5729 &rule_domains,
5730 rows,
5731 &format!("{path} projected set"),
5732 )?;
5733 if transient > required {
5734 required = transient;
5735 witness = format!(
5736 "{path} transient_bound={transient} proof={} body={:?}",
5737 proof.peak_detail, rule.body
5738 );
5739 }
5740 let total = additions.entry(rule.head.clone()).or_default();
5741 *total = total.checked_add(set_rows).ok_or_else(|| {
5742 resident_workspace_decline(format!(
5743 "{path} rule-output addition overflow: partial={total} addition={set_rows} raw_rows={rows}"
5744 ))
5745 })?;
5746 let raw_total = raw_additions.entry(rule.head.clone()).or_default();
5747 *raw_total = raw_total.checked_add(rows).ok_or_else(|| {
5748 resident_workspace_decline(format!(
5749 "{path} raw rule-output addition overflow: partial={raw_total} addition={rows}"
5750 ))
5751 })?;
5752 addition_witnesses.insert(
5753 rule.head.clone(),
5754 format!(
5755 "{path} body={:?} raw_rows={rows} per_clause_set_cap={set_rows} domains={}",
5756 rule.body,
5757 resident_domain_description(&rule_domains)
5758 ),
5759 );
5760 }
5761
5762 for (head, addition) in additions {
5763 let (current, relation) = if let Some(relation) = self.name_to_rel.get(&head) {
5764 (
5765 relation_set_bounds.get(relation).copied().unwrap_or(0),
5766 Some(*relation),
5767 )
5768 } else {
5769 (synthetic_set_bounds.get(&head).copied().unwrap_or(0), None)
5770 };
5771 let path = addition_witnesses
5772 .get(&head)
5773 .map(String::as_str)
5774 .unwrap_or("unknown rule");
5775 let raw_addition = raw_additions.get(&head).copied().unwrap_or(0);
5776 let raw_candidate = current.checked_add(raw_addition).ok_or_else(|| {
5777 resident_workspace_decline(format!(
5778 "{path} staged candidate addition overflow: prior={current} raw_rule_rows={raw_addition}"
5779 ))
5780 })?;
5781 if raw_candidate > required {
5782 required = raw_candidate;
5783 witness = format!(
5784 "{path} staged_candidate_bound={raw_candidate} prior={current} raw_rule_rows={raw_addition}"
5785 );
5786 }
5787 if scc.is_recursive {
5788 let candidate = current.checked_add(addition).ok_or_else(|| {
5789 resident_workspace_decline(format!(
5790 "{path} recursive set candidate addition overflow: finite_cap={current} set_rule_rows={addition}"
5791 ))
5792 })?;
5793 if candidate > required {
5794 required = candidate;
5795 witness = format!(
5796 "{path} recursive_candidate_bound={candidate} finite_cap={current} raw_rule_rows={addition}"
5797 );
5798 }
5799 } else {
5800 let summed_bound = current.checked_add(addition).ok_or_else(|| {
5801 resident_workspace_decline(format!(
5802 "{path} nonrecursive set addition overflow: prior={current} rule_rows={addition}"
5803 ))
5804 })?;
5805 let columns = if let Some(relation) = relation {
5806 relation_domains.get(&relation)
5807 } else {
5808 synthetic_domains.get(&head)
5809 }
5810 .ok_or_else(|| {
5811 resident_workspace_decline(format!(
5812 "{path} nonrecursive head {head} has no active-domain proof"
5813 ))
5814 })?;
5815 let set_bound = resident_domain_product_capped(
5816 columns,
5817 summed_bound,
5818 &format!("{path} merged set {head}"),
5819 )?;
5820 if let Some(relation) = relation {
5821 relation_set_bounds.insert(relation, set_bound);
5822 } else {
5823 synthetic_set_bounds.insert(head.clone(), set_bound);
5824 }
5825 if set_bound > required {
5826 required = set_bound;
5827 witness = format!(
5828 "{path} nonrecursive_set_bound={set_bound} prior={current} rule_rows={addition}"
5829 );
5830 }
5831 }
5832 }
5833 }
5834 if seen_sccs.len() != plan.sccs.len() {
5835 return Err(resident_workspace_decline(
5836 "resident row proof schedule omits one or more SCCs",
5837 ));
5838 }
5839 Ok((required, witness))
5840 }
5841}
5842
5843type ResidentDomain = BTreeMap<String, u64>;
5844type ResidentColumnDomains = Vec<ResidentDomain>;
5845
5846fn resident_plan_relation_ids<'a>(
5847 rules: impl Iterator<Item = &'a CompiledRule>,
5848 name_to_rel: &HashMap<String, RelId>,
5849) -> HashSet<RelId> {
5850 let mut relations = HashSet::new();
5851 for rule in rules {
5852 relations.extend(rule.body.referenced_relations());
5853 if let Some(head) = name_to_rel.get(&rule.head) {
5854 relations.insert(*head);
5855 }
5856 }
5857 relations
5858}
5859
5860fn resident_workspace_decline(detail: impl Into<String>) -> ResidentGraphDeclineReason {
5861 ResidentGraphDeclineReason::WorkspaceUnbounded {
5862 detail: detail.into(),
5863 }
5864}
5865
5866fn merge_resident_domains(
5867 relation: &str,
5868 target: &mut ResidentColumnDomains,
5869 contribution: ResidentColumnDomains,
5870) -> std::result::Result<bool, ResidentGraphDeclineReason> {
5871 if target.len() != contribution.len() {
5872 return Err(resident_workspace_decline(format!(
5873 "resident rule head {relation} domain arity changed"
5874 )));
5875 }
5876 let mut changed = false;
5877 for (target_domain, contribution_domain) in target.iter_mut().zip(contribution) {
5878 for (source, bound) in contribution_domain {
5879 match target_domain.entry(source) {
5880 std::collections::btree_map::Entry::Vacant(entry) => {
5881 entry.insert(bound);
5882 changed = true;
5883 }
5884 std::collections::btree_map::Entry::Occupied(mut entry) if *entry.get() < bound => {
5885 entry.insert(bound);
5886 changed = true;
5887 }
5888 std::collections::btree_map::Entry::Occupied(_) => {}
5889 }
5890 }
5891 }
5892 Ok(changed)
5893}
5894
5895fn resident_node_domains(
5896 node: &RirNode,
5897 relations: &HashMap<RelId, ResidentColumnDomains>,
5898) -> std::result::Result<ResidentColumnDomains, ResidentGraphDeclineReason> {
5899 match node {
5900 RirNode::Unit => Ok(Vec::new()),
5901 RirNode::Scan { rel } => relations.get(rel).cloned().ok_or_else(|| {
5902 resident_workspace_decline(format!(
5903 "resident scan {rel:?} has no source-domain certificate"
5904 ))
5905 }),
5906 RirNode::Filter { input, predicate } => {
5907 let mut domains = resident_node_domains(input, relations)?;
5908 resident_refine_filter_domains(&mut domains, predicate)?;
5909 Ok(domains)
5910 }
5911 RirNode::Distinct { input, .. } => resident_node_domains(input, relations),
5912 RirNode::Project { input, columns } => {
5913 let input_domains = resident_node_domains(input, relations)?;
5914 resident_project_domains(&input_domains, columns)
5915 }
5916 RirNode::Join {
5917 left,
5918 right,
5919 join_type,
5920 ..
5921 } => {
5922 let mut left_domains = resident_node_domains(left, relations)?;
5923 match join_type {
5924 JoinType::Inner => {
5925 left_domains.extend(resident_node_domains(right, relations)?);
5926 Ok(left_domains)
5927 }
5928 JoinType::Semi => Ok(left_domains),
5929 other => Err(resident_workspace_decline(format!(
5930 "resident domain proof does not support {other:?} joins"
5931 ))),
5932 }
5933 }
5934 RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
5935 resident_node_domains(fallback, relations)
5936 }
5937 RirNode::Union { inputs } => {
5938 let mut merged: Option<ResidentColumnDomains> = None;
5939 for input in inputs {
5940 let contribution = resident_node_domains(input, relations)?;
5941 if let Some(domains) = merged.as_mut() {
5942 if domains.len() != contribution.len() {
5943 return Err(resident_workspace_decline(
5944 "resident union domain arity mismatch",
5945 ));
5946 }
5947 for (domain, addition) in domains.iter_mut().zip(contribution) {
5948 for (source, bound) in addition {
5949 domain
5950 .entry(source)
5951 .and_modify(|current| *current = (*current).max(bound))
5952 .or_insert(bound);
5953 }
5954 }
5955 } else {
5956 merged = Some(contribution);
5957 }
5958 }
5959 merged.ok_or_else(|| resident_workspace_decline("resident union has no inputs"))
5960 }
5961 RirNode::Diff { left, .. } => resident_node_domains(left, relations),
5962 RirNode::GroupBy { .. } | RirNode::Fixpoint { .. } | RirNode::TensorMaskedJoin { .. } => {
5963 Err(resident_workspace_decline(
5964 "resident domain proof encountered an uncertified operator",
5965 ))
5966 }
5967 }
5968}
5969
5970fn resident_refine_filter_domains(
5971 domains: &mut ResidentColumnDomains,
5972 predicate: &Expr,
5973) -> std::result::Result<(), ResidentGraphDeclineReason> {
5974 match predicate {
5975 Expr::And(parts) => {
5976 for part in parts {
5977 resident_refine_filter_domains(domains, part)?;
5978 }
5979 }
5980 Expr::Compare {
5981 left,
5982 op: RirCompareOp::Eq,
5983 right,
5984 } => {
5985 let (column, value) = match (&**left, &**right) {
5986 (Expr::Column(column), Expr::Const(value))
5987 | (Expr::Const(value), Expr::Column(column)) => (*column, value),
5988 _ => return Ok(()),
5989 };
5990 let arity = domains.len();
5991 let domain = domains.get_mut(column).ok_or_else(|| {
5992 resident_workspace_decline(format!(
5993 "resident equality filter column {column} exceeds input arity {}",
5994 arity
5995 ))
5996 })?;
5997 *domain = BTreeMap::from([(format!("constant:{value:?}"), 1)]);
5998 }
5999 _ => {}
6000 }
6001 Ok(())
6002}
6003
6004fn resident_project_domains(
6005 input: &[ResidentDomain],
6006 columns: &[ProjectExpr],
6007) -> std::result::Result<ResidentColumnDomains, ResidentGraphDeclineReason> {
6008 columns
6009 .iter()
6010 .map(|column| match column {
6011 ProjectExpr::Column(index) => input.get(*index).cloned().ok_or_else(|| {
6012 resident_workspace_decline(format!(
6013 "resident projection column {index} exceeds input arity {}",
6014 input.len()
6015 ))
6016 }),
6017 ProjectExpr::Computed(Expr::Const(value), scalar) => Ok(BTreeMap::from([(
6018 format!("constant:{scalar:?}:{value:?}"),
6019 1,
6020 )])),
6021 _ => Err(resident_workspace_decline(
6022 "resident projection domain is not a column or constant",
6023 )),
6024 })
6025 .collect()
6026}
6027
6028fn resident_domain_product(
6029 columns: &[ResidentDomain],
6030 context: &str,
6031) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6032 if columns.is_empty() {
6033 return Ok(1);
6034 }
6035 columns
6036 .iter()
6037 .enumerate()
6038 .try_fold(1u64, |product, (column_index, domain)| {
6039 let values = domain.values().try_fold(0u64, |total, bound| {
6040 total.checked_add(*bound).ok_or_else(|| {
6041 resident_workspace_decline(format!(
6042 "{context} active-domain sum overflow at column {column_index}: partial={total} addition={bound} lineage={domain:?}"
6043 ))
6044 })
6045 })?;
6046 product.checked_mul(values).ok_or_else(|| {
6047 resident_workspace_decline(format!(
6048 "{context} active-domain product overflow at column {column_index}: partial={product} factor={values} lineage={domain:?}"
6049 ))
6050 })
6051 })
6052}
6053
6054fn resident_domain_product_capped(
6055 columns: &[ResidentDomain],
6056 cap: u64,
6057 context: &str,
6058) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6059 if cap == 0 {
6060 return Ok(0);
6061 }
6062 if columns.is_empty() {
6063 return Ok(1);
6064 }
6065 let mut product = 1u64;
6066 for (column_index, domain) in columns.iter().enumerate() {
6067 let mut values = 0u64;
6068 for bound in domain.values() {
6069 let addition = (*bound).min(cap.saturating_sub(values));
6070 values = values.checked_add(addition).ok_or_else(|| {
6071 resident_workspace_decline(format!(
6072 "{context} capped active-domain sum overflow at column {column_index}: partial={values} addition={addition}"
6073 ))
6074 })?;
6075 if values == cap {
6076 break;
6077 }
6078 }
6079 if values == 0 {
6080 return Ok(0);
6081 }
6082 if product > cap / values {
6083 return Ok(cap);
6084 }
6085 product = product.checked_mul(values).ok_or_else(|| {
6086 resident_workspace_decline(format!(
6087 "{context} capped active-domain product overflow at column {column_index}: partial={product} factor={values}"
6088 ))
6089 })?;
6090 }
6091 Ok(product.min(cap))
6092}
6093
6094fn resident_domain_description(columns: &[ResidentDomain]) -> String {
6095 columns
6096 .iter()
6097 .enumerate()
6098 .map(|(column, domain)| {
6099 let sum = domain.values().copied().fold(0u64, u64::saturating_add);
6100 format!("column[{column}] sum={sum} lineage={domain:?}")
6101 })
6102 .collect::<Vec<_>>()
6103 .join("; ")
6104}
6105
6106#[derive(Clone, Debug, PartialEq, Eq)]
6107enum ResidentCapacityBound {
6108 Finite { rows: u64, proof: String },
6109 AboveResidentLimit { proof: String },
6110}
6111
6112fn resident_capacity_product_bound(proof: &'static str, factors: &[u64]) -> ResidentCapacityBound {
6113 if factors.contains(&0) {
6114 return ResidentCapacityBound::Finite {
6115 rows: 0,
6116 proof: proof.to_owned(),
6117 };
6118 }
6119
6120 let limit = u64::from(MAX_RESIDENT_CAPACITY);
6121 let mut rows = 1u64;
6122 for factor in factors {
6123 if rows > limit / factor {
6124 return ResidentCapacityBound::AboveResidentLimit {
6125 proof: format!("{proof} factors={factors:?}"),
6126 };
6127 }
6128 rows *= factor;
6129 }
6130 ResidentCapacityBound::Finite {
6131 rows,
6132 proof: proof.to_owned(),
6133 }
6134}
6135
6136fn resident_join_capacity_bound(
6137 left_rows: u64,
6138 right_rows: u64,
6139 left_fanout: u64,
6140 right_fanout: u64,
6141 matching_key_values: u64,
6142) -> ResidentCapacityBound {
6143 let candidates = [
6144 resident_capacity_product_bound("left_rows*right_rows", &[left_rows, right_rows]),
6145 resident_capacity_product_bound("left_rows*right_fanout", &[left_rows, right_fanout]),
6146 resident_capacity_product_bound("right_rows*left_fanout", &[right_rows, left_fanout]),
6147 resident_capacity_product_bound(
6148 "matching_keys*left_fanout*right_fanout",
6149 &[matching_key_values, left_fanout, right_fanout],
6150 ),
6151 ];
6152
6153 let mut tightest: Option<(u64, String)> = None;
6154 let mut above = Vec::new();
6155 for candidate in candidates {
6156 match candidate {
6157 ResidentCapacityBound::Finite { rows, proof } => {
6158 if tightest.as_ref().is_none_or(|(current, _)| rows < *current) {
6159 tightest = Some((rows, proof));
6160 }
6161 }
6162 ResidentCapacityBound::AboveResidentLimit { proof } => above.push(proof),
6163 }
6164 }
6165 if let Some((rows, proof)) = tightest {
6166 ResidentCapacityBound::Finite { rows, proof }
6167 } else {
6168 ResidentCapacityBound::AboveResidentLimit {
6169 proof: above.join("; "),
6170 }
6171 }
6172}
6173
6174struct ResidentRowProof {
6175 rows: u64,
6176 peak: u64,
6177 domains: ResidentColumnDomains,
6178 full_row_unique: bool,
6179 peak_detail: String,
6180}
6181
6182fn resident_node_row_bound(
6183 node: &RirNode,
6184 source_rows: &HashMap<RelId, u64>,
6185 relation_set_bounds: &HashMap<RelId, u64>,
6186 relation_domains: &HashMap<RelId, ResidentColumnDomains>,
6187 path: &str,
6188) -> std::result::Result<ResidentRowProof, ResidentGraphDeclineReason> {
6189 match node {
6190 RirNode::Unit => Ok(ResidentRowProof {
6191 rows: 1,
6192 peak: 1,
6193 domains: Vec::new(),
6194 full_row_unique: true,
6195 peak_detail: format!("{path}.unit rows=1"),
6196 }),
6197 RirNode::Scan { rel } => {
6198 let rows = source_rows
6199 .get(rel)
6200 .copied()
6201 .unwrap_or(0)
6202 .max(relation_set_bounds.get(rel).copied().unwrap_or(0));
6203 let domains = relation_domains.get(rel).cloned().ok_or_else(|| {
6204 resident_workspace_decline(format!(
6205 "{path}.scan {rel:?} has no active-domain proof"
6206 ))
6207 })?;
6208 Ok(ResidentRowProof {
6209 rows,
6210 peak: rows,
6211 domains,
6212 full_row_unique: true,
6213 peak_detail: format!("{path}.scan rel={rel:?} rows={rows}"),
6214 })
6215 }
6216 RirNode::Filter { input, predicate } => {
6217 let mut proof = resident_node_row_bound(
6218 input,
6219 source_rows,
6220 relation_set_bounds,
6221 relation_domains,
6222 &format!("{path}.filter.input"),
6223 )?;
6224 resident_refine_filter_domains(&mut proof.domains, predicate)?;
6225 if proof.full_row_unique {
6226 proof.rows = resident_domain_product_capped(
6227 &proof.domains,
6228 proof.rows,
6229 &format!("{path}.filter"),
6230 )?;
6231 }
6232 Ok(proof)
6233 }
6234 RirNode::Project { input, columns } => {
6235 let input = resident_node_row_bound(
6236 input,
6237 source_rows,
6238 relation_set_bounds,
6239 relation_domains,
6240 &format!("{path}.project.input"),
6241 )?;
6242 let domains = resident_project_domains(&input.domains, columns)?;
6243 let full_row_unique = input.full_row_unique
6244 && resident_projection_is_injective(&input.domains, columns, path)?;
6245 Ok(ResidentRowProof {
6246 rows: input.rows,
6247 peak: input.peak.max(input.rows),
6248 domains,
6249 full_row_unique,
6250 peak_detail: input.peak_detail,
6251 })
6252 }
6253 RirNode::Distinct { input, .. } => {
6254 let mut proof = resident_node_row_bound(
6255 input,
6256 source_rows,
6257 relation_set_bounds,
6258 relation_domains,
6259 &format!("{path}.distinct.input"),
6260 )?;
6261 proof.rows = resident_domain_product_capped(
6262 &proof.domains,
6263 proof.rows,
6264 &format!("{path}.distinct"),
6265 )?;
6266 proof.full_row_unique = true;
6267 Ok(proof)
6268 }
6269 RirNode::Join {
6270 left,
6271 right,
6272 left_keys,
6273 right_keys,
6274 join_type,
6275 } => {
6276 let left = resident_node_row_bound(
6277 left,
6278 source_rows,
6279 relation_set_bounds,
6280 relation_domains,
6281 &format!("{path}.join.left"),
6282 )?;
6283 let right = resident_node_row_bound(
6284 right,
6285 source_rows,
6286 relation_set_bounds,
6287 relation_domains,
6288 &format!("{path}.join.right"),
6289 )?;
6290 match join_type {
6291 JoinType::Semi => Ok(ResidentRowProof {
6292 rows: left.rows,
6293 peak: left.peak.max(right.peak).max(left.rows),
6294 domains: left.domains,
6295 full_row_unique: left.full_row_unique,
6296 peak_detail: if left.peak >= right.peak {
6297 left.peak_detail
6298 } else {
6299 right.peak_detail
6300 },
6301 }),
6302 JoinType::Inner => {
6303 if left_keys.len() != 1 || right_keys.len() != 1 {
6304 return Err(resident_workspace_decline(format!(
6305 "{path} resident row proof requires exactly one join key per side"
6306 )));
6307 }
6308 let left_key = left_keys[0];
6309 let right_key = right_keys[0];
6310 let left_fanout = resident_key_fanout_bound(
6311 &left.domains,
6312 left_key,
6313 left.rows,
6314 left.full_row_unique,
6315 &format!("{path}.join.left"),
6316 )?;
6317 let right_fanout = resident_key_fanout_bound(
6318 &right.domains,
6319 right_key,
6320 right.rows,
6321 right.full_row_unique,
6322 &format!("{path}.join.right"),
6323 )?;
6324 let left_key_values = resident_domain_cardinality_capped(
6325 left.domains.get(left_key).ok_or_else(|| {
6326 resident_workspace_decline(format!(
6327 "{path} left join key {left_key} exceeds arity {}",
6328 left.domains.len()
6329 ))
6330 })?,
6331 left.rows,
6332 &format!("{path}.join.left_key"),
6333 )?;
6334 let right_key_values = resident_domain_cardinality_capped(
6335 right.domains.get(right_key).ok_or_else(|| {
6336 resident_workspace_decline(format!(
6337 "{path} right join key {right_key} exceeds arity {}",
6338 right.domains.len()
6339 ))
6340 })?,
6341 right.rows,
6342 &format!("{path}.join.right_key"),
6343 )?;
6344 let matching_key_values = left_key_values.min(right_key_values);
6345 let (rows, bound_proof) = match resident_join_capacity_bound(
6346 left.rows,
6347 right.rows,
6348 left_fanout,
6349 right_fanout,
6350 matching_key_values,
6351 ) {
6352 ResidentCapacityBound::Finite { rows, proof } => (rows, proof),
6353 ResidentCapacityBound::AboveResidentLimit { proof } => {
6354 return Err(resident_workspace_decline(format!(
6355 "{path} inner join exceeds the fixed resident capacity limit: left_rows={} right_rows={} left_fanout={left_fanout} right_fanout={right_fanout} matching_keys={matching_key_values} products={proof}",
6356 left.rows, right.rows
6357 )));
6358 }
6359 };
6360 let join_detail = format!(
6361 "{path}.join left_rows={} right_rows={} left_fanout={left_fanout} right_fanout={right_fanout} matching_keys={matching_key_values} bound={rows} proof={bound_proof}",
6362 left.rows, right.rows
6363 );
6364 let (peak, peak_detail) = if rows >= left.peak && rows >= right.peak {
6365 (rows, join_detail)
6366 } else if left.peak >= right.peak {
6367 (left.peak, left.peak_detail)
6368 } else {
6369 (right.peak, right.peak_detail)
6370 };
6371 let mut domains = left.domains;
6372 domains.extend(right.domains);
6373 Ok(ResidentRowProof {
6374 rows,
6375 peak,
6376 domains,
6377 full_row_unique: left.full_row_unique && right.full_row_unique,
6378 peak_detail,
6379 })
6380 }
6381 other => Err(resident_workspace_decline(format!(
6382 "resident row proof does not support {other:?} joins"
6383 ))),
6384 }
6385 }
6386 RirNode::ChainJoin { fallback, .. } | RirNode::MultiWayJoin { fallback, .. } => {
6387 resident_node_row_bound(
6388 fallback,
6389 source_rows,
6390 relation_set_bounds,
6391 relation_domains,
6392 &format!("{path}.fallback"),
6393 )
6394 }
6395 RirNode::Union { inputs } => {
6396 let mut rows = 0u64;
6397 let mut peak = 0u64;
6398 let mut peak_detail = format!("{path}.union empty");
6399 let mut domains: Option<ResidentColumnDomains> = None;
6400 for (index, input) in inputs.iter().enumerate() {
6401 let proof = resident_node_row_bound(
6402 input,
6403 source_rows,
6404 relation_set_bounds,
6405 relation_domains,
6406 &format!("{path}.union[{index}]"),
6407 )?;
6408 rows = rows.checked_add(proof.rows).ok_or_else(|| {
6409 resident_workspace_decline(format!(
6410 "{path} union addition overflow: partial={rows} addition={}",
6411 proof.rows
6412 ))
6413 })?;
6414 if proof.peak > peak {
6415 peak = proof.peak;
6416 peak_detail = proof.peak_detail;
6417 }
6418 if let Some(merged) = domains.as_mut() {
6419 merge_domain_columns(merged, proof.domains, path)?;
6420 } else {
6421 domains = Some(proof.domains);
6422 }
6423 }
6424 if rows > peak {
6425 peak = rows;
6426 peak_detail = format!("{path}.union raw_rows={rows}");
6427 }
6428 Ok(ResidentRowProof {
6429 rows,
6430 peak,
6431 domains: domains
6432 .ok_or_else(|| resident_workspace_decline("resident union has no inputs"))?,
6433 full_row_unique: false,
6434 peak_detail,
6435 })
6436 }
6437 RirNode::Diff { left, right } => {
6438 let left = resident_node_row_bound(
6439 left,
6440 source_rows,
6441 relation_set_bounds,
6442 relation_domains,
6443 &format!("{path}.diff.left"),
6444 )?;
6445 let right = resident_node_row_bound(
6446 right,
6447 source_rows,
6448 relation_set_bounds,
6449 relation_domains,
6450 &format!("{path}.diff.right"),
6451 )?;
6452 let (peak, peak_detail) = if left.peak >= right.peak {
6453 (left.peak.max(left.rows), left.peak_detail)
6454 } else {
6455 (right.peak.max(left.rows), right.peak_detail)
6456 };
6457 Ok(ResidentRowProof {
6458 rows: left.rows,
6459 peak,
6460 domains: left.domains,
6461 full_row_unique: left.full_row_unique,
6462 peak_detail,
6463 })
6464 }
6465 RirNode::GroupBy { .. } | RirNode::Fixpoint { .. } | RirNode::TensorMaskedJoin { .. } => {
6466 Err(resident_workspace_decline(
6467 "resident row proof encountered an uncertified operator",
6468 ))
6469 }
6470 }
6471}
6472
6473fn resident_projection_is_injective(
6474 input_domains: &[ResidentDomain],
6475 columns: &[ProjectExpr],
6476 path: &str,
6477) -> std::result::Result<bool, ResidentGraphDeclineReason> {
6478 let mut retained = vec![false; input_domains.len()];
6479 for column in columns {
6480 if let ProjectExpr::Column(index) = column {
6481 let slot = retained.get_mut(*index).ok_or_else(|| {
6482 resident_workspace_decline(format!(
6483 "{path} projection column {index} exceeds input arity {}",
6484 input_domains.len()
6485 ))
6486 })?;
6487 *slot = true;
6488 }
6489 }
6490 for (index, (retained, domain)) in retained.iter().zip(input_domains).enumerate() {
6491 if !retained
6492 && resident_domain_cardinality_capped(
6493 domain,
6494 2,
6495 &format!("{path}.project.omitted[{index}]"),
6496 )? > 1
6497 {
6498 return Ok(false);
6499 }
6500 }
6501 Ok(true)
6502}
6503
6504fn resident_domain_cardinality_capped(
6505 domain: &ResidentDomain,
6506 cap: u64,
6507 context: &str,
6508) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6509 let mut total = 0u64;
6510 for bound in domain.values() {
6511 let addition = (*bound).min(cap.saturating_sub(total));
6512 total = total.checked_add(addition).ok_or_else(|| {
6513 resident_workspace_decline(format!(
6514 "{context} domain-cardinality addition overflow: partial={total} addition={addition}"
6515 ))
6516 })?;
6517 if total == cap {
6518 break;
6519 }
6520 }
6521 Ok(total)
6522}
6523
6524fn resident_key_fanout_bound(
6525 domains: &[ResidentDomain],
6526 key: usize,
6527 rows: u64,
6528 full_row_unique: bool,
6529 context: &str,
6530) -> std::result::Result<u64, ResidentGraphDeclineReason> {
6531 if key >= domains.len() {
6532 return Err(resident_workspace_decline(format!(
6533 "{context} join key {key} exceeds input arity {}",
6534 domains.len()
6535 )));
6536 }
6537 if !full_row_unique {
6538 return Ok(rows);
6539 }
6540 let non_key = domains
6541 .iter()
6542 .enumerate()
6543 .filter_map(|(index, domain)| (index != key).then_some(domain.clone()))
6544 .collect::<Vec<_>>();
6545 resident_domain_product_capped(&non_key, rows, &format!("{context}.fanout"))
6546}
6547
6548fn merge_domain_columns(
6549 target: &mut ResidentColumnDomains,
6550 addition: ResidentColumnDomains,
6551 path: &str,
6552) -> std::result::Result<(), ResidentGraphDeclineReason> {
6553 if target.len() != addition.len() {
6554 return Err(resident_workspace_decline(format!(
6555 "{path} union domain arity mismatch: left={} right={}",
6556 target.len(),
6557 addition.len()
6558 )));
6559 }
6560 for (target, addition) in target.iter_mut().zip(addition) {
6561 for (source, bound) in addition {
6562 target
6563 .entry(source)
6564 .and_modify(|current| *current = (*current).max(bound))
6565 .or_insert(bound);
6566 }
6567 }
6568 Ok(())
6569}
6570
6571fn checked_capacity_class(source_capacity: u32) -> Result<u32> {
6572 let capacity = source_capacity
6573 .max(1)
6574 .checked_next_power_of_two()
6575 .ok_or_else(|| XlogError::Execution("resident capacity class overflow".to_string()))?;
6576 if capacity > MAX_RESIDENT_CAPACITY {
6577 return Err(XlogError::Execution(format!(
6578 "resident capacity class {capacity} exceeds the fixed scan envelope {MAX_RESIDENT_CAPACITY}"
6579 )));
6580 }
6581 Ok(capacity)
6582}
6583
6584#[cfg(test)]
6585mod tests {
6586 use super::{
6587 checked_capacity_class, resident_compact_allocation_bytes,
6588 resident_compact_filter_descriptors, resident_compact_preflight_device_bytes,
6589 resident_compact_project_descriptors, resident_compact_regions,
6590 resident_compact_schedule_metadata_bytes, resident_compact_schema_defaults,
6591 resident_compact_tables, resident_compact_topology, resident_join_capacity_bound,
6592 resident_lower_compact_regions, resident_new_phase_unit, resident_output_indices,
6593 resident_phase_merge, resident_plan_relation_ids, resident_record_lifetimes,
6594 resident_record_scan_leaf, resident_record_schema_winner_mark, resident_record_unit_leaf,
6595 resident_register_schema_selection, resident_resolve_output_schema,
6596 resident_semantic_trace_guard, resident_source_binding_route,
6597 resident_source_logical_count, resident_source_slot_map, resident_union_fold_mode,
6598 resident_validate_conditional_body_node_kinds, resident_validate_exact_reservation,
6599 resident_validate_parent_graph_kinds, resident_validate_slot_assignments,
6600 validate_compact_resident_node_envelope, CudaGraphNodeKind, ResidentAllocationManifest,
6601 ResidentBufferRef, ResidentCapacityBound, ResidentCaptureParentKind, ResidentCapturePhase,
6602 ResidentCompactDescriptorTables, ResidentCompactLogicalRegion, ResidentFilterPlan,
6603 ResidentGraphDeclineReason, ResidentHeadSchemaSelection, ResidentLogicalRelation,
6604 ResidentOpDescriptor, ResidentOutputSchemaPlan, ResidentOutputSchemaSelection,
6605 ResidentPhaseMergeStep, ResidentPhysicalSlotPlan, ResidentProjectPlan, ResidentRecordedOp,
6606 ResidentSlotAssignment, ResidentUnionFoldMode, ScalarType, Schema,
6607 RESIDENT_DYNAMIC_SCHEMA_ID,
6608 };
6609 use std::collections::{BTreeMap, HashMap, HashSet};
6610 use xlog_core::RelId;
6611 use xlog_ir::{
6612 CompareOp, CompiledRule, ConstValue, Expr, JoinType, ProjectExpr, RirMeta, RirNode,
6613 };
6614
6615 #[test]
6616 fn resident_plan_relations_exclude_registered_but_unreferenced_relations() {
6617 let source = RelId(1);
6618 let head = RelId(2);
6619 let unrelated = RelId(3);
6620 let rules = [CompiledRule {
6621 head: "reachable".to_owned(),
6622 body: RirNode::Scan { rel: source },
6623 meta: RirMeta::default(),
6624 }];
6625 let names = HashMap::from([
6626 ("source".to_owned(), source),
6627 ("reachable".to_owned(), head),
6628 ("unrelated".to_owned(), unrelated),
6629 ]);
6630
6631 let relations = resident_plan_relation_ids(rules.iter(), &names);
6632
6633 assert_eq!(relations, HashSet::from([source, head]));
6634 }
6635
6636 #[test]
6637 fn capacity_class_is_source_bounded_and_checked() {
6638 assert_eq!(checked_capacity_class(0).unwrap(), 1);
6639 assert_eq!(checked_capacity_class(4_393).unwrap(), 8_192);
6640 assert_eq!(checked_capacity_class(65_536).unwrap(), 65_536);
6641 assert!(checked_capacity_class(65_537).is_err());
6642 }
6643
6644 #[test]
6645 fn schema_lineage_fails_closed_for_cycles_and_missing_sources() {
6646 let schema = Schema::new(vec![("value".to_owned(), ScalarType::U32)]);
6647 let selection = |source: &str| ResidentHeadSchemaSelection {
6648 source_head: source.to_owned(),
6649 output_schemas_by_source_winner: vec![schema.clone()],
6650 };
6651 let choices = BTreeMap::from([
6652 ("a".to_owned(), vec![schema.clone()]),
6653 ("b".to_owned(), vec![schema.clone()]),
6654 ("c".to_owned(), vec![schema.clone()]),
6655 ("empty".to_owned(), Vec::new()),
6656 ]);
6657
6658 let direct_cycle =
6659 resident_register_schema_selection("a", selection("a"), &choices, &mut BTreeMap::new())
6660 .expect_err("direct schema lineage cycle must decline");
6661 assert!(matches!(
6662 direct_cycle,
6663 ResidentGraphDeclineReason::WorkspaceUnbounded { detail }
6664 if detail.contains("contains a cycle")
6665 ));
6666
6667 let mut long_cycle = BTreeMap::new();
6668 resident_register_schema_selection("a", selection("b"), &choices, &mut long_cycle)
6669 .expect("first acyclic lineage edge");
6670 resident_register_schema_selection("b", selection("c"), &choices, &mut long_cycle)
6671 .expect("second acyclic lineage edge");
6672 let long_cycle =
6673 resident_register_schema_selection("c", selection("a"), &choices, &mut long_cycle)
6674 .expect_err("transitive schema lineage cycle must decline");
6675 assert!(matches!(
6676 long_cycle,
6677 ResidentGraphDeclineReason::WorkspaceUnbounded { detail }
6678 if detail.contains("contains a cycle")
6679 ));
6680
6681 for source in ["missing", "empty"] {
6682 let missing = resident_register_schema_selection(
6683 "target",
6684 selection(source),
6685 &choices,
6686 &mut BTreeMap::new(),
6687 )
6688 .expect_err("missing schema source must decline");
6689 assert!(matches!(
6690 missing,
6691 ResidentGraphDeclineReason::WorkspaceUnbounded { detail }
6692 if detail.contains("has no admitted candidate")
6693 ));
6694 }
6695
6696 let dynamic_plan = |source_output| ResidentOutputSchemaPlan {
6697 candidates: vec![schema.clone()],
6698 selection: ResidentOutputSchemaSelection::SourceWinner {
6699 source_output,
6700 schemas: vec![schema.clone()],
6701 },
6702 };
6703 let resolver_cycle = resident_resolve_output_schema(
6704 &[dynamic_plan(1), dynamic_plan(0)],
6705 &[RESIDENT_DYNAMIC_SCHEMA_ID, RESIDENT_DYNAMIC_SCHEMA_ID],
6706 0,
6707 &mut HashSet::new(),
6708 )
6709 .expect_err("receipt schema resolver cycle must fail");
6710 assert!(resolver_cycle.to_string().contains("contains a cycle"));
6711
6712 let missing_plan = resident_resolve_output_schema(
6713 &[dynamic_plan(1)],
6714 &[RESIDENT_DYNAMIC_SCHEMA_ID],
6715 0,
6716 &mut HashSet::new(),
6717 )
6718 .expect_err("missing receipt schema source plan must fail");
6719 assert!(missing_plan.to_string().contains("output 1 is missing"));
6720 }
6721
6722 #[test]
6723 fn join_capacity_uses_a_finite_tight_bound_when_loose_products_overflow() {
6724 let bound = resident_join_capacity_bound(u64::MAX, u64::MAX, 1, 1, 1);
6725 assert_eq!(
6726 bound,
6727 ResidentCapacityBound::Finite {
6728 rows: 1,
6729 proof: "matching_keys*left_fanout*right_fanout".to_owned(),
6730 }
6731 );
6732 }
6733
6734 #[test]
6735 fn true_cartesian_join_remains_above_the_resident_limit() {
6736 let bound = resident_join_capacity_bound(257, 257, 257, 257, 1);
6737 assert!(matches!(
6738 bound,
6739 ResidentCapacityBound::AboveResidentLimit { .. }
6740 ));
6741 }
6742
6743 #[test]
6744 fn lifetime_scan_rejects_an_out_of_range_logical_input() {
6745 let relations = vec![ResidentLogicalRelation {
6746 schema: Schema::new(vec![("value".to_string(), ScalarType::U32)]),
6747 initial_count: 0,
6748 permanent: false,
6749 }];
6750 let mut definitions = vec![None];
6751 let mut last_uses = vec![None];
6752 let mut ordinal = 0;
6753 let error = resident_record_lifetimes(
6754 &[ResidentRecordedOp::Filter {
6755 input: ResidentBufferRef::Private(1),
6756 output: 0,
6757 workspace: 0,
6758 op_id: 0,
6759 }],
6760 &relations,
6761 &mut definitions,
6762 &mut last_uses,
6763 &mut ordinal,
6764 )
6765 .expect_err("an invalid logical input must fail before allocation");
6766 assert!(error
6767 .to_string()
6768 .contains("logical input relation 1 is missing"));
6769 }
6770
6771 #[test]
6772 fn stale_scratch_generation_is_rejected_before_materialization() {
6773 let schema = Schema::new(vec![("value".to_string(), ScalarType::U32)]);
6774 let relations = vec![
6775 ResidentLogicalRelation {
6776 schema: schema.clone(),
6777 initial_count: 0,
6778 permanent: false,
6779 },
6780 ResidentLogicalRelation {
6781 schema: schema.clone(),
6782 initial_count: 0,
6783 permanent: false,
6784 },
6785 ];
6786 let slots = vec![ResidentPhysicalSlotPlan {
6787 schema,
6788 initial_count: 0,
6789 permanent: false,
6790 }];
6791 let assignments = vec![
6792 ResidentSlotAssignment {
6793 slot: 0,
6794 generation: 0,
6795 },
6796 ResidentSlotAssignment {
6797 slot: 0,
6798 generation: 0,
6799 },
6800 ];
6801 let error = resident_validate_slot_assignments(
6802 &relations,
6803 &[Some(0), Some(2)],
6804 &[Some(0), Some(2)],
6805 &slots,
6806 &assignments,
6807 )
6808 .expect_err("a stale scratch generation must fail before allocation");
6809 assert!(error.to_string().contains("generation 0 but expected 1"));
6810 }
6811
6812 #[test]
6813 fn authored_scan_and_unit_are_explicit_lifetime_operations() {
6814 let relations = vec![ResidentLogicalRelation {
6815 schema: Schema::new(Vec::<(String, ScalarType)>::new()),
6816 initial_count: 0,
6817 permanent: false,
6818 }];
6819 let mut definitions = vec![None];
6820 let mut last_uses = vec![None];
6821 let mut ordinal = 0;
6822 let operations = vec![
6823 ResidentRecordedOp::Unit {
6824 output: 0,
6825 op_id: 17,
6826 },
6827 ResidentRecordedOp::Scan {
6828 relation: ResidentBufferRef::Private(0),
6829 op_id: 18,
6830 },
6831 ResidentRecordedOp::TraceDelta {
6832 scan_delta: 1,
6833 filter_delta: 0,
6834 semantic_guard: None,
6835 },
6836 ];
6837
6838 resident_record_lifetimes(
6839 &operations,
6840 &relations,
6841 &mut definitions,
6842 &mut last_uses,
6843 &mut ordinal,
6844 )
6845 .expect("an emitted Unit defines the relation consumed by the emitted Scan");
6846
6847 assert_eq!(definitions, vec![Some(0)]);
6848 assert_eq!(last_uses, vec![Some(1)]);
6849 assert_eq!(ordinal, operations.len());
6850 }
6851
6852 #[test]
6853 fn authored_leaf_emitters_preserve_scan_identity_and_trace_order() {
6854 let mut operations = Vec::new();
6855 let push = |ops: &mut Vec<_>, op, _op_id| ops.push(op);
6856
6857 let unit = resident_record_unit_leaf(3, 17, &mut operations, push);
6858 let first = resident_record_scan_leaf(
6859 ResidentBufferRef::Private(4),
6860 18,
6861 None,
6862 &mut operations,
6863 push,
6864 );
6865 let second = resident_record_scan_leaf(
6866 ResidentBufferRef::Private(5),
6867 19,
6868 None,
6869 &mut operations,
6870 push,
6871 );
6872
6873 assert!(matches!(unit, ResidentBufferRef::Private(3)));
6874 assert!(matches!(first, ResidentBufferRef::Private(4)));
6875 assert!(matches!(second, ResidentBufferRef::Private(5)));
6876 assert!(matches!(
6877 operations.as_slice(),
6878 [
6879 ResidentRecordedOp::Unit {
6880 output: 3,
6881 op_id: 17
6882 },
6883 ResidentRecordedOp::Scan {
6884 relation: ResidentBufferRef::Private(4),
6885 op_id: 18
6886 },
6887 ResidentRecordedOp::TraceDelta {
6888 scan_delta: 1,
6889 filter_delta: 0,
6890 semantic_guard: None,
6891 },
6892 ResidentRecordedOp::Scan {
6893 relation: ResidentBufferRef::Private(5),
6894 op_id: 19
6895 },
6896 ResidentRecordedOp::TraceDelta {
6897 scan_delta: 1,
6898 filter_delta: 0,
6899 semantic_guard: None,
6900 },
6901 ]
6902 ));
6903 }
6904
6905 #[test]
6906 fn recursive_trace_semantics_use_the_selected_delta_and_leave_seed_unguarded() {
6907 let selected_delta = ResidentBufferRef::Private(9);
6908 assert!(resident_semantic_trace_guard(None).is_none());
6909 assert!(matches!(
6910 resident_semantic_trace_guard(Some((RelId(4), 2, 9))),
6911 Some(ResidentBufferRef::Private(9))
6912 ));
6913
6914 let mut operations = Vec::new();
6915 let push = |ops: &mut Vec<_>, op, _op_id| ops.push(op);
6916 resident_record_scan_leaf(
6917 ResidentBufferRef::Private(4),
6918 18,
6919 None,
6920 &mut operations,
6921 push,
6922 );
6923 resident_record_scan_leaf(
6924 ResidentBufferRef::Private(5),
6925 19,
6926 Some(selected_delta.clone()),
6927 &mut operations,
6928 push,
6929 );
6930
6931 assert!(matches!(
6932 operations.as_slice(),
6933 [
6934 ResidentRecordedOp::Scan { op_id: 18, .. },
6935 ResidentRecordedOp::TraceDelta {
6936 semantic_guard: None,
6937 ..
6938 },
6939 ResidentRecordedOp::Scan { op_id: 19, .. },
6940 ResidentRecordedOp::TraceDelta {
6941 semantic_guard: Some(ResidentBufferRef::Private(9)),
6942 ..
6943 },
6944 ]
6945 ));
6946 }
6947
6948 #[test]
6949 fn phase_unit_allocator_creates_fresh_scratch_values() {
6950 let mut relations = Vec::new();
6951
6952 let (first, first_op) = resident_new_phase_unit(&mut relations, 17).unwrap();
6953 let (second, second_op) = resident_new_phase_unit(&mut relations, 18).unwrap();
6954
6955 assert!(matches!(first, ResidentBufferRef::Private(0)));
6956 assert!(matches!(second, ResidentBufferRef::Private(1)));
6957 assert!(matches!(
6958 first_op,
6959 ResidentRecordedOp::Unit {
6960 output: 0,
6961 op_id: 17
6962 }
6963 ));
6964 assert!(matches!(
6965 second_op,
6966 ResidentRecordedOp::Unit {
6967 output: 1,
6968 op_id: 18
6969 }
6970 ));
6971 assert!(relations
6972 .iter()
6973 .all(|relation| !relation.permanent && relation.initial_count == 0));
6974 }
6975
6976 #[test]
6977 fn phase_local_merge_deduplicates_each_contribution_before_ordered_union() {
6978 let mut steps = Vec::new();
6979 let first = resident_phase_merge(None, 2_u32, |step| match step {
6980 ResidentPhaseMergeStep::Deduplicate(value) => {
6981 steps.push(("dedup", value, 0));
6982 Ok::<_, &'static str>(value * 10)
6983 }
6984 ResidentPhaseMergeStep::Union(left, right) => {
6985 steps.push(("union", left, right));
6986 Ok(left + right)
6987 }
6988 })
6989 .unwrap();
6990 let second = resident_phase_merge(Some(first), 3_u32, |step| match step {
6991 ResidentPhaseMergeStep::Deduplicate(value) => {
6992 steps.push(("dedup", value, 0));
6993 Ok::<_, &'static str>(value * 10)
6994 }
6995 ResidentPhaseMergeStep::Union(left, right) => {
6996 steps.push(("union", left, right));
6997 Ok(left + right)
6998 }
6999 })
7000 .unwrap();
7001
7002 assert_eq!(first, 20);
7003 assert_eq!(second, 50);
7004 assert_eq!(steps, [("dedup", 2, 0), ("dedup", 3, 0), ("union", 20, 30)]);
7005 }
7006
7007 #[test]
7008 fn initial_copy_schema_marker_is_ordered_after_its_count_producer() {
7009 let source = ResidentBufferRef::Source("head".to_owned());
7010 let mut operations = vec![ResidentRecordedOp::Project {
7011 input: source.clone(),
7012 output: 0,
7013 workspace: 0,
7014 op_id: 7,
7015 }];
7016
7017 resident_record_schema_winner_mark(&mut operations, ResidentBufferRef::Private(0), 0, 3);
7018
7019 assert!(matches!(
7020 operations.as_slice(),
7021 [
7022 ResidentRecordedOp::Project { op_id: 7, .. },
7023 ResidentRecordedOp::SchemaWinnerMark {
7024 contribution: ResidentBufferRef::Private(0),
7025 head_index: 0,
7026 schema_id: 3,
7027 }
7028 ]
7029 ));
7030 }
7031
7032 #[test]
7033 fn source_slots_are_deduplicated_sorted_and_distinct_from_private_targets() {
7034 let slots = resident_source_slot_map(3, ["zeta", "head", "zeta", "head"].into_iter())
7035 .expect("source slots");
7036
7037 assert_eq!(slots.get("head"), Some(&3));
7038 assert_eq!(slots.get("zeta"), Some(&4));
7039 assert_eq!(slots.len(), 2);
7040 assert_ne!(slots["head"], 0, "stored source and staged target differ");
7041 }
7042
7043 #[test]
7044 fn empty_untracked_sources_are_normalized_before_slot_binding() {
7045 let logical_count = resident_source_logical_count(Some(0)).unwrap();
7046 assert_eq!(
7047 resident_source_binding_route(logical_count, false, false).unwrap(),
7048 super::ResidentSourceBindingRoute::NormalizeEmpty
7049 );
7050 assert_eq!(
7051 resident_source_binding_route(0, true, true).unwrap(),
7052 super::ResidentSourceBindingRoute::Direct
7053 );
7054 assert!(resident_source_binding_route(1, false, true).is_err());
7055 assert!(resident_source_binding_route(1, true, false).is_err());
7056 assert!(resident_source_logical_count(None).is_err());
7057 }
7058
7059 #[test]
7060 fn compact_regions_preserve_phase_order_and_form_five_parent_nodes() {
7061 let unit = |output, op_id| ResidentRecordedOp::Unit { output, op_id };
7062 let phases = vec![
7063 ResidentCapturePhase::Segment {
7064 ops: vec![unit(10, 10)],
7065 scc_begin: None,
7066 },
7067 ResidentCapturePhase::Segment {
7068 ops: vec![unit(11, 11)],
7069 scc_begin: Some((64, 101)),
7070 },
7071 ResidentCapturePhase::ConditionalWhile {
7072 ops: vec![unit(12, 12)],
7073 iteration_limit: 64,
7074 convergence_op_id: 102,
7075 },
7076 ResidentCapturePhase::Segment {
7077 ops: vec![unit(20, 20)],
7078 scc_begin: None,
7079 },
7080 ResidentCapturePhase::Segment {
7081 ops: vec![ResidentRecordedOp::Scan {
7082 relation: ResidentBufferRef::Private(20),
7083 op_id: 21,
7084 }],
7085 scc_begin: Some((32, 201)),
7086 },
7087 ResidentCapturePhase::ConditionalWhile {
7088 ops: vec![unit(22, 22)],
7089 iteration_limit: 32,
7090 convergence_op_id: 202,
7091 },
7092 ResidentCapturePhase::Segment {
7093 ops: vec![unit(30, 30)],
7094 scc_begin: None,
7095 },
7096 ];
7097
7098 let regions = resident_compact_regions(vec![unit(0, 0)], phases, 999).expect("regions");
7099
7100 assert_eq!(regions.len(), 5);
7101 assert_eq!(
7102 regions.iter().filter(|region| region.recursive()).count(),
7103 2
7104 );
7105 assert_eq!(regions.len() + 2, 7, "hierarchical node inventory");
7106 let topology = resident_compact_topology(®ions).unwrap();
7107 assert_eq!(
7108 topology.parent_kinds,
7109 vec![
7110 ResidentCaptureParentKind::Kernel,
7111 ResidentCaptureParentKind::Conditional,
7112 ResidentCaptureParentKind::Kernel,
7113 ResidentCaptureParentKind::Conditional,
7114 ResidentCaptureParentKind::Kernel,
7115 ]
7116 );
7117 assert_eq!(topology.conditional_body_kernel_counts, vec![1, 1]);
7118 assert_eq!(topology.hierarchical_node_count, 7);
7119 assert!(regions[0].initializes());
7120 assert!(regions[0].begins_scc());
7121 assert_eq!(regions[0].op_id, regions[1].op_id);
7122 assert_eq!(regions[0].iteration_limit, regions[1].iteration_limit);
7123 assert_eq!(regions[2].op_id, regions[3].op_id);
7124 assert_eq!(regions[2].iteration_limit, regions[3].iteration_limit);
7125 assert!(regions[4].finalizes());
7126 assert_eq!(regions[4].op_id, 999);
7127 assert!(matches!(
7128 regions[2].ops.as_slice(),
7129 [
7130 ResidentRecordedOp::Unit { output: 20, .. },
7131 ResidentRecordedOp::Scan {
7132 relation: ResidentBufferRef::Private(20),
7133 ..
7134 }
7135 ]
7136 ));
7137 }
7138
7139 #[test]
7140 fn conditional_body_inventory_requires_one_actual_kernel_per_body() {
7141 let exact = vec![
7142 vec![CudaGraphNodeKind::Kernel],
7143 vec![CudaGraphNodeKind::Kernel],
7144 ];
7145 assert_eq!(
7146 resident_validate_conditional_body_node_kinds(&exact, 2).unwrap(),
7147 vec![1, 1]
7148 );
7149 assert!(resident_validate_conditional_body_node_kinds(&exact[..1], 2).is_err());
7150 assert!(resident_validate_conditional_body_node_kinds(&[Vec::new()], 1).is_err());
7151 assert!(resident_validate_conditional_body_node_kinds(
7152 &[vec![CudaGraphNodeKind::Kernel, CudaGraphNodeKind::Kernel]],
7153 1,
7154 )
7155 .is_err());
7156 assert!(resident_validate_conditional_body_node_kinds(
7157 &[vec![CudaGraphNodeKind::Memcpy]],
7158 1,
7159 )
7160 .is_err());
7161 }
7162
7163 #[test]
7164 fn compact_parent_graph_kind_validation_is_exact() {
7165 let expected = vec![
7166 ResidentCaptureParentKind::Kernel,
7167 ResidentCaptureParentKind::Conditional,
7168 ResidentCaptureParentKind::Kernel,
7169 ];
7170 assert!(resident_validate_parent_graph_kinds(
7171 &[
7172 CudaGraphNodeKind::Kernel,
7173 CudaGraphNodeKind::Conditional,
7174 CudaGraphNodeKind::Kernel
7175 ],
7176 &expected,
7177 )
7178 .is_ok());
7179 assert!(resident_validate_parent_graph_kinds(
7180 &[
7181 CudaGraphNodeKind::Kernel,
7182 CudaGraphNodeKind::Kernel,
7183 CudaGraphNodeKind::Conditional
7184 ],
7185 &expected,
7186 )
7187 .is_err());
7188 assert!(resident_validate_parent_graph_kinds(
7189 &[CudaGraphNodeKind::Kernel, CudaGraphNodeKind::Conditional],
7190 &expected,
7191 )
7192 .is_err());
7193 }
7194
7195 #[test]
7196 fn compact_descriptors_preserve_physical_generations_and_source_slots() {
7197 let regions = vec![ResidentCompactLogicalRegion {
7198 ops: vec![
7199 ResidentRecordedOp::Unit {
7200 output: 0,
7201 op_id: 7,
7202 },
7203 ResidentRecordedOp::SchemaWinnerMark {
7204 contribution: ResidentBufferRef::Private(0),
7205 head_index: 0,
7206 schema_id: 3,
7207 },
7208 ResidentRecordedOp::TraceDelta {
7209 scan_delta: 0,
7210 filter_delta: 1,
7211 semantic_guard: None,
7212 },
7213 ResidentRecordedOp::Scan {
7214 relation: ResidentBufferRef::Source("source".to_owned()),
7215 op_id: 8,
7216 },
7217 ResidentRecordedOp::TraceDelta {
7218 scan_delta: 1,
7219 filter_delta: 0,
7220 semantic_guard: None,
7221 },
7222 ],
7223 iteration_limit: 1,
7224 op_id: 0,
7225 flags: super::RESIDENT_SCHEDULE_REGION_INITIALIZE
7226 | super::RESIDENT_SCHEDULE_REGION_FINALIZE,
7227 }];
7228 let slots = vec![
7229 ResidentPhysicalSlotPlan {
7230 schema: Schema::new(Vec::new()),
7231 initial_count: 0,
7232 permanent: true,
7233 },
7234 ResidentPhysicalSlotPlan {
7235 schema: Schema::new(Vec::new()),
7236 initial_count: 0,
7237 permanent: false,
7238 },
7239 ];
7240 let assignments = vec![ResidentSlotAssignment {
7241 slot: 1,
7242 generation: 4,
7243 }];
7244
7245 let plan = resident_lower_compact_regions(
7246 regions,
7247 &slots,
7248 &assignments,
7249 ["source"].into_iter(),
7250 Default::default(),
7251 )
7252 .expect("compact descriptors");
7253
7254 assert_eq!(plan.source_slots["source"], 2);
7255 assert_eq!(plan.ops.len(), 4);
7256 assert_eq!(plan.ops[0].kind, super::ResidentScheduleOpKind::Unit);
7257 assert_eq!(plan.ops[0].out, 1);
7258 assert_eq!(plan.ops[0].out_generation, 4);
7259 assert_eq!(plan.ops[0].schema_winner_head, 0);
7260 assert_eq!(plan.ops[0].schema_winner_id, 3);
7261 assert_eq!(plan.ops[2].kind, super::ResidentScheduleOpKind::Scan);
7262 assert_eq!(plan.ops[2].out, 2);
7263 assert_eq!(plan.ops[2].in0_generation, 0);
7264 assert_eq!(plan.waves.len(), plan.ops.len());
7265 assert_eq!(plan.regions.len(), 1);
7266 assert_eq!(plan.generation_bases, vec![0, 4, 0]);
7267 }
7268
7269 #[test]
7270 fn compact_novelty_marker_observes_the_completed_delta_copy() {
7271 let regions = vec![ResidentCompactLogicalRegion {
7272 ops: vec![
7273 ResidentRecordedOp::ChangedReset,
7274 ResidentRecordedOp::Diff {
7275 left: ResidentBufferRef::Private(0),
7276 right: ResidentBufferRef::Private(1),
7277 output: 2,
7278 op_id: 10,
7279 },
7280 ResidentRecordedOp::Project {
7281 input: ResidentBufferRef::Private(2),
7282 output: 3,
7283 workspace: 0,
7284 op_id: 11,
7285 },
7286 ResidentRecordedOp::ChangedMark { relation: 3 },
7287 ],
7288 iteration_limit: 3,
7289 op_id: 12,
7290 flags: super::RESIDENT_SCHEDULE_REGION_RECURSIVE,
7291 }];
7292 let slots = (0..4)
7293 .map(|slot| ResidentPhysicalSlotPlan {
7294 schema: Schema::new(Vec::new()),
7295 initial_count: 0,
7296 permanent: slot != 2,
7297 })
7298 .collect::<Vec<_>>();
7299 let assignments = (0..4)
7300 .map(|slot| ResidentSlotAssignment {
7301 slot,
7302 generation: 0,
7303 })
7304 .collect::<Vec<_>>();
7305 let tables = ResidentCompactDescriptorTables {
7306 project_expressions: vec![super::ResidentProjectExpressionDescriptor::column(0, 4)],
7307 project_ranges: vec![(0, 1)],
7308 ..Default::default()
7309 };
7310
7311 let plan = resident_lower_compact_regions(
7312 regions,
7313 &slots,
7314 &assignments,
7315 std::iter::empty(),
7316 tables,
7317 )
7318 .expect("the final delta copy must carry the novelty marker");
7319
7320 assert_eq!(plan.ops.len(), 2);
7321 assert_eq!(plan.ops[0].kind, super::ResidentScheduleOpKind::Diff);
7322 assert_eq!(plan.ops[0].flags, 0);
7323 assert_eq!(plan.ops[1].kind, super::ResidentScheduleOpKind::Project);
7324 assert_eq!(plan.ops[1].flags, super::RESIDENT_SCHEDULE_OP_MARK_NOVELTY);
7325 assert_eq!(plan.ops[1].out, 3);
7326 }
7327
7328 #[test]
7329 fn compact_schema_marker_must_name_the_producer_output() {
7330 let regions = vec![ResidentCompactLogicalRegion {
7331 ops: vec![
7332 ResidentRecordedOp::Project {
7333 input: ResidentBufferRef::Source("source".to_owned()),
7334 output: 0,
7335 workspace: 0,
7336 op_id: 7,
7337 },
7338 ResidentRecordedOp::SchemaWinnerMark {
7339 contribution: ResidentBufferRef::Source("source".to_owned()),
7340 head_index: 0,
7341 schema_id: 0,
7342 },
7343 ],
7344 iteration_limit: 1,
7345 op_id: 9,
7346 flags: super::RESIDENT_SCHEDULE_REGION_INITIALIZE
7347 | super::RESIDENT_SCHEDULE_REGION_FINALIZE,
7348 }];
7349 let slots = vec![ResidentPhysicalSlotPlan {
7350 schema: Schema::new(Vec::new()),
7351 initial_count: 0,
7352 permanent: true,
7353 }];
7354 let assignments = vec![ResidentSlotAssignment {
7355 slot: 0,
7356 generation: 0,
7357 }];
7358 let tables = ResidentCompactDescriptorTables {
7359 project_ranges: vec![(0, 0)],
7360 ..Default::default()
7361 };
7362
7363 assert!(resident_lower_compact_regions(
7364 regions,
7365 &slots,
7366 &assignments,
7367 ["source"].into_iter(),
7368 tables,
7369 )
7370 .is_err());
7371 }
7372
7373 #[test]
7374 fn compact_schema_defaults_follow_first_marker_per_head() {
7375 let ops = vec![
7376 ResidentOpDescriptor::unit(1, 0, 0).with_schema_winner(1, u32::MAX),
7377 ResidentOpDescriptor::unit(2, 1, 0).with_schema_winner(0, 7),
7378 ResidentOpDescriptor::unit(3, 2, 0).with_schema_winner(1, 9),
7379 ];
7380 assert_eq!(
7381 resident_compact_schema_defaults(&ops, 2).unwrap(),
7382 [7, u32::MAX]
7383 );
7384 assert!(resident_compact_schema_defaults(&ops[..1], 2).is_err());
7385 }
7386
7387 #[test]
7388 fn compact_changed_reset_is_only_absorbed_at_recursive_region_entry() {
7389 let slot = ResidentPhysicalSlotPlan {
7390 schema: Schema::new(Vec::new()),
7391 initial_count: 0,
7392 permanent: false,
7393 };
7394 let assignment = ResidentSlotAssignment {
7395 slot: 0,
7396 generation: 0,
7397 };
7398 let region = |ops, flags| ResidentCompactLogicalRegion {
7399 ops,
7400 iteration_limit: 4,
7401 op_id: 9,
7402 flags,
7403 };
7404
7405 assert!(resident_lower_compact_regions(
7406 vec![region(vec![ResidentRecordedOp::ChangedReset], 0)],
7407 std::slice::from_ref(&slot),
7408 std::slice::from_ref(&assignment),
7409 std::iter::empty(),
7410 Default::default(),
7411 )
7412 .is_err());
7413 assert!(resident_lower_compact_regions(
7414 vec![region(
7415 vec![
7416 ResidentRecordedOp::Unit {
7417 output: 0,
7418 op_id: 1,
7419 },
7420 ResidentRecordedOp::ChangedReset,
7421 ],
7422 super::RESIDENT_SCHEDULE_REGION_RECURSIVE,
7423 )],
7424 std::slice::from_ref(&slot),
7425 std::slice::from_ref(&assignment),
7426 std::iter::empty(),
7427 Default::default(),
7428 )
7429 .is_err());
7430 assert!(resident_lower_compact_regions(
7431 vec![region(
7432 vec![
7433 ResidentRecordedOp::ChangedReset,
7434 ResidentRecordedOp::Unit {
7435 output: 0,
7436 op_id: 1,
7437 },
7438 ],
7439 super::RESIDENT_SCHEDULE_REGION_RECURSIVE,
7440 )],
7441 &[slot],
7442 &[assignment],
7443 std::iter::empty(),
7444 Default::default(),
7445 )
7446 .is_ok());
7447 }
7448
7449 #[test]
7450 fn compact_filter_project_tables_preserve_types_widths_and_order() {
7451 let input = Schema::new(vec![
7452 ("symbol".to_owned(), ScalarType::Symbol),
7453 ("number".to_owned(), ScalarType::U64),
7454 ]);
7455 let predicate = Expr::And(vec![
7456 Expr::Compare {
7457 left: Box::new(Expr::Column(1)),
7458 op: CompareOp::Ge,
7459 right: Box::new(Expr::Const(ConstValue::U64(9))),
7460 },
7461 Expr::Compare {
7462 left: Box::new(Expr::Column(0)),
7463 op: CompareOp::Eq,
7464 right: Box::new(Expr::Const(ConstValue::Symbol("x".to_owned()))),
7465 },
7466 ]);
7467 let comparisons =
7468 resident_compact_filter_descriptors(&predicate, &input).expect("filter table");
7469 assert_eq!(comparisons.len(), 2);
7470 assert_eq!(comparisons[0].left_column, 1);
7471 assert_eq!(comparisons[0].width, 8);
7472 assert_eq!(comparisons[0].right_constant, 9);
7473 assert_eq!(comparisons[1].left_column, 0);
7474 assert_eq!(comparisons[1].width, 4);
7475
7476 let output = Schema::new(vec![
7477 ("number".to_owned(), ScalarType::U64),
7478 ("symbol".to_owned(), ScalarType::Symbol),
7479 ]);
7480 let expressions = resident_compact_project_descriptors(
7481 &[
7482 ProjectExpr::Column(1),
7483 ProjectExpr::Computed(
7484 Expr::Const(ConstValue::Symbol("y".to_owned())),
7485 ScalarType::Symbol,
7486 ),
7487 ],
7488 &input,
7489 &output,
7490 )
7491 .expect("project table");
7492 assert_eq!(expressions.len(), output.arity());
7493 assert_eq!(expressions[0].column, 1);
7494 assert_eq!(expressions[0].width, 8);
7495 assert_eq!(expressions[1].kind, 1);
7496 assert_eq!(expressions[1].width, 4);
7497
7498 let mismatch = Expr::Compare {
7499 left: Box::new(Expr::Column(1)),
7500 op: CompareOp::Eq,
7501 right: Box::new(Expr::Const(ConstValue::Symbol("wrong".to_owned()))),
7502 };
7503 assert!(resident_compact_filter_descriptors(&mismatch, &input).is_err());
7504 }
7505
7506 #[test]
7507 fn compact_tables_cover_multiple_empty_occurrences_without_duplicate_payloads() {
7508 let filters = vec![
7509 ResidentFilterPlan {
7510 compact_comparisons: Vec::new(),
7511 },
7512 ResidentFilterPlan {
7513 compact_comparisons: Vec::new(),
7514 },
7515 ];
7516 let projects = vec![ResidentProjectPlan {
7517 compact_expressions: Vec::new(),
7518 }];
7519
7520 let tables = resident_compact_tables(&filters, &projects).expect("compact tables");
7521
7522 assert!(tables.filter_comparisons.is_empty());
7523 assert_eq!(tables.filter_ranges, [(0, 0), (0, 0)]);
7524 assert!(tables.project_expressions.is_empty());
7525 assert_eq!(tables.project_ranges, [(0, 0)]);
7526 }
7527
7528 #[test]
7529 fn compact_descriptor_ranges_follow_operation_order_exactly() {
7530 let regions = vec![ResidentCompactLogicalRegion {
7531 ops: vec![
7532 ResidentRecordedOp::Project {
7533 input: ResidentBufferRef::Source("source".to_owned()),
7534 output: 0,
7535 workspace: 1,
7536 op_id: 1,
7537 },
7538 ResidentRecordedOp::Project {
7539 input: ResidentBufferRef::Source("source".to_owned()),
7540 output: 1,
7541 workspace: 0,
7542 op_id: 2,
7543 },
7544 ],
7545 iteration_limit: 1,
7546 op_id: 3,
7547 flags: super::RESIDENT_SCHEDULE_REGION_INITIALIZE
7548 | super::RESIDENT_SCHEDULE_REGION_FINALIZE,
7549 }];
7550 let slots = vec![
7551 ResidentPhysicalSlotPlan {
7552 schema: Schema::new(Vec::new()),
7553 initial_count: 0,
7554 permanent: false,
7555 },
7556 ResidentPhysicalSlotPlan {
7557 schema: Schema::new(Vec::new()),
7558 initial_count: 0,
7559 permanent: false,
7560 },
7561 ];
7562 let assignments = vec![
7563 ResidentSlotAssignment {
7564 slot: 0,
7565 generation: 0,
7566 },
7567 ResidentSlotAssignment {
7568 slot: 1,
7569 generation: 0,
7570 },
7571 ];
7572 let tables = ResidentCompactDescriptorTables {
7573 project_expressions: vec![Default::default(), Default::default()],
7574 project_ranges: vec![(0, 1), (1, 1)],
7575 ..Default::default()
7576 };
7577
7578 assert!(resident_lower_compact_regions(
7579 regions,
7580 &slots,
7581 &assignments,
7582 ["source"].into_iter(),
7583 tables,
7584 )
7585 .is_err());
7586 }
7587
7588 #[test]
7589 fn compact_metadata_reservation_counts_generation_bases_and_schema_defaults() {
7590 let actual = resident_compact_schedule_metadata_bytes(3, 5, 5, 2, 6, 4, 0, 0)
7591 .expect("metadata bytes");
7592 let expected = super::resident_schedule_metadata_device_bytes(3, 5, 5, 2, 10, 0, 0)
7593 .expect("expected metadata bytes");
7594 let missing_defaults = super::resident_schedule_metadata_device_bytes(3, 5, 5, 2, 6, 0, 0)
7595 .expect("generation-only bytes");
7596
7597 assert_eq!(actual, expected);
7598 assert_ne!(actual, missing_defaults);
7599 }
7600
7601 #[test]
7602 fn compact_manifest_replaces_per_occurrence_descriptor_allocations() {
7603 let plan = super::ResidentCompactSchedulePlan {
7604 source_slots: [("source".to_owned(), 2)].into_iter().collect(),
7605 ops: Vec::new(),
7606 waves: Vec::new(),
7607 regions: Vec::new(),
7608 generation_bases: Vec::new(),
7609 filter_comparisons: Vec::new(),
7610 project_expressions: Vec::new(),
7611 };
7612 let (required, metadata) =
7613 resident_compact_allocation_bytes(1_000, 2_000, 3_000, 2, 2, &plan).unwrap();
7614 let expected_metadata =
7615 resident_compact_schedule_metadata_bytes(3, 0, 0, 0, 0, 2, 0, 0).unwrap();
7616 assert_eq!(metadata, expected_metadata);
7617 assert_eq!(required, 1_000 + 2_000 + 3_000 + expected_metadata);
7618 assert_ne!(
7619 required,
7620 1_000 + 48 + 2_000 + 24 + 3_000 + expected_metadata
7621 );
7622 }
7623
7624 #[test]
7625 fn compact_preflight_bytes_report_flattened_tables_without_double_counting() {
7626 let assert_components = |filter_count: usize, project_count: usize| {
7627 let plan = super::ResidentCompactSchedulePlan {
7628 source_slots: [("source".to_owned(), 2)].into_iter().collect(),
7629 ops: Vec::new(),
7630 waves: Vec::new(),
7631 regions: Vec::new(),
7632 generation_bases: Vec::new(),
7633 filter_comparisons: vec![
7634 super::ResidentFilterComparisonDescriptor::default();
7635 filter_count
7636 ],
7637 project_expressions: vec![
7638 super::ResidentProjectExpressionDescriptor::default();
7639 project_count
7640 ],
7641 };
7642 let mut manifest = ResidentAllocationManifest {
7643 slots: Vec::new(),
7644 logical_to_slot: Vec::new(),
7645 required_bytes: 0,
7646 relation_bytes: 1_000,
7647 filter_scratch_bytes: 2_000,
7648 schedule_metadata_bytes: 0,
7649 fixed_workspace_bytes: 3_000,
7650 logical_relation_values: 0,
7651 permanent_slots: 0,
7652 scratch_slots: 0,
7653 filter_scratch_allocations: 1,
7654 max_row_bytes: 0,
7655 };
7656 manifest.finalize_compact_schedule(&plan, 2).unwrap();
7657
7658 let (filter_bytes, project_bytes, fixed_bytes) =
7659 resident_compact_preflight_device_bytes(&manifest, &plan).unwrap();
7660 assert_eq!(
7661 filter_bytes,
7662 48 * u64::try_from(filter_count.max(1)).unwrap()
7663 );
7664 assert_eq!(
7665 project_bytes,
7666 24 * u64::try_from(project_count.max(1)).unwrap()
7667 );
7668 assert_eq!(
7669 manifest.relation_bytes
7670 + manifest.filter_scratch_bytes
7671 + filter_bytes
7672 + project_bytes
7673 + fixed_bytes,
7674 manifest.required_bytes
7675 );
7676 };
7677
7678 assert_components(0, 0);
7679 assert_components(3, 2);
7680 }
7681
7682 #[test]
7683 fn compact_manifest_is_exact_and_preserves_permanent_head_mapping() {
7684 let mut manifest = ResidentAllocationManifest {
7685 slots: vec![
7686 ResidentPhysicalSlotPlan {
7687 schema: Schema::new(Vec::new()),
7688 initial_count: 0,
7689 permanent: true,
7690 },
7691 ResidentPhysicalSlotPlan {
7692 schema: Schema::new(Vec::new()),
7693 initial_count: 0,
7694 permanent: true,
7695 },
7696 ],
7697 logical_to_slot: vec![
7698 ResidentSlotAssignment {
7699 slot: 0,
7700 generation: 0,
7701 },
7702 ResidentSlotAssignment {
7703 slot: 1,
7704 generation: 0,
7705 },
7706 ],
7707 required_bytes: 0,
7708 relation_bytes: 1_000,
7709 filter_scratch_bytes: 2_000,
7710 schedule_metadata_bytes: 0,
7711 fixed_workspace_bytes: 3_000,
7712 logical_relation_values: 2,
7713 permanent_slots: 2,
7714 scratch_slots: 0,
7715 filter_scratch_allocations: 1,
7716 max_row_bytes: 1,
7717 };
7718 let plan = super::ResidentCompactSchedulePlan {
7719 source_slots: [("external".to_owned(), 2)].into_iter().collect(),
7720 ops: Vec::new(),
7721 waves: Vec::new(),
7722 regions: Vec::new(),
7723 generation_bases: Vec::new(),
7724 filter_comparisons: Vec::new(),
7725 project_expressions: Vec::new(),
7726 };
7727 manifest.finalize_compact_schedule(&plan, 1).unwrap();
7728 let without_source =
7729 resident_compact_schedule_metadata_bytes(2, 0, 0, 0, 0, 1, 0, 0).unwrap();
7730 assert_eq!(manifest.schedule_metadata_bytes - without_source, 240);
7731 assert_eq!(
7732 manifest.required_bytes,
7733 manifest.relation_bytes
7734 + manifest.filter_scratch_bytes
7735 + manifest.fixed_workspace_bytes
7736 + manifest.schedule_metadata_bytes
7737 );
7738 resident_validate_exact_reservation(manifest.required_bytes, manifest.required_bytes, 0)
7739 .unwrap();
7740 assert!(resident_validate_exact_reservation(manifest.required_bytes, 5_999, 1).is_err());
7741 let heads = [("head".to_owned(), 0_usize)].into_iter().collect();
7742 assert_eq!(
7743 resident_output_indices(&heads, &manifest.logical_to_slot, &manifest.slots).unwrap(),
7744 vec![("head".to_owned(), 0)]
7745 );
7746 let mut scratch_slots = manifest.slots.clone();
7747 scratch_slots[0].permanent = false;
7748 assert!(
7749 resident_output_indices(&heads, &manifest.logical_to_slot, &scratch_slots,).is_err()
7750 );
7751 }
7752
7753 #[test]
7754 fn compact_distinct_declines_partial_keys_before_planning() {
7755 let node = RirNode::Distinct {
7756 input: Box::new(RirNode::Scan { rel: RelId(1) }),
7757 key_cols: vec![0],
7758 };
7759 let schema = Schema::new(vec![
7760 ("key".to_owned(), ScalarType::U32),
7761 ("value".to_owned(), ScalarType::U32),
7762 ]);
7763
7764 let error = validate_compact_resident_node_envelope(&node, &|_| Ok(schema.clone()))
7765 .expect_err("a partial-key distinct must decline before allocation");
7766
7767 let crate::resident_graph::ResidentGraphDeclineReason::WorkspaceUnbounded { detail } =
7768 error
7769 else {
7770 panic!("expected a workspace-envelope decline");
7771 };
7772 assert!(detail.contains("canonical full-row key columns"));
7773 }
7774
7775 #[test]
7776 fn compact_distinct_accepts_empty_keys_for_nullary_rows() {
7777 let node = RirNode::Distinct {
7778 input: Box::new(RirNode::Unit),
7779 key_cols: vec![],
7780 };
7781
7782 validate_compact_resident_node_envelope(&node, &|_| Ok(Schema::new(Vec::new())))
7783 .expect("the canonical full-row key for an arity-zero relation is empty");
7784 }
7785
7786 #[test]
7787 fn compact_join_declines_same_width_different_key_types() {
7788 let node = RirNode::Join {
7789 left: Box::new(RirNode::Scan { rel: RelId(1) }),
7790 right: Box::new(RirNode::Scan { rel: RelId(2) }),
7791 left_keys: vec![0],
7792 right_keys: vec![0],
7793 join_type: JoinType::Inner,
7794 };
7795
7796 let error = validate_compact_resident_node_envelope(&node, &|node| match node {
7797 RirNode::Scan { rel: RelId(1) } => {
7798 Ok(Schema::new(vec![("key".to_owned(), ScalarType::U32)]))
7799 }
7800 RirNode::Scan { rel: RelId(2) } => {
7801 Ok(Schema::new(vec![("key".to_owned(), ScalarType::Symbol)]))
7802 }
7803 _ => unreachable!("the envelope asks only for operand schemas"),
7804 })
7805 .expect_err("same-width but different key types must decline");
7806
7807 let crate::resident_graph::ResidentGraphDeclineReason::WorkspaceUnbounded { detail } =
7808 error
7809 else {
7810 panic!("expected a workspace-envelope decline");
7811 };
7812 assert!(detail.contains("matching U32, U64, or Symbol key types"));
7813 }
7814
7815 #[test]
7816 fn compact_union_shape_requires_unary_self_union() {
7817 assert!(resident_union_fold_mode(0).is_err());
7818 assert_eq!(
7819 resident_union_fold_mode(1).unwrap(),
7820 ResidentUnionFoldMode::SelfUnion
7821 );
7822 assert_eq!(
7823 resident_union_fold_mode(4).unwrap(),
7824 ResidentUnionFoldMode::LeftAssociated
7825 );
7826 }
7827}