Skip to main content

xlog_cuda/provider/
wcoj.rs

1//! v0.6.2 GPU 3-way Worst-Case Optimal Join — provider entries.
2//!
3//! Public methods (parallel u32 and u64 entries — see
4//! [`CudaKernelProvider::wcoj_triangle_u32_recorded`] and
5//! [`CudaKernelProvider::wcoj_triangle_u64_recorded`]):
6//!
7//!   * **U32 / Symbol** entry takes 2-column inputs whose columns
8//!     may be [`xlog_core::ScalarType::U32`] or
9//!     [`xlog_core::ScalarType::Symbol`] (both share the same
10//!     4-byte physical layout). It routes through the
11//!     histogram-guided block-slice triangle path.
12//!   * **U64** entry takes 2-column inputs whose columns are
13//!     [`xlog_core::ScalarType::U64`] only. Backed by parallel
14//!     `_u64` count + materialize kernels in `wcoj.cu`; counters
15//!     and the `wcoj_compute_total` reducer are reused unchanged
16//!     (they're bounded by `u32::MAX` rows).
17//!   * Mixed-width inputs in the same triangle are rejected at
18//!     the provider level — each entry's schema guard requires
19//!     all three relations match its width.
20//!   * **Sorted, deduped inputs.** Caller-supplied:
21//!     - `e_xy` lex-sorted+deduped by (X, Y),
22//!     - `e_yz` lex-sorted+deduped by (Y, Z),
23//!     - `e_xz` lex-sorted+deduped by (X, Z).
24//!       Physical layout construction is a separate slice — this
25//!       entry assumes the caller has already arranged input layout.
26//!   * **Two-phase count → device-scan → materialize.** Mirrors
27//!     SRDatalog (Sun et al., arXiv 2604.20073) Section 4's
28//!     deterministic two-phase pipeline. Row counts are
29//!     prefix-summed on device; the only host visit between the
30//!     two phases is a single 4-byte `dtoh_scalar_untracked` of
31//!     the inclusive total (sanctioned metadata read, exempt from
32//!     the strict deterministic-D2H gate).
33//!   * **Strict [`LaunchRecorder`] discipline.** Two recorders
34//!     run sequentially on the caller-supplied launch stream:
35//!     1. count+scan recorder: reads `e_xy` / `e_yz` / `e_xz`
36//!        columns + their `d_num_rows`; writes `count_buf`,
37//!        `offsets_buf`, `d_total`. Spans the count kernel,
38//!        the dtod copy `count_buf → offsets_buf`, the
39//!        device-side prefix-sum on `offsets_buf`, and
40//!        `wcoj_compute_total`.
41//!     2. materialize recorder: reads same inputs + `offsets_buf`;
42//!        writes the three output columns + output `d_num_rows`.
43//!   * **Output deterministic and lex-sorted by (X, Y, Z).** Locked
44//!     by [`tests/test_wcoj_triangle_u32.rs`].
45//!   * **Set semantics on deduped input.** If the caller violates
46//!     the dedup contract on an input, the kernel may emit
47//!     duplicates; the test suite documents this as caller
48//!     responsibility.
49//!
50use std::ffi::c_void;
51use std::time::Instant;
52
53use cudarc::driver::sys;
54use xlog_core::{Result, ScalarType, Schema, XlogError};
55
56use super::{wcoj_kernels, CudaKernelProvider, WCOJ_MODULE};
57use crate::device_runtime::StreamId;
58use crate::launch::LaunchRecorder;
59use crate::memory::{CudaColumn, DeviceMemoryView, DeviceRead, TrackedCudaSlice};
60use crate::wcoj_metadata::WcojRelationMetadata;
61use crate::CudaBuffer;
62use crate::{AsKernelParam, LaunchAsync, LaunchConfig};
63
64const BLOCK_SIZE: u32 = 256;
65
66fn column_u32(input: &CudaBuffer, col_idx: usize) -> Result<DeviceMemoryView<u32>> {
67    let col = input.column(col_idx).ok_or_else(|| {
68        XlogError::Kernel(format!(
69            "wcoj_layout_u32_recorded: column {col_idx} not found"
70        ))
71    })?;
72    // SAFETY: unsigned integers accept every bit pattern. The checked cast
73    // preserves the storage owner and validates byte length and alignment.
74    unsafe { col.device_view().cast() }
75        .ok_or_else(|| XlogError::Kernel("invalid u32 device column layout".into()))
76}
77
78fn column_u64(input: &CudaBuffer, col_idx: usize) -> Result<DeviceMemoryView<u64>> {
79    let col = input.column(col_idx).ok_or_else(|| {
80        XlogError::Kernel(format!(
81            "wcoj_layout_u64_recorded: column {col_idx} not found"
82        ))
83    })?;
84    // SAFETY: unsigned integers accept every bit pattern. The checked cast
85    // preserves the storage owner and validates byte length and alignment.
86    unsafe { col.device_view().cast() }
87        .ok_or_else(|| XlogError::Kernel("invalid u64 device column layout".into()))
88}
89
90impl CudaKernelProvider {
91    /// Build the sorted+deduped WCOJ physical layout for a 2-column
92    /// u32 relation.
93    ///
94    /// Output: a 2-column u32 [`CudaBuffer`] sorted lexicographically
95    /// by `(col0, col1)` and deduplicated. The output is suitable for
96    /// direct consumption by [`Self::wcoj_triangle_u32_recorded`] in
97    /// any of the three slot positions (`e_xy`, `e_yz`, `e_xz`); the
98    /// caller chooses which logical relation each input represents
99    /// by the slot it passes the layout into.
100    ///
101    /// Fast-path: if the input is already strictly lex-sorted and
102    /// full-row unique, a recorded checker proves that property and
103    /// the method returns a recorded device-side clone. Otherwise it
104    /// falls back to [`Self::dedup_full_row_recorded`], which invokes
105    /// [`Self::sort_recorded`] (typed multi-column radix sort on
106    /// `(col0, col1)`) followed by an on-stream
107    /// `mark_unique_full_row_bytewise` mask + counted compaction.
108    /// Both paths are launch-recorder disciplined and preserve the
109    /// sorted+deduped output contract.
110    ///
111    /// This entry exists for two reasons:
112    ///   1. Narrowing the input contract to 2-column u32 lets the
113    ///      WCOJ-specific call site fail fast with a clear error
114    ///      rather than the more generic dedup error if the caller
115    ///      passes the wrong arity / type.
116    ///   2. Naming the WCOJ pipeline boundary makes downstream
117    ///      callers (planner / executor wiring, cert harness)
118    ///      target the WCOJ-specific layout API rather than the
119    ///      general-purpose dedup primitive — separating concerns
120    ///      that may diverge as the WCOJ stack grows.
121    ///
122    /// # Errors
123    /// * `XlogError::Kernel` if the manager has no runtime
124    ///   (`with_runtime` is required), the input is not 2-column,
125    ///   any column is not [`ScalarType::U32`] or
126    ///   [`ScalarType::Symbol`] (both share the same 4-byte
127    ///   physical layout, so the underlying sort/dedup primitives
128    ///   handle either with no kernel changes), or any inner
129    ///   sort/dedup primitive fails.
130    pub fn wcoj_layout_u32_recorded(
131        &self,
132        input: &CudaBuffer,
133        launch_stream: StreamId,
134    ) -> Result<CudaBuffer> {
135        // Manager must be runtime-backed — the inner
136        // dedup_full_row_recorded enforces the same constraint, but
137        // checking here gives a WCOJ-specific error message.
138        if self.memory().runtime().is_none() {
139            return Err(XlogError::Kernel(
140                "wcoj_layout_u32_recorded requires a runtime-backed \
141                 GpuMemoryManager (constructed via with_runtime)"
142                    .to_string(),
143            ));
144        }
145        // 2-column 4-byte-key contract: U32 or Symbol per column.
146        // Both share the same 4-byte physical layout
147        // (`ScalarType::size_bytes` == 4); the sort+dedup
148        // primitives we delegate to already accept either.
149        if input.arity() != 2 {
150            return Err(XlogError::Kernel(format!(
151                "wcoj_layout_u32_recorded: input must be 2-column, got arity {}",
152                input.arity()
153            )));
154        }
155        for col_idx in 0..2 {
156            let ty = input.schema.column_type(col_idx).ok_or_else(|| {
157                XlogError::Kernel(format!(
158                    "wcoj_layout_u32_recorded: column {} type missing",
159                    col_idx
160                ))
161            })?;
162            if !matches!(ty, ScalarType::U32 | ScalarType::Symbol) {
163                return Err(XlogError::Kernel(format!(
164                    "wcoj_layout_u32_recorded: column {} must be U32 or Symbol, got {:?}",
165                    col_idx, ty
166                )));
167            }
168        }
169        // Layout fast-path: if the input is already strictly
170        // lex-sorted AND full-row unique, we can skip the
171        // (expensive) sort + mark-unique + compact pipeline
172        // and emit a recorded device-side clone. The phase
173        // report (docs/evidence/2026-05-01-wcoj-bench-baseline/phase-timing-report.md)
174        // measured layout at 91-97% of WCOJ adaptive dispatch
175        // wall clock; this branch is the targeted overhead
176        // reduction.
177        match self.try_wcoj_layout_fast_path_u32(input, launch_stream) {
178            Ok(Some(out)) => {
179                self.record_wcoj_layout_fast_path_hit();
180                return Ok(out);
181            }
182            Ok(None) => {
183                // Empty input handled by dedup_full_row_recorded
184                // (which has its own n==0 short-circuit returning
185                // create_empty_buffer). Fall through.
186            }
187            Err(_) => {
188                // Checker failed unexpectedly. Fall through to
189                // the safe path; correctness is preserved.
190            }
191        }
192        // Fall through: the input wasn't proven sorted+unique.
193        // Delegate to the existing typed sort + full-row dedup.
194        // Both primitives are fully recorder-disciplined; the
195        // resulting CudaBuffer is sorted lex by (col0, col1) and
196        // deduplicated.
197        self.dedup_full_row_recorded(input, launch_stream)
198    }
199
200    /// Generic full-row WCOJ layout sort+dedup for relations of any
201    /// arity ≥ 2 in the 4-byte width-class (`U32`, `Symbol`,
202    /// mixable within the class).
203    ///
204    /// **Design**: this entry point leaves the existing arity-2
205    /// [`Self::wcoj_layout_u32_recorded`] is **unchanged** for
206    /// the triangle / 4-cycle / project-then-layout callers — it
207    /// retains its arity-2-specific fast-path branch. This generic
208    /// surface delegates straight to
209    /// [`Self::dedup_full_row_recorded`] for any arity ≥ 2.
210    ///
211    /// **Validation order** (`runtime → arity ≥ 2 → per-column
212    /// width-class → delegate`):
213    ///   1. Manager runtime-backed.
214    ///   2. `input.arity() >= 2`.
215    ///   3. Every column type ∈ `{U32, Symbol}` (4-byte
216    ///      width-class). Mixed `U32` + `Symbol` within one
217    ///      relation is permitted; `U64` is rejected — use
218    ///      [`Self::wcoj_layout_sort_u64_recorded`] instead.
219    ///   4. Delegate to `dedup_full_row_recorded(input, launch_stream)`.
220    ///
221    /// Stream resolution is owned by `dedup_full_row_recorded`
222    /// and is NOT in this entry point's validation list. The
223    /// `n == 0` short-circuit (returns
224    /// `create_empty_buffer(input.schema().clone())`) is also
225    /// owned downstream — single source of truth, no duplicated
226    /// empty-buffer semantics.
227    ///
228    /// **Composition**: `dedup_full_row_recorded` only — there
229    /// is no fast-path branch for arity ≥ 3 in this generic
230    /// full-row layout-sort accessor (the
231    /// existing arity-2 fast-path stays untouched and reachable
232    /// only via `wcoj_layout_u32_recorded`).
233    ///
234    /// # Errors
235    /// * `XlogError::Kernel` if the manager has no runtime
236    ///   (`with_runtime` is required).
237    /// * `XlogError::Kernel` if `input.arity() < 2`.
238    /// * `XlogError::Kernel` if any column is not `U32` /
239    ///   `Symbol`.
240    /// * Whatever `dedup_full_row_recorded` returns for
241    ///   stream-resolution / kernel-launch failures.
242    pub fn wcoj_layout_sort_u32_recorded(
243        &self,
244        input: &CudaBuffer,
245        launch_stream: StreamId,
246    ) -> Result<CudaBuffer> {
247        self.record_wcoj_layout_sort_invocation();
248        if self.memory().runtime().is_none() {
249            return Err(XlogError::Kernel(
250                "wcoj_layout_sort_u32_recorded requires a runtime-backed \
251                 GpuMemoryManager (constructed via with_runtime)"
252                    .to_string(),
253            ));
254        }
255        if input.arity() < 2 {
256            return Err(XlogError::Kernel(format!(
257                "wcoj_layout_sort_u32_recorded: input must have arity >= 2, got {}",
258                input.arity()
259            )));
260        }
261        for col_idx in 0..input.arity() {
262            let ty = input.schema.column_type(col_idx).ok_or_else(|| {
263                XlogError::Kernel(format!(
264                    "wcoj_layout_sort_u32_recorded: column {} type missing",
265                    col_idx
266                ))
267            })?;
268            if !matches!(ty, ScalarType::U32 | ScalarType::Symbol) {
269                return Err(XlogError::Kernel(format!(
270                    "wcoj_layout_sort_u32_recorded: column {} must be U32 or Symbol \
271                     (4-byte width-class), got {:?}",
272                    col_idx, ty
273                )));
274            }
275        }
276        self.dedup_full_row_recorded(input, launch_stream)
277    }
278
279    /// Evaluate `tri(X, Y, Z) :- e_xy(X,Y), e_yz(Y,Z), e_xz(X,Z)`
280    /// on already-sorted, already-deduped binary 4-byte-key
281    /// relations. See module-level docs for the full contract.
282    ///
283    /// Each column may be [`ScalarType::U32`] or
284    /// [`ScalarType::Symbol`] — both share the same 4-byte
285    /// physical layout, so the kernel reads the bits unchanged.
286    /// Cross-relation type compatibility (e.g., that Y is the
287    /// same type in `e_xy.col1` and `e_yz.col0`) is the
288    /// planner's responsibility upstream; this entry only
289    /// enforces width.
290    ///
291    /// The output schema preserves per-head-position scalar types
292    /// from the inputs:
293    ///   * `out.col0` = `e_xy.col0` type (X)
294    ///   * `out.col1` = `e_xy.col1` type (Y)
295    ///   * `out.col2` = `e_yz.col1` type (Z)
296    ///
297    /// # Errors
298    /// * `XlogError::Kernel` if the manager has no runtime
299    ///   (`with_runtime` is required), the launch stream does
300    ///   not resolve, an input is not 2-column with U32/Symbol
301    ///   columns, or any kernel launch fails.
302    pub fn wcoj_triangle_u32_recorded(
303        &self,
304        e_xy: &CudaBuffer,
305        e_yz: &CudaBuffer,
306        e_xz: &CudaBuffer,
307        launch_stream: StreamId,
308    ) -> Result<CudaBuffer> {
309        self.wcoj_triangle_hg_u32_recorded(
310            e_xy,
311            e_yz,
312            e_xz,
313            crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
314            launch_stream,
315        )
316    }
317
318    /// Evaluate `cyc4(W, X, Y, Z) :- e1(W,X), e2(X,Y), e3(Y,Z), e4(Z,W)`
319    /// on already-sorted, already-deduped binary 4-byte-key
320    /// relations. Structural mirror of [`Self::wcoj_triangle_u32_recorded`]
321    /// for the 4-cycle case; see that entry's contract and the
322    /// module-level docs for the shared two-phase recorder
323    /// discipline.
324    ///
325    /// Each column may be [`ScalarType::U32`] or
326    /// [`ScalarType::Symbol`] — both share the same 4-byte
327    /// physical layout, so the kernel reads the bits unchanged.
328    /// Cross-relation type compatibility (e.g., that X is the
329    /// same type in `e1.col1` and `e2.col0`) is the planner's
330    /// responsibility upstream; this entry only enforces width.
331    ///
332    /// The 4-cycle slot order is `[e1(W,X), e2(X,Y), e3(Y,Z), e4(Z,W)]`.
333    /// The output schema preserves per-head-position scalar types
334    /// from the inputs:
335    ///   * `out.col0` = `e1.col0` type (W)
336    ///   * `out.col1` = `e1.col1` type (X)
337    ///   * `out.col2` = `e2.col1` type (Y)
338    ///   * `out.col3` = `e3.col1` type (Z)
339    ///
340    /// # Errors
341    /// * `XlogError::Kernel` if the manager has no runtime
342    ///   (`with_runtime` is required), the launch stream does
343    ///   not resolve, an input is not 2-column with U32/Symbol
344    ///   columns, or any kernel launch fails.
345    pub fn wcoj_4cycle_u32_recorded(
346        &self,
347        e1: &CudaBuffer,
348        e2: &CudaBuffer,
349        e3: &CudaBuffer,
350        e4: &CudaBuffer,
351        launch_stream: StreamId,
352    ) -> Result<CudaBuffer> {
353        self.wcoj_4cycle_hg_u32_recorded(
354            e1,
355            e2,
356            e3,
357            e4,
358            crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
359            launch_stream,
360        )
361    }
362
363    fn logical_row_count_u32(&self, buf: &CudaBuffer) -> Result<u32> {
364        if let Some(c) = buf.cached_row_count() {
365            return Ok(c);
366        }
367        self.dtoh_scalar_untracked::<u32>(buf.num_rows_device(), 0)
368    }
369
370    /// Build the sorted+deduped WCOJ physical layout for a 2-column
371    /// U64 relation. Output: a 2-column U64 [`CudaBuffer`] sorted
372    /// lexicographically by `(col0, col1)` and deduplicated. Suitable
373    /// for direct consumption by [`Self::wcoj_triangle_u64_recorded`].
374    ///
375    /// Composition mirrors [`Self::wcoj_layout_u32_recorded`]:
376    /// already sorted+unique inputs take the recorded fast-path clone;
377    /// other inputs fall back to [`Self::dedup_full_row_recorded`],
378    /// whose U64 `sort_recorded` path ports the legacy `sort()` hi/lo
379    /// radix-pass strategy into recorded launch discipline.
380    ///
381    /// # Errors
382    /// * `XlogError::Kernel` if the manager has no runtime, the
383    ///   input is not 2-column, any column is not
384    ///   [`ScalarType::U64`], or any inner sort/dedup primitive
385    ///   fails.
386    pub fn wcoj_layout_u64_recorded(
387        &self,
388        input: &CudaBuffer,
389        launch_stream: StreamId,
390    ) -> Result<CudaBuffer> {
391        if self.memory().runtime().is_none() {
392            return Err(XlogError::Kernel(
393                "wcoj_layout_u64_recorded requires a runtime-backed \
394                 GpuMemoryManager (constructed via with_runtime)"
395                    .to_string(),
396            ));
397        }
398        if input.arity() != 2 {
399            return Err(XlogError::Kernel(format!(
400                "wcoj_layout_u64_recorded: input must be 2-column, got arity {}",
401                input.arity()
402            )));
403        }
404        for col_idx in 0..2 {
405            let ty = input.schema.column_type(col_idx).ok_or_else(|| {
406                XlogError::Kernel(format!(
407                    "wcoj_layout_u64_recorded: column {} type missing",
408                    col_idx
409                ))
410            })?;
411            if !matches!(ty, ScalarType::U64) {
412                return Err(XlogError::Kernel(format!(
413                    "wcoj_layout_u64_recorded: column {} must be U64, got {:?}",
414                    col_idx, ty
415                )));
416            }
417        }
418        // Fast-path: see u32 entry for rationale + measurement
419        // basis. Strictly lex-sorted AND full-row unique inputs
420        // skip dedup_full_row_recorded.
421        if let Ok(Some(out)) = self.try_wcoj_layout_fast_path_u64(input, launch_stream) {
422            self.record_wcoj_layout_fast_path_hit();
423            return Ok(out);
424        }
425        // dedup_full_row_recorded internally invokes sort_recorded
426        // (U64-aware after commit 1) and the bytewise mask kernel
427        // (already width-generic via col_sizes upload).
428        self.dedup_full_row_recorded(input, launch_stream)
429    }
430
431    /// Generic full-row WCOJ layout sort+dedup for relations of any
432    /// arity ≥ 2 in the 8-byte width-class (`U64` only).
433    ///
434    /// **Design**: this entry point leaves the existing arity-2
435    /// [`Self::wcoj_layout_u64_recorded`] is **unchanged** for
436    /// the existing 2-column callers — it retains its
437    /// arity-2-specific fast-path branch. This generic surface
438    /// delegates straight to [`Self::dedup_full_row_recorded`]
439    /// for any arity ≥ 2.
440    ///
441    /// Mirrors [`Self::wcoj_layout_sort_u32_recorded`]'s contract
442    /// at the 8-byte width-class — see that entry's doc for the
443    /// validation order, stream-resolution ownership, n==0
444    /// semantics, and "no fast-path for arity ≥ 3" lock.
445    ///
446    /// **Validation order** (`runtime → arity ≥ 2 → per-column
447    /// width-class → delegate`):
448    ///   1. Manager runtime-backed.
449    ///   2. `input.arity() >= 2`.
450    ///   3. Every column type = `U64`. `U32` / `Symbol` are
451    ///      rejected — use [`Self::wcoj_layout_sort_u32_recorded`]
452    ///      instead.
453    ///   4. Delegate to `dedup_full_row_recorded(input, launch_stream)`.
454    ///
455    /// # Errors
456    /// * `XlogError::Kernel` if the manager has no runtime
457    ///   (`with_runtime` is required).
458    /// * `XlogError::Kernel` if `input.arity() < 2`.
459    /// * `XlogError::Kernel` if any column is not `U64`.
460    /// * Whatever `dedup_full_row_recorded` returns for
461    ///   stream-resolution / kernel-launch failures.
462    pub fn wcoj_layout_sort_u64_recorded(
463        &self,
464        input: &CudaBuffer,
465        launch_stream: StreamId,
466    ) -> Result<CudaBuffer> {
467        self.record_wcoj_layout_sort_invocation();
468        if self.memory().runtime().is_none() {
469            return Err(XlogError::Kernel(
470                "wcoj_layout_sort_u64_recorded requires a runtime-backed \
471                 GpuMemoryManager (constructed via with_runtime)"
472                    .to_string(),
473            ));
474        }
475        if input.arity() < 2 {
476            return Err(XlogError::Kernel(format!(
477                "wcoj_layout_sort_u64_recorded: input must have arity >= 2, got {}",
478                input.arity()
479            )));
480        }
481        for col_idx in 0..input.arity() {
482            let ty = input.schema.column_type(col_idx).ok_or_else(|| {
483                XlogError::Kernel(format!(
484                    "wcoj_layout_sort_u64_recorded: column {} type missing",
485                    col_idx
486                ))
487            })?;
488            if !matches!(ty, ScalarType::U64) {
489                return Err(XlogError::Kernel(format!(
490                    "wcoj_layout_sort_u64_recorded: column {} must be U64 \
491                     (8-byte width-class), got {:?}",
492                    col_idx, ty
493                )));
494            }
495        }
496        self.dedup_full_row_recorded(input, launch_stream)
497    }
498
499    /// Evaluate `tri(X, Y, Z) :- e_xy(X,Y), e_yz(Y,Z), e_xz(X,Z)`
500    /// on already-sorted, already-deduped binary U64 relations.
501    /// Mirrors [`Self::wcoj_triangle_u32_recorded`]'s contract;
502    /// the only differences are the 8-byte join-key reads/writes
503    /// and the U64-specific count/materialize kernels. Counters
504    /// and the total reducer remain u32.
505    ///
506    /// # Errors
507    /// * `XlogError::Kernel` if the manager has no runtime, the
508    ///   launch stream does not resolve, an input is not 2-column
509    ///   with U64 columns, or any kernel launch fails.
510    pub fn wcoj_triangle_u64_recorded(
511        &self,
512        e_xy: &CudaBuffer,
513        e_yz: &CudaBuffer,
514        e_xz: &CudaBuffer,
515        launch_stream: StreamId,
516    ) -> Result<CudaBuffer> {
517        self.wcoj_triangle_hg_u64_recorded(
518            e_xy,
519            e_yz,
520            e_xz,
521            crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
522            launch_stream,
523        )
524    }
525
526    /// Evaluate
527    /// `cycle4(W, X, Y, Z) :- e1(W,X), e2(X,Y), e3(Y,Z), e4(Z,W)`
528    /// on already-sorted, already-deduped binary U64 relations.
529    /// Mirrors [`Self::wcoj_4cycle_u32_recorded`]'s contract; the
530    /// only differences are the 8-byte join-key reads/writes and
531    /// the U64 HG planner/count/materialize kernels. Counters and
532    /// the total reducer remain u32 (bounded by the upstream host-
533    /// side row-count guard).
534    ///
535    /// 4-cycle slot order:
536    /// `[e1(W,X), e2(X,Y), e3(Y,Z), e4(Z,W)]`.
537    ///
538    /// # Errors
539    /// * `XlogError::Kernel` if the manager has no runtime, the
540    ///   launch stream does not resolve, an input is not 2-column
541    ///   with U64 columns, or any kernel launch fails.
542    pub fn wcoj_4cycle_u64_recorded(
543        &self,
544        e1: &CudaBuffer,
545        e2: &CudaBuffer,
546        e3: &CudaBuffer,
547        e4: &CudaBuffer,
548        launch_stream: StreamId,
549    ) -> Result<CudaBuffer> {
550        self.wcoj_4cycle_hg_u64_recorded(
551            e1,
552            e2,
553            e3,
554            e4,
555            crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
556            launch_stream,
557        )
558    }
559}
560
561// ===============================================================
562// v0.6.2 — WCOJ layout fast-path implementation.
563//
564// Goal: when an input is already strictly lex-sorted AND full-row
565// unique, skip `dedup_full_row_recorded` (sort + mark-unique +
566// compact) and emit a recorded device-side clone instead. The
567// existing layout API surface is unchanged; the fast-path is a
568// purely additive optimization with proof-based correctness.
569//
570// Flow:
571//   1. Validate (caller already did this).
572//   2. Resolve LOGICAL row count via `logical_row_count_u32`
573//      (NOT `input.num_rows()` — that returns row_cap on
574//      compacted buffers).
575//   3. n == 0  → return Ok(None); caller falls through to the
576//      existing `dedup_full_row_recorded` n==0 short-circuit
577//      (`create_empty_buffer`). We don't mint an empty clone
578//      here; we preserve existing semantics exactly.
579//   4. n == 1  → recorded clone (trivially sorted+unique).
580//   5. n >= 2  → launch the checker kernel under a fresh
581//      `LaunchRecorder`; sync; D2H the 4-byte flag.
582//   6. Flag == 1 → recorded clone. Flag == 0 → return
583//      Ok(None); caller falls through to the dedup path.
584//
585// Strict-D2H: the 4-byte flag read uses
586// `dtoh_scalar_untracked::<u32>` (whitelisted by the strict
587// gate, same class as `wcoj_compute_total`'s d_total read).
588// ===============================================================
589
590impl CudaKernelProvider {
591    /// Try to short-circuit a u32/Symbol layout call by proving
592    /// the input is already sorted+unique. Returns:
593    ///   * `Ok(Some(out))` — fast-path hit; `out` is the layout.
594    ///   * `Ok(None)`      — fast-path missed (n==0 or proof
595    ///     failed). Caller falls through to
596    ///     `dedup_full_row_recorded`.
597    ///   * `Err(e)`        — checker pipeline error. Caller
598    ///     treats this as "fall through" to preserve correctness.
599    fn try_wcoj_layout_fast_path_u32(
600        &self,
601        input: &CudaBuffer,
602        launch_stream: StreamId,
603    ) -> Result<Option<CudaBuffer>> {
604        let runtime = self
605            .memory()
606            .runtime()
607            .ok_or_else(|| XlogError::Kernel("wcoj_layout fast-path: no runtime".to_string()))?;
608        let cu_stream = runtime
609            .stream_pool()
610            .resolve(launch_stream)
611            .ok_or_else(|| {
612                XlogError::Kernel("wcoj_layout fast-path: stream resolve".to_string())
613            })?;
614
615        // LOGICAL row count, not row_cap: compacted buffers
616        // (e.g. dedup outputs) have row_cap > logical.
617        let n = self.logical_row_count_u32(input)?;
618        if n == 0 {
619            // Preserve existing semantics: dedup_full_row_recorded's
620            // n==0 path returns create_empty_buffer(schema). Don't
621            // shadow that here.
622            return Ok(None);
623        }
624        if n == 1 {
625            // Trivially sorted+unique; skip the checker entirely.
626            return Ok(Some(self.recorded_clone_2col_4byte(
627                input,
628                n,
629                launch_stream,
630                &cu_stream,
631                runtime,
632            )?));
633        }
634
635        // n >= 2: run the checker. Output flag in u32 (4 bytes).
636        let mut flag_buf = self.memory.alloc::<u32>(1)?;
637
638        let col0 = &column_u32(input, 0)?;
639        let col1 = &column_u32(input, 1)?;
640
641        // Resolve the kernel before queueing launch-stream work.
642        // If the module lookup fails, `flag_buf` can drop without
643        // racing any in-flight H2D / kernel work.
644        let device = self.device.inner();
645        let kernel = device
646            .get_func(
647                WCOJ_MODULE,
648                wcoj_kernels::WCOJ_LAYOUT_CHECK_SORTED_UNIQUE_U32,
649            )
650            .ok_or_else(|| {
651                XlogError::Kernel("wcoj_layout_check_sorted_unique_u32 kernel not found".into())
652            })?;
653
654        let mut rec = LaunchRecorder::new_strict(launch_stream);
655        rec.read(input.num_rows_device());
656        rec.read_column(input.column(0).expect("col0"));
657        rec.read_column(input.column(1).expect("col1"));
658        rec.write(&flag_buf);
659        rec.preflight(runtime)
660            .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path: preflight {e}")))?;
661
662        // Initialize flag = 1 on stream via cuMemsetD32Async-
663        // equivalent. The simplest portable path is a 4-byte
664        // host->device async copy (sequenced before the kernel
665        // by stream order). Doing it as part of the recorded
666        // window keeps the dealloc-safety chain intact.
667        let one: u32 = 1;
668        let grid = n.div_ceil(BLOCK_SIZE);
669        let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
670            self.initialize_launch_metadata_u32(
671                one,
672                &flag_buf,
673                stream,
674                "wcoj_layout fast-path flag init",
675            )?;
676
677            // SAFETY: 4-arg signature
678            //   wcoj_layout_check_sorted_unique_u32(
679            //     const u32* col0, const u32* col1, u32 n, u32* flag)
680            unsafe {
681                kernel
682                    .clone()
683                    .launch_in(
684                        stream,
685                        LaunchConfig {
686                            grid_dim: (grid, 1, 1),
687                            block_dim: (BLOCK_SIZE, 1, 1),
688                            shared_mem_bytes: 0,
689                        },
690                        (col0, col1, n, &mut flag_buf),
691                    )
692                    .map_err(|e| {
693                        XlogError::Kernel(format!(
694                            "wcoj_layout_check_sorted_unique_u32 launch: {e}"
695                        ))
696                    })?;
697            }
698            Ok(())
699        };
700        // SAFETY: preflight retained the accessed buffers and bound this stream.
701        let rec = unsafe { rec.enqueue_prepared_with(&cu_stream, enqueue) }
702            .map_err(|error| error.into_xlog_error())?;
703
704        rec.commit()
705            .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path: commit {e}")))?;
706
707        cu_stream
708            .synchronize()
709            .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path: sync {e}")))?;
710
711        let flag_val = self.dtoh_scalar_untracked::<u32>(&flag_buf, 0)?;
712        if flag_val == 1 {
713            Ok(Some(self.recorded_clone_2col_4byte(
714                input,
715                n,
716                launch_stream,
717                &cu_stream,
718                runtime,
719            )?))
720        } else {
721            Ok(None)
722        }
723    }
724
725    /// U64 variant. Mirrors `try_wcoj_layout_fast_path_u32`.
726    fn try_wcoj_layout_fast_path_u64(
727        &self,
728        input: &CudaBuffer,
729        launch_stream: StreamId,
730    ) -> Result<Option<CudaBuffer>> {
731        let runtime = self
732            .memory()
733            .runtime()
734            .ok_or_else(|| XlogError::Kernel("wcoj_layout fast-path: no runtime".to_string()))?;
735        let cu_stream = runtime
736            .stream_pool()
737            .resolve(launch_stream)
738            .ok_or_else(|| {
739                XlogError::Kernel("wcoj_layout fast-path: stream resolve".to_string())
740            })?;
741
742        let n = self.logical_row_count_u32(input)?;
743        if n == 0 {
744            return Ok(None);
745        }
746        if n == 1 {
747            return Ok(Some(self.recorded_clone_2col_8byte(
748                input,
749                n,
750                launch_stream,
751                &cu_stream,
752                runtime,
753            )?));
754        }
755
756        let mut flag_buf = self.memory.alloc::<u32>(1)?;
757        let col0 = &column_u64(input, 0)?;
758        let col1 = &column_u64(input, 1)?;
759
760        let device = self.device.inner();
761        let kernel = device
762            .get_func(
763                WCOJ_MODULE,
764                wcoj_kernels::WCOJ_LAYOUT_CHECK_SORTED_UNIQUE_U64,
765            )
766            .ok_or_else(|| {
767                XlogError::Kernel("wcoj_layout_check_sorted_unique_u64 kernel not found".into())
768            })?;
769
770        let mut rec = LaunchRecorder::new_strict(launch_stream);
771        rec.read(input.num_rows_device());
772        rec.read_column(input.column(0).expect("col0"));
773        rec.read_column(input.column(1).expect("col1"));
774        rec.write(&flag_buf);
775        rec.preflight(runtime)
776            .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path u64: preflight {e}")))?;
777
778        let one: u32 = 1;
779        let grid = n.div_ceil(BLOCK_SIZE);
780        let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
781            self.initialize_launch_metadata_u32(
782                one,
783                &flag_buf,
784                stream,
785                "wcoj_layout fast-path u64 flag init",
786            )?;
787
788            unsafe {
789                kernel
790                    .clone()
791                    .launch_in(
792                        stream,
793                        LaunchConfig {
794                            grid_dim: (grid, 1, 1),
795                            block_dim: (BLOCK_SIZE, 1, 1),
796                            shared_mem_bytes: 0,
797                        },
798                        (col0, col1, n, &mut flag_buf),
799                    )
800                    .map_err(|e| {
801                        XlogError::Kernel(format!(
802                            "wcoj_layout_check_sorted_unique_u64 launch: {e}"
803                        ))
804                    })?;
805            }
806            Ok(())
807        };
808        // SAFETY: preflight retained the accessed buffers and bound this stream.
809        let rec = unsafe { rec.enqueue_prepared_with(&cu_stream, enqueue) }
810            .map_err(|error| error.into_xlog_error())?;
811
812        rec.commit()
813            .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path u64: commit {e}")))?;
814
815        cu_stream
816            .synchronize()
817            .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path u64: sync {e}")))?;
818
819        let flag_val = self.dtoh_scalar_untracked::<u32>(&flag_buf, 0)?;
820        if flag_val == 1 {
821            Ok(Some(self.recorded_clone_2col_8byte(
822                input,
823                n,
824                launch_stream,
825                &cu_stream,
826                runtime,
827            )?))
828        } else {
829            Ok(None)
830        }
831    }
832
833    /// Recorded device-side clone of a 2-column 4-byte-per-key
834    /// buffer, sized to `n` logical rows. Allocates fresh
835    /// columns + d_num_rows on the runtime allocator and copies
836    /// via `cuMemcpyDtoDAsync_v2` on `launch_stream` under a
837    /// `LaunchRecorder` window. NOT a view: the output buffer
838    /// owns its bytes; input lifetime independence is preserved.
839    fn recorded_clone_2col_4byte(
840        &self,
841        input: &CudaBuffer,
842        n: u32,
843        launch_stream: StreamId,
844        cu_stream: &cudarc::driver::CudaStream,
845        runtime: &std::sync::Arc<crate::device_runtime::XlogDeviceRuntime>,
846    ) -> Result<CudaBuffer> {
847        let bpc = (n as usize) * 4;
848        let out_col0 = self.memory.alloc::<u8>(bpc)?;
849        let out_col1 = self.memory.alloc::<u8>(bpc)?;
850        let out_d_num_rows = self.memory.alloc::<u32>(1)?;
851
852        let src_col0 = input.column(0).expect("col0");
853        let src_col1 = input.column(1).expect("col1");
854
855        let mut rec = LaunchRecorder::new_strict(launch_stream);
856        rec.read(input.num_rows_device());
857        rec.read_column(src_col0);
858        rec.read_column(src_col1);
859        rec.write(&out_col0);
860        rec.write(&out_col1);
861        rec.write(&out_d_num_rows);
862        rec.preflight(runtime)
863            .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 4B: preflight {e}")))?;
864
865        let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
866            unsafe {
867                let r0 = sys::cuMemcpyDtoDAsync_v2(
868                    *out_col0.device_ptr(),
869                    *src_col0.device_ptr(),
870                    bpc,
871                    stream.stream().cu_stream(),
872                );
873                if r0 != sys::cudaError_enum::CUDA_SUCCESS {
874                    return Err(XlogError::Kernel(format!(
875                        "wcoj_layout clone 4B: dtod col0 failed: {r0:?}"
876                    )));
877                }
878                let r1 = sys::cuMemcpyDtoDAsync_v2(
879                    *out_col1.device_ptr(),
880                    *src_col1.device_ptr(),
881                    bpc,
882                    stream.stream().cu_stream(),
883                );
884                if r1 != sys::cudaError_enum::CUDA_SUCCESS {
885                    return Err(XlogError::Kernel(format!(
886                        "wcoj_layout clone 4B: dtod col1 failed: {r1:?}"
887                    )));
888                }
889            }
890            self.initialize_launch_metadata_u32(
891                n,
892                &out_d_num_rows,
893                stream,
894                "wcoj_layout clone 4B d_num_rows",
895            )?;
896            Ok(())
897        };
898        // SAFETY: preflight retained the accessed buffers and bound this stream.
899        let rec = unsafe { rec.enqueue_prepared_with(cu_stream, enqueue) }
900            .map_err(|error| error.into_xlog_error())?;
901
902        rec.commit()
903            .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 4B: commit {e}")))?;
904
905        Ok(CudaBuffer::from_columns_with_host_count(
906            vec![out_col0.into(), out_col1.into()],
907            n as u64,
908            out_d_num_rows,
909            input.schema().clone(),
910            n,
911        ))
912    }
913
914    /// 8-byte-per-key sibling. Same recorder discipline.
915    fn recorded_clone_2col_8byte(
916        &self,
917        input: &CudaBuffer,
918        n: u32,
919        launch_stream: StreamId,
920        cu_stream: &cudarc::driver::CudaStream,
921        runtime: &std::sync::Arc<crate::device_runtime::XlogDeviceRuntime>,
922    ) -> Result<CudaBuffer> {
923        let bpc = (n as usize) * 8;
924        let out_col0 = self.memory.alloc::<u8>(bpc)?;
925        let out_col1 = self.memory.alloc::<u8>(bpc)?;
926        let out_d_num_rows = self.memory.alloc::<u32>(1)?;
927
928        let src_col0 = input.column(0).expect("col0");
929        let src_col1 = input.column(1).expect("col1");
930
931        let mut rec = LaunchRecorder::new_strict(launch_stream);
932        rec.read(input.num_rows_device());
933        rec.read_column(src_col0);
934        rec.read_column(src_col1);
935        rec.write(&out_col0);
936        rec.write(&out_col1);
937        rec.write(&out_d_num_rows);
938        rec.preflight(runtime)
939            .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 8B: preflight {e}")))?;
940
941        let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
942            unsafe {
943                let r0 = sys::cuMemcpyDtoDAsync_v2(
944                    *out_col0.device_ptr(),
945                    *src_col0.device_ptr(),
946                    bpc,
947                    stream.stream().cu_stream(),
948                );
949                if r0 != sys::cudaError_enum::CUDA_SUCCESS {
950                    return Err(XlogError::Kernel(format!(
951                        "wcoj_layout clone 8B: dtod col0 failed: {r0:?}"
952                    )));
953                }
954                let r1 = sys::cuMemcpyDtoDAsync_v2(
955                    *out_col1.device_ptr(),
956                    *src_col1.device_ptr(),
957                    bpc,
958                    stream.stream().cu_stream(),
959                );
960                if r1 != sys::cudaError_enum::CUDA_SUCCESS {
961                    return Err(XlogError::Kernel(format!(
962                        "wcoj_layout clone 8B: dtod col1 failed: {r1:?}"
963                    )));
964                }
965            }
966            self.initialize_launch_metadata_u32(
967                n,
968                &out_d_num_rows,
969                stream,
970                "wcoj_layout clone 8B d_num_rows",
971            )?;
972            Ok(())
973        };
974        // SAFETY: preflight retained the accessed buffers and bound this stream.
975        let rec = unsafe { rec.enqueue_prepared_with(cu_stream, enqueue) }
976            .map_err(|error| error.into_xlog_error())?;
977
978        rec.commit()
979            .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 8B: commit {e}")))?;
980
981        Ok(CudaBuffer::from_columns_with_host_count(
982            vec![out_col0.into(), out_col1.into()],
983            n as u64,
984            out_d_num_rows,
985            input.schema().clone(),
986            n,
987        ))
988    }
989}
990
991// ===============================================================
992// General-arity clique WCOJ provider for K=5..8.
993//
994// Thin public methods (k=5..8 × u32/u64) delegate to a
995// single generic helper `wcoj_clique_recorded_inner`. Width-class
996// (4-byte = U32+Symbol mixable, 8-byte = U64) and K
997// drive kernel-name selection and per-row element-size; otherwise
998// the orchestration is identical: validate → upload edge-pointer
999// arrays → count → scan → total → materialize → output.
1000//
1001// Each public entry assumes sorted+deduped input as a
1002// pre-condition (same contract as `wcoj_triangle_u32_recorded` /
1003// `wcoj_4cycle_u32_recorded`); the runtime dispatcher routes
1004// every edge through the generic full-row layout-sort accessors before
1005// invoking these entries.
1006// ===============================================================
1007
1008#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1009enum CliqueWidthClass {
1010    FourByte,  // U32 / Symbol
1011    EightByte, // U64
1012}
1013
1014impl CliqueWidthClass {
1015    fn elem_bytes(self) -> usize {
1016        match self {
1017            CliqueWidthClass::FourByte => 4,
1018            CliqueWidthClass::EightByte => 8,
1019        }
1020    }
1021    fn label(self) -> &'static str {
1022        match self {
1023            CliqueWidthClass::FourByte => "u32",
1024            CliqueWidthClass::EightByte => "u64",
1025        }
1026    }
1027    fn validate_col_type(self, ty: ScalarType) -> bool {
1028        match self {
1029            CliqueWidthClass::FourByte => matches!(ty, ScalarType::U32 | ScalarType::Symbol),
1030            CliqueWidthClass::EightByte => matches!(ty, ScalarType::U64),
1031        }
1032    }
1033}
1034
1035/// Resolve the kernel name for a given `(K, count_or_materialize, width_class)`.
1036fn clique_kernel_name(k: usize, materialize: bool, w: CliqueWidthClass) -> &'static str {
1037    match (k, materialize, w) {
1038        (5, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE5_COUNT_HG_U32,
1039        (5, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE5_MATERIALIZE_HG_U32,
1040        (5, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE5_COUNT_HG_U64,
1041        (5, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE5_MATERIALIZE_HG_U64,
1042        (6, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE6_COUNT_HG_U32,
1043        (6, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE6_MATERIALIZE_HG_U32,
1044        (6, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE6_COUNT_HG_U64,
1045        (6, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE6_MATERIALIZE_HG_U64,
1046        (7, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE7_COUNT_HG_U32,
1047        (7, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE7_MATERIALIZE_HG_U32,
1048        (7, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE7_COUNT_HG_U64,
1049        (7, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE7_MATERIALIZE_HG_U64,
1050        (8, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE8_COUNT_HG_U32,
1051        (8, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE8_MATERIALIZE_HG_U32,
1052        (8, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE8_COUNT_HG_U64,
1053        (8, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE8_MATERIALIZE_HG_U64,
1054        _ => panic!("clique_kernel_name: K must be 5..8, got {}", k),
1055    }
1056}
1057
1058enum CliqueLeaderMetadata {
1059    U32(WcojRelationMetadata<u32>),
1060    U64(WcojRelationMetadata<u64>),
1061}
1062
1063impl CliqueLeaderMetadata {
1064    fn total_rows_u32(&self, entry_label: &str) -> Result<u32> {
1065        let total = match self {
1066            CliqueLeaderMetadata::U32(metadata) => metadata.total,
1067            CliqueLeaderMetadata::U64(metadata) => metadata.total,
1068        };
1069        u32::try_from(total).map_err(|_| {
1070            XlogError::Kernel(format!(
1071                "{}: leader metadata total {} exceeds u32 kernel surface",
1072                entry_label, total
1073            ))
1074        })
1075    }
1076
1077    fn key_count(&self) -> u32 {
1078        match self {
1079            CliqueLeaderMetadata::U32(metadata) => metadata.key_count,
1080            CliqueLeaderMetadata::U64(metadata) => metadata.key_count,
1081        }
1082    }
1083}
1084
1085fn validate_clique_metadata_leader<'a>(
1086    k: usize,
1087    edges: &'a [&CudaBuffer],
1088    leader_edge_idx: u32,
1089    width_class: CliqueWidthClass,
1090    entry_label: &str,
1091) -> Result<&'a CudaBuffer> {
1092    if !(5..=8).contains(&k) {
1093        return Err(XlogError::Kernel(format!(
1094            "{}: k must be 5..8, got {}",
1095            entry_label, k
1096        )));
1097    }
1098    let expected_edges = k * (k - 1) / 2;
1099    if edges.len() != expected_edges {
1100        return Err(XlogError::Kernel(format!(
1101            "{}: expected {} edges (= C({}, 2)), got {}",
1102            entry_label,
1103            expected_edges,
1104            k,
1105            edges.len()
1106        )));
1107    }
1108    let leader_slot = usize::try_from(leader_edge_idx)
1109        .ok()
1110        .filter(|idx| *idx < expected_edges)
1111        .ok_or_else(|| {
1112            XlogError::Kernel(format!(
1113                "{}: leader_edge_idx {} out of range for {} edges",
1114                entry_label, leader_edge_idx, expected_edges
1115            ))
1116        })?;
1117    let leader = edges[leader_slot];
1118    if leader.arity() != 2 {
1119        return Err(XlogError::Kernel(format!(
1120            "{}: leader edge must be 2-column, got arity {}",
1121            entry_label,
1122            leader.arity()
1123        )));
1124    }
1125    let ty = leader.schema.column_type(0).ok_or_else(|| {
1126        XlogError::Kernel(format!(
1127            "{}: leader edge column 0 type missing",
1128            entry_label
1129        ))
1130    })?;
1131    if !width_class.validate_col_type(ty) {
1132        return Err(XlogError::Kernel(format!(
1133            "{}: leader edge column 0 type {:?} not in {} width-class",
1134            entry_label,
1135            ty,
1136            width_class.label()
1137        )));
1138    }
1139    Ok(leader)
1140}
1141
1142impl CudaKernelProvider {
1143    fn wcoj_clique_metadata_recorded_u32_inner(
1144        &self,
1145        k: usize,
1146        edges: &[&CudaBuffer],
1147        leader_edge_idx: u32,
1148        launch_stream: StreamId,
1149        entry_label: &str,
1150    ) -> Result<WcojRelationMetadata<u32>> {
1151        let leader = validate_clique_metadata_leader(
1152            k,
1153            edges,
1154            leader_edge_idx,
1155            CliqueWidthClass::FourByte,
1156            entry_label,
1157        )?;
1158        self.wcoj_build_metadata_u32_recorded(leader, 0, launch_stream)
1159    }
1160
1161    fn wcoj_clique_metadata_recorded_u64_inner(
1162        &self,
1163        k: usize,
1164        edges: &[&CudaBuffer],
1165        leader_edge_idx: u32,
1166        launch_stream: StreamId,
1167        entry_label: &str,
1168    ) -> Result<WcojRelationMetadata<u64>> {
1169        let leader = validate_clique_metadata_leader(
1170            k,
1171            edges,
1172            leader_edge_idx,
1173            CliqueWidthClass::EightByte,
1174            entry_label,
1175        )?;
1176        self.wcoj_build_metadata_u64_recorded(leader, 0, launch_stream)
1177    }
1178
1179    /// Build leader-edge runtime metadata for a 5-clique 4-byte-width dispatch.
1180    pub fn wcoj_clique5_metadata_recorded_u32(
1181        &self,
1182        edges: &[&CudaBuffer; 10],
1183        leader_edge_idx: u32,
1184        launch_stream: StreamId,
1185    ) -> Result<WcojRelationMetadata<u32>> {
1186        self.wcoj_clique_metadata_recorded_u32_inner(
1187            5,
1188            edges,
1189            leader_edge_idx,
1190            launch_stream,
1191            "wcoj_clique5_metadata_recorded_u32",
1192        )
1193    }
1194
1195    /// Build leader-edge runtime metadata for a 5-clique 8-byte-width dispatch.
1196    pub fn wcoj_clique5_metadata_recorded_u64(
1197        &self,
1198        edges: &[&CudaBuffer; 10],
1199        leader_edge_idx: u32,
1200        launch_stream: StreamId,
1201    ) -> Result<WcojRelationMetadata<u64>> {
1202        self.wcoj_clique_metadata_recorded_u64_inner(
1203            5,
1204            edges,
1205            leader_edge_idx,
1206            launch_stream,
1207            "wcoj_clique5_metadata_recorded_u64",
1208        )
1209    }
1210
1211    /// Build leader-edge runtime metadata for a 6-clique 4-byte-width dispatch.
1212    pub fn wcoj_clique6_metadata_recorded_u32(
1213        &self,
1214        edges: &[&CudaBuffer; 15],
1215        leader_edge_idx: u32,
1216        launch_stream: StreamId,
1217    ) -> Result<WcojRelationMetadata<u32>> {
1218        self.wcoj_clique_metadata_recorded_u32_inner(
1219            6,
1220            edges,
1221            leader_edge_idx,
1222            launch_stream,
1223            "wcoj_clique6_metadata_recorded_u32",
1224        )
1225    }
1226
1227    /// Build leader-edge runtime metadata for a 6-clique 8-byte-width dispatch.
1228    pub fn wcoj_clique6_metadata_recorded_u64(
1229        &self,
1230        edges: &[&CudaBuffer; 15],
1231        leader_edge_idx: u32,
1232        launch_stream: StreamId,
1233    ) -> Result<WcojRelationMetadata<u64>> {
1234        self.wcoj_clique_metadata_recorded_u64_inner(
1235            6,
1236            edges,
1237            leader_edge_idx,
1238            launch_stream,
1239            "wcoj_clique6_metadata_recorded_u64",
1240        )
1241    }
1242
1243    /// Generic clique provider helper. Orchestrates count
1244    /// → scan → total → materialize for K-clique on K*(K-1)/2
1245    /// 2-column edges in the given width-class.
1246    ///
1247    /// Caller pre-conditions:
1248    ///   * Manager runtime-backed (validated here too).
1249    ///   * `K ∈ {5, 6, 7, 8}` (validated; panic-free `Err` otherwise).
1250    ///   * `edges.len() == K * (K - 1) / 2`.
1251    ///   * Each edge is 2-column with all columns in `width_class`.
1252    ///   * Each edge is lex-sorted+deduped on `(col0, col1)` —
1253    ///     same contract as `wcoj_triangle_*_recorded`. The
1254    ///     runtime dispatcher routes every edge through the generic
1255    ///     full-row layout-sort accessors before
1256    ///     calling here; provider does NOT layout-sort itself.
1257    #[expect(
1258        clippy::too_many_arguments,
1259        reason = "the clique launch core mirrors the kernel's explicit inputs, variable layout, output projection, limits, and stream"
1260    )]
1261    fn wcoj_clique_recorded_inner(
1262        &self,
1263        k: usize,
1264        edges: &[&CudaBuffer],
1265        leader_edge_idx: u32,
1266        edge_order: Option<&[u8]>,
1267        iteration_order: Option<&[u8]>,
1268        width_class: CliqueWidthClass,
1269        launch_stream: StreamId,
1270        entry_label: &str,
1271    ) -> Result<CudaBuffer> {
1272        let runtime = self.memory().runtime().ok_or_else(|| {
1273            XlogError::Kernel(format!(
1274                "{} requires a runtime-backed GpuMemoryManager (with_runtime)",
1275                entry_label
1276            ))
1277        })?;
1278        let cu_stream = runtime
1279            .stream_pool()
1280            .resolve(launch_stream)
1281            .ok_or_else(|| {
1282                XlogError::Kernel(format!(
1283                    "{}: launch_stream StreamId({}) does not resolve",
1284                    entry_label, launch_stream.0
1285                ))
1286            })?;
1287        if !(5..=8).contains(&k) {
1288            return Err(XlogError::Kernel(format!(
1289                "{}: k must be 5..8, got {}",
1290                entry_label, k
1291            )));
1292        }
1293        let expected_edges = k * (k - 1) / 2;
1294        if edges.len() != expected_edges {
1295            return Err(XlogError::Kernel(format!(
1296                "{}: expected {} edges (= C({}, 2)), got {}",
1297                entry_label,
1298                expected_edges,
1299                k,
1300                edges.len()
1301            )));
1302        }
1303        if usize::try_from(leader_edge_idx)
1304            .ok()
1305            .is_none_or(|idx| idx >= expected_edges)
1306        {
1307            return Err(XlogError::Kernel(format!(
1308                "{}: leader_edge_idx {} out of range for {} edges",
1309                entry_label, leader_edge_idx, expected_edges
1310            )));
1311        }
1312        match (edge_order, iteration_order) {
1313            (Some(edge_order), Some(iteration_order)) => {
1314                validate_clique_u8_permutation(
1315                    edge_order,
1316                    expected_edges,
1317                    "edge_order",
1318                    entry_label,
1319                )?;
1320                validate_clique_u8_permutation(iteration_order, k, "iteration_order", entry_label)?;
1321            }
1322            (None, None) => {}
1323            _ => {
1324                return Err(XlogError::Kernel(format!(
1325                    "{}: edge_order and iteration_order must both be present or both be omitted",
1326                    entry_label
1327                )));
1328            }
1329        }
1330
1331        // Validate every edge: 2-column, in width-class.
1332        for (i, buf) in edges.iter().enumerate() {
1333            if buf.arity() != 2 {
1334                return Err(XlogError::Kernel(format!(
1335                    "{}: edge[{}] must be 2-column, got arity {}",
1336                    entry_label,
1337                    i,
1338                    buf.arity()
1339                )));
1340            }
1341            for col_idx in 0..2 {
1342                let ty = buf.schema.column_type(col_idx).ok_or_else(|| {
1343                    XlogError::Kernel(format!(
1344                        "{}: edge[{}] column {} type missing",
1345                        entry_label, i, col_idx
1346                    ))
1347                })?;
1348                if !width_class.validate_col_type(ty) {
1349                    return Err(XlogError::Kernel(format!(
1350                        "{}: edge[{}] column {} type {:?} not in {} width-class",
1351                        entry_label,
1352                        i,
1353                        col_idx,
1354                        ty,
1355                        width_class.label()
1356                    )));
1357                }
1358            }
1359        }
1360
1361        // Build output schema in kernel binding order. The runtime
1362        // projects this buffer back to rule-head order when the plan
1363        // chooses a non-identity variable order.
1364        let mut head_types = Vec::with_capacity(k);
1365        let leader_slot = edge_order.map(|order| order[0] as usize).unwrap_or(0);
1366        head_types.push(edges[leader_slot].schema.column_type(0).expect("validated"));
1367        head_types.push(edges[leader_slot].schema.column_type(1).expect("validated"));
1368        for i in 2..k {
1369            let logical_edge = i - 1;
1370            let edge_slot = edge_order
1371                .map(|order| order[logical_edge] as usize)
1372                .unwrap_or(logical_edge);
1373            head_types.push(edges[edge_slot].schema.column_type(1).expect("validated"));
1374        }
1375        let out_schema = Schema::new(
1376            head_types
1377                .iter()
1378                .enumerate()
1379                .map(|(i, t)| (format!("col{}", i), *t))
1380                .collect(),
1381        );
1382
1383        let leader_slot = usize::try_from(leader_edge_idx).expect("validated");
1384        let n_leader = self.logical_row_count_u32(edges[leader_slot])?;
1385        if n_leader == 0 {
1386            return self.create_empty_buffer(out_schema);
1387        }
1388        // Paper section 5 Algorithm 1 Phase 1: histograms are maintained
1389        // alongside data and refreshed during recursive merge handling.
1390        let metadata_start = Instant::now();
1391        let leader_metadata = match width_class {
1392            CliqueWidthClass::FourByte => {
1393                CliqueLeaderMetadata::U32(self.wcoj_clique_metadata_recorded_u32_inner(
1394                    k,
1395                    edges,
1396                    leader_edge_idx,
1397                    launch_stream,
1398                    entry_label,
1399                )?)
1400            }
1401            CliqueWidthClass::EightByte => {
1402                CliqueLeaderMetadata::U64(self.wcoj_clique_metadata_recorded_u64_inner(
1403                    k,
1404                    edges,
1405                    leader_edge_idx,
1406                    launch_stream,
1407                    entry_label,
1408                )?)
1409            }
1410        };
1411        self.record_kclique_metadata_build_nanos(metadata_start.elapsed().as_nanos());
1412        let leader_work_total = leader_metadata.total_rows_u32(entry_label)?;
1413        if leader_work_total != n_leader {
1414            return Err(XlogError::Kernel(format!(
1415                "{}: leader metadata total {} does not match leader row count {}",
1416                entry_label, leader_work_total, n_leader
1417            )));
1418        }
1419        let leader_metadata_key_count = leader_metadata.key_count();
1420        if leader_metadata_key_count == 0 {
1421            return self.create_empty_buffer(out_schema);
1422        }
1423
1424        // Build host-side per-edge pointer arrays + row-count
1425        // array. These get htod'd to small device buffers that
1426        // the kernel reads as `const T* const* edge_col0` etc.
1427        let mut edge_col0_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
1428        let mut edge_col1_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
1429        let mut edge_n_host: Vec<u32> = Vec::with_capacity(expected_edges);
1430        for buf in edges.iter() {
1431            let col0 = buf.column(0).expect("validated");
1432            let col1 = buf.column(1).expect("validated");
1433            edge_col0_ptrs.push(*col0.device_ptr());
1434            edge_col1_ptrs.push(*col1.device_ptr());
1435            edge_n_host.push(self.logical_row_count_u32(buf)?);
1436        }
1437
1438        // Allocate device-side pointer arrays + row counts.
1439        let mut d_edge_col0 = self.memory.alloc::<u64>(expected_edges)?;
1440        let mut d_edge_col1 = self.memory.alloc::<u64>(expected_edges)?;
1441        let mut d_edge_n = self.memory.alloc::<u32>(expected_edges)?;
1442        let device = self.device.inner();
1443        self.htod_launch_metadata_sync_copy_into(&edge_col0_ptrs, &mut d_edge_col0)
1444            .map_err(|e| {
1445                XlogError::Kernel(format!(
1446                    "{}: htod edge_col0_ptrs failed: {}",
1447                    entry_label, e
1448                ))
1449            })?;
1450        self.htod_launch_metadata_sync_copy_into(&edge_col1_ptrs, &mut d_edge_col1)
1451            .map_err(|e| {
1452                XlogError::Kernel(format!(
1453                    "{}: htod edge_col1_ptrs failed: {}",
1454                    entry_label, e
1455                ))
1456            })?;
1457        self.htod_launch_metadata_sync_copy_into(&edge_n_host, &mut d_edge_n)
1458            .map_err(|e| {
1459                XlogError::Kernel(format!("{}: htod edge_n failed: {}", entry_label, e))
1460            })?;
1461        let d_edge_order = if let Some(edge_order) = edge_order {
1462            let mut buf = self.memory.alloc::<u8>(expected_edges)?;
1463            self.htod_launch_metadata_sync_copy_into(edge_order, &mut buf)
1464                .map_err(|e| {
1465                    XlogError::Kernel(format!("{}: htod edge_order failed: {}", entry_label, e))
1466                })?;
1467            Some(buf)
1468        } else {
1469            None
1470        };
1471        let d_iteration_order = if let Some(iteration_order) = iteration_order {
1472            let mut buf = self.memory.alloc::<u8>(k)?;
1473            self.htod_launch_metadata_sync_copy_into(iteration_order, &mut buf)
1474                .map_err(|e| {
1475                    XlogError::Kernel(format!(
1476                        "{}: htod iteration_order failed: {}",
1477                        entry_label, e
1478                    ))
1479                })?;
1480            Some(buf)
1481        } else {
1482            None
1483        };
1484
1485        // Phase 1: HG block counts + scan + total.
1486        let block_work_unit = crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT;
1487        let grid = leader_work_total.div_ceil(block_work_unit);
1488        let count_buf = self.memory.alloc::<u32>(grid as usize)?;
1489        let thread_counts_buf = self
1490            .memory
1491            .alloc::<u32>((grid as usize) * (BLOCK_SIZE as usize))?;
1492        let mut offsets_buf = self.memory.alloc::<u32>(grid as usize)?;
1493        let d_total = self.memory.alloc::<u32>(1)?;
1494
1495        let mut scan_scratch = self.multiblock_scan_u32_scratch_for_len(grid)?;
1496        let mut rec_count = LaunchRecorder::new_strict(launch_stream);
1497        for buf in edges.iter() {
1498            rec_count.read(buf.num_rows_device());
1499            rec_count.read_column(buf.column(0).expect("validated"));
1500            rec_count.read_column(buf.column(1).expect("validated"));
1501        }
1502        rec_count.read(&d_edge_col0);
1503        rec_count.read(&d_edge_col1);
1504        rec_count.read(&d_edge_n);
1505        if let Some(buf) = d_edge_order.as_ref() {
1506            rec_count.read(buf);
1507        }
1508        if let Some(buf) = d_iteration_order.as_ref() {
1509            rec_count.read(buf);
1510        }
1511        match &leader_metadata {
1512            CliqueLeaderMetadata::U32(leader_metadata) => {
1513                rec_count.read(&leader_metadata.unique_keys);
1514                rec_count.read(&leader_metadata.fan_out);
1515                rec_count.read(&leader_metadata.prefix_sum);
1516            }
1517            CliqueLeaderMetadata::U64(leader_metadata) => {
1518                rec_count.read(&leader_metadata.unique_keys);
1519                rec_count.read(&leader_metadata.fan_out);
1520                rec_count.read(&leader_metadata.prefix_sum);
1521            }
1522        }
1523        rec_count.write(&count_buf);
1524        rec_count.write(&thread_counts_buf);
1525        rec_count.write(&offsets_buf);
1526        rec_count.write(&d_total);
1527        for level in scan_scratch.levels() {
1528            rec_count.read_write(level);
1529        }
1530        rec_count.preflight(runtime).map_err(|e| {
1531            XlogError::Kernel(format!("{}: count preflight failed: {}", entry_label, e))
1532        })?;
1533
1534        let count_kernel = device
1535            .get_func(WCOJ_MODULE, clique_kernel_name(k, false, width_class))
1536            .ok_or_else(|| {
1537                XlogError::Kernel(format!(
1538                    "{}: count kernel '{}' not found",
1539                    entry_label,
1540                    clique_kernel_name(k, false, width_class)
1541                ))
1542            })?;
1543        let count_config = LaunchConfig {
1544            grid_dim: (grid, 1, 1),
1545            block_dim: (BLOCK_SIZE, 1, 1),
1546            shared_mem_bytes: 0,
1547        };
1548
1549        // SAFETY: kernel signature
1550        //   wcoj_clique{K}_count_*(
1551        //     const T* const* edge_col0,
1552        //     const T* const* edge_col1,
1553        //     const u32* edge_n,
1554        //     u32 leader_edge_idx,
1555        //     const u8* edge_order,
1556        //     const u8* iteration_order,
1557        //     u32 leader_count,
1558        //     const T* unique_keys,
1559        //     const u32* fan_out,
1560        //     const u32* prefix_sum,
1561        //     u32 metadata_key_count,
1562        //     u32 block_work_unit,
1563        //     u32* out_block_counts,
1564        //     u32* out_thread_counts)
1565        // Pointers all device-resident; preflight verified
1566        // cross-stream tracking. Raw params are required because
1567        // the metadata-extended ABI exceeds the tuple-launch arity.
1568        let null_order_ptr = 0_u64;
1569        let edge_order_param = match d_edge_order.as_ref() {
1570            Some(buf) => buf.as_kernel_param(),
1571            None => null_order_ptr.as_kernel_param(),
1572        };
1573        let iteration_order_param = match d_iteration_order.as_ref() {
1574            Some(buf) => buf.as_kernel_param(),
1575            None => null_order_ptr.as_kernel_param(),
1576        };
1577        let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
1578            unsafe {
1579                let mut params: Vec<*mut c_void> = match &leader_metadata {
1580                    CliqueLeaderMetadata::U32(leader_metadata) => vec![
1581                        (&d_edge_col0).as_kernel_param(),
1582                        (&d_edge_col1).as_kernel_param(),
1583                        (&d_edge_n).as_kernel_param(),
1584                        leader_edge_idx.as_kernel_param(),
1585                        edge_order_param,
1586                        iteration_order_param,
1587                        n_leader.as_kernel_param(),
1588                        (&leader_metadata.unique_keys).as_kernel_param(),
1589                        (&leader_metadata.fan_out).as_kernel_param(),
1590                        (&leader_metadata.prefix_sum).as_kernel_param(),
1591                        leader_metadata_key_count.as_kernel_param(),
1592                        block_work_unit.as_kernel_param(),
1593                        (&count_buf).as_kernel_param(),
1594                        (&thread_counts_buf).as_kernel_param(),
1595                    ],
1596                    CliqueLeaderMetadata::U64(leader_metadata) => vec![
1597                        (&d_edge_col0).as_kernel_param(),
1598                        (&d_edge_col1).as_kernel_param(),
1599                        (&d_edge_n).as_kernel_param(),
1600                        leader_edge_idx.as_kernel_param(),
1601                        edge_order_param,
1602                        iteration_order_param,
1603                        n_leader.as_kernel_param(),
1604                        (&leader_metadata.unique_keys).as_kernel_param(),
1605                        (&leader_metadata.fan_out).as_kernel_param(),
1606                        (&leader_metadata.prefix_sum).as_kernel_param(),
1607                        leader_metadata_key_count.as_kernel_param(),
1608                        block_work_unit.as_kernel_param(),
1609                        (&count_buf).as_kernel_param(),
1610                        (&thread_counts_buf).as_kernel_param(),
1611                    ],
1612                };
1613                count_kernel
1614                    .clone()
1615                    .launch_in(stream, count_config, &mut params)
1616                    .map_err(|e| {
1617                        XlogError::Kernel(format!(
1618                            "{}: count kernel launch failed: {}",
1619                            entry_label, e
1620                        ))
1621                    })?;
1622            }
1623
1624            // dtod count → offsets (scan modifies offsets in place).
1625            let bytes_count = (grid as usize) * std::mem::size_of::<u32>();
1626            unsafe {
1627                let res = sys::cuMemcpyDtoDAsync_v2(
1628                    *offsets_buf.device_ptr(),
1629                    *count_buf.device_ptr(),
1630                    bytes_count,
1631                    stream.stream().cu_stream(),
1632                );
1633                if res != sys::cudaError_enum::CUDA_SUCCESS {
1634                    return Err(XlogError::Kernel(format!(
1635                        "{}: dtod count → offsets failed: {:?}",
1636                        entry_label, res
1637                    )));
1638                }
1639            }
1640
1641            // Device-side exclusive prefix-sum on offsets.
1642            self.multiblock_scan_u32_inplace_on_stream(
1643                &mut offsets_buf,
1644                grid,
1645                stream,
1646                &mut scan_scratch,
1647            )?;
1648
1649            // Compute total = counts[n-1] + offsets[n-1].
1650            let total_kernel = device
1651                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_COMPUTE_TOTAL)
1652                .ok_or_else(|| {
1653                    XlogError::Kernel("wcoj_compute_total kernel not found".to_string())
1654                })?;
1655            unsafe {
1656                total_kernel
1657                    .clone()
1658                    .launch_in(
1659                        stream,
1660                        LaunchConfig {
1661                            grid_dim: (1, 1, 1),
1662                            block_dim: (1, 1, 1),
1663                            shared_mem_bytes: 0,
1664                        },
1665                        (&count_buf, &offsets_buf, grid, &d_total),
1666                    )
1667                    .map_err(|e| {
1668                        XlogError::Kernel(format!("wcoj_compute_total launch failed: {}", e))
1669                    })?;
1670            }
1671
1672            Ok(())
1673        };
1674        // SAFETY: preflight retained the accessed buffers and bound this stream.
1675        let rec_count = unsafe { rec_count.enqueue_prepared_with(&cu_stream, enqueue) }
1676            .map_err(|error| error.into_xlog_error())?;
1677
1678        rec_count.commit().map_err(|e| {
1679            XlogError::Kernel(format!(
1680                "{}: count+scan+total commit failed: {}",
1681                entry_label, e
1682            ))
1683        })?;
1684
1685        cu_stream.synchronize().map_err(|e| {
1686            XlogError::Kernel(format!(
1687                "{}: stream sync after total failed: {}",
1688                entry_label, e
1689            ))
1690        })?;
1691        let total_rows = self
1692            .dtoh_scalar_untracked::<u32>(&d_total, 0)
1693            .map_err(|e| {
1694                XlogError::Kernel(format!("{}: read d_total failed: {}", entry_label, e))
1695            })?;
1696
1697        if total_rows == 0 {
1698            return self.create_empty_buffer(out_schema);
1699        }
1700
1701        // Phase 2: materialize. Allocate K output column buffers.
1702        let elem_bytes = width_class.elem_bytes();
1703        let bytes_per_col = (total_rows as usize) * elem_bytes;
1704        let mut out_col_bufs: Vec<TrackedCudaSlice<u8>> = Vec::with_capacity(k);
1705        let mut out_col_ptrs: Vec<u64> = Vec::with_capacity(k);
1706        for _ in 0..k {
1707            let buf = self.memory.alloc::<u8>(bytes_per_col)?;
1708            out_col_ptrs.push(*buf.device_ptr());
1709            out_col_bufs.push(buf);
1710        }
1711        let mut d_out_cols = self.memory.alloc::<u64>(k)?;
1712        self.htod_launch_metadata_sync_copy_into(&out_col_ptrs, &mut d_out_cols)
1713            .map_err(|e| {
1714                XlogError::Kernel(format!("{}: htod out_col_ptrs failed: {}", entry_label, e))
1715            })?;
1716        let out_d_num_rows = self.memory.alloc::<u32>(1)?;
1717
1718        let mut rec_mat = LaunchRecorder::new_strict(launch_stream);
1719        for buf in edges.iter() {
1720            rec_mat.read(buf.num_rows_device());
1721            rec_mat.read_column(buf.column(0).expect("validated"));
1722            rec_mat.read_column(buf.column(1).expect("validated"));
1723        }
1724        rec_mat.read(&d_edge_col0);
1725        rec_mat.read(&d_edge_col1);
1726        rec_mat.read(&d_edge_n);
1727        if let Some(buf) = d_edge_order.as_ref() {
1728            rec_mat.read(buf);
1729        }
1730        if let Some(buf) = d_iteration_order.as_ref() {
1731            rec_mat.read(buf);
1732        }
1733        match &leader_metadata {
1734            CliqueLeaderMetadata::U32(leader_metadata) => {
1735                rec_mat.read(&leader_metadata.unique_keys);
1736                rec_mat.read(&leader_metadata.fan_out);
1737                rec_mat.read(&leader_metadata.prefix_sum);
1738            }
1739            CliqueLeaderMetadata::U64(leader_metadata) => {
1740                rec_mat.read(&leader_metadata.unique_keys);
1741                rec_mat.read(&leader_metadata.fan_out);
1742                rec_mat.read(&leader_metadata.prefix_sum);
1743            }
1744        }
1745        rec_mat.read(&thread_counts_buf);
1746        rec_mat.read(&offsets_buf);
1747        rec_mat.read(&d_out_cols);
1748        for buf in out_col_bufs.iter() {
1749            rec_mat.write(buf);
1750        }
1751        rec_mat.write(&out_d_num_rows);
1752        rec_mat.preflight(runtime).map_err(|e| {
1753            XlogError::Kernel(format!(
1754                "{}: materialize preflight failed: {}",
1755                entry_label, e
1756            ))
1757        })?;
1758
1759        let materialize_kernel = device
1760            .get_func(WCOJ_MODULE, clique_kernel_name(k, true, width_class))
1761            .ok_or_else(|| {
1762                XlogError::Kernel(format!(
1763                    "{}: materialize kernel '{}' not found",
1764                    entry_label,
1765                    clique_kernel_name(k, true, width_class)
1766                ))
1767            })?;
1768
1769        // SAFETY: kernel signature
1770        //   wcoj_clique{K}_materialize_*(
1771        //     const T* const* edge_col0,
1772        //     const T* const* edge_col1,
1773        //     const u32* edge_n,
1774        //     u32 leader_edge_idx,
1775        //     const u8* edge_order,
1776        //     const u8* iteration_order,
1777        //     u32 leader_count,
1778        //     const T* unique_keys,
1779        //     const u32* fan_out,
1780        //     const u32* prefix_sum,
1781        //     u32 metadata_key_count,
1782        //     u32 block_work_unit,
1783        //     const u32* thread_counts,
1784        //     const u32* block_offsets,
1785        //     u32 total_rows,
1786        //     T* const* out_cols)
1787        // Raw params are required because the metadata-extended ABI
1788        // exceeds the tuple-launch arity.
1789        let mat_config = LaunchConfig {
1790            grid_dim: (grid, 1, 1),
1791            block_dim: (BLOCK_SIZE, 1, 1),
1792            shared_mem_bytes: 0,
1793        };
1794        let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
1795            self.initialize_launch_metadata_u32(
1796                total_rows,
1797                &out_d_num_rows,
1798                stream,
1799                &format!("{entry_label}: out_d_num_rows"),
1800            )?;
1801            unsafe {
1802                let mut params: Vec<*mut c_void> = match &leader_metadata {
1803                    CliqueLeaderMetadata::U32(leader_metadata) => vec![
1804                        (&d_edge_col0).as_kernel_param(),
1805                        (&d_edge_col1).as_kernel_param(),
1806                        (&d_edge_n).as_kernel_param(),
1807                        leader_edge_idx.as_kernel_param(),
1808                        edge_order_param,
1809                        iteration_order_param,
1810                        n_leader.as_kernel_param(),
1811                        (&leader_metadata.unique_keys).as_kernel_param(),
1812                        (&leader_metadata.fan_out).as_kernel_param(),
1813                        (&leader_metadata.prefix_sum).as_kernel_param(),
1814                        leader_metadata_key_count.as_kernel_param(),
1815                        block_work_unit.as_kernel_param(),
1816                        (&thread_counts_buf).as_kernel_param(),
1817                        (&offsets_buf).as_kernel_param(),
1818                        total_rows.as_kernel_param(),
1819                        (&d_out_cols).as_kernel_param(),
1820                    ],
1821                    CliqueLeaderMetadata::U64(leader_metadata) => vec![
1822                        (&d_edge_col0).as_kernel_param(),
1823                        (&d_edge_col1).as_kernel_param(),
1824                        (&d_edge_n).as_kernel_param(),
1825                        leader_edge_idx.as_kernel_param(),
1826                        edge_order_param,
1827                        iteration_order_param,
1828                        n_leader.as_kernel_param(),
1829                        (&leader_metadata.unique_keys).as_kernel_param(),
1830                        (&leader_metadata.fan_out).as_kernel_param(),
1831                        (&leader_metadata.prefix_sum).as_kernel_param(),
1832                        leader_metadata_key_count.as_kernel_param(),
1833                        block_work_unit.as_kernel_param(),
1834                        (&thread_counts_buf).as_kernel_param(),
1835                        (&offsets_buf).as_kernel_param(),
1836                        total_rows.as_kernel_param(),
1837                        (&d_out_cols).as_kernel_param(),
1838                    ],
1839                };
1840                materialize_kernel
1841                    .clone()
1842                    .launch_in(stream, mat_config, &mut params)
1843                    .map_err(|e| {
1844                        XlogError::Kernel(format!(
1845                            "{}: materialize launch failed: {}",
1846                            entry_label, e
1847                        ))
1848                    })?;
1849            }
1850            Ok(())
1851        };
1852        // SAFETY: preflight retained the accessed buffers and bound this stream.
1853        let rec_mat = unsafe { rec_mat.enqueue_prepared_with(&cu_stream, enqueue) }
1854            .map_err(|error| error.into_xlog_error())?;
1855
1856        rec_mat.commit().map_err(|e| {
1857            XlogError::Kernel(format!("{}: materialize commit failed: {}", entry_label, e))
1858        })?;
1859
1860        let columns: Vec<CudaColumn> = out_col_bufs.into_iter().map(|b| b.into()).collect();
1861        Ok(CudaBuffer::from_columns_with_host_count(
1862            columns,
1863            total_rows as u64,
1864            out_d_num_rows,
1865            out_schema,
1866            total_rows,
1867        ))
1868    }
1869
1870    /// 5-clique WCOJ at 4-byte width-class.
1871    ///
1872    /// `edges` must contain exactly **10** 2-column buffers in
1873    /// canonical lex `(i, j)` order (i < j): `(0,1), (0,2), (0,3),
1874    /// (0,4), (1,2), (1,3), (1,4), (2,3), (2,4), (3,4)`. Each
1875    /// column may be `U32` or `Symbol` (mixable within the
1876    /// 4-byte width-class). All edges must be lex-sorted+deduped
1877    /// on `(col0, col1)` — the runtime dispatcher routes through
1878    /// `wcoj_layout_sort_u32_recorded` before calling here.
1879    pub fn wcoj_clique5_u32_recorded(
1880        &self,
1881        edges: &[&CudaBuffer; 10],
1882        launch_stream: StreamId,
1883    ) -> Result<CudaBuffer> {
1884        self.wcoj_clique_recorded_inner(
1885            5,
1886            edges,
1887            0,
1888            None,
1889            None,
1890            CliqueWidthClass::FourByte,
1891            launch_stream,
1892            "wcoj_clique5_u32_recorded",
1893        )
1894    }
1895
1896    /// 5-clique WCOJ at 4-byte width-class using plan-derived launch params.
1897    pub fn wcoj_clique5_u32_recorded_planned(
1898        &self,
1899        edges: &[&CudaBuffer; 10],
1900        leader_edge_idx: u32,
1901        edge_order: &[u8],
1902        iteration_order: &[u8],
1903        launch_stream: StreamId,
1904    ) -> Result<CudaBuffer> {
1905        self.wcoj_clique_recorded_inner(
1906            5,
1907            edges,
1908            leader_edge_idx,
1909            Some(edge_order),
1910            Some(iteration_order),
1911            CliqueWidthClass::FourByte,
1912            launch_stream,
1913            "wcoj_clique5_u32_recorded_planned",
1914        )
1915    }
1916
1917    /// 5-clique WCOJ at 8-byte width-class (U64 only).
1918    pub fn wcoj_clique5_u64_recorded(
1919        &self,
1920        edges: &[&CudaBuffer; 10],
1921        launch_stream: StreamId,
1922    ) -> Result<CudaBuffer> {
1923        self.wcoj_clique_recorded_inner(
1924            5,
1925            edges,
1926            0,
1927            None,
1928            None,
1929            CliqueWidthClass::EightByte,
1930            launch_stream,
1931            "wcoj_clique5_u64_recorded",
1932        )
1933    }
1934
1935    /// 5-clique WCOJ at 8-byte width-class using plan-derived launch params.
1936    pub fn wcoj_clique5_u64_recorded_planned(
1937        &self,
1938        edges: &[&CudaBuffer; 10],
1939        leader_edge_idx: u32,
1940        edge_order: &[u8],
1941        iteration_order: &[u8],
1942        launch_stream: StreamId,
1943    ) -> Result<CudaBuffer> {
1944        self.wcoj_clique_recorded_inner(
1945            5,
1946            edges,
1947            leader_edge_idx,
1948            Some(edge_order),
1949            Some(iteration_order),
1950            CliqueWidthClass::EightByte,
1951            launch_stream,
1952            "wcoj_clique5_u64_recorded_planned",
1953        )
1954    }
1955
1956    /// 6-clique WCOJ at 4-byte width-class.
1957    ///
1958    /// `edges` must contain exactly **15** 2-column buffers in
1959    /// canonical lex `(i, j)` order. Width-class + sort+dedup
1960    /// pre-condition match `wcoj_clique5_u32_recorded`.
1961    pub fn wcoj_clique6_u32_recorded(
1962        &self,
1963        edges: &[&CudaBuffer; 15],
1964        launch_stream: StreamId,
1965    ) -> Result<CudaBuffer> {
1966        self.wcoj_clique_recorded_inner(
1967            6,
1968            edges,
1969            0,
1970            None,
1971            None,
1972            CliqueWidthClass::FourByte,
1973            launch_stream,
1974            "wcoj_clique6_u32_recorded",
1975        )
1976    }
1977
1978    /// 6-clique WCOJ at 4-byte width-class using plan-derived launch params.
1979    pub fn wcoj_clique6_u32_recorded_planned(
1980        &self,
1981        edges: &[&CudaBuffer; 15],
1982        leader_edge_idx: u32,
1983        edge_order: &[u8],
1984        iteration_order: &[u8],
1985        launch_stream: StreamId,
1986    ) -> Result<CudaBuffer> {
1987        self.wcoj_clique_recorded_inner(
1988            6,
1989            edges,
1990            leader_edge_idx,
1991            Some(edge_order),
1992            Some(iteration_order),
1993            CliqueWidthClass::FourByte,
1994            launch_stream,
1995            "wcoj_clique6_u32_recorded_planned",
1996        )
1997    }
1998
1999    /// 6-clique WCOJ at 8-byte width-class (U64 only).
2000    pub fn wcoj_clique6_u64_recorded(
2001        &self,
2002        edges: &[&CudaBuffer; 15],
2003        launch_stream: StreamId,
2004    ) -> Result<CudaBuffer> {
2005        self.wcoj_clique_recorded_inner(
2006            6,
2007            edges,
2008            0,
2009            None,
2010            None,
2011            CliqueWidthClass::EightByte,
2012            launch_stream,
2013            "wcoj_clique6_u64_recorded",
2014        )
2015    }
2016
2017    /// 6-clique WCOJ at 8-byte width-class using plan-derived launch params.
2018    pub fn wcoj_clique6_u64_recorded_planned(
2019        &self,
2020        edges: &[&CudaBuffer; 15],
2021        leader_edge_idx: u32,
2022        edge_order: &[u8],
2023        iteration_order: &[u8],
2024        launch_stream: StreamId,
2025    ) -> Result<CudaBuffer> {
2026        self.wcoj_clique_recorded_inner(
2027            6,
2028            edges,
2029            leader_edge_idx,
2030            Some(edge_order),
2031            Some(iteration_order),
2032            CliqueWidthClass::EightByte,
2033            launch_stream,
2034            "wcoj_clique6_u64_recorded_planned",
2035        )
2036    }
2037
2038    /// 7-clique WCOJ at 4-byte width-class.
2039    pub fn wcoj_clique7_u32_recorded(
2040        &self,
2041        edges: &[&CudaBuffer; 21],
2042        launch_stream: StreamId,
2043    ) -> Result<CudaBuffer> {
2044        self.wcoj_clique_recorded_inner(
2045            7,
2046            edges,
2047            0,
2048            None,
2049            None,
2050            CliqueWidthClass::FourByte,
2051            launch_stream,
2052            "wcoj_clique7_u32_recorded",
2053        )
2054    }
2055
2056    /// 7-clique WCOJ at 4-byte width-class using plan-derived launch params.
2057    pub fn wcoj_clique7_u32_recorded_planned(
2058        &self,
2059        edges: &[&CudaBuffer; 21],
2060        leader_edge_idx: u32,
2061        edge_order: &[u8],
2062        iteration_order: &[u8],
2063        launch_stream: StreamId,
2064    ) -> Result<CudaBuffer> {
2065        self.wcoj_clique_recorded_inner(
2066            7,
2067            edges,
2068            leader_edge_idx,
2069            Some(edge_order),
2070            Some(iteration_order),
2071            CliqueWidthClass::FourByte,
2072            launch_stream,
2073            "wcoj_clique7_u32_recorded_planned",
2074        )
2075    }
2076
2077    /// 7-clique WCOJ at 8-byte width-class (U64 only).
2078    pub fn wcoj_clique7_u64_recorded(
2079        &self,
2080        edges: &[&CudaBuffer; 21],
2081        launch_stream: StreamId,
2082    ) -> Result<CudaBuffer> {
2083        self.wcoj_clique_recorded_inner(
2084            7,
2085            edges,
2086            0,
2087            None,
2088            None,
2089            CliqueWidthClass::EightByte,
2090            launch_stream,
2091            "wcoj_clique7_u64_recorded",
2092        )
2093    }
2094
2095    /// 7-clique WCOJ at 8-byte width-class using plan-derived launch params.
2096    pub fn wcoj_clique7_u64_recorded_planned(
2097        &self,
2098        edges: &[&CudaBuffer; 21],
2099        leader_edge_idx: u32,
2100        edge_order: &[u8],
2101        iteration_order: &[u8],
2102        launch_stream: StreamId,
2103    ) -> Result<CudaBuffer> {
2104        self.wcoj_clique_recorded_inner(
2105            7,
2106            edges,
2107            leader_edge_idx,
2108            Some(edge_order),
2109            Some(iteration_order),
2110            CliqueWidthClass::EightByte,
2111            launch_stream,
2112            "wcoj_clique7_u64_recorded_planned",
2113        )
2114    }
2115
2116    /// 8-clique WCOJ at 4-byte width-class.
2117    pub fn wcoj_clique8_u32_recorded(
2118        &self,
2119        edges: &[&CudaBuffer; 28],
2120        launch_stream: StreamId,
2121    ) -> Result<CudaBuffer> {
2122        self.wcoj_clique_recorded_inner(
2123            8,
2124            edges,
2125            0,
2126            None,
2127            None,
2128            CliqueWidthClass::FourByte,
2129            launch_stream,
2130            "wcoj_clique8_u32_recorded",
2131        )
2132    }
2133
2134    /// 8-clique WCOJ at 4-byte width-class using plan-derived launch params.
2135    pub fn wcoj_clique8_u32_recorded_planned(
2136        &self,
2137        edges: &[&CudaBuffer; 28],
2138        leader_edge_idx: u32,
2139        edge_order: &[u8],
2140        iteration_order: &[u8],
2141        launch_stream: StreamId,
2142    ) -> Result<CudaBuffer> {
2143        self.wcoj_clique_recorded_inner(
2144            8,
2145            edges,
2146            leader_edge_idx,
2147            Some(edge_order),
2148            Some(iteration_order),
2149            CliqueWidthClass::FourByte,
2150            launch_stream,
2151            "wcoj_clique8_u32_recorded_planned",
2152        )
2153    }
2154
2155    /// 8-clique WCOJ at 8-byte width-class (U64 only).
2156    pub fn wcoj_clique8_u64_recorded(
2157        &self,
2158        edges: &[&CudaBuffer; 28],
2159        launch_stream: StreamId,
2160    ) -> Result<CudaBuffer> {
2161        self.wcoj_clique_recorded_inner(
2162            8,
2163            edges,
2164            0,
2165            None,
2166            None,
2167            CliqueWidthClass::EightByte,
2168            launch_stream,
2169            "wcoj_clique8_u64_recorded",
2170        )
2171    }
2172
2173    /// 8-clique WCOJ at 8-byte width-class using plan-derived launch params.
2174    pub fn wcoj_clique8_u64_recorded_planned(
2175        &self,
2176        edges: &[&CudaBuffer; 28],
2177        leader_edge_idx: u32,
2178        edge_order: &[u8],
2179        iteration_order: &[u8],
2180        launch_stream: StreamId,
2181    ) -> Result<CudaBuffer> {
2182        self.wcoj_clique_recorded_inner(
2183            8,
2184            edges,
2185            leader_edge_idx,
2186            Some(edge_order),
2187            Some(iteration_order),
2188            CliqueWidthClass::EightByte,
2189            launch_stream,
2190            "wcoj_clique8_u64_recorded_planned",
2191        )
2192    }
2193
2194    /// Aggregate-fused K-clique count-by-root (u32 width-class,
2195    /// K ∈ {5, 6}). For `q(R, count(*)) :- <complete K-clique body>`
2196    /// grouped by the plan's position-0 root variable, computes the
2197    /// (root, count) row set WITHOUT materializing the clique rows.
2198    ///
2199    /// Pipeline (mirrors `wcoj_4cycle_groupby_root_count_u32_recorded`):
2200    ///   1. Layout-normalize every edge per dispatch (sorted-fast-path
2201    ///      clone when already lex-sorted + unique) — the fused path must
2202    ///      give the same guarantee as the unfused pipeline instead of
2203    ///      trusting store-buffer sortedness.
2204    ///   2. Build the leader-edge runtime metadata, htod the per-edge
2205    ///      pointer arrays + plan orders (same surface as the unfused
2206    ///      planned clique entries).
2207    ///   3. `wcoj_clique{K}_groupby_root_count_hg_u32` accumulates, per
2208    ///      leader-edge row, the row's clique completion count via
2209    ///      atomicAdd (order-insensitive, deterministic values). The
2210    ///      row's group key is the oriented leader edge's col0 — the
2211    ///      kernel's binding[0] root.
2212    ///   4. Staging (root, count) over the n_leader input rows, compact
2213    ///      count>0, reduce per root with the recorded groupby Sum.
2214    ///
2215    /// All reduction work is O(n_leader) — input-sized, never
2216    /// join-output-sized. Output schema (root: U32/Symbol, count: U64)
2217    /// matches the unfused materialize+groupby baseline.
2218    #[expect(
2219        clippy::too_many_arguments,
2220        reason = "fused clique aggregation keeps every relation, variable layout, aggregate selector, output bound, and stream explicit"
2221    )]
2222    fn wcoj_clique_groupby_root_count_recorded_inner(
2223        &self,
2224        k: usize,
2225        edges: &[&CudaBuffer],
2226        leader_edge_idx: u32,
2227        edge_order: &[u8],
2228        iteration_order: &[u8],
2229        launch_stream: StreamId,
2230        entry_label: &str,
2231    ) -> Result<CudaBuffer> {
2232        let runtime = self.memory().runtime().ok_or_else(|| {
2233            XlogError::Kernel(format!(
2234                "{} requires a runtime-backed GpuMemoryManager (with_runtime)",
2235                entry_label
2236            ))
2237        })?;
2238        let cu_stream = runtime
2239            .stream_pool()
2240            .resolve(launch_stream)
2241            .ok_or_else(|| {
2242                XlogError::Kernel(format!(
2243                    "{}: launch_stream StreamId({}) does not resolve",
2244                    entry_label, launch_stream.0
2245                ))
2246            })?;
2247        if !matches!(k, 5 | 6) {
2248            return Err(XlogError::Kernel(format!(
2249                "{}: fused count-by-root supports k 5..6, got {}",
2250                entry_label, k
2251            )));
2252        }
2253        let expected_edges = k * (k - 1) / 2;
2254        if edges.len() != expected_edges {
2255            return Err(XlogError::Kernel(format!(
2256                "{}: expected {} edges (= C({}, 2)), got {}",
2257                entry_label,
2258                expected_edges,
2259                k,
2260                edges.len()
2261            )));
2262        }
2263        let leader_slot = usize::try_from(leader_edge_idx)
2264            .ok()
2265            .filter(|idx| *idx < expected_edges)
2266            .ok_or_else(|| {
2267                XlogError::Kernel(format!(
2268                    "{}: leader_edge_idx {} out of range for {} edges",
2269                    entry_label, leader_edge_idx, expected_edges
2270                ))
2271            })?;
2272        validate_clique_u8_permutation(edge_order, expected_edges, "edge_order", entry_label)?;
2273        validate_clique_u8_permutation(iteration_order, k, "iteration_order", entry_label)?;
2274
2275        // Layout-normalize per dispatch (commit 31b0ccf0 contract for
2276        // ALL fused group-by-root entries). Also enforces the 2-column
2277        // 4-byte width-class per edge.
2278        let mut laid_out: Vec<CudaBuffer> = Vec::with_capacity(expected_edges);
2279        for buf in edges {
2280            laid_out.push(self.wcoj_layout_u32_recorded(buf, launch_stream)?);
2281        }
2282        let edges: Vec<&CudaBuffer> = laid_out.iter().collect();
2283        let leader = edges[leader_slot];
2284
2285        let w_type = leader.schema.column_type(0).ok_or_else(|| {
2286            XlogError::Kernel(format!(
2287                "{}: leader edge column 0 type missing",
2288                entry_label
2289            ))
2290        })?;
2291        let out_schema = Schema::new(vec![
2292            ("root".to_string(), w_type),
2293            ("count".to_string(), ScalarType::U64),
2294        ]);
2295        let n_leader = self.logical_row_count_u32(leader)?;
2296        if n_leader == 0 {
2297            return self.create_empty_buffer(out_schema);
2298        }
2299
2300        let leader_metadata = self.wcoj_clique_metadata_recorded_u32_inner(
2301            k,
2302            &edges,
2303            leader_edge_idx,
2304            launch_stream,
2305            entry_label,
2306        )?;
2307        let leader_work_total = u32::try_from(leader_metadata.total).map_err(|_| {
2308            XlogError::Kernel(format!(
2309                "{}: leader metadata total {} exceeds u32 kernel surface",
2310                entry_label, leader_metadata.total
2311            ))
2312        })?;
2313        if leader_work_total != n_leader {
2314            return Err(XlogError::Kernel(format!(
2315                "{}: leader metadata total {} does not match leader row count {}",
2316                entry_label, leader_work_total, n_leader
2317            )));
2318        }
2319        if leader_metadata.key_count == 0 {
2320            return self.create_empty_buffer(out_schema);
2321        }
2322
2323        // Host-side per-edge pointer arrays + row counts, htod'd to
2324        // small device buffers (same surface as the unfused planned
2325        // clique entries).
2326        let mut edge_col0_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
2327        let mut edge_col1_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
2328        let mut edge_n_host: Vec<u32> = Vec::with_capacity(expected_edges);
2329        for buf in edges.iter() {
2330            let col0 = buf.column(0).ok_or_else(|| {
2331                XlogError::Kernel(format!("{}: edge column 0 missing", entry_label))
2332            })?;
2333            let col1 = buf.column(1).ok_or_else(|| {
2334                XlogError::Kernel(format!("{}: edge column 1 missing", entry_label))
2335            })?;
2336            edge_col0_ptrs.push(*col0.device_ptr());
2337            edge_col1_ptrs.push(*col1.device_ptr());
2338            edge_n_host.push(self.logical_row_count_u32(buf)?);
2339        }
2340        let mut d_edge_col0 = self.memory.alloc::<u64>(expected_edges)?;
2341        let mut d_edge_col1 = self.memory.alloc::<u64>(expected_edges)?;
2342        let mut d_edge_n = self.memory.alloc::<u32>(expected_edges)?;
2343        self.htod_launch_metadata_sync_copy_into(&edge_col0_ptrs, &mut d_edge_col0)
2344            .map_err(|e| {
2345                XlogError::Kernel(format!(
2346                    "{}: htod edge_col0_ptrs failed: {}",
2347                    entry_label, e
2348                ))
2349            })?;
2350        self.htod_launch_metadata_sync_copy_into(&edge_col1_ptrs, &mut d_edge_col1)
2351            .map_err(|e| {
2352                XlogError::Kernel(format!(
2353                    "{}: htod edge_col1_ptrs failed: {}",
2354                    entry_label, e
2355                ))
2356            })?;
2357        self.htod_launch_metadata_sync_copy_into(&edge_n_host, &mut d_edge_n)
2358            .map_err(|e| {
2359                XlogError::Kernel(format!("{}: htod edge_n failed: {}", entry_label, e))
2360            })?;
2361        let mut d_edge_order = self.memory.alloc::<u8>(expected_edges)?;
2362        self.htod_launch_metadata_sync_copy_into(edge_order, &mut d_edge_order)
2363            .map_err(|e| {
2364                XlogError::Kernel(format!("{}: htod edge_order failed: {}", entry_label, e))
2365            })?;
2366        let mut d_iteration_order = self.memory.alloc::<u8>(k)?;
2367        self.htod_launch_metadata_sync_copy_into(iteration_order, &mut d_iteration_order)
2368            .map_err(|e| {
2369                XlogError::Kernel(format!(
2370                    "{}: htod iteration_order failed: {}",
2371                    entry_label, e
2372                ))
2373            })?;
2374
2375        // Per-leader-row match counters, zero-initialized. Allocated as
2376        // the u8-backed column layout so the array doubles as the
2377        // staging buffer's count column after the kernel fills it.
2378        let mut row_counts = self
2379            .memory()
2380            .alloc::<u8>(n_leader as usize * std::mem::size_of::<u32>())?;
2381        self.device()
2382            .inner()
2383            .memset_zeros(&mut row_counts)
2384            .map_err(|e| {
2385                XlogError::Kernel(format!("{}: zero row counts failed: {}", entry_label, e))
2386            })?;
2387
2388        let block_work_unit = crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT;
2389        let grid = leader_work_total.div_ceil(block_work_unit);
2390        let mut rec = LaunchRecorder::new_strict(launch_stream);
2391        for buf in edges.iter() {
2392            rec.read(buf.num_rows_device());
2393            rec.read_column(buf.column(0).expect("validated"));
2394            rec.read_column(buf.column(1).expect("validated"));
2395        }
2396        rec.read(&d_edge_col0);
2397        rec.read(&d_edge_col1);
2398        rec.read(&d_edge_n);
2399        rec.read(&d_edge_order);
2400        rec.read(&d_iteration_order);
2401        rec.read(&leader_metadata.unique_keys);
2402        rec.read(&leader_metadata.fan_out);
2403        rec.read(&leader_metadata.prefix_sum);
2404        rec.write(&row_counts);
2405        rec.preflight(runtime)
2406            .map_err(|e| XlogError::Kernel(format!("{}: preflight failed: {}", entry_label, e)))?;
2407
2408        let kernel_name = match k {
2409            5 => wcoj_kernels::WCOJ_CLIQUE5_GROUPBY_ROOT_COUNT_HG_U32,
2410            _ => wcoj_kernels::WCOJ_CLIQUE6_GROUPBY_ROOT_COUNT_HG_U32,
2411        };
2412        let kernel = self
2413            .device()
2414            .inner()
2415            .get_func(WCOJ_MODULE, kernel_name)
2416            .ok_or_else(|| {
2417                XlogError::Kernel(format!(
2418                    "{}: kernel '{}' not found",
2419                    entry_label, kernel_name
2420                ))
2421            })?;
2422        // SAFETY: kernel signature
2423        //   wcoj_clique{K}_groupby_root_count_hg_u32(
2424        //     const u32* const* edge_col0,
2425        //     const u32* const* edge_col1,
2426        //     const u32* edge_n,
2427        //     u32 leader_edge_idx,
2428        //     const u8* edge_order,
2429        //     const u8* iteration_order,
2430        //     u32 leader_count,
2431        //     const u32* unique_keys,
2432        //     const u32* fan_out,
2433        //     const u32* prefix_sum,
2434        //     u32 metadata_key_count,
2435        //     u32 block_work_unit,
2436        //     u32* out_row_counts)
2437        // Pointers all device-resident; preflight verified cross-stream
2438        // tracking. Raw params are required because the
2439        // metadata-extended ABI exceeds the tuple-launch arity.
2440        let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
2441            unsafe {
2442                let mut params: Vec<*mut c_void> = vec![
2443                    (&d_edge_col0).as_kernel_param(),
2444                    (&d_edge_col1).as_kernel_param(),
2445                    (&d_edge_n).as_kernel_param(),
2446                    leader_edge_idx.as_kernel_param(),
2447                    (&d_edge_order).as_kernel_param(),
2448                    (&d_iteration_order).as_kernel_param(),
2449                    n_leader.as_kernel_param(),
2450                    (&leader_metadata.unique_keys).as_kernel_param(),
2451                    (&leader_metadata.fan_out).as_kernel_param(),
2452                    (&leader_metadata.prefix_sum).as_kernel_param(),
2453                    leader_metadata.key_count.as_kernel_param(),
2454                    block_work_unit.as_kernel_param(),
2455                    (&row_counts).as_kernel_param(),
2456                ];
2457                kernel
2458                    .clone()
2459                    .launch_in(
2460                        stream,
2461                        LaunchConfig {
2462                            grid_dim: (grid, 1, 1),
2463                            block_dim: (BLOCK_SIZE, 1, 1),
2464                            shared_mem_bytes: 0,
2465                        },
2466                        &mut params,
2467                    )
2468                    .map_err(|e| {
2469                        XlogError::Kernel(format!(
2470                            "{}: groupby-count launch failed: {}",
2471                            entry_label, e
2472                        ))
2473                    })?;
2474            }
2475            Ok(())
2476        };
2477        // SAFETY: preflight retained the accessed buffers and bound this stream.
2478        let rec = unsafe { rec.enqueue_prepared_with(&cu_stream, enqueue) }
2479            .map_err(|error| error.into_xlog_error())?;
2480        rec.commit()
2481            .map_err(|e| XlogError::Kernel(format!("{}: commit failed: {}", entry_label, e)))?;
2482
2483        // Staging buffer (root, count) over the n_leader input rows:
2484        // root is a device-to-device copy of the oriented leader edge's
2485        // col0; the count column is the kernel-filled array. Rows stay
2486        // lex-sorted by root.
2487        let root_src = match leader.column(0).expect("validated") {
2488            CudaColumn::Owned(slice) => slice,
2489            _ => {
2490                return Err(XlogError::Kernel(format!(
2491                    "{}: leader.col0 must be an owned CudaColumn",
2492                    entry_label
2493                )))
2494            }
2495        };
2496        let root_copy = self
2497            .memory()
2498            .alloc::<u8>(n_leader as usize * std::mem::size_of::<u32>())?;
2499        // Explicit-length copy: layout-normalized columns are allocated at
2500        // capacity, which can exceed the logical n_leader * 4 bytes a
2501        // full-slice typed copy would assert on.
2502        unsafe {
2503            let res = sys::cuMemcpyDtoD_v2(
2504                *root_copy.device_ptr(),
2505                *root_src.device_ptr(),
2506                n_leader as usize * std::mem::size_of::<u32>(),
2507            );
2508            if res != sys::cudaError_enum::CUDA_SUCCESS {
2509                return Err(XlogError::Kernel(format!(
2510                    "{}: copy root column failed: {:?}",
2511                    entry_label, res
2512                )));
2513            }
2514        }
2515        let mut d_num_rows = self.memory().alloc::<u32>(1)?;
2516        self.device()
2517            .inner()
2518            .dtod_copy(leader.num_rows_device(), &mut d_num_rows)
2519            .map_err(|e| {
2520                XlogError::Kernel(format!("{}: copy row count failed: {}", entry_label, e))
2521            })?;
2522        let staging_schema = Schema::new(vec![
2523            ("root".to_string(), w_type),
2524            ("count".to_string(), ScalarType::U32),
2525        ]);
2526        let staging = CudaBuffer::from_columns_with_host_count(
2527            vec![root_copy.into(), row_counts.into()],
2528            n_leader as u64,
2529            d_num_rows,
2530            staging_schema,
2531            n_leader,
2532        );
2533
2534        // Keep only roots with at least one completed clique, then
2535        // reduce per root. Both steps run over input-sized data.
2536        let mask = self.compare_const_mask_recorded::<u32>(
2537            &staging,
2538            1,
2539            0u32,
2540            crate::CompareOp::Gt,
2541            launch_stream,
2542        )?;
2543        let compacted =
2544            self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)?;
2545        self.groupby_multi_agg_recorded(
2546            &compacted,
2547            &[0],
2548            &[(1, xlog_core::AggOp::Sum)],
2549            launch_stream,
2550        )
2551    }
2552
2553    /// Fused 5-clique count-by-root at the 4-byte width-class,
2554    /// using plan-derived launch params. See
2555    /// [`Self::wcoj_clique_groupby_root_count_recorded_inner`].
2556    pub fn wcoj_clique5_groupby_root_count_u32_recorded_planned(
2557        &self,
2558        edges: &[&CudaBuffer; 10],
2559        leader_edge_idx: u32,
2560        edge_order: &[u8],
2561        iteration_order: &[u8],
2562        launch_stream: StreamId,
2563    ) -> Result<CudaBuffer> {
2564        self.wcoj_clique_groupby_root_count_recorded_inner(
2565            5,
2566            edges,
2567            leader_edge_idx,
2568            edge_order,
2569            iteration_order,
2570            launch_stream,
2571            "wcoj_clique5_groupby_root_count_u32_recorded_planned",
2572        )
2573    }
2574
2575    /// Fused 6-clique count-by-root at the 4-byte width-class,
2576    /// using plan-derived launch params. See
2577    /// [`Self::wcoj_clique_groupby_root_count_recorded_inner`].
2578    pub fn wcoj_clique6_groupby_root_count_u32_recorded_planned(
2579        &self,
2580        edges: &[&CudaBuffer; 15],
2581        leader_edge_idx: u32,
2582        edge_order: &[u8],
2583        iteration_order: &[u8],
2584        launch_stream: StreamId,
2585    ) -> Result<CudaBuffer> {
2586        self.wcoj_clique_groupby_root_count_recorded_inner(
2587            6,
2588            edges,
2589            leader_edge_idx,
2590            edge_order,
2591            iteration_order,
2592            launch_stream,
2593            "wcoj_clique6_groupby_root_count_u32_recorded_planned",
2594        )
2595    }
2596}
2597
2598fn validate_clique_u8_permutation(
2599    values: &[u8],
2600    len: usize,
2601    label: &str,
2602    entry_label: &str,
2603) -> Result<()> {
2604    if values.len() != len {
2605        return Err(XlogError::Kernel(format!(
2606            "{}: {} length {} must equal {}",
2607            entry_label,
2608            label,
2609            values.len(),
2610            len
2611        )));
2612    }
2613    let mut seen = vec![false; len];
2614    for &value in values {
2615        let idx = usize::from(value);
2616        if idx >= len {
2617            return Err(XlogError::Kernel(format!(
2618                "{}: {} value {} out of range 0..{}",
2619                entry_label, label, value, len
2620            )));
2621        }
2622        if seen[idx] {
2623            return Err(XlogError::Kernel(format!(
2624                "{}: {} duplicates value {}",
2625                entry_label, label, value
2626            )));
2627        }
2628        seen[idx] = true;
2629    }
2630    Ok(())
2631}