xlog_induce/
nary_engine.rs

1//! Engine orchestration for bounded exact n-ary induction.
2//!
3//! The n-ary counterpart of the binary [`crate::induce_exact`] pipeline:
4//! enumerate every canonical pattern ([`crate::nary::enumerate_patterns`]),
5//! flatten the batch into the device layout
6//! ([`crate::nary_layout::flatten_patterns`]), score it through the
7//! device-resident launcher
8//! ([`CudaKernelProvider::ilp_exact_nary_score_device`]), and reduce the
9//! per-pattern coverage counts deterministically
10//! ([`crate::reduce::reduce_nary`]).
11//!
12//! Transfer budget: relation and example values never visit the host —
13//! ingestion is device-to-device column copies inside the launcher; row
14//! counts come from cached host-side metadata (a pure struct read); the
15//! only device-to-host transfers are the two per-pattern count arrays.
16//! The reduction then runs on those counts entirely host-side.
17
18use xlog_core::{RelId, Result, ScalarType, XlogError};
19use xlog_cuda::{CudaBuffer, CudaKernelProvider, IlpExactNaryPatterns};
20
21use crate::nary::{enumerate_patterns, NaryEnumerationConfig, NaryRulePattern};
22use crate::nary_layout::{flatten_patterns, NARY_MAX_ATOM_ARITY};
23use crate::reduce::reduce_nary;
24
25/// Head arity bound of the device contract: the kernel gathers one example
26/// tuple into a fixed per-thread array of this size, and the launcher
27/// refuses wider heads fail-closed. Mirrored here so the engine refuses
28/// with a typed error before enumerating a batch no kernel could score.
29const NARY_MAX_HEAD_ARITY: usize = 8;
30
31/// Bounds and selection size for one [`induce_exact_nary`] call.
32#[derive(Debug, Clone, Copy)]
33pub struct NaryInductionConfig {
34    /// Pattern-space bounds (body atoms, join variables, hard pattern cap).
35    pub enumeration: NaryEnumerationConfig,
36    /// Number of top-ranked patterns to keep.
37    pub k: u32,
38}
39
40/// Inputs to one [`induce_exact_nary`] call.
41///
42/// Each candidate is a `(RelId, &CudaBuffer)` pair exactly as in the binary
43/// engine: the `RelId` is a label that flows through to the scored output,
44/// and the buffer carries the relation's facts as all-`U64` columns. The
45/// head arity is the arity of `positives`; `negatives`, when present, must
46/// match it. Name resolution and relation-store lookup happen at the
47/// pyxlog boundary — the engine only sees indices + handles.
48pub struct InduceExactNaryRequest<'a> {
49    pub head_rel_idx: RelId,
50    pub candidates: &'a [(RelId, &'a CudaBuffer)],
51    pub positives: &'a CudaBuffer,
52    pub negatives: Option<&'a CudaBuffer>,
53    pub config: NaryInductionConfig,
54}
55
56/// One kept pattern with full metadata and diagnostics.
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct ScoredNaryCandidate {
59    pub head_rel_idx: RelId,
60    /// Relation identity of each body atom, in body order — the
61    /// `candidate_slot` of the corresponding [`NaryRulePattern`] atom
62    /// resolved against the request's candidate list.
63    pub body_rel_idxs: Vec<RelId>,
64    /// The full canonical pattern (bindings included), for rule assembly.
65    pub pattern: NaryRulePattern,
66    pub positives_covered: u32,
67    pub negatives_covered: u32,
68    pub local_rank: u32,
69    pub next_positives_covered: u32,
70    pub next_negatives_covered: u32,
71    pub tie_class_size: u32,
72}
73
74/// Result of one [`induce_exact_nary`] call.
75///
76/// `total_scored` counts patterns actually scored by the kernel; the
77/// trivial early-outs (no candidates, no patterns, `k == 0`, no positive
78/// examples) report `0` and an empty candidate list.
79#[derive(Debug, Clone, Default, PartialEq, Eq)]
80pub struct NaryInductionResult {
81    pub candidates: Vec<ScoredNaryCandidate>,
82    pub total_scored: u32,
83    pub candidate_count: u32,
84    pub positive_count: u32,
85    pub negative_count: u32,
86}
87
88/// Run exact n-ary induction against one request.
89///
90/// The `provider` owns the kernel launcher and is also used to
91/// materialize the short-lived empty negatives buffer when
92/// `request.negatives` is `None` — the same normalization the binary
93/// engine performs.
94///
95/// Diagnostic asymmetry, stated rather than left to be discovered: the
96/// EMPTY-CANDIDATES early-out returns a fully default result (zero
97/// example counts) because it precedes buffer validation — with no
98/// candidate there is nothing whose arity the examples must agree with.
99/// Every other trivial early-out (`k == 0`, zero positives) validates
100/// first and therefore reports the real `positive_count` /
101/// `negative_count`.
102pub fn induce_exact_nary(
103    provider: &CudaKernelProvider,
104    request: &InduceExactNaryRequest<'_>,
105) -> Result<NaryInductionResult> {
106    // Empty candidates is a trivial dead-end and needs no CUDA inspection.
107    if request.candidates.is_empty() {
108        return Ok(NaryInductionResult::default());
109    }
110
111    // ── Buffer validation (typed, before any enumeration cost). The
112    //    launcher re-validates fail-closed — this pass exists to fail loud
113    //    on pyxlog-side assembly bugs with engine-shaped errors.
114    let (head_arity_u32, pos_count) = require_u64_columnar(request.positives, "positives")?;
115    let head_arity = head_arity_u32 as usize;
116    if head_arity == 0 || head_arity > NARY_MAX_HEAD_ARITY {
117        return Err(XlogError::Type(format!(
118            "induce_exact_nary: positives arity {head_arity} outside supported \
119             head arity 1..={NARY_MAX_HEAD_ARITY}",
120        )));
121    }
122    let neg_count = match request.negatives {
123        Some(neg) => {
124            let (neg_arity, neg_count) = require_u64_columnar(neg, "negatives")?;
125            if neg_arity != head_arity_u32 {
126                return Err(XlogError::Type(format!(
127                    "induce_exact_nary: negatives arity {neg_arity} != head arity {head_arity}",
128                )));
129            }
130            neg_count
131        }
132        None => 0,
133    };
134    let mut candidate_arities: Vec<u8> = Vec::with_capacity(request.candidates.len());
135    for (i, (_, buf)) in request.candidates.iter().enumerate() {
136        let (arity, _) = require_u64_columnar(buf, &format!("candidate[{i}]"))?;
137        if arity == 0 || arity as usize > NARY_MAX_ATOM_ARITY {
138            return Err(XlogError::Type(format!(
139                "induce_exact_nary: candidate[{i}] arity {arity} outside supported \
140                 atom arity 1..={NARY_MAX_ATOM_ARITY}",
141            )));
142        }
143        candidate_arities.push(arity as u8);
144    }
145    let candidate_count = request.candidates.len() as u32;
146
147    // ── Enumerate the canonical pattern space (typed refusal on blow-up,
148    //    never a silent cap).
149    let patterns = enumerate_patterns(
150        head_arity as u8,
151        &candidate_arities,
152        &request.config.enumeration,
153    )?;
154
155    let counts_only = NaryInductionResult {
156        candidates: Vec::new(),
157        total_scored: 0,
158        candidate_count,
159        positive_count: pos_count,
160        negative_count: neg_count,
161    };
162    if patterns.is_empty() || request.config.k == 0 || pos_count == 0 {
163        // No pattern can be kept: nothing to enumerate, nothing requested,
164        // or no positive example can ever push coverage above zero. All
165        // three are provable host-side without a launch.
166        return Ok(counts_only);
167    }
168
169    // ── Flatten into the device layout. The enumeration already
170    //    guarantees canonical form; this refuses only device-bound
171    //    violations (a config wider than the kernel's fixed state).
172    let layout = flatten_patterns(&patterns)
173        .map_err(|e| XlogError::Type(format!("induce_exact_nary: flatten: {e}")))?;
174    let patterns_view = IlpExactNaryPatterns {
175        body_offset: &layout.body_offset,
176        body_len: &layout.body_len,
177        atom_candidate_slot: &layout.atom_candidate_slot,
178        atom_arity: &layout.atom_arity,
179        atom_binding_offset: &layout.atom_binding_offset,
180        binding_codes: &layout.binding_codes,
181        head_arity: head_arity_u32,
182    };
183
184    // ── Normalize negatives: the launcher expects an always-present
185    //    buffer. Same construction as the binary engine — an empty buffer
186    //    with the positives' schema.
187    let empty_neg_holder: Option<CudaBuffer> = if request.negatives.is_none() {
188        Some(provider.create_empty_buffer(request.positives.schema().clone())?)
189    } else {
190        None
191    };
192    let negatives: &CudaBuffer = match request.negatives {
193        Some(b) => b,
194        None => empty_neg_holder
195            .as_ref()
196            .expect("holder populated in the None branch above"),
197    };
198
199    // ── Score on device (D2D ingest inside; the two count arrays are the
200    //    only D2H) and reduce deterministically on the counts.
201    let candidate_buffers: Vec<&CudaBuffer> = request.candidates.iter().map(|(_, b)| *b).collect();
202    let (pos_covered, neg_covered) = provider.ilp_exact_nary_score_device(
203        &patterns_view,
204        &candidate_buffers,
205        request.positives,
206        negatives,
207    )?;
208    if pos_covered.len() != patterns.len() || neg_covered.len() != patterns.len() {
209        return Err(XlogError::Execution(format!(
210            "induce_exact_nary: launcher returned {}/{} counts for {} patterns",
211            pos_covered.len(),
212            neg_covered.len(),
213            patterns.len(),
214        )));
215    }
216    let coverage: Vec<(u32, u32)> = pos_covered.into_iter().zip(neg_covered).collect();
217    let kept = reduce_nary(&coverage, request.config.k);
218
219    let mut candidates = Vec::with_capacity(kept.len());
220    for keep in kept {
221        let pattern = patterns
222            .get(keep.pattern_idx)
223            .ok_or_else(|| {
224                XlogError::Execution(format!(
225                    "induce_exact_nary: reduction returned pattern index {} for {} patterns",
226                    keep.pattern_idx,
227                    patterns.len(),
228                ))
229            })?
230            .clone();
231        let body_rel_idxs = pattern
232            .body
233            .iter()
234            .map(|atom| request.candidates[atom.candidate_slot as usize].0)
235            .collect();
236        candidates.push(ScoredNaryCandidate {
237            head_rel_idx: request.head_rel_idx,
238            body_rel_idxs,
239            pattern,
240            positives_covered: keep.positives_covered,
241            negatives_covered: keep.negatives_covered,
242            local_rank: keep.local_rank,
243            next_positives_covered: keep.next_positives_covered,
244            next_negatives_covered: keep.next_negatives_covered,
245            tie_class_size: keep.tie_class_size,
246        });
247    }
248
249    Ok(NaryInductionResult {
250        candidates,
251        total_scored: patterns.len() as u32,
252        candidate_count,
253        positive_count: pos_count,
254        negative_count: neg_count,
255    })
256}
257
258fn require_u64_columnar(buf: &CudaBuffer, label: &str) -> Result<(u32, u32)> {
259    let arity = buf.arity();
260    for col_idx in 0..arity {
261        let t = buf.schema().column_type(col_idx).ok_or_else(|| {
262            XlogError::Type(format!(
263                "induce_exact_nary: {label} buffer column {col_idx} has no schema type",
264            ))
265        })?;
266        if t != ScalarType::U64 {
267            return Err(XlogError::Type(format!(
268                "induce_exact_nary: {label} buffer column {col_idx} has type {t:?}, expected U64",
269            )));
270        }
271    }
272    let rows = buf.cached_row_count().ok_or_else(|| {
273        XlogError::Execution(format!(
274            "induce_exact_nary: {label} buffer has no cached row count \
275             (device-resident ingest should populate it; required to avoid \
276             a hot-loop device-to-host transfer)",
277        ))
278    })?;
279    let arity = u32::try_from(arity)
280        .map_err(|_| XlogError::Type(format!("induce_exact_nary: {label} arity exceeds u32")))?;
281    Ok((arity, rows))
282}
xlog_induce/nary_engine.rs

xlog_induce/
nary_engine.rs
