xlog_cuda/provider/
wcoj_metadata.rs

1use std::collections::BTreeMap;
2use std::ffi::c_void;
3
4use cudarc::driver::sys;
5use xlog_core::{AggOp, Result, ScalarType, Schema, XlogError};
6
7use super::{arith_kernels, wcoj_kernels, CudaKernelProvider, ARITH_MODULE, WCOJ_MODULE};
8use crate::device_runtime::StreamId;
9use crate::launch::LaunchRecorder;
10use crate::memory::{CudaColumn, TrackedCudaSlice};
11use crate::wcoj_metadata::{
12    Wcoj4CycleRootAggValue, WcojCycle4HgWorkPlanU32, WcojCycle4HgWorkPlanU64, WcojRelationMetadata,
13    WcojRootAggValue, WcojTriangleHgCountPhaseU32, WcojTriangleHgWorkPlanU32,
14    WcojTriangleHgWorkPlanU64,
15};
16use crate::{AsKernelParam, CudaBuffer, LaunchAsync, LaunchConfig};
17
18const BLOCK_SIZE: u32 = 256;
19const HG_COUNT_BLOCK_SIZE: u32 = 512;
20
21impl CudaKernelProvider {
22    pub fn wcoj_build_metadata_u32_recorded(
23        &self,
24        input: &CudaBuffer,
25        key_col_idx: usize,
26        launch_stream: StreamId,
27    ) -> Result<WcojRelationMetadata<u32>> {
28        self.validate_metadata_column(input, key_col_idx, MetadataWidth::U32)?;
29        let keys = metadata_column_u32(input, key_col_idx)?;
30        self.build_metadata_u32_from_column(input, key_col_idx, keys, launch_stream)
31    }
32
33    pub fn wcoj_build_metadata_u64_recorded(
34        &self,
35        input: &CudaBuffer,
36        key_col_idx: usize,
37        launch_stream: StreamId,
38    ) -> Result<WcojRelationMetadata<u64>> {
39        self.validate_metadata_column(input, key_col_idx, MetadataWidth::U64)?;
40        let keys = metadata_column_u64(input, key_col_idx)?;
41        self.build_metadata_u64_from_column(input, key_col_idx, keys, launch_stream)
42    }
43
44    pub fn wcoj_triangle_hg_work_plan_u32_recorded(
45        &self,
46        e_xy: &CudaBuffer,
47        e_yz: &CudaBuffer,
48        e_xz: &CudaBuffer,
49        block_work_unit: u32,
50        launch_stream: StreamId,
51    ) -> Result<WcojTriangleHgWorkPlanU32> {
52        let ctx = "wcoj_triangle_hg_work_plan_u32_recorded";
53        if block_work_unit == 0 {
54            return Err(XlogError::Kernel(format!(
55                "{ctx}: block_work_unit must be nonzero"
56            )));
57        }
58        validate_binary_u32(ctx, "e_xy", e_xy)?;
59        validate_binary_u32(ctx, "e_yz", e_yz)?;
60        validate_binary_u32(ctx, "e_xz", e_xz)?;
61
62        let n_xy = self.metadata_logical_rows(e_xy)?;
63        let n_yz = self.metadata_logical_rows(e_yz)?;
64        let n_xz = self.metadata_logical_rows(e_xz)?;
65        let prefix_len = n_xy
66            .checked_add(1)
67            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: prefix length overflow")))?;
68        let mut xy_work_prefix = self.memory().alloc::<u32>(prefix_len as usize)?;
69        let mut xy_yz_start = self.memory().alloc::<u32>(n_xy as usize)?;
70        let mut xy_yz_end = self.memory().alloc::<u32>(n_xy as usize)?;
71        let mut xy_xz_start = self.memory().alloc::<u32>(n_xy as usize)?;
72        let mut xy_xz_end = self.memory().alloc::<u32>(n_xy as usize)?;
73
74        if n_xy == 0 {
75            let block_counts = self.memory().alloc::<u32>(1)?;
76            let block_offsets = self.memory().alloc::<u32>(1)?;
77            let scratch_x = self.memory().alloc::<u32>(1)?;
78            let scratch_y = self.memory().alloc::<u32>(1)?;
79            let scratch_z = self.memory().alloc::<u32>(1)?;
80            return Ok(WcojTriangleHgWorkPlanU32 {
81                xy_work_prefix,
82                xy_yz_start,
83                xy_yz_end,
84                xy_xz_start,
85                xy_xz_end,
86                block_counts,
87                block_offsets,
88                scratch_x,
89                scratch_y,
90                scratch_z,
91                total_work: 0,
92                block_work_unit,
93                row_count: 0,
94            });
95        }
96
97        let runtime = self.memory().runtime().ok_or_else(|| {
98            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
99        })?;
100        let cu_stream = runtime
101            .stream_pool()
102            .resolve(launch_stream)
103            .ok_or_else(|| {
104                XlogError::Kernel(format!(
105                    "{ctx}: launch_stream StreamId({}) does not resolve",
106                    launch_stream.0
107                ))
108            })?;
109
110        let xy_col0 = metadata_column_u32(e_xy, 0)?;
111        let xy_col1 = metadata_column_u32(e_xy, 1)?;
112        let yz_col0 = metadata_column_u32(e_yz, 0)?;
113        let xz_col0 = metadata_column_u32(e_xz, 0)?;
114
115        let mut rec = LaunchRecorder::new_strict(launch_stream);
116        rec.read(e_xy.num_rows_device());
117        rec.read(e_yz.num_rows_device());
118        rec.read(e_xz.num_rows_device());
119        rec.read_column(e_xy.column(0).expect("xy.col0"));
120        rec.read_column(e_xy.column(1).expect("xy.col1"));
121        rec.read_column(e_yz.column(0).expect("yz.col0"));
122        rec.read_column(e_xz.column(0).expect("xz.col0"));
123        rec.write(&xy_work_prefix);
124        rec.write(&xy_yz_start);
125        rec.write(&xy_yz_end);
126        rec.write(&xy_xz_start);
127        rec.write(&xy_xz_end);
128        rec.preflight(runtime)
129            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
130
131        let kernel = self
132            .device()
133            .inner()
134            .get_func(
135                WCOJ_MODULE,
136                wcoj_kernels::WCOJ_TRIANGLE_BUILD_HG_WORK_PLAN_U32,
137            )
138            .ok_or_else(|| {
139                XlogError::Kernel(
140                    "wcoj_triangle_build_hg_work_plan_u32 kernel not found".to_string(),
141                )
142            })?;
143        let grid = n_xy.div_ceil(BLOCK_SIZE);
144        unsafe {
145            kernel
146                .clone()
147                .launch_on_stream(
148                    &cu_stream,
149                    LaunchConfig {
150                        grid_dim: (grid, 1, 1),
151                        block_dim: (BLOCK_SIZE, 1, 1),
152                        shared_mem_bytes: 0,
153                    },
154                    (
155                        xy_col0,
156                        xy_col1,
157                        n_xy,
158                        yz_col0,
159                        n_yz,
160                        xz_col0,
161                        n_xz,
162                        &mut xy_work_prefix,
163                        &mut xy_yz_start,
164                        &mut xy_yz_end,
165                        &mut xy_xz_start,
166                        &mut xy_xz_end,
167                    ),
168                )
169                .map_err(|e| {
170                    XlogError::Kernel(format!(
171                        "wcoj_triangle_build_hg_work_plan_u32 launch failed: {e}"
172                    ))
173                })?;
174        }
175        self.multiblock_scan_u32_inplace_on_stream(
176            &mut xy_work_prefix,
177            prefix_len,
178            &cu_stream,
179            launch_stream,
180            runtime,
181        )?;
182        rec.commit(runtime)
183            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
184        cu_stream
185            .synchronize()
186            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
187        let total_work = self.dtoh_scalar_untracked::<u32>(&xy_work_prefix, n_xy as usize)?;
188        let scratch_slots = if total_work == 0 {
189            1usize
190        } else {
191            let grid = total_work.div_ceil(block_work_unit);
192            (grid as usize)
193                .checked_mul(block_work_unit as usize)
194                .ok_or_else(|| XlogError::Kernel(format!("{ctx}: scratch slot overflow")))?
195        };
196        let scratch_x = self.memory().alloc::<u32>(scratch_slots)?;
197        let scratch_y = self.memory().alloc::<u32>(scratch_slots)?;
198        let scratch_z = self.memory().alloc::<u32>(scratch_slots)?;
199        let grid = if total_work == 0 {
200            1
201        } else {
202            total_work.div_ceil(block_work_unit)
203        };
204        let block_counts = self.memory().alloc::<u32>(grid as usize)?;
205        let block_offsets = self.memory().alloc::<u32>(grid as usize)?;
206
207        Ok(WcojTriangleHgWorkPlanU32 {
208            xy_work_prefix,
209            xy_yz_start,
210            xy_yz_end,
211            xy_xz_start,
212            xy_xz_end,
213            block_counts,
214            block_offsets,
215            scratch_x,
216            scratch_y,
217            scratch_z,
218            total_work,
219            block_work_unit,
220            row_count: n_xy,
221        })
222    }
223
224    pub fn wcoj_triangle_count_hg_u32_recorded(
225        &self,
226        e_yz: &CudaBuffer,
227        e_xz: &CudaBuffer,
228        plan: &WcojTriangleHgWorkPlanU32,
229        launch_stream: StreamId,
230    ) -> Result<CudaBuffer> {
231        let ctx = "wcoj_triangle_count_hg_u32_recorded";
232        validate_binary_u32(ctx, "e_yz", e_yz)?;
233        validate_binary_u32(ctx, "e_xz", e_xz)?;
234        let n_yz = self.metadata_logical_rows(e_yz)?;
235        let n_xz = self.metadata_logical_rows(e_xz)?;
236        let grid = if plan.total_work == 0 {
237            1
238        } else {
239            plan.total_work.div_ceil(plan.block_work_unit)
240        };
241        let bytes_count = (grid as usize)
242            .checked_mul(std::mem::size_of::<u32>())
243            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: count byte size overflow")))?;
244        let mut count_bytes = self.memory().alloc::<u8>(bytes_count)?;
245        let d_num_rows = self.memory().alloc::<u32>(1)?;
246
247        let runtime = self.memory().runtime().ok_or_else(|| {
248            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
249        })?;
250        let cu_stream = runtime
251            .stream_pool()
252            .resolve(launch_stream)
253            .ok_or_else(|| {
254                XlogError::Kernel(format!(
255                    "{ctx}: launch_stream StreamId({}) does not resolve",
256                    launch_stream.0
257                ))
258            })?;
259        let yz_col1 = metadata_column_u32(e_yz, 1)?;
260        let xz_col1 = metadata_column_u32(e_xz, 1)?;
261
262        let mut rec = LaunchRecorder::new_strict(launch_stream);
263        rec.read(e_yz.num_rows_device());
264        rec.read(e_xz.num_rows_device());
265        rec.read_column(e_yz.column(1).expect("yz.col1"));
266        rec.read_column(e_xz.column(1).expect("xz.col1"));
267        rec.read(&plan.xy_work_prefix);
268        rec.read(&plan.xy_yz_start);
269        rec.read(&plan.xy_yz_end);
270        rec.read(&plan.xy_xz_start);
271        rec.read(&plan.xy_xz_end);
272        rec.write(&count_bytes);
273        rec.write(&d_num_rows);
274        rec.preflight(runtime)
275            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
276
277        self.htod_launch_metadata_async_copy_one(
278            &grid,
279            &d_num_rows,
280            &cu_stream,
281            &format!("{ctx}: d_num_rows"),
282        )?;
283
284        let kernel = self
285            .device()
286            .inner()
287            .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_TRIANGLE_COUNT_HG_U32)
288            .ok_or_else(|| {
289                XlogError::Kernel("wcoj_triangle_count_hg_u32 kernel not found".to_string())
290            })?;
291        let count_u32 = unsafe { reinterpret_u8_as_u32(&mut count_bytes) };
292        let mut params: Vec<*mut c_void> = vec![
293            yz_col1.as_kernel_param(),
294            n_yz.as_kernel_param(),
295            xz_col1.as_kernel_param(),
296            n_xz.as_kernel_param(),
297            (&plan.xy_work_prefix).as_kernel_param(),
298            (&plan.xy_yz_start).as_kernel_param(),
299            (&plan.xy_yz_end).as_kernel_param(),
300            (&plan.xy_xz_start).as_kernel_param(),
301            (&plan.xy_xz_end).as_kernel_param(),
302            plan.row_count.as_kernel_param(),
303            plan.total_work.as_kernel_param(),
304            plan.block_work_unit.as_kernel_param(),
305            count_u32.as_kernel_param(),
306        ];
307        unsafe {
308            kernel
309                .clone()
310                .launch_on_stream(
311                    &cu_stream,
312                    LaunchConfig {
313                        grid_dim: (grid, 1, 1),
314                        block_dim: (BLOCK_SIZE, 1, 1),
315                        shared_mem_bytes: 0,
316                    },
317                    &mut params,
318                )
319                .map_err(|e| XlogError::Kernel(format!("{ctx}: launch failed: {e}")))?;
320        }
321        rec.commit(runtime)
322            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
323        cu_stream
324            .synchronize()
325            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
326
327        let schema = Schema::new(vec![("count".to_string(), ScalarType::U32)]);
328        Ok(CudaBuffer::from_columns_with_host_count(
329            vec![count_bytes.into()],
330            grid as u64,
331            d_num_rows,
332            schema,
333            grid,
334        ))
335    }
336
337    pub fn wcoj_triangle_hg_u32_recorded(
338        &self,
339        e_xy: &CudaBuffer,
340        e_yz: &CudaBuffer,
341        e_xz: &CudaBuffer,
342        block_work_unit: u32,
343        launch_stream: StreamId,
344    ) -> Result<CudaBuffer> {
345        let ctx = "wcoj_triangle_hg_u32_recorded";
346        validate_binary_u32(ctx, "e_xy", e_xy)?;
347        validate_binary_u32(ctx, "e_yz", e_yz)?;
348        validate_binary_u32(ctx, "e_xz", e_xz)?;
349        let plan = self.wcoj_triangle_hg_work_plan_u32_recorded(
350            e_xy,
351            e_yz,
352            e_xz,
353            block_work_unit,
354            launch_stream,
355        )?;
356        self.wcoj_triangle_hg_u32_with_plan_recorded(e_xy, e_yz, e_xz, &plan, launch_stream)
357    }
358
359    /// Aggregate-fused triangle group-by-root count: evaluate
360    /// `q(X, count) :- e_xy(X,Y), e_yz(Y,Z), e_xz(X,Z)` grouped by the
361    /// variable-order root X, WITHOUT materializing the triangle rows.
362    ///
363    /// Pipeline (all recorded; the triangle result never exists as rows):
364    /// 1. the standard histogram-guided work plan;
365    /// 2. `wcoj_triangle_groupby_root_count_hg_u32` accumulates per-e_xy-row
366    ///    match counts (integer atomicAdd — order-insensitive, deterministic
367    ///    values) into a zero-initialized `n_xy`-long array;
368    /// 3. a 2-column (X, count) staging buffer over the *input* rows is
369    ///    compacted to count>0 rows (group-by over the join result must not
370    ///    emit roots with no completion) and reduced per X via the recorded
371    ///    groupby Sum (rows are already X-sorted because e_xy is lex-sorted).
372    ///
373    /// All reduction work is O(n_xy) — input-sized, never join-output-sized.
374    ///
375    /// Output schema matches the unfused materialize+groupby-count baseline:
376    /// `col0` = X (e_xy.col0 type, U32/Symbol), `col1` = count (U64).
377    ///
378    /// # Errors
379    /// * `XlogError::Kernel` if the manager has no runtime, the launch
380    ///   stream does not resolve, an input is not 2-column U32/Symbol, or
381    ///   any kernel launch fails.
382    pub fn wcoj_triangle_groupby_root_count_u32_recorded(
383        &self,
384        e_xy: &CudaBuffer,
385        e_yz: &CudaBuffer,
386        e_xz: &CudaBuffer,
387        block_work_unit: u32,
388        launch_stream: StreamId,
389    ) -> Result<CudaBuffer> {
390        let ctx = "wcoj_triangle_groupby_root_count_u32_recorded";
391        // Layout-normalize per dispatch (sorted-fast-path clone when the
392        // input is already lex-sorted + unique): the fused path must give
393        // the same guarantee as the unfused pipeline instead of trusting
394        // store-buffer sortedness — unsorted/duplicated inputs previously
395        // produced silently wrong (empty) fused results.
396        let e_xy = &self.wcoj_layout_u32_recorded(e_xy, launch_stream)?;
397        let e_yz = &self.wcoj_layout_u32_recorded(e_yz, launch_stream)?;
398        let e_xz = &self.wcoj_layout_u32_recorded(e_xz, launch_stream)?;
399        validate_binary_u32(ctx, "e_xy", e_xy)?;
400        validate_binary_u32(ctx, "e_yz", e_yz)?;
401        validate_binary_u32(ctx, "e_xz", e_xz)?;
402        let plan = self.wcoj_triangle_hg_work_plan_u32_recorded(
403            e_xy,
404            e_yz,
405            e_xz,
406            block_work_unit,
407            launch_stream,
408        )?;
409        let n_xy = plan.row_count;
410        let x_type = e_xy.schema().column_type(0).expect("xy.col0 type");
411        let out_schema = Schema::new(vec![
412            ("x".to_string(), x_type),
413            ("count".to_string(), ScalarType::U64),
414        ]);
415        if n_xy == 0 || plan.total_work == 0 {
416            return self.create_empty_buffer(out_schema);
417        }
418
419        let runtime = self.memory().runtime().ok_or_else(|| {
420            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
421        })?;
422        let cu_stream = runtime
423            .stream_pool()
424            .resolve(launch_stream)
425            .ok_or_else(|| {
426                XlogError::Kernel(format!(
427                    "{ctx}: launch_stream StreamId({}) does not resolve",
428                    launch_stream.0
429                ))
430            })?;
431
432        let yz_col1 = metadata_column_u32(e_yz, 1)?;
433        let xz_col1 = metadata_column_u32(e_xz, 1)?;
434        let n_yz = self.metadata_logical_rows(e_yz)?;
435        let n_xz = self.metadata_logical_rows(e_xz)?;
436
437        // Per-e_xy-row match counters, zero-initialized. Allocated as the
438        // u8-backed column layout so the array doubles as the staging
439        // buffer's count column after the kernel fills it.
440        let mut row_counts = self
441            .memory()
442            .alloc::<u8>(n_xy as usize * std::mem::size_of::<u32>())?;
443        self.device()
444            .inner()
445            .memset_zeros(&mut row_counts)
446            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row counts failed: {e}")))?;
447
448        let grid = plan.total_work.div_ceil(plan.block_work_unit);
449        let mut rec = LaunchRecorder::new_strict(launch_stream);
450        rec.read(e_xy.num_rows_device());
451        rec.read(e_yz.num_rows_device());
452        rec.read(e_xz.num_rows_device());
453        rec.read_column(e_yz.column(1).expect("yz.col1"));
454        rec.read_column(e_xz.column(1).expect("xz.col1"));
455        rec.read(&plan.xy_work_prefix);
456        rec.read(&plan.xy_yz_start);
457        rec.read(&plan.xy_yz_end);
458        rec.read(&plan.xy_xz_start);
459        rec.read(&plan.xy_xz_end);
460        rec.write(&row_counts);
461        rec.preflight(runtime)
462            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
463        {
464            let kernel = self
465                .device()
466                .inner()
467                .get_func(
468                    WCOJ_MODULE,
469                    wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_COUNT_HG_U32,
470                )
471                .ok_or_else(|| {
472                    XlogError::Kernel(
473                        "wcoj_triangle_groupby_root_count_hg_u32 kernel not found".to_string(),
474                    )
475                })?;
476            let mut params: Vec<*mut c_void> = vec![
477                yz_col1.as_kernel_param(),
478                n_yz.as_kernel_param(),
479                xz_col1.as_kernel_param(),
480                n_xz.as_kernel_param(),
481                (&plan.xy_work_prefix).as_kernel_param(),
482                (&plan.xy_yz_start).as_kernel_param(),
483                (&plan.xy_yz_end).as_kernel_param(),
484                (&plan.xy_xz_start).as_kernel_param(),
485                (&plan.xy_xz_end).as_kernel_param(),
486                plan.row_count.as_kernel_param(),
487                plan.total_work.as_kernel_param(),
488                plan.block_work_unit.as_kernel_param(),
489                (&row_counts).as_kernel_param(),
490            ];
491            unsafe {
492                kernel
493                    .clone()
494                    .launch_on_stream(
495                        &cu_stream,
496                        LaunchConfig {
497                            grid_dim: (grid, 1, 1),
498                            block_dim: (BLOCK_SIZE, 1, 1),
499                            shared_mem_bytes: 0,
500                        },
501                        &mut params,
502                    )
503                    .map_err(|e| {
504                        XlogError::Kernel(format!("{ctx}: groupby-count launch failed: {e}"))
505                    })?;
506            }
507        }
508        rec.commit(runtime)
509            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
510
511        // Staging buffer (X, count) over the n_xy input rows: X is a
512        // device-to-device copy of e_xy.col0; the count column is the
513        // kernel-filled array. Rows stay lex-sorted by X.
514        let x_src = match e_xy.column(0).expect("xy.col0") {
515            CudaColumn::Owned(slice) => slice,
516            _ => {
517                return Err(XlogError::Kernel(format!(
518                    "{ctx}: e_xy.col0 must be an owned CudaColumn"
519                )))
520            }
521        };
522        let x_copy = self
523            .memory()
524            .alloc::<u8>(n_xy as usize * std::mem::size_of::<u32>())?;
525        // Explicit-length copy: layout-normalized columns are allocated at
526        // capacity, which can exceed the logical n_xy * 4 bytes a full-slice
527        // typed copy would assert on.
528        unsafe {
529            let res = sys::cuMemcpyDtoD_v2(
530                *x_copy.device_ptr(),
531                *x_src.device_ptr(),
532                n_xy as usize * std::mem::size_of::<u32>(),
533            );
534            if res != sys::cudaError_enum::CUDA_SUCCESS {
535                return Err(XlogError::Kernel(format!(
536                    "{ctx}: copy X column failed: {res:?}"
537                )));
538            }
539        }
540        let mut d_num_rows = self.memory().alloc::<u32>(1)?;
541        self.device()
542            .inner()
543            .dtod_copy(e_xy.num_rows_device(), &mut d_num_rows)
544            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy row count failed: {e}")))?;
545        let staging_schema = Schema::new(vec![
546            ("x".to_string(), x_type),
547            ("count".to_string(), ScalarType::U32),
548        ]);
549        let staging = CudaBuffer::from_columns_with_host_count(
550            vec![x_copy.into(), row_counts.into()],
551            n_xy as u64,
552            d_num_rows,
553            staging_schema,
554            n_xy,
555        );
556
557        // Keep only roots with at least one completed triangle, then reduce
558        // per X. Both steps run over input-sized data.
559        let mask = self.compare_const_mask_recorded::<u32>(
560            &staging,
561            1,
562            0u32,
563            crate::CompareOp::Gt,
564            launch_stream,
565        )?;
566        let compacted =
567            self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)?;
568        self.groupby_multi_agg_recorded(
569            &compacted,
570            &[0],
571            &[(1, xlog_core::AggOp::Sum)],
572            launch_stream,
573        )
574    }
575
576    /// Aggregate-fused triangle group-by-root sum/min/max: evaluate
577    /// `q(X, agg(V)) :- e_xy(X,Y), e_yz(Y,Z), e_xz(X,Z)` with
578    /// `agg ∈ {Sum, Min, Max}` and `V ∈ {Y, Z}` grouped by the
579    /// variable-order root X, WITHOUT materializing the triangle rows.
580    ///
581    /// Pipeline (all recorded; the triangle result never exists as rows):
582    /// 1. the standard histogram-guided work plan;
583    /// 2. the per-op fused kernel accumulates, per e_xy row, a match count
584    ///    (compaction mask) and the per-row partial aggregate (integer
585    ///    atomics — order-insensitive, deterministic values). Sum partials
586    ///    are u64 (a per-row partial can exceed `u32::MAX`); min partials
587    ///    start at `u32::MAX`, max partials at 0;
588    /// 3. a 3-column (X, count, agg) staging buffer over the *input* rows
589    ///    is compacted to count>0 rows (groups with no completion must be
590    ///    absent) and reduced per X via the recorded groupby with the same
591    ///    `AggOp` (Sum over the u64 partials; Min/Max over u32).
592    ///
593    /// All reduction work is O(n_xy) — input-sized, never join-output-sized.
594    ///
595    /// Output schema matches the unfused materialize+groupby baseline:
596    /// `col0` = X (e_xy.col0 type, U32/Symbol), `col1` = U64 for Sum,
597    /// U32 for Min/Max.
598    ///
599    /// Bag semantics: every (Y, Z) completion contributes its value,
600    /// exactly like aggregating the materialized projection.
601    ///
602    /// # Errors
603    /// * `XlogError::Kernel` if `agg_op` is not Sum/Min/Max, the value
604    ///   columns are not plain U32, the manager has no runtime, the launch
605    ///   stream does not resolve, an input is not 2-column U32/Symbol, or
606    ///   any kernel launch fails.
607    #[expect(
608        clippy::too_many_arguments,
609        reason = "three relation inputs and launch controls mirror the recorded kernel contract"
610    )]
611    pub fn wcoj_triangle_groupby_root_agg_u32_recorded(
612        &self,
613        e_xy: &CudaBuffer,
614        e_yz: &CudaBuffer,
615        e_xz: &CudaBuffer,
616        agg_op: AggOp,
617        value: WcojRootAggValue,
618        block_work_unit: u32,
619        launch_stream: StreamId,
620    ) -> Result<CudaBuffer> {
621        let ctx = "wcoj_triangle_groupby_root_agg_u32_recorded";
622        // Layout-normalize per dispatch (sorted-fast-path clone when the
623        // input is already lex-sorted + unique): the fused path must give
624        // the same guarantee as the unfused pipeline instead of trusting
625        // store-buffer sortedness — unsorted/duplicated inputs previously
626        // produced silently wrong (empty) fused results.
627        let e_xy = &self.wcoj_layout_u32_recorded(e_xy, launch_stream)?;
628        let e_yz = &self.wcoj_layout_u32_recorded(e_yz, launch_stream)?;
629        let e_xz = &self.wcoj_layout_u32_recorded(e_xz, launch_stream)?;
630        let (kernel_name, agg_elem_size, agg_scalar, agg_name) = match agg_op {
631            AggOp::Sum => (
632                wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_SUM_HG_U32,
633                std::mem::size_of::<u64>(),
634                ScalarType::U64,
635                "sum_0",
636            ),
637            AggOp::Min => (
638                wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_MIN_HG_U32,
639                std::mem::size_of::<u32>(),
640                ScalarType::U32,
641                "min_0",
642            ),
643            AggOp::Max => (
644                wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_MAX_HG_U32,
645                std::mem::size_of::<u32>(),
646                ScalarType::U32,
647                "max_0",
648            ),
649            other => {
650                return Err(XlogError::Kernel(format!(
651                    "{ctx}: unsupported AggOp {other:?} (Sum/Min/Max only; use \
652                     wcoj_triangle_groupby_root_count_u32_recorded for Count)"
653                )))
654            }
655        };
656        validate_binary_u32(ctx, "e_xy", e_xy)?;
657        validate_binary_u32(ctx, "e_yz", e_yz)?;
658        validate_binary_u32(ctx, "e_xz", e_xz)?;
659        // The aggregate value is arithmetic: require plain U32 value
660        // columns (Symbol ids are not summable/orderable data).
661        let value_cols: &[(&CudaBuffer, &str)] = match value {
662            WcojRootAggValue::Y => &[(e_xy, "e_xy")],
663            WcojRootAggValue::Z => &[(e_yz, "e_yz"), (e_xz, "e_xz")],
664        };
665        for (buf, label) in value_cols {
666            let ty = buf.schema().column_type(1).expect("validated 2-col");
667            if ty != ScalarType::U32 {
668                return Err(XlogError::Kernel(format!(
669                    "{ctx}: {label}.col1 supplies the aggregate value and must be U32, got {ty:?}"
670                )));
671            }
672        }
673
674        let plan = self.wcoj_triangle_hg_work_plan_u32_recorded(
675            e_xy,
676            e_yz,
677            e_xz,
678            block_work_unit,
679            launch_stream,
680        )?;
681        let n_xy = plan.row_count;
682        let x_type = e_xy.schema().column_type(0).expect("xy.col0 type");
683        let out_schema = Schema::new(vec![
684            ("x".to_string(), x_type),
685            (agg_name.to_string(), agg_scalar),
686        ]);
687        if n_xy == 0 || plan.total_work == 0 {
688            return self.create_empty_buffer(out_schema);
689        }
690
691        let runtime = self.memory().runtime().ok_or_else(|| {
692            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
693        })?;
694        let cu_stream = runtime
695            .stream_pool()
696            .resolve(launch_stream)
697            .ok_or_else(|| {
698                XlogError::Kernel(format!(
699                    "{ctx}: launch_stream StreamId({}) does not resolve",
700                    launch_stream.0
701                ))
702            })?;
703
704        let yz_col1 = metadata_column_u32(e_yz, 1)?;
705        let xz_col1 = metadata_column_u32(e_xz, 1)?;
706        let xy_col1 = metadata_column_u32(e_xy, 1)?;
707        let n_yz = self.metadata_logical_rows(e_yz)?;
708        let n_xz = self.metadata_logical_rows(e_xz)?;
709        let value_from_z: u32 = match value {
710            WcojRootAggValue::Y => 0,
711            WcojRootAggValue::Z => 1,
712        };
713
714        // Per-e_xy-row match counters + aggregate partials, allocated as
715        // the u8-backed column layout so the arrays double as the staging
716        // buffer's columns after the kernel fills them.
717        let mut row_counts = self
718            .memory()
719            .alloc::<u8>(n_xy as usize * std::mem::size_of::<u32>())?;
720        self.device()
721            .inner()
722            .memset_zeros(&mut row_counts)
723            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row counts failed: {e}")))?;
724        let mut row_agg = self.memory().alloc::<u8>(n_xy as usize * agg_elem_size)?;
725        self.device()
726            .inner()
727            .memset_zeros(&mut row_agg)
728            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row aggregates failed: {e}")))?;
729
730        let grid = plan.total_work.div_ceil(plan.block_work_unit);
731        let mut rec = LaunchRecorder::new_strict(launch_stream);
732        rec.read(e_xy.num_rows_device());
733        rec.read(e_yz.num_rows_device());
734        rec.read(e_xz.num_rows_device());
735        rec.read_column(e_yz.column(1).expect("yz.col1"));
736        rec.read_column(e_xz.column(1).expect("xz.col1"));
737        rec.read_column(e_xy.column(1).expect("xy.col1"));
738        rec.read(&plan.xy_work_prefix);
739        rec.read(&plan.xy_yz_start);
740        rec.read(&plan.xy_yz_end);
741        rec.read(&plan.xy_xz_start);
742        rec.read(&plan.xy_xz_end);
743        rec.write(&row_counts);
744        rec.write(&row_agg);
745        rec.preflight(runtime)
746            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
747        if matches!(agg_op, AggOp::Min) {
748            // Min identity: u32::MAX (compaction drops untouched rows).
749            let fill = self
750                .device()
751                .inner()
752                .get_func(ARITH_MODULE, arith_kernels::ARITH_FILL_CONST_U32)
753                .ok_or_else(|| {
754                    XlogError::Kernel("arith_fill_const_u32 kernel not found".to_string())
755                })?;
756            let row_agg_u32 = unsafe { reinterpret_u8_as_u32(&mut row_agg) };
757            // SAFETY: arith_fill_const_u32(value, n, output)
758            unsafe {
759                fill.clone()
760                    .launch_on_stream(
761                        &cu_stream,
762                        LaunchConfig::for_num_elems(n_xy),
763                        (u32::MAX, n_xy, &mut *row_agg_u32),
764                    )
765                    .map_err(|e| {
766                        XlogError::Kernel(format!("{ctx}: min identity fill failed: {e}"))
767                    })?;
768            }
769        }
770        {
771            let kernel = self
772                .device()
773                .inner()
774                .get_func(WCOJ_MODULE, kernel_name)
775                .ok_or_else(|| XlogError::Kernel(format!("{kernel_name} kernel not found")))?;
776            let mut params: Vec<*mut c_void> = vec![
777                yz_col1.as_kernel_param(),
778                n_yz.as_kernel_param(),
779                xz_col1.as_kernel_param(),
780                n_xz.as_kernel_param(),
781                xy_col1.as_kernel_param(),
782                value_from_z.as_kernel_param(),
783                (&plan.xy_work_prefix).as_kernel_param(),
784                (&plan.xy_yz_start).as_kernel_param(),
785                (&plan.xy_yz_end).as_kernel_param(),
786                (&plan.xy_xz_start).as_kernel_param(),
787                (&plan.xy_xz_end).as_kernel_param(),
788                plan.row_count.as_kernel_param(),
789                plan.total_work.as_kernel_param(),
790                plan.block_work_unit.as_kernel_param(),
791                (&row_counts).as_kernel_param(),
792                (&row_agg).as_kernel_param(),
793            ];
794            unsafe {
795                kernel
796                    .clone()
797                    .launch_on_stream(
798                        &cu_stream,
799                        LaunchConfig {
800                            grid_dim: (grid, 1, 1),
801                            block_dim: (BLOCK_SIZE, 1, 1),
802                            shared_mem_bytes: 0,
803                        },
804                        &mut params,
805                    )
806                    .map_err(|e| {
807                        XlogError::Kernel(format!("{ctx}: groupby-agg launch failed: {e}"))
808                    })?;
809            }
810        }
811        rec.commit(runtime)
812            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
813
814        // Staging buffer (X, count, agg) over the n_xy input rows: X is a
815        // device-to-device copy of e_xy.col0; count and agg are the
816        // kernel-filled arrays. Rows stay lex-sorted by X.
817        let x_src = match e_xy.column(0).expect("xy.col0") {
818            CudaColumn::Owned(slice) => slice,
819            _ => {
820                return Err(XlogError::Kernel(format!(
821                    "{ctx}: e_xy.col0 must be an owned CudaColumn"
822                )))
823            }
824        };
825        let x_copy = self
826            .memory()
827            .alloc::<u8>(n_xy as usize * std::mem::size_of::<u32>())?;
828        // Explicit-length copy: layout-normalized columns are allocated at
829        // capacity, which can exceed the logical n_xy * 4 bytes a full-slice
830        // typed copy would assert on.
831        unsafe {
832            let res = sys::cuMemcpyDtoD_v2(
833                *x_copy.device_ptr(),
834                *x_src.device_ptr(),
835                n_xy as usize * std::mem::size_of::<u32>(),
836            );
837            if res != sys::cudaError_enum::CUDA_SUCCESS {
838                return Err(XlogError::Kernel(format!(
839                    "{ctx}: copy X column failed: {res:?}"
840                )));
841            }
842        }
843        let mut d_num_rows = self.memory().alloc::<u32>(1)?;
844        self.device()
845            .inner()
846            .dtod_copy(e_xy.num_rows_device(), &mut d_num_rows)
847            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy row count failed: {e}")))?;
848        let staging_schema = Schema::new(vec![
849            ("x".to_string(), x_type),
850            ("count".to_string(), ScalarType::U32),
851            ("agg".to_string(), agg_scalar),
852        ]);
853        let staging = CudaBuffer::from_columns_with_host_count(
854            vec![x_copy.into(), row_counts.into(), row_agg.into()],
855            n_xy as u64,
856            d_num_rows,
857            staging_schema,
858            n_xy,
859        );
860
861        // Keep only roots with at least one completed triangle, then reduce
862        // per X with the same AggOp. Both steps run over input-sized data.
863        let mask = self.compare_const_mask_recorded::<u32>(
864            &staging,
865            1,
866            0u32,
867            crate::CompareOp::Gt,
868            launch_stream,
869        )?;
870        let compacted =
871            self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)?;
872        self.groupby_multi_agg_recorded(&compacted, &[0], &[(2, agg_op)], launch_stream)
873    }
874
875    /// U64-key aggregate-fused triangle count sibling of
876    /// [`Self::wcoj_triangle_groupby_root_count_u32_recorded`]: evaluate
877    /// `q(X, count)` over the triangle shape grouped by the root X for U64
878    /// relations, WITHOUT materializing the triangle rows.
879    ///
880    /// The recorded groupby is U32/Symbol-key only, so the per-X reduction
881    /// reuses the WCOJ relation metadata instead: e_xy is lex-sorted, so
882    /// `wcoj_build_metadata_u64_recorded` yields one (unique X, group start)
883    /// pair per root, and `wcoj_groupby_root_segment_sum_counts_u32`
884    /// accumulates the per-row match counts into per-unique-root u64
885    /// totals (integer atomicAdd — deterministic). Roots with zero
886    /// completions are compacted away. All reduction work is O(n_xy).
887    ///
888    /// Output schema matches the unfused materialize+groupby baseline:
889    /// `col0` = X (U64), `col1` = count (U64).
890    ///
891    /// # Errors
892    /// * `XlogError::Kernel` if the manager has no runtime, the launch
893    ///   stream does not resolve, an input is not 2-column U64, or any
894    ///   kernel launch fails.
895    pub fn wcoj_triangle_groupby_root_count_u64_recorded(
896        &self,
897        e_xy: &CudaBuffer,
898        e_yz: &CudaBuffer,
899        e_xz: &CudaBuffer,
900        block_work_unit: u32,
901        launch_stream: StreamId,
902    ) -> Result<CudaBuffer> {
903        let ctx = "wcoj_triangle_groupby_root_count_u64_recorded";
904        // Layout-normalize per dispatch (sorted-fast-path clone when the
905        // input is already lex-sorted + unique): the fused path must give
906        // the same guarantee as the unfused pipeline instead of trusting
907        // store-buffer sortedness — unsorted/duplicated inputs previously
908        // produced silently wrong (empty) fused results.
909        let e_xy = &self.wcoj_layout_u64_recorded(e_xy, launch_stream)?;
910        let e_yz = &self.wcoj_layout_u64_recorded(e_yz, launch_stream)?;
911        let e_xz = &self.wcoj_layout_u64_recorded(e_xz, launch_stream)?;
912        validate_binary_u64(ctx, "e_xy", e_xy)?;
913        validate_binary_u64(ctx, "e_yz", e_yz)?;
914        validate_binary_u64(ctx, "e_xz", e_xz)?;
915        let plan = self.wcoj_triangle_hg_work_plan_u64_recorded(
916            e_xy,
917            e_yz,
918            e_xz,
919            block_work_unit,
920            launch_stream,
921        )?;
922        let n_xy = plan.row_count;
923        let out_schema = Schema::new(vec![
924            ("x".to_string(), ScalarType::U64),
925            ("count".to_string(), ScalarType::U64),
926        ]);
927        if n_xy == 0 || plan.total_work == 0 {
928            return self.create_empty_buffer(out_schema);
929        }
930
931        let runtime = self.memory().runtime().ok_or_else(|| {
932            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
933        })?;
934        let cu_stream = runtime
935            .stream_pool()
936            .resolve(launch_stream)
937            .ok_or_else(|| {
938                XlogError::Kernel(format!(
939                    "{ctx}: launch_stream StreamId({}) does not resolve",
940                    launch_stream.0
941                ))
942            })?;
943
944        let yz_col1 = metadata_column_u64(e_yz, 1)?;
945        let xz_col1 = metadata_column_u64(e_xz, 1)?;
946        let n_yz = self.metadata_logical_rows(e_yz)?;
947        let n_xz = self.metadata_logical_rows(e_xz)?;
948
949        // Per-e_xy-row match counters, zero-initialized.
950        let mut row_counts = self.memory().alloc::<u32>(n_xy as usize)?;
951        self.device()
952            .inner()
953            .memset_zeros(&mut row_counts)
954            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row counts failed: {e}")))?;
955
956        let grid = plan.total_work.div_ceil(plan.block_work_unit);
957        let mut rec = LaunchRecorder::new_strict(launch_stream);
958        rec.read(e_xy.num_rows_device());
959        rec.read(e_yz.num_rows_device());
960        rec.read(e_xz.num_rows_device());
961        rec.read_column(e_yz.column(1).expect("yz.col1"));
962        rec.read_column(e_xz.column(1).expect("xz.col1"));
963        rec.read(&plan.xy_work_prefix);
964        rec.read(&plan.xy_yz_start);
965        rec.read(&plan.xy_yz_end);
966        rec.read(&plan.xy_xz_start);
967        rec.read(&plan.xy_xz_end);
968        rec.write(&row_counts);
969        rec.preflight(runtime)
970            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
971        {
972            let kernel = self
973                .device()
974                .inner()
975                .get_func(
976                    WCOJ_MODULE,
977                    wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_COUNT_HG_U64,
978                )
979                .ok_or_else(|| {
980                    XlogError::Kernel(
981                        "wcoj_triangle_groupby_root_count_hg_u64 kernel not found".to_string(),
982                    )
983                })?;
984            let mut params: Vec<*mut c_void> = vec![
985                yz_col1.as_kernel_param(),
986                n_yz.as_kernel_param(),
987                xz_col1.as_kernel_param(),
988                n_xz.as_kernel_param(),
989                (&plan.xy_work_prefix).as_kernel_param(),
990                (&plan.xy_yz_start).as_kernel_param(),
991                (&plan.xy_yz_end).as_kernel_param(),
992                (&plan.xy_xz_start).as_kernel_param(),
993                (&plan.xy_xz_end).as_kernel_param(),
994                plan.row_count.as_kernel_param(),
995                plan.total_work.as_kernel_param(),
996                plan.block_work_unit.as_kernel_param(),
997                (&row_counts).as_kernel_param(),
998            ];
999            unsafe {
1000                kernel
1001                    .clone()
1002                    .launch_on_stream(
1003                        &cu_stream,
1004                        LaunchConfig {
1005                            grid_dim: (grid, 1, 1),
1006                            block_dim: (BLOCK_SIZE, 1, 1),
1007                            shared_mem_bytes: 0,
1008                        },
1009                        &mut params,
1010                    )
1011                    .map_err(|e| {
1012                        XlogError::Kernel(format!("{ctx}: groupby-count launch failed: {e}"))
1013                    })?;
1014            }
1015        }
1016        rec.commit(runtime)
1017            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
1018
1019        // Per-X reduction via the relation metadata: one (unique X, group
1020        // start) pair per root; e_xy is lex-sorted by X so group rows are
1021        // contiguous.
1022        let meta = self.wcoj_build_metadata_u64_recorded(e_xy, 0, launch_stream)?;
1023        let key_count = meta.key_count;
1024        if key_count == 0 {
1025            return self.create_empty_buffer(out_schema);
1026        }
1027        let mut sums = self
1028            .memory()
1029            .alloc::<u8>(key_count as usize * std::mem::size_of::<u64>())?;
1030        self.device()
1031            .inner()
1032            .memset_zeros(&mut sums)
1033            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero group sums failed: {e}")))?;
1034
1035        let mut rec_sum = LaunchRecorder::new_strict(launch_stream);
1036        rec_sum.read(&row_counts);
1037        rec_sum.read(&meta.prefix_sum);
1038        rec_sum.write(&sums);
1039        rec_sum
1040            .preflight(runtime)
1041            .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce preflight failed: {e}")))?;
1042        {
1043            let kernel = self
1044                .device()
1045                .inner()
1046                .get_func(
1047                    WCOJ_MODULE,
1048                    wcoj_kernels::WCOJ_GROUPBY_ROOT_SEGMENT_SUM_COUNTS_U32,
1049                )
1050                .ok_or_else(|| {
1051                    XlogError::Kernel(
1052                        "wcoj_groupby_root_segment_sum_counts_u32 kernel not found".to_string(),
1053                    )
1054                })?;
1055            let reduce_grid = n_xy.div_ceil(BLOCK_SIZE);
1056            let mut params: Vec<*mut c_void> = vec![
1057                (&row_counts).as_kernel_param(),
1058                n_xy.as_kernel_param(),
1059                (&meta.prefix_sum).as_kernel_param(),
1060                key_count.as_kernel_param(),
1061                (&sums).as_kernel_param(),
1062            ];
1063            unsafe {
1064                kernel
1065                    .clone()
1066                    .launch_on_stream(
1067                        &cu_stream,
1068                        LaunchConfig {
1069                            grid_dim: (reduce_grid, 1, 1),
1070                            block_dim: (BLOCK_SIZE, 1, 1),
1071                            shared_mem_bytes: 0,
1072                        },
1073                        &mut params,
1074                    )
1075                    .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce launch failed: {e}")))?;
1076            }
1077        }
1078        rec_sum
1079            .commit(runtime)
1080            .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce commit failed: {e}")))?;
1081
1082        // (unique X, total) buffer over the key_count roots, then drop the
1083        // roots with no completion. The copies run on launch_stream and the
1084        // fresh destination blocks are registered through the strict
1085        // recorder BEFORE the enqueue — a raw async copy into a freshly
1086        // pool-allocated block without recording is a visibility race.
1087        let x_copy = self
1088            .memory()
1089            .alloc::<u8>(key_count as usize * std::mem::size_of::<u64>())?;
1090        let d_num_rows = self.memory().alloc::<u32>(1)?;
1091        let mut rec_copy = LaunchRecorder::new_strict(launch_stream);
1092        rec_copy.read(&meta.unique_keys);
1093        rec_copy.write(&x_copy);
1094        rec_copy.write(&d_num_rows);
1095        rec_copy
1096            .preflight(runtime)
1097            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy preflight failed: {e}")))?;
1098        unsafe {
1099            let res = sys::cuMemcpyDtoDAsync_v2(
1100                *x_copy.device_ptr(),
1101                *meta.unique_keys.device_ptr(),
1102                key_count as usize * std::mem::size_of::<u64>(),
1103                cu_stream.cu_stream(),
1104            );
1105            if res != sys::cudaError_enum::CUDA_SUCCESS {
1106                return Err(XlogError::Kernel(format!(
1107                    "{ctx}: DtoD unique keys copy failed: {res:?}"
1108                )));
1109            }
1110        }
1111        self.htod_launch_metadata_async_copy_one(
1112            &key_count,
1113            &d_num_rows,
1114            &cu_stream,
1115            &format!("{ctx}: d_num_rows"),
1116        )?;
1117        rec_copy
1118            .commit(runtime)
1119            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy commit failed: {e}")))?;
1120        cu_stream
1121            .synchronize()
1122            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
1123        let staging_schema = Schema::new(vec![
1124            ("x".to_string(), ScalarType::U64),
1125            ("count".to_string(), ScalarType::U64),
1126        ]);
1127        let staging = CudaBuffer::from_columns_with_host_count(
1128            vec![x_copy.into(), sums.into()],
1129            u64::from(key_count),
1130            d_num_rows,
1131            staging_schema,
1132            key_count,
1133        );
1134        let mask = self.compare_const_mask_recorded::<u64>(
1135            &staging,
1136            1,
1137            0u64,
1138            crate::CompareOp::Gt,
1139            launch_stream,
1140        )?;
1141        self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)
1142    }
1143
1144    /// U64-key aggregate-fused triangle sum/min/max sibling of
1145    /// [`Self::wcoj_triangle_groupby_root_agg_u32_recorded`]: evaluate
1146    /// `q(X, agg(V)) :- e_xy(X,Y), e_yz(Y,Z), e_xz(X,Z)` with
1147    /// `agg ∈ {Sum, Min, Max}` and `V ∈ {Y, Z}` over U64 relations,
1148    /// grouped by the variable-order root X, WITHOUT materializing the
1149    /// triangle rows.
1150    ///
1151    /// The recorded groupby is U32/Symbol-key only, so the per-X reduction
1152    /// reuses the WCOJ relation metadata (one unique root per group, e_xy
1153    /// lex-sorted) like the u64 count path:
1154    /// 1. the per-op fused kernel accumulates, per e_xy row, a match count
1155    ///    and a u64 aggregate partial (integer atomics — deterministic;
1156    ///    sum wraps on overflow exactly like `groupby_sum_u64`; min
1157    ///    partials start at `u64::MAX`, max partials at 0);
1158    /// 2. `wcoj_groupby_root_segment_sum_counts_u32` reduces per-row match
1159    ///    counts to per-unique-root totals (the presence mask), and the
1160    ///    per-op `wcoj_groupby_root_segment_{sum,min,max}_values_u64`
1161    ///    kernel folds the per-row partials into per-unique-root u64
1162    ///    aggregates, skipping zero-match rows;
1163    /// 3. a (X, agg) staging buffer over the unique roots is compacted to
1164    ///    count>0 groups.
1165    ///
1166    /// All reduction work is O(n_xy) — input-sized, never join-output-sized.
1167    ///
1168    /// Output schema matches the unfused materialize+groupby baseline
1169    /// (legacy groupby widened to u64 values): `col0` = X (U64),
1170    /// `col1` = U64 for sum, min and max alike.
1171    ///
1172    /// # Errors
1173    /// * `XlogError::Kernel` if `agg_op` is not Sum/Min/Max, the manager
1174    ///   has no runtime, the launch stream does not resolve, an input is
1175    ///   not 2-column U64, or any kernel launch fails.
1176    #[expect(
1177        clippy::too_many_arguments,
1178        reason = "three relation inputs and launch controls mirror the recorded kernel contract"
1179    )]
1180    pub fn wcoj_triangle_groupby_root_agg_u64_recorded(
1181        &self,
1182        e_xy: &CudaBuffer,
1183        e_yz: &CudaBuffer,
1184        e_xz: &CudaBuffer,
1185        agg_op: AggOp,
1186        value: WcojRootAggValue,
1187        block_work_unit: u32,
1188        launch_stream: StreamId,
1189    ) -> Result<CudaBuffer> {
1190        let ctx = "wcoj_triangle_groupby_root_agg_u64_recorded";
1191        // Layout-normalize per dispatch (sorted-fast-path clone when the
1192        // input is already lex-sorted + unique): the fused path must give
1193        // the same guarantee as the unfused pipeline instead of trusting
1194        // store-buffer sortedness — unsorted/duplicated inputs previously
1195        // produced silently wrong (empty) fused results.
1196        let e_xy = &self.wcoj_layout_u64_recorded(e_xy, launch_stream)?;
1197        let e_yz = &self.wcoj_layout_u64_recorded(e_yz, launch_stream)?;
1198        let e_xz = &self.wcoj_layout_u64_recorded(e_xz, launch_stream)?;
1199        let (kernel_name, segment_kernel_name, agg_name) = match agg_op {
1200            AggOp::Sum => (
1201                wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_SUM_HG_U64,
1202                wcoj_kernels::WCOJ_GROUPBY_ROOT_SEGMENT_SUM_VALUES_U64,
1203                "sum_0",
1204            ),
1205            AggOp::Min => (
1206                wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_MIN_HG_U64,
1207                wcoj_kernels::WCOJ_GROUPBY_ROOT_SEGMENT_MIN_VALUES_U64,
1208                "min_0",
1209            ),
1210            AggOp::Max => (
1211                wcoj_kernels::WCOJ_TRIANGLE_GROUPBY_ROOT_MAX_HG_U64,
1212                wcoj_kernels::WCOJ_GROUPBY_ROOT_SEGMENT_MAX_VALUES_U64,
1213                "max_0",
1214            ),
1215            other => {
1216                return Err(XlogError::Kernel(format!(
1217                    "{ctx}: unsupported AggOp {other:?} (Sum/Min/Max only; use \
1218                     wcoj_triangle_groupby_root_count_u64_recorded for Count)"
1219                )))
1220            }
1221        };
1222        validate_binary_u64(ctx, "e_xy", e_xy)?;
1223        validate_binary_u64(ctx, "e_yz", e_yz)?;
1224        validate_binary_u64(ctx, "e_xz", e_xz)?;
1225        let plan = self.wcoj_triangle_hg_work_plan_u64_recorded(
1226            e_xy,
1227            e_yz,
1228            e_xz,
1229            block_work_unit,
1230            launch_stream,
1231        )?;
1232        let n_xy = plan.row_count;
1233        let out_schema = Schema::new(vec![
1234            ("x".to_string(), ScalarType::U64),
1235            (agg_name.to_string(), ScalarType::U64),
1236        ]);
1237        if n_xy == 0 || plan.total_work == 0 {
1238            return self.create_empty_buffer(out_schema);
1239        }
1240
1241        let runtime = self.memory().runtime().ok_or_else(|| {
1242            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
1243        })?;
1244        let cu_stream = runtime
1245            .stream_pool()
1246            .resolve(launch_stream)
1247            .ok_or_else(|| {
1248                XlogError::Kernel(format!(
1249                    "{ctx}: launch_stream StreamId({}) does not resolve",
1250                    launch_stream.0
1251                ))
1252            })?;
1253
1254        let yz_col1 = metadata_column_u64(e_yz, 1)?;
1255        let xz_col1 = metadata_column_u64(e_xz, 1)?;
1256        let xy_col1 = metadata_column_u64(e_xy, 1)?;
1257        let n_yz = self.metadata_logical_rows(e_yz)?;
1258        let n_xz = self.metadata_logical_rows(e_xz)?;
1259        let value_from_z: u32 = match value {
1260            WcojRootAggValue::Y => 0,
1261            WcojRootAggValue::Z => 1,
1262        };
1263
1264        // Per-e_xy-row match counters + u64 aggregate partials.
1265        let mut row_counts = self.memory().alloc::<u32>(n_xy as usize)?;
1266        self.device()
1267            .inner()
1268            .memset_zeros(&mut row_counts)
1269            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row counts failed: {e}")))?;
1270        let mut row_agg = self
1271            .memory()
1272            .alloc::<u8>(n_xy as usize * std::mem::size_of::<u64>())?;
1273        self.device()
1274            .inner()
1275            .memset_zeros(&mut row_agg)
1276            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row aggregates failed: {e}")))?;
1277
1278        let grid = plan.total_work.div_ceil(plan.block_work_unit);
1279        let mut rec = LaunchRecorder::new_strict(launch_stream);
1280        rec.read(e_xy.num_rows_device());
1281        rec.read(e_yz.num_rows_device());
1282        rec.read(e_xz.num_rows_device());
1283        rec.read_column(e_yz.column(1).expect("yz.col1"));
1284        rec.read_column(e_xz.column(1).expect("xz.col1"));
1285        rec.read_column(e_xy.column(1).expect("xy.col1"));
1286        rec.read(&plan.xy_work_prefix);
1287        rec.read(&plan.xy_yz_start);
1288        rec.read(&plan.xy_yz_end);
1289        rec.read(&plan.xy_xz_start);
1290        rec.read(&plan.xy_xz_end);
1291        rec.write(&row_counts);
1292        rec.write(&row_agg);
1293        rec.preflight(runtime)
1294            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
1295        if matches!(agg_op, AggOp::Min) {
1296            // Min identity: u64::MAX (compaction drops untouched groups).
1297            let fill = self
1298                .device()
1299                .inner()
1300                .get_func(ARITH_MODULE, arith_kernels::ARITH_FILL_CONST_U64)
1301                .ok_or_else(|| {
1302                    XlogError::Kernel("arith_fill_const_u64 kernel not found".to_string())
1303                })?;
1304            let row_agg_u64 = unsafe { reinterpret_u8_as_u64(&mut row_agg) };
1305            // SAFETY: arith_fill_const_u64(value, n, output)
1306            unsafe {
1307                fill.clone()
1308                    .launch_on_stream(
1309                        &cu_stream,
1310                        LaunchConfig::for_num_elems(n_xy),
1311                        (u64::MAX, n_xy, &mut *row_agg_u64),
1312                    )
1313                    .map_err(|e| {
1314                        XlogError::Kernel(format!("{ctx}: min identity fill failed: {e}"))
1315                    })?;
1316            }
1317        }
1318        {
1319            let kernel = self
1320                .device()
1321                .inner()
1322                .get_func(WCOJ_MODULE, kernel_name)
1323                .ok_or_else(|| XlogError::Kernel(format!("{kernel_name} kernel not found")))?;
1324            let mut params: Vec<*mut c_void> = vec![
1325                yz_col1.as_kernel_param(),
1326                n_yz.as_kernel_param(),
1327                xz_col1.as_kernel_param(),
1328                n_xz.as_kernel_param(),
1329                xy_col1.as_kernel_param(),
1330                value_from_z.as_kernel_param(),
1331                (&plan.xy_work_prefix).as_kernel_param(),
1332                (&plan.xy_yz_start).as_kernel_param(),
1333                (&plan.xy_yz_end).as_kernel_param(),
1334                (&plan.xy_xz_start).as_kernel_param(),
1335                (&plan.xy_xz_end).as_kernel_param(),
1336                plan.row_count.as_kernel_param(),
1337                plan.total_work.as_kernel_param(),
1338                plan.block_work_unit.as_kernel_param(),
1339                (&row_counts).as_kernel_param(),
1340                (&row_agg).as_kernel_param(),
1341            ];
1342            unsafe {
1343                kernel
1344                    .clone()
1345                    .launch_on_stream(
1346                        &cu_stream,
1347                        LaunchConfig {
1348                            grid_dim: (grid, 1, 1),
1349                            block_dim: (BLOCK_SIZE, 1, 1),
1350                            shared_mem_bytes: 0,
1351                        },
1352                        &mut params,
1353                    )
1354                    .map_err(|e| {
1355                        XlogError::Kernel(format!("{ctx}: groupby-agg launch failed: {e}"))
1356                    })?;
1357            }
1358        }
1359        rec.commit(runtime)
1360            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
1361
1362        // Per-X reduction via the relation metadata: one (unique X, group
1363        // start) pair per root; e_xy is lex-sorted by X so group rows are
1364        // contiguous.
1365        let meta = self.wcoj_build_metadata_u64_recorded(e_xy, 0, launch_stream)?;
1366        let key_count = meta.key_count;
1367        if key_count == 0 {
1368            return self.create_empty_buffer(out_schema);
1369        }
1370        let mut count_sums = self
1371            .memory()
1372            .alloc::<u8>(key_count as usize * std::mem::size_of::<u64>())?;
1373        self.device()
1374            .inner()
1375            .memset_zeros(&mut count_sums)
1376            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero group counts failed: {e}")))?;
1377        let mut group_agg = self
1378            .memory()
1379            .alloc::<u8>(key_count as usize * std::mem::size_of::<u64>())?;
1380        self.device()
1381            .inner()
1382            .memset_zeros(&mut group_agg)
1383            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero group aggregates failed: {e}")))?;
1384
1385        let mut rec_reduce = LaunchRecorder::new_strict(launch_stream);
1386        rec_reduce.read(&row_counts);
1387        rec_reduce.read(&row_agg);
1388        rec_reduce.read(&meta.prefix_sum);
1389        rec_reduce.write(&count_sums);
1390        rec_reduce.write(&group_agg);
1391        rec_reduce
1392            .preflight(runtime)
1393            .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce preflight failed: {e}")))?;
1394        if matches!(agg_op, AggOp::Min) {
1395            let fill = self
1396                .device()
1397                .inner()
1398                .get_func(ARITH_MODULE, arith_kernels::ARITH_FILL_CONST_U64)
1399                .ok_or_else(|| {
1400                    XlogError::Kernel("arith_fill_const_u64 kernel not found".to_string())
1401                })?;
1402            let group_agg_u64 = unsafe { reinterpret_u8_as_u64(&mut group_agg) };
1403            // SAFETY: arith_fill_const_u64(value, n, output)
1404            unsafe {
1405                fill.clone()
1406                    .launch_on_stream(
1407                        &cu_stream,
1408                        LaunchConfig::for_num_elems(key_count),
1409                        (u64::MAX, key_count, &mut *group_agg_u64),
1410                    )
1411                    .map_err(|e| {
1412                        XlogError::Kernel(format!("{ctx}: group min identity fill failed: {e}"))
1413                    })?;
1414            }
1415        }
1416        let reduce_grid = n_xy.div_ceil(BLOCK_SIZE);
1417        {
1418            let kernel = self
1419                .device()
1420                .inner()
1421                .get_func(
1422                    WCOJ_MODULE,
1423                    wcoj_kernels::WCOJ_GROUPBY_ROOT_SEGMENT_SUM_COUNTS_U32,
1424                )
1425                .ok_or_else(|| {
1426                    XlogError::Kernel(
1427                        "wcoj_groupby_root_segment_sum_counts_u32 kernel not found".to_string(),
1428                    )
1429                })?;
1430            let mut params: Vec<*mut c_void> = vec![
1431                (&row_counts).as_kernel_param(),
1432                n_xy.as_kernel_param(),
1433                (&meta.prefix_sum).as_kernel_param(),
1434                key_count.as_kernel_param(),
1435                (&count_sums).as_kernel_param(),
1436            ];
1437            unsafe {
1438                kernel
1439                    .clone()
1440                    .launch_on_stream(
1441                        &cu_stream,
1442                        LaunchConfig {
1443                            grid_dim: (reduce_grid, 1, 1),
1444                            block_dim: (BLOCK_SIZE, 1, 1),
1445                            shared_mem_bytes: 0,
1446                        },
1447                        &mut params,
1448                    )
1449                    .map_err(|e| {
1450                        XlogError::Kernel(format!("{ctx}: count reduce launch failed: {e}"))
1451                    })?;
1452            }
1453        }
1454        {
1455            let kernel = self
1456                .device()
1457                .inner()
1458                .get_func(WCOJ_MODULE, segment_kernel_name)
1459                .ok_or_else(|| {
1460                    XlogError::Kernel(format!("{segment_kernel_name} kernel not found"))
1461                })?;
1462            let mut params: Vec<*mut c_void> = vec![
1463                (&row_counts).as_kernel_param(),
1464                (&row_agg).as_kernel_param(),
1465                n_xy.as_kernel_param(),
1466                (&meta.prefix_sum).as_kernel_param(),
1467                key_count.as_kernel_param(),
1468                (&group_agg).as_kernel_param(),
1469            ];
1470            unsafe {
1471                kernel
1472                    .clone()
1473                    .launch_on_stream(
1474                        &cu_stream,
1475                        LaunchConfig {
1476                            grid_dim: (reduce_grid, 1, 1),
1477                            block_dim: (BLOCK_SIZE, 1, 1),
1478                            shared_mem_bytes: 0,
1479                        },
1480                        &mut params,
1481                    )
1482                    .map_err(|e| {
1483                        XlogError::Kernel(format!("{ctx}: agg reduce launch failed: {e}"))
1484                    })?;
1485            }
1486        }
1487        rec_reduce
1488            .commit(runtime)
1489            .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce commit failed: {e}")))?;
1490
1491        // (unique X, agg) staging plus a counts-only buffer whose mask
1492        // drops groups with no completion. Fresh destination blocks are
1493        // registered through the strict recorder BEFORE the enqueue (a raw
1494        // async copy into a freshly pool-allocated block without recording
1495        // is a visibility race).
1496        let x_copy = self
1497            .memory()
1498            .alloc::<u8>(key_count as usize * std::mem::size_of::<u64>())?;
1499        let d_num_rows_agg = self.memory().alloc::<u32>(1)?;
1500        let d_num_rows_counts = self.memory().alloc::<u32>(1)?;
1501        let mut rec_copy = LaunchRecorder::new_strict(launch_stream);
1502        rec_copy.read(&meta.unique_keys);
1503        rec_copy.write(&x_copy);
1504        rec_copy.write(&d_num_rows_agg);
1505        rec_copy.write(&d_num_rows_counts);
1506        rec_copy
1507            .preflight(runtime)
1508            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy preflight failed: {e}")))?;
1509        unsafe {
1510            let res = sys::cuMemcpyDtoDAsync_v2(
1511                *x_copy.device_ptr(),
1512                *meta.unique_keys.device_ptr(),
1513                key_count as usize * std::mem::size_of::<u64>(),
1514                cu_stream.cu_stream(),
1515            );
1516            if res != sys::cudaError_enum::CUDA_SUCCESS {
1517                return Err(XlogError::Kernel(format!(
1518                    "{ctx}: DtoD unique keys copy failed: {res:?}"
1519                )));
1520            }
1521        }
1522        self.htod_launch_metadata_async_copy_one(
1523            &key_count,
1524            &d_num_rows_agg,
1525            &cu_stream,
1526            &format!("{ctx}: d_num_rows_agg"),
1527        )?;
1528        self.htod_launch_metadata_async_copy_one(
1529            &key_count,
1530            &d_num_rows_counts,
1531            &cu_stream,
1532            &format!("{ctx}: d_num_rows_counts"),
1533        )?;
1534        rec_copy
1535            .commit(runtime)
1536            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy commit failed: {e}")))?;
1537        cu_stream
1538            .synchronize()
1539            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
1540
1541        let counts_schema = Schema::new(vec![("count".to_string(), ScalarType::U64)]);
1542        let counts_buf = CudaBuffer::from_columns_with_host_count(
1543            vec![count_sums.into()],
1544            u64::from(key_count),
1545            d_num_rows_counts,
1546            counts_schema,
1547            key_count,
1548        );
1549        let staging = CudaBuffer::from_columns_with_host_count(
1550            vec![x_copy.into(), group_agg.into()],
1551            u64::from(key_count),
1552            d_num_rows_agg,
1553            out_schema,
1554            key_count,
1555        );
1556        let mask = self.compare_const_mask_recorded::<u64>(
1557            &counts_buf,
1558            0,
1559            0u64,
1560            crate::CompareOp::Gt,
1561            launch_stream,
1562        )?;
1563        self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)
1564    }
1565
1566    pub fn wcoj_triangle_hg_count_phase_u32_recorded(
1567        &self,
1568        e_xy: &CudaBuffer,
1569        e_yz: &CudaBuffer,
1570        e_xz: &CudaBuffer,
1571        plan: &WcojTriangleHgWorkPlanU32,
1572        launch_stream: StreamId,
1573    ) -> Result<WcojTriangleHgCountPhaseU32> {
1574        let ctx = "wcoj_triangle_hg_count_phase_u32_recorded";
1575        validate_binary_u32(ctx, "e_xy", e_xy)?;
1576        validate_binary_u32(ctx, "e_yz", e_yz)?;
1577        validate_binary_u32(ctx, "e_xz", e_xz)?;
1578        let grid = plan.total_work.div_ceil(plan.block_work_unit);
1579        if grid > 1024 {
1580            return Err(XlogError::Kernel(format!(
1581                "{ctx}: spike phase path requires grid <= 1024, got {grid}"
1582            )));
1583        }
1584        let total_rows_device = self.memory().alloc::<u32>(1)?;
1585        if plan.total_work == 0 {
1586            return Ok(WcojTriangleHgCountPhaseU32 {
1587                total_rows_device,
1588                total_rows: 0,
1589            });
1590        }
1591
1592        let runtime = self.memory().runtime().ok_or_else(|| {
1593            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
1594        })?;
1595        let cu_stream = runtime
1596            .stream_pool()
1597            .resolve(launch_stream)
1598            .ok_or_else(|| {
1599                XlogError::Kernel(format!(
1600                    "{ctx}: launch_stream StreamId({}) does not resolve",
1601                    launch_stream.0
1602                ))
1603            })?;
1604
1605        let yz_col1 = metadata_column_u32(e_yz, 1)?;
1606        let xz_col1 = metadata_column_u32(e_xz, 1)?;
1607        let n_yz = self.metadata_logical_rows(e_yz)?;
1608        let n_xz = self.metadata_logical_rows(e_xz)?;
1609
1610        let mut rec_hg = LaunchRecorder::new_strict(launch_stream);
1611        rec_hg.read(e_xy.num_rows_device());
1612        rec_hg.read(e_yz.num_rows_device());
1613        rec_hg.read(e_xz.num_rows_device());
1614        rec_hg.read_column(e_yz.column(1).expect("yz.col1"));
1615        rec_hg.read_column(e_xz.column(1).expect("xz.col1"));
1616        rec_hg.read(&plan.xy_work_prefix);
1617        rec_hg.read(&plan.xy_yz_start);
1618        rec_hg.read(&plan.xy_yz_end);
1619        rec_hg.read(&plan.xy_xz_start);
1620        rec_hg.read(&plan.xy_xz_end);
1621        rec_hg.read_write(&plan.block_counts);
1622        rec_hg.read_write(&plan.block_offsets);
1623        rec_hg.write(&total_rows_device);
1624        rec_hg
1625            .preflight(runtime)
1626            .map_err(|e| XlogError::Kernel(format!("{ctx}: count preflight failed: {e}")))?;
1627        {
1628            let kernel = self
1629                .device()
1630                .inner()
1631                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_TRIANGLE_COUNT_HG_U32)
1632                .ok_or_else(|| {
1633                    XlogError::Kernel("wcoj_triangle_count_hg_u32 kernel not found".to_string())
1634                })?;
1635            let mut params: Vec<*mut c_void> = vec![
1636                yz_col1.as_kernel_param(),
1637                n_yz.as_kernel_param(),
1638                xz_col1.as_kernel_param(),
1639                n_xz.as_kernel_param(),
1640                (&plan.xy_work_prefix).as_kernel_param(),
1641                (&plan.xy_yz_start).as_kernel_param(),
1642                (&plan.xy_yz_end).as_kernel_param(),
1643                (&plan.xy_xz_start).as_kernel_param(),
1644                (&plan.xy_xz_end).as_kernel_param(),
1645                plan.row_count.as_kernel_param(),
1646                plan.total_work.as_kernel_param(),
1647                plan.block_work_unit.as_kernel_param(),
1648                (&plan.block_counts).as_kernel_param(),
1649            ];
1650            unsafe {
1651                kernel
1652                    .clone()
1653                    .launch_on_stream(
1654                        &cu_stream,
1655                        LaunchConfig {
1656                            grid_dim: (grid, 1, 1),
1657                            block_dim: (BLOCK_SIZE, 1, 1),
1658                            shared_mem_bytes: 0,
1659                        },
1660                        &mut params,
1661                    )
1662                    .map_err(|e| XlogError::Kernel(format!("{ctx}: count launch failed: {e}")))?;
1663            }
1664        }
1665        {
1666            let kernel = self
1667                .device()
1668                .inner()
1669                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_SCAN_HG_BLOCK_COUNTS_U32)
1670                .ok_or_else(|| {
1671                    XlogError::Kernel("wcoj_scan_hg_block_counts_u32 kernel not found".to_string())
1672                })?;
1673            let mut params: Vec<*mut c_void> = vec![
1674                (&plan.block_counts).as_kernel_param(),
1675                grid.as_kernel_param(),
1676                (&plan.block_offsets).as_kernel_param(),
1677                (&total_rows_device).as_kernel_param(),
1678            ];
1679            unsafe {
1680                kernel
1681                    .clone()
1682                    .launch_on_stream(
1683                        &cu_stream,
1684                        LaunchConfig {
1685                            grid_dim: (1, 1, 1),
1686                            block_dim: (1024, 1, 1),
1687                            shared_mem_bytes: 0,
1688                        },
1689                        &mut params,
1690                    )
1691                    .map_err(|e| XlogError::Kernel(format!("{ctx}: scan launch failed: {e}")))?;
1692            }
1693        }
1694        rec_hg
1695            .commit(runtime)
1696            .map_err(|e| XlogError::Kernel(format!("{ctx}: count commit failed: {e}")))?;
1697        cu_stream
1698            .synchronize()
1699            .map_err(|e| XlogError::Kernel(format!("{ctx}: count stream sync failed: {e}")))?;
1700        let total_rows = self
1701            .dtoh_scalar_untracked::<u32>(&total_rows_device, 0)
1702            .map_err(|e| XlogError::Kernel(format!("{ctx}: read total rows failed: {e}")))?;
1703        Ok(WcojTriangleHgCountPhaseU32 {
1704            total_rows_device,
1705            total_rows,
1706        })
1707    }
1708
1709    pub fn wcoj_triangle_hg_materialize_phase_u32_recorded(
1710        &self,
1711        e_xy: &CudaBuffer,
1712        e_yz: &CudaBuffer,
1713        e_xz: &CudaBuffer,
1714        plan: &WcojTriangleHgWorkPlanU32,
1715        count: WcojTriangleHgCountPhaseU32,
1716        launch_stream: StreamId,
1717    ) -> Result<CudaBuffer> {
1718        let ctx = "wcoj_triangle_hg_materialize_phase_u32_recorded";
1719        validate_binary_u32(ctx, "e_xy", e_xy)?;
1720        validate_binary_u32(ctx, "e_yz", e_yz)?;
1721        validate_binary_u32(ctx, "e_xz", e_xz)?;
1722        let out_schema = Schema::new(vec![
1723            (
1724                "x".to_string(),
1725                e_xy.schema().column_type(0).expect("xy.col0 type"),
1726            ),
1727            (
1728                "y".to_string(),
1729                e_xy.schema().column_type(1).expect("xy.col1 type"),
1730            ),
1731            (
1732                "z".to_string(),
1733                e_yz.schema().column_type(1).expect("yz.col1 type"),
1734            ),
1735        ]);
1736        if count.total_rows == 0 {
1737            return self.create_empty_buffer(out_schema);
1738        }
1739        let grid = plan.total_work.div_ceil(plan.block_work_unit);
1740        if grid > 1024 {
1741            return Err(XlogError::Kernel(format!(
1742                "{ctx}: spike phase path requires grid <= 1024, got {grid}"
1743            )));
1744        }
1745        let runtime = self.memory().runtime().ok_or_else(|| {
1746            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
1747        })?;
1748        let cu_stream = runtime
1749            .stream_pool()
1750            .resolve(launch_stream)
1751            .ok_or_else(|| {
1752                XlogError::Kernel(format!(
1753                    "{ctx}: launch_stream StreamId({}) does not resolve",
1754                    launch_stream.0
1755                ))
1756            })?;
1757        let xy_col0 = metadata_column_u32(e_xy, 0)?;
1758        let xy_col1 = metadata_column_u32(e_xy, 1)?;
1759        let yz_col1 = metadata_column_u32(e_yz, 1)?;
1760        let xz_col1 = metadata_column_u32(e_xz, 1)?;
1761        let n_yz = self.metadata_logical_rows(e_yz)?;
1762        let n_xz = self.metadata_logical_rows(e_xz)?;
1763        let bytes_per_col = (count.total_rows as usize)
1764            .checked_mul(std::mem::size_of::<u32>())
1765            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: output byte size overflow")))?;
1766        let mut out_x = self.memory().alloc::<u8>(bytes_per_col)?;
1767        let mut out_y = self.memory().alloc::<u8>(bytes_per_col)?;
1768        let mut out_z = self.memory().alloc::<u8>(bytes_per_col)?;
1769        let mut rec_mat = LaunchRecorder::new_strict(launch_stream);
1770        rec_mat.read(e_xy.num_rows_device());
1771        rec_mat.read(e_yz.num_rows_device());
1772        rec_mat.read(e_xz.num_rows_device());
1773        rec_mat.read_column(e_xy.column(0).expect("xy.col0"));
1774        rec_mat.read_column(e_xy.column(1).expect("xy.col1"));
1775        rec_mat.read_column(e_yz.column(1).expect("yz.col1"));
1776        rec_mat.read_column(e_xz.column(1).expect("xz.col1"));
1777        rec_mat.read(&plan.xy_work_prefix);
1778        rec_mat.read(&plan.xy_yz_start);
1779        rec_mat.read(&plan.xy_yz_end);
1780        rec_mat.read(&plan.xy_xz_start);
1781        rec_mat.read(&plan.xy_xz_end);
1782        rec_mat.read(&plan.block_offsets);
1783        rec_mat.write(&out_x);
1784        rec_mat.write(&out_y);
1785        rec_mat.write(&out_z);
1786        rec_mat
1787            .preflight(runtime)
1788            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize preflight failed: {e}")))?;
1789        {
1790            let kernel = self
1791                .device()
1792                .inner()
1793                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_TRIANGLE_MATERIALIZE_HG_U32)
1794                .ok_or_else(|| {
1795                    XlogError::Kernel(
1796                        "wcoj_triangle_materialize_hg_u32 kernel not found".to_string(),
1797                    )
1798                })?;
1799            let out_x_u32 = unsafe { reinterpret_u8_as_u32(&mut out_x) };
1800            let out_y_u32 = unsafe { reinterpret_u8_as_u32(&mut out_y) };
1801            let out_z_u32 = unsafe { reinterpret_u8_as_u32(&mut out_z) };
1802            let mut params: Vec<*mut c_void> = vec![
1803                xy_col0.as_kernel_param(),
1804                xy_col1.as_kernel_param(),
1805                yz_col1.as_kernel_param(),
1806                n_yz.as_kernel_param(),
1807                xz_col1.as_kernel_param(),
1808                n_xz.as_kernel_param(),
1809                (&plan.xy_work_prefix).as_kernel_param(),
1810                (&plan.xy_yz_start).as_kernel_param(),
1811                (&plan.xy_yz_end).as_kernel_param(),
1812                (&plan.xy_xz_start).as_kernel_param(),
1813                (&plan.xy_xz_end).as_kernel_param(),
1814                plan.row_count.as_kernel_param(),
1815                plan.total_work.as_kernel_param(),
1816                plan.block_work_unit.as_kernel_param(),
1817                (&plan.block_offsets).as_kernel_param(),
1818                count.total_rows.as_kernel_param(),
1819                out_x_u32.as_kernel_param(),
1820                out_y_u32.as_kernel_param(),
1821                out_z_u32.as_kernel_param(),
1822            ];
1823            unsafe {
1824                kernel
1825                    .clone()
1826                    .launch_on_stream(
1827                        &cu_stream,
1828                        LaunchConfig {
1829                            grid_dim: (grid, 1, 1),
1830                            block_dim: (BLOCK_SIZE, 1, 1),
1831                            shared_mem_bytes: 0,
1832                        },
1833                        &mut params,
1834                    )
1835                    .map_err(|e| {
1836                        XlogError::Kernel(format!("{ctx}: materialize launch failed: {e}"))
1837                    })?;
1838            }
1839        }
1840        rec_mat
1841            .commit(runtime)
1842            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize commit failed: {e}")))?;
1843        cu_stream.synchronize().map_err(|e| {
1844            XlogError::Kernel(format!("{ctx}: materialize stream sync failed: {e}"))
1845        })?;
1846        Ok(CudaBuffer::from_columns_with_host_count(
1847            vec![out_x.into(), out_y.into(), out_z.into()],
1848            count.total_rows as u64,
1849            count.total_rows_device,
1850            out_schema,
1851            count.total_rows,
1852        ))
1853    }
1854
1855    pub fn wcoj_triangle_hg_u32_with_plan_recorded(
1856        &self,
1857        e_xy: &CudaBuffer,
1858        e_yz: &CudaBuffer,
1859        e_xz: &CudaBuffer,
1860        plan: &WcojTriangleHgWorkPlanU32,
1861        launch_stream: StreamId,
1862    ) -> Result<CudaBuffer> {
1863        let ctx = "wcoj_triangle_hg_u32_with_plan_recorded";
1864        validate_binary_u32(ctx, "e_xy", e_xy)?;
1865        validate_binary_u32(ctx, "e_yz", e_yz)?;
1866        validate_binary_u32(ctx, "e_xz", e_xz)?;
1867        let out_schema = Schema::new(vec![
1868            (
1869                "x".to_string(),
1870                e_xy.schema().column_type(0).expect("xy.col0 type"),
1871            ),
1872            (
1873                "y".to_string(),
1874                e_xy.schema().column_type(1).expect("xy.col1 type"),
1875            ),
1876            (
1877                "z".to_string(),
1878                e_yz.schema().column_type(1).expect("yz.col1 type"),
1879            ),
1880        ]);
1881        if plan.total_work == 0 {
1882            return self.create_empty_buffer(out_schema);
1883        }
1884
1885        let grid = plan.total_work.div_ceil(plan.block_work_unit);
1886        let bytes_count = (grid as usize)
1887            .checked_mul(std::mem::size_of::<u32>())
1888            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: count byte size overflow")))?;
1889        let mut local_counts = None;
1890        let mut local_offsets = None;
1891        if grid > 1024 {
1892            local_counts = Some(self.memory().alloc::<u32>(grid as usize)?);
1893            local_offsets = Some(self.memory().alloc::<u32>(grid as usize)?);
1894        }
1895        let total_rows_device = self.memory().alloc::<u32>(1)?;
1896
1897        let runtime = self.memory().runtime().ok_or_else(|| {
1898            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
1899        })?;
1900        let cu_stream = runtime
1901            .stream_pool()
1902            .resolve(launch_stream)
1903            .ok_or_else(|| {
1904                XlogError::Kernel(format!(
1905                    "{ctx}: launch_stream StreamId({}) does not resolve",
1906                    launch_stream.0
1907                ))
1908            })?;
1909
1910        let xy_col0 = metadata_column_u32(e_xy, 0)?;
1911        let xy_col1 = metadata_column_u32(e_xy, 1)?;
1912        let yz_col1 = metadata_column_u32(e_yz, 1)?;
1913        let xz_col1 = metadata_column_u32(e_xz, 1)?;
1914        let n_yz = self.metadata_logical_rows(e_yz)?;
1915        let n_xz = self.metadata_logical_rows(e_xz)?;
1916
1917        let count_u32 = if grid <= 1024 {
1918            &plan.block_counts
1919        } else {
1920            local_counts
1921                .as_ref()
1922                .expect("local HG counts allocated when grid exceeds single-block scan")
1923        };
1924        let mut rec_hg = LaunchRecorder::new_strict(launch_stream);
1925        rec_hg.read(e_xy.num_rows_device());
1926        rec_hg.read(e_yz.num_rows_device());
1927        rec_hg.read(e_xz.num_rows_device());
1928        rec_hg.read_column(e_xy.column(0).expect("xy.col0"));
1929        rec_hg.read_column(e_xy.column(1).expect("xy.col1"));
1930        rec_hg.read_column(e_yz.column(1).expect("yz.col1"));
1931        rec_hg.read_column(e_xz.column(1).expect("xz.col1"));
1932        rec_hg.read(&plan.xy_work_prefix);
1933        rec_hg.read(&plan.xy_yz_start);
1934        rec_hg.read(&plan.xy_yz_end);
1935        rec_hg.read(&plan.xy_xz_start);
1936        rec_hg.read(&plan.xy_xz_end);
1937        rec_hg.read_write(count_u32);
1938        if grid <= 1024 {
1939            rec_hg.read_write(&plan.block_offsets);
1940        } else {
1941            rec_hg.read_write(
1942                local_offsets
1943                    .as_ref()
1944                    .expect("local HG offsets allocated when grid exceeds single-block scan"),
1945            );
1946        }
1947        rec_hg.write(&total_rows_device);
1948        rec_hg.read_write(&plan.scratch_x);
1949        rec_hg.read_write(&plan.scratch_y);
1950        rec_hg.read_write(&plan.scratch_z);
1951        rec_hg
1952            .preflight(runtime)
1953            .map_err(|e| XlogError::Kernel(format!("{ctx}: HG preflight failed: {e}")))?;
1954        {
1955            let kernel = self
1956                .device()
1957                .inner()
1958                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_TRIANGLE_COUNT_HG_CACHED_U32)
1959                .ok_or_else(|| {
1960                    XlogError::Kernel(
1961                        "wcoj_triangle_count_hg_cached_u32 kernel not found".to_string(),
1962                    )
1963                })?;
1964            let mut params: Vec<*mut c_void> = vec![
1965                xy_col0.as_kernel_param(),
1966                xy_col1.as_kernel_param(),
1967                yz_col1.as_kernel_param(),
1968                n_yz.as_kernel_param(),
1969                xz_col1.as_kernel_param(),
1970                n_xz.as_kernel_param(),
1971                (&plan.xy_work_prefix).as_kernel_param(),
1972                (&plan.xy_yz_start).as_kernel_param(),
1973                (&plan.xy_yz_end).as_kernel_param(),
1974                (&plan.xy_xz_start).as_kernel_param(),
1975                (&plan.xy_xz_end).as_kernel_param(),
1976                plan.row_count.as_kernel_param(),
1977                plan.total_work.as_kernel_param(),
1978                plan.block_work_unit.as_kernel_param(),
1979                count_u32.as_kernel_param(),
1980                (&plan.scratch_x).as_kernel_param(),
1981                (&plan.scratch_y).as_kernel_param(),
1982                (&plan.scratch_z).as_kernel_param(),
1983            ];
1984            unsafe {
1985                kernel
1986                    .clone()
1987                    .launch_on_stream(
1988                        &cu_stream,
1989                        LaunchConfig {
1990                            grid_dim: (grid, 1, 1),
1991                            block_dim: (HG_COUNT_BLOCK_SIZE, 1, 1),
1992                            shared_mem_bytes: 0,
1993                        },
1994                        &mut params,
1995                    )
1996                    .map_err(|e| {
1997                        XlogError::Kernel(format!("{ctx}: cached count launch failed: {e}"))
1998                    })?;
1999            }
2000        }
2001        if grid <= 1024 {
2002            let kernel = self
2003                .device()
2004                .inner()
2005                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_SCAN_HG_BLOCK_COUNTS_U32)
2006                .ok_or_else(|| {
2007                    XlogError::Kernel("wcoj_scan_hg_block_counts_u32 kernel not found".to_string())
2008                })?;
2009            let mut params: Vec<*mut c_void> = vec![
2010                count_u32.as_kernel_param(),
2011                grid.as_kernel_param(),
2012                (&plan.block_offsets).as_kernel_param(),
2013                (&total_rows_device).as_kernel_param(),
2014            ];
2015            unsafe {
2016                kernel
2017                    .clone()
2018                    .launch_on_stream(
2019                        &cu_stream,
2020                        LaunchConfig {
2021                            grid_dim: (1, 1, 1),
2022                            block_dim: (1024, 1, 1),
2023                            shared_mem_bytes: 0,
2024                        },
2025                        &mut params,
2026                    )
2027                    .map_err(|e| {
2028                        XlogError::Kernel(format!("{ctx}: HG block-count scan failed: {e}"))
2029                    })?;
2030            }
2031        } else {
2032            let offsets_mut = local_offsets
2033                .as_mut()
2034                .expect("local HG offsets allocated when grid exceeds single-block scan");
2035            unsafe {
2036                let res = sys::cuMemcpyDtoDAsync_v2(
2037                    *offsets_mut.device_ptr(),
2038                    *count_u32.device_ptr(),
2039                    bytes_count,
2040                    cu_stream.cu_stream(),
2041                );
2042                if res != sys::cudaError_enum::CUDA_SUCCESS {
2043                    return Err(XlogError::Kernel(format!(
2044                        "{ctx}: DtoD count to offsets failed: {res:?}"
2045                    )));
2046                }
2047            }
2048            self.multiblock_scan_u32_inplace_on_stream(
2049                offsets_mut,
2050                grid,
2051                &cu_stream,
2052                launch_stream,
2053                runtime,
2054            )?;
2055            let total_kernel = self
2056                .device()
2057                .inner()
2058                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_COMPUTE_TOTAL)
2059                .ok_or_else(|| {
2060                    XlogError::Kernel("wcoj_compute_total kernel not found".to_string())
2061                })?;
2062            let mut params: Vec<*mut c_void> = vec![
2063                count_u32.as_kernel_param(),
2064                (&*offsets_mut).as_kernel_param(),
2065                grid.as_kernel_param(),
2066                (&total_rows_device).as_kernel_param(),
2067            ];
2068            unsafe {
2069                total_kernel
2070                    .clone()
2071                    .launch_on_stream(
2072                        &cu_stream,
2073                        LaunchConfig {
2074                            grid_dim: (1, 1, 1),
2075                            block_dim: (1, 1, 1),
2076                            shared_mem_bytes: 0,
2077                        },
2078                        &mut params,
2079                    )
2080                    .map_err(|e| {
2081                        XlogError::Kernel(format!("{ctx}: HG total reducer failed: {e}"))
2082                    })?;
2083            }
2084        }
2085        rec_hg
2086            .commit(runtime)
2087            .map_err(|e| XlogError::Kernel(format!("{ctx}: HG count commit failed: {e}")))?;
2088        cu_stream
2089            .synchronize()
2090            .map_err(|e| XlogError::Kernel(format!("{ctx}: count stream sync failed: {e}")))?;
2091        let total_rows = self
2092            .dtoh_scalar_untracked::<u32>(&total_rows_device, 0)
2093            .map_err(|e| XlogError::Kernel(format!("{ctx}: read total rows failed: {e}")))?;
2094        if total_rows == 0 {
2095            return self.create_empty_buffer(out_schema);
2096        }
2097
2098        let bytes_per_col = (total_rows as usize)
2099            .checked_mul(std::mem::size_of::<u32>())
2100            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: output byte size overflow")))?;
2101        let mut out_x = self.memory().alloc::<u8>(bytes_per_col)?;
2102        let mut out_y = self.memory().alloc::<u8>(bytes_per_col)?;
2103        let mut out_z = self.memory().alloc::<u8>(bytes_per_col)?;
2104        let materialize_offsets = if grid <= 1024 {
2105            &plan.block_offsets
2106        } else {
2107            local_offsets
2108                .as_ref()
2109                .expect("local HG offsets allocated when grid exceeds single-block scan")
2110        };
2111        let mut rec_mat = LaunchRecorder::new_strict(launch_stream);
2112        rec_mat.read(count_u32);
2113        rec_mat.read(materialize_offsets);
2114        rec_mat.read(&plan.scratch_x);
2115        rec_mat.read(&plan.scratch_y);
2116        rec_mat.read(&plan.scratch_z);
2117        rec_mat.write(&out_x);
2118        rec_mat.write(&out_y);
2119        rec_mat.write(&out_z);
2120        rec_mat
2121            .preflight(runtime)
2122            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize preflight failed: {e}")))?;
2123        {
2124            let kernel = self
2125                .device()
2126                .inner()
2127                .get_func(
2128                    WCOJ_MODULE,
2129                    wcoj_kernels::WCOJ_TRIANGLE_MATERIALIZE_HG_CACHED_U32,
2130                )
2131                .ok_or_else(|| {
2132                    XlogError::Kernel(
2133                        "wcoj_triangle_materialize_hg_cached_u32 kernel not found".to_string(),
2134                    )
2135                })?;
2136            let out_x_u32 = unsafe { reinterpret_u8_as_u32(&mut out_x) };
2137            let out_y_u32 = unsafe { reinterpret_u8_as_u32(&mut out_y) };
2138            let out_z_u32 = unsafe { reinterpret_u8_as_u32(&mut out_z) };
2139            let mut params: Vec<*mut c_void> = vec![
2140                count_u32.as_kernel_param(),
2141                materialize_offsets.as_kernel_param(),
2142                plan.block_work_unit.as_kernel_param(),
2143                total_rows.as_kernel_param(),
2144                (&plan.scratch_x).as_kernel_param(),
2145                (&plan.scratch_y).as_kernel_param(),
2146                (&plan.scratch_z).as_kernel_param(),
2147                out_x_u32.as_kernel_param(),
2148                out_y_u32.as_kernel_param(),
2149                out_z_u32.as_kernel_param(),
2150            ];
2151            unsafe {
2152                kernel
2153                    .clone()
2154                    .launch_on_stream(
2155                        &cu_stream,
2156                        LaunchConfig {
2157                            grid_dim: (grid, 1, 1),
2158                            block_dim: (BLOCK_SIZE, 1, 1),
2159                            shared_mem_bytes: 0,
2160                        },
2161                        &mut params,
2162                    )
2163                    .map_err(|e| {
2164                        XlogError::Kernel(format!("{ctx}: materialize launch failed: {e}"))
2165                    })?;
2166            }
2167        }
2168        rec_mat
2169            .commit(runtime)
2170            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize commit failed: {e}")))?;
2171        cu_stream.synchronize().map_err(|e| {
2172            XlogError::Kernel(format!("{ctx}: materialize stream sync failed: {e}"))
2173        })?;
2174
2175        Ok(CudaBuffer::from_columns_with_host_count(
2176            vec![out_x.into(), out_y.into(), out_z.into()],
2177            total_rows as u64,
2178            total_rows_device,
2179            out_schema,
2180            total_rows,
2181        ))
2182    }
2183
2184    pub fn wcoj_triangle_hg_work_plan_u64_recorded(
2185        &self,
2186        e_xy: &CudaBuffer,
2187        e_yz: &CudaBuffer,
2188        e_xz: &CudaBuffer,
2189        block_work_unit: u32,
2190        launch_stream: StreamId,
2191    ) -> Result<WcojTriangleHgWorkPlanU64> {
2192        let ctx = "wcoj_triangle_hg_work_plan_u64_recorded";
2193        if block_work_unit == 0 {
2194            return Err(XlogError::Kernel(format!(
2195                "{ctx}: block_work_unit must be nonzero"
2196            )));
2197        }
2198        validate_binary_u64(ctx, "e_xy", e_xy)?;
2199        validate_binary_u64(ctx, "e_yz", e_yz)?;
2200        validate_binary_u64(ctx, "e_xz", e_xz)?;
2201
2202        let n_xy = self.metadata_logical_rows(e_xy)?;
2203        let n_yz = self.metadata_logical_rows(e_yz)?;
2204        let n_xz = self.metadata_logical_rows(e_xz)?;
2205        let prefix_len = n_xy
2206            .checked_add(1)
2207            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: prefix length overflow")))?;
2208        let mut xy_work_prefix = self.memory().alloc::<u32>(prefix_len as usize)?;
2209        let mut xy_yz_start = self.memory().alloc::<u32>(n_xy as usize)?;
2210        let mut xy_yz_end = self.memory().alloc::<u32>(n_xy as usize)?;
2211        let mut xy_xz_start = self.memory().alloc::<u32>(n_xy as usize)?;
2212        let mut xy_xz_end = self.memory().alloc::<u32>(n_xy as usize)?;
2213
2214        if n_xy == 0 {
2215            let block_counts = self.memory().alloc::<u32>(1)?;
2216            let block_offsets = self.memory().alloc::<u32>(1)?;
2217            return Ok(WcojTriangleHgWorkPlanU64 {
2218                xy_work_prefix,
2219                xy_yz_start,
2220                xy_yz_end,
2221                xy_xz_start,
2222                xy_xz_end,
2223                block_counts,
2224                block_offsets,
2225                total_work: 0,
2226                block_work_unit,
2227                row_count: n_xy,
2228            });
2229        }
2230
2231        let runtime = self.memory().runtime().ok_or_else(|| {
2232            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
2233        })?;
2234        let cu_stream = runtime
2235            .stream_pool()
2236            .resolve(launch_stream)
2237            .ok_or_else(|| {
2238                XlogError::Kernel(format!(
2239                    "{ctx}: launch_stream StreamId({}) does not resolve",
2240                    launch_stream.0
2241                ))
2242            })?;
2243
2244        let xy_col0 = metadata_column_u64(e_xy, 0)?;
2245        let xy_col1 = metadata_column_u64(e_xy, 1)?;
2246        let yz_col0 = metadata_column_u64(e_yz, 0)?;
2247        let xz_col0 = metadata_column_u64(e_xz, 0)?;
2248
2249        let mut rec = LaunchRecorder::new_strict(launch_stream);
2250        rec.read(e_xy.num_rows_device());
2251        rec.read(e_yz.num_rows_device());
2252        rec.read(e_xz.num_rows_device());
2253        rec.read_column(e_xy.column(0).expect("xy.col0"));
2254        rec.read_column(e_xy.column(1).expect("xy.col1"));
2255        rec.read_column(e_yz.column(0).expect("yz.col0"));
2256        rec.read_column(e_xz.column(0).expect("xz.col0"));
2257        rec.write(&xy_work_prefix);
2258        rec.write(&xy_yz_start);
2259        rec.write(&xy_yz_end);
2260        rec.write(&xy_xz_start);
2261        rec.write(&xy_xz_end);
2262        rec.preflight(runtime)
2263            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
2264
2265        let kernel = self
2266            .device()
2267            .inner()
2268            .get_func(
2269                WCOJ_MODULE,
2270                wcoj_kernels::WCOJ_TRIANGLE_BUILD_HG_WORK_PLAN_U64,
2271            )
2272            .ok_or_else(|| {
2273                XlogError::Kernel(
2274                    "wcoj_triangle_build_hg_work_plan_u64 kernel not found".to_string(),
2275                )
2276            })?;
2277        let grid = n_xy.div_ceil(BLOCK_SIZE);
2278        unsafe {
2279            kernel
2280                .clone()
2281                .launch_on_stream(
2282                    &cu_stream,
2283                    LaunchConfig {
2284                        grid_dim: (grid, 1, 1),
2285                        block_dim: (BLOCK_SIZE, 1, 1),
2286                        shared_mem_bytes: 0,
2287                    },
2288                    (
2289                        xy_col0,
2290                        xy_col1,
2291                        n_xy,
2292                        yz_col0,
2293                        n_yz,
2294                        xz_col0,
2295                        n_xz,
2296                        &mut xy_work_prefix,
2297                        &mut xy_yz_start,
2298                        &mut xy_yz_end,
2299                        &mut xy_xz_start,
2300                        &mut xy_xz_end,
2301                    ),
2302                )
2303                .map_err(|e| {
2304                    XlogError::Kernel(format!(
2305                        "wcoj_triangle_build_hg_work_plan_u64 launch failed: {e}"
2306                    ))
2307                })?;
2308        }
2309        self.multiblock_scan_u32_inplace_on_stream(
2310            &mut xy_work_prefix,
2311            prefix_len,
2312            &cu_stream,
2313            launch_stream,
2314            runtime,
2315        )?;
2316        rec.commit(runtime)
2317            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
2318        cu_stream
2319            .synchronize()
2320            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
2321        let total_work = self.dtoh_scalar_untracked::<u32>(&xy_work_prefix, n_xy as usize)?;
2322        let grid = if total_work == 0 {
2323            1
2324        } else {
2325            total_work.div_ceil(block_work_unit)
2326        };
2327        let block_counts = self.memory().alloc::<u32>(grid as usize)?;
2328        let block_offsets = self.memory().alloc::<u32>(grid as usize)?;
2329
2330        Ok(WcojTriangleHgWorkPlanU64 {
2331            xy_work_prefix,
2332            xy_yz_start,
2333            xy_yz_end,
2334            xy_xz_start,
2335            xy_xz_end,
2336            block_counts,
2337            block_offsets,
2338            total_work,
2339            block_work_unit,
2340            row_count: n_xy,
2341        })
2342    }
2343
2344    pub fn wcoj_triangle_hg_u64_recorded(
2345        &self,
2346        e_xy: &CudaBuffer,
2347        e_yz: &CudaBuffer,
2348        e_xz: &CudaBuffer,
2349        block_work_unit: u32,
2350        launch_stream: StreamId,
2351    ) -> Result<CudaBuffer> {
2352        let ctx = "wcoj_triangle_hg_u64_recorded";
2353        validate_binary_u64(ctx, "e_xy", e_xy)?;
2354        validate_binary_u64(ctx, "e_yz", e_yz)?;
2355        validate_binary_u64(ctx, "e_xz", e_xz)?;
2356        let plan = self.wcoj_triangle_hg_work_plan_u64_recorded(
2357            e_xy,
2358            e_yz,
2359            e_xz,
2360            block_work_unit,
2361            launch_stream,
2362        )?;
2363        let out_schema = Schema::new(vec![
2364            ("col0".to_string(), ScalarType::U64),
2365            ("col1".to_string(), ScalarType::U64),
2366            ("col2".to_string(), ScalarType::U64),
2367        ]);
2368        if plan.total_work == 0 {
2369            return self.create_empty_buffer(out_schema);
2370        }
2371
2372        let grid = plan.total_work.div_ceil(plan.block_work_unit);
2373        let bytes_count = (grid as usize)
2374            .checked_mul(std::mem::size_of::<u32>())
2375            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: count byte size overflow")))?;
2376        let mut local_counts = None;
2377        let mut local_offsets = None;
2378        if grid > 1024 {
2379            local_counts = Some(self.memory().alloc::<u32>(grid as usize)?);
2380            local_offsets = Some(self.memory().alloc::<u32>(grid as usize)?);
2381        }
2382        let total_rows_device = self.memory().alloc::<u32>(1)?;
2383
2384        let runtime = self.memory().runtime().ok_or_else(|| {
2385            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
2386        })?;
2387        let cu_stream = runtime
2388            .stream_pool()
2389            .resolve(launch_stream)
2390            .ok_or_else(|| {
2391                XlogError::Kernel(format!(
2392                    "{ctx}: launch_stream StreamId({}) does not resolve",
2393                    launch_stream.0
2394                ))
2395            })?;
2396
2397        let xy_col0 = metadata_column_u64(e_xy, 0)?;
2398        let xy_col1 = metadata_column_u64(e_xy, 1)?;
2399        let yz_col1 = metadata_column_u64(e_yz, 1)?;
2400        let xz_col1 = metadata_column_u64(e_xz, 1)?;
2401        let n_yz = self.metadata_logical_rows(e_yz)?;
2402        let n_xz = self.metadata_logical_rows(e_xz)?;
2403
2404        let count_u32 = if grid <= 1024 {
2405            &plan.block_counts
2406        } else {
2407            local_counts
2408                .as_ref()
2409                .expect("local HG counts allocated when grid exceeds single-block scan")
2410        };
2411        let mut rec_hg = LaunchRecorder::new_strict(launch_stream);
2412        rec_hg.read(e_xy.num_rows_device());
2413        rec_hg.read(e_yz.num_rows_device());
2414        rec_hg.read(e_xz.num_rows_device());
2415        rec_hg.read_column(e_yz.column(1).expect("yz.col1"));
2416        rec_hg.read_column(e_xz.column(1).expect("xz.col1"));
2417        rec_hg.read(&plan.xy_work_prefix);
2418        rec_hg.read(&plan.xy_yz_start);
2419        rec_hg.read(&plan.xy_yz_end);
2420        rec_hg.read(&plan.xy_xz_start);
2421        rec_hg.read(&plan.xy_xz_end);
2422        rec_hg.read_write(count_u32);
2423        if grid <= 1024 {
2424            rec_hg.read_write(&plan.block_offsets);
2425        } else {
2426            rec_hg.read_write(
2427                local_offsets
2428                    .as_ref()
2429                    .expect("local HG offsets allocated when grid exceeds single-block scan"),
2430            );
2431        }
2432        rec_hg.write(&total_rows_device);
2433        rec_hg
2434            .preflight(runtime)
2435            .map_err(|e| XlogError::Kernel(format!("{ctx}: HG preflight failed: {e}")))?;
2436        {
2437            let kernel = self
2438                .device()
2439                .inner()
2440                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_TRIANGLE_COUNT_HG_U64)
2441                .ok_or_else(|| {
2442                    XlogError::Kernel("wcoj_triangle_count_hg_u64 kernel not found".to_string())
2443                })?;
2444            let mut params: Vec<*mut c_void> = vec![
2445                yz_col1.as_kernel_param(),
2446                n_yz.as_kernel_param(),
2447                xz_col1.as_kernel_param(),
2448                n_xz.as_kernel_param(),
2449                (&plan.xy_work_prefix).as_kernel_param(),
2450                (&plan.xy_yz_start).as_kernel_param(),
2451                (&plan.xy_yz_end).as_kernel_param(),
2452                (&plan.xy_xz_start).as_kernel_param(),
2453                (&plan.xy_xz_end).as_kernel_param(),
2454                plan.row_count.as_kernel_param(),
2455                plan.total_work.as_kernel_param(),
2456                plan.block_work_unit.as_kernel_param(),
2457                count_u32.as_kernel_param(),
2458            ];
2459            unsafe {
2460                kernel
2461                    .clone()
2462                    .launch_on_stream(
2463                        &cu_stream,
2464                        LaunchConfig {
2465                            grid_dim: (grid, 1, 1),
2466                            block_dim: (BLOCK_SIZE, 1, 1),
2467                            shared_mem_bytes: 0,
2468                        },
2469                        &mut params,
2470                    )
2471                    .map_err(|e| XlogError::Kernel(format!("{ctx}: count launch failed: {e}")))?;
2472            }
2473        }
2474        if grid <= 1024 {
2475            let kernel = self
2476                .device()
2477                .inner()
2478                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_SCAN_HG_BLOCK_COUNTS_U32)
2479                .ok_or_else(|| {
2480                    XlogError::Kernel("wcoj_scan_hg_block_counts_u32 kernel not found".to_string())
2481                })?;
2482            let mut params: Vec<*mut c_void> = vec![
2483                count_u32.as_kernel_param(),
2484                grid.as_kernel_param(),
2485                (&plan.block_offsets).as_kernel_param(),
2486                (&total_rows_device).as_kernel_param(),
2487            ];
2488            unsafe {
2489                kernel
2490                    .clone()
2491                    .launch_on_stream(
2492                        &cu_stream,
2493                        LaunchConfig {
2494                            grid_dim: (1, 1, 1),
2495                            block_dim: (1024, 1, 1),
2496                            shared_mem_bytes: 0,
2497                        },
2498                        &mut params,
2499                    )
2500                    .map_err(|e| {
2501                        XlogError::Kernel(format!("{ctx}: HG block-count scan failed: {e}"))
2502                    })?;
2503            }
2504        } else {
2505            let offsets_mut = local_offsets
2506                .as_mut()
2507                .expect("local HG offsets allocated when grid exceeds single-block scan");
2508            unsafe {
2509                let res = sys::cuMemcpyDtoDAsync_v2(
2510                    *offsets_mut.device_ptr(),
2511                    *count_u32.device_ptr(),
2512                    bytes_count,
2513                    cu_stream.cu_stream(),
2514                );
2515                if res != sys::cudaError_enum::CUDA_SUCCESS {
2516                    return Err(XlogError::Kernel(format!(
2517                        "{ctx}: DtoD count to offsets failed: {res:?}"
2518                    )));
2519                }
2520            }
2521            self.multiblock_scan_u32_inplace_on_stream(
2522                offsets_mut,
2523                grid,
2524                &cu_stream,
2525                launch_stream,
2526                runtime,
2527            )?;
2528            let total_kernel = self
2529                .device()
2530                .inner()
2531                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_COMPUTE_TOTAL)
2532                .ok_or_else(|| {
2533                    XlogError::Kernel("wcoj_compute_total kernel not found".to_string())
2534                })?;
2535            let mut params: Vec<*mut c_void> = vec![
2536                count_u32.as_kernel_param(),
2537                (&*offsets_mut).as_kernel_param(),
2538                grid.as_kernel_param(),
2539                (&total_rows_device).as_kernel_param(),
2540            ];
2541            unsafe {
2542                total_kernel
2543                    .clone()
2544                    .launch_on_stream(
2545                        &cu_stream,
2546                        LaunchConfig {
2547                            grid_dim: (1, 1, 1),
2548                            block_dim: (1, 1, 1),
2549                            shared_mem_bytes: 0,
2550                        },
2551                        &mut params,
2552                    )
2553                    .map_err(|e| {
2554                        XlogError::Kernel(format!("{ctx}: HG total reducer failed: {e}"))
2555                    })?;
2556            }
2557        }
2558        rec_hg
2559            .commit(runtime)
2560            .map_err(|e| XlogError::Kernel(format!("{ctx}: HG count commit failed: {e}")))?;
2561        cu_stream
2562            .synchronize()
2563            .map_err(|e| XlogError::Kernel(format!("{ctx}: count stream sync failed: {e}")))?;
2564        let total_rows = self
2565            .dtoh_scalar_untracked::<u32>(&total_rows_device, 0)
2566            .map_err(|e| XlogError::Kernel(format!("{ctx}: read total rows failed: {e}")))?;
2567        if total_rows == 0 {
2568            return self.create_empty_buffer(out_schema);
2569        }
2570
2571        let bytes_per_col = (total_rows as usize)
2572            .checked_mul(std::mem::size_of::<u64>())
2573            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: output byte size overflow")))?;
2574        let mut out_x = self.memory().alloc::<u8>(bytes_per_col)?;
2575        let mut out_y = self.memory().alloc::<u8>(bytes_per_col)?;
2576        let mut out_z = self.memory().alloc::<u8>(bytes_per_col)?;
2577        let materialize_offsets = if grid <= 1024 {
2578            &plan.block_offsets
2579        } else {
2580            local_offsets
2581                .as_ref()
2582                .expect("local HG offsets allocated when grid exceeds single-block scan")
2583        };
2584        let mut rec_mat = LaunchRecorder::new_strict(launch_stream);
2585        rec_mat.read(materialize_offsets);
2586        rec_mat.read(e_xy.num_rows_device());
2587        rec_mat.read(e_yz.num_rows_device());
2588        rec_mat.read(e_xz.num_rows_device());
2589        rec_mat.read_column(e_xy.column(0).expect("xy.col0"));
2590        rec_mat.read_column(e_xy.column(1).expect("xy.col1"));
2591        rec_mat.read_column(e_yz.column(1).expect("yz.col1"));
2592        rec_mat.read_column(e_xz.column(1).expect("xz.col1"));
2593        rec_mat.read(&plan.xy_work_prefix);
2594        rec_mat.read(&plan.xy_yz_start);
2595        rec_mat.read(&plan.xy_yz_end);
2596        rec_mat.read(&plan.xy_xz_start);
2597        rec_mat.read(&plan.xy_xz_end);
2598        rec_mat.write(&out_x);
2599        rec_mat.write(&out_y);
2600        rec_mat.write(&out_z);
2601        rec_mat
2602            .preflight(runtime)
2603            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize preflight failed: {e}")))?;
2604        {
2605            let kernel = self
2606                .device()
2607                .inner()
2608                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_TRIANGLE_MATERIALIZE_HG_U64)
2609                .ok_or_else(|| {
2610                    XlogError::Kernel(
2611                        "wcoj_triangle_materialize_hg_u64 kernel not found".to_string(),
2612                    )
2613                })?;
2614            let out_x_u64 = unsafe { reinterpret_u8_as_u64(&mut out_x) };
2615            let out_y_u64 = unsafe { reinterpret_u8_as_u64(&mut out_y) };
2616            let out_z_u64 = unsafe { reinterpret_u8_as_u64(&mut out_z) };
2617            let mut params: Vec<*mut c_void> = vec![
2618                xy_col0.as_kernel_param(),
2619                xy_col1.as_kernel_param(),
2620                yz_col1.as_kernel_param(),
2621                n_yz.as_kernel_param(),
2622                xz_col1.as_kernel_param(),
2623                n_xz.as_kernel_param(),
2624                (&plan.xy_work_prefix).as_kernel_param(),
2625                (&plan.xy_yz_start).as_kernel_param(),
2626                (&plan.xy_yz_end).as_kernel_param(),
2627                (&plan.xy_xz_start).as_kernel_param(),
2628                (&plan.xy_xz_end).as_kernel_param(),
2629                plan.row_count.as_kernel_param(),
2630                plan.total_work.as_kernel_param(),
2631                plan.block_work_unit.as_kernel_param(),
2632                materialize_offsets.as_kernel_param(),
2633                total_rows.as_kernel_param(),
2634                out_x_u64.as_kernel_param(),
2635                out_y_u64.as_kernel_param(),
2636                out_z_u64.as_kernel_param(),
2637            ];
2638            unsafe {
2639                kernel
2640                    .clone()
2641                    .launch_on_stream(
2642                        &cu_stream,
2643                        LaunchConfig {
2644                            grid_dim: (grid, 1, 1),
2645                            block_dim: (BLOCK_SIZE, 1, 1),
2646                            shared_mem_bytes: 0,
2647                        },
2648                        &mut params,
2649                    )
2650                    .map_err(|e| {
2651                        XlogError::Kernel(format!("{ctx}: materialize launch failed: {e}"))
2652                    })?;
2653            }
2654        }
2655        rec_mat
2656            .commit(runtime)
2657            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize commit failed: {e}")))?;
2658        cu_stream.synchronize().map_err(|e| {
2659            XlogError::Kernel(format!("{ctx}: materialize stream sync failed: {e}"))
2660        })?;
2661
2662        Ok(CudaBuffer::from_columns_with_host_count(
2663            vec![out_x.into(), out_y.into(), out_z.into()],
2664            total_rows as u64,
2665            total_rows_device,
2666            out_schema,
2667            total_rows,
2668        ))
2669    }
2670
2671    pub fn wcoj_4cycle_hg_work_plan_u32_recorded(
2672        &self,
2673        e1: &CudaBuffer,
2674        e2: &CudaBuffer,
2675        e3: &CudaBuffer,
2676        e4: &CudaBuffer,
2677        block_work_unit: u32,
2678        launch_stream: StreamId,
2679    ) -> Result<WcojCycle4HgWorkPlanU32> {
2680        let ctx = "wcoj_4cycle_hg_work_plan_u32_recorded";
2681        if block_work_unit == 0 {
2682            return Err(XlogError::Kernel(format!(
2683                "{ctx}: block_work_unit must be nonzero"
2684            )));
2685        }
2686        validate_binary_u32(ctx, "e1", e1)?;
2687        validate_binary_u32(ctx, "e2", e2)?;
2688        validate_binary_u32(ctx, "e3", e3)?;
2689        validate_binary_u32(ctx, "e4", e4)?;
2690
2691        let n_e1 = self.metadata_logical_rows(e1)?;
2692        let n_e2 = self.metadata_logical_rows(e2)?;
2693        let n_e3 = self.metadata_logical_rows(e3)?;
2694        let prefix_len = n_e1
2695            .checked_add(1)
2696            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: prefix length overflow")))?;
2697        let e2_prefix_len = n_e2
2698            .checked_add(1)
2699            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: e2 prefix length overflow")))?;
2700        let mut e1_work_prefix = self.memory().alloc::<u32>(prefix_len as usize)?;
2701        let mut e2_work_prefix = self.memory().alloc::<u32>(e2_prefix_len as usize)?;
2702        let mut e1_e2_start = self.memory().alloc::<u32>(n_e1 as usize)?;
2703        let mut e1_e2_end = self.memory().alloc::<u32>(n_e1 as usize)?;
2704
2705        if n_e1 == 0 || n_e2 == 0 || n_e3 == 0 || self.metadata_logical_rows(e4)? == 0 {
2706            let block_counts = self.memory().alloc::<u32>(1)?;
2707            let block_offsets = self.memory().alloc::<u32>(1)?;
2708            return Ok(WcojCycle4HgWorkPlanU32 {
2709                e1_work_prefix,
2710                e2_work_prefix,
2711                e1_e2_start,
2712                e1_e2_end,
2713                block_counts,
2714                block_offsets,
2715                total_work: 0,
2716                block_work_unit,
2717                row_count: n_e1,
2718            });
2719        }
2720
2721        let runtime = self.memory().runtime().ok_or_else(|| {
2722            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
2723        })?;
2724        let cu_stream = runtime
2725            .stream_pool()
2726            .resolve(launch_stream)
2727            .ok_or_else(|| {
2728                XlogError::Kernel(format!(
2729                    "{ctx}: launch_stream StreamId({}) does not resolve",
2730                    launch_stream.0
2731                ))
2732            })?;
2733
2734        let e1_col1 = metadata_column_u32(e1, 1)?;
2735        let e2_col0 = metadata_column_u32(e2, 0)?;
2736        let e2_col1 = metadata_column_u32(e2, 1)?;
2737        let e3_col0 = metadata_column_u32(e3, 0)?;
2738
2739        let mut rec = LaunchRecorder::new_strict(launch_stream);
2740        rec.read(e1.num_rows_device());
2741        rec.read(e2.num_rows_device());
2742        rec.read(e3.num_rows_device());
2743        rec.read_column(e1.column(1).expect("e1.col1"));
2744        rec.read_column(e2.column(0).expect("e2.col0"));
2745        rec.read_column(e2.column(1).expect("e2.col1"));
2746        rec.read_column(e3.column(0).expect("e3.col0"));
2747        rec.read_write(&e2_work_prefix);
2748        rec.write(&e1_work_prefix);
2749        rec.write(&e1_e2_start);
2750        rec.write(&e1_e2_end);
2751        rec.preflight(runtime)
2752            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
2753
2754        let e2_kernel = self
2755            .device()
2756            .inner()
2757            .get_func(
2758                WCOJ_MODULE,
2759                wcoj_kernels::WCOJ_4CYCLE_BUILD_E2_WORK_PREFIX_U32,
2760            )
2761            .ok_or_else(|| {
2762                XlogError::Kernel(
2763                    "wcoj_4cycle_build_e2_work_prefix_u32 kernel not found".to_string(),
2764                )
2765            })?;
2766        let e2_grid = n_e2.div_ceil(BLOCK_SIZE);
2767        unsafe {
2768            e2_kernel
2769                .clone()
2770                .launch_on_stream(
2771                    &cu_stream,
2772                    LaunchConfig {
2773                        grid_dim: (e2_grid, 1, 1),
2774                        block_dim: (BLOCK_SIZE, 1, 1),
2775                        shared_mem_bytes: 0,
2776                    },
2777                    (e2_col1, n_e2, e3_col0, n_e3, &mut e2_work_prefix),
2778                )
2779                .map_err(|e| {
2780                    XlogError::Kernel(format!(
2781                        "wcoj_4cycle_build_e2_work_prefix_u32 launch failed: {e}"
2782                    ))
2783                })?;
2784        }
2785        self.multiblock_scan_u32_inplace_on_stream(
2786            &mut e2_work_prefix,
2787            e2_prefix_len,
2788            &cu_stream,
2789            launch_stream,
2790            runtime,
2791        )?;
2792
2793        let kernel = self
2794            .device()
2795            .inner()
2796            .get_func(
2797                WCOJ_MODULE,
2798                wcoj_kernels::WCOJ_4CYCLE_BUILD_HG_WORK_PLAN_U32,
2799            )
2800            .ok_or_else(|| {
2801                XlogError::Kernel("wcoj_4cycle_build_hg_work_plan_u32 kernel not found".to_string())
2802            })?;
2803        let grid = n_e1.div_ceil(BLOCK_SIZE);
2804        unsafe {
2805            kernel
2806                .clone()
2807                .launch_on_stream(
2808                    &cu_stream,
2809                    LaunchConfig {
2810                        grid_dim: (grid, 1, 1),
2811                        block_dim: (BLOCK_SIZE, 1, 1),
2812                        shared_mem_bytes: 0,
2813                    },
2814                    (
2815                        e1_col1,
2816                        n_e1,
2817                        e2_col0,
2818                        n_e2,
2819                        &e2_work_prefix,
2820                        &mut e1_work_prefix,
2821                        &mut e1_e2_start,
2822                        &mut e1_e2_end,
2823                    ),
2824                )
2825                .map_err(|e| {
2826                    XlogError::Kernel(format!(
2827                        "wcoj_4cycle_build_hg_work_plan_u32 launch failed: {e}"
2828                    ))
2829                })?;
2830        }
2831        self.multiblock_scan_u32_inplace_on_stream(
2832            &mut e1_work_prefix,
2833            prefix_len,
2834            &cu_stream,
2835            launch_stream,
2836            runtime,
2837        )?;
2838        rec.commit(runtime)
2839            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
2840        cu_stream
2841            .synchronize()
2842            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
2843        let total_work = self.dtoh_scalar_untracked::<u32>(&e1_work_prefix, n_e1 as usize)?;
2844        let grid = if total_work == 0 {
2845            1
2846        } else {
2847            total_work.div_ceil(block_work_unit)
2848        };
2849        let block_counts = self.memory().alloc::<u32>(grid as usize)?;
2850        let block_offsets = self.memory().alloc::<u32>(grid as usize)?;
2851
2852        Ok(WcojCycle4HgWorkPlanU32 {
2853            e1_work_prefix,
2854            e2_work_prefix,
2855            e1_e2_start,
2856            e1_e2_end,
2857            block_counts,
2858            block_offsets,
2859            total_work,
2860            block_work_unit,
2861            row_count: n_e1,
2862        })
2863    }
2864
2865    pub fn wcoj_4cycle_hg_u32_recorded(
2866        &self,
2867        e1: &CudaBuffer,
2868        e2: &CudaBuffer,
2869        e3: &CudaBuffer,
2870        e4: &CudaBuffer,
2871        block_work_unit: u32,
2872        launch_stream: StreamId,
2873    ) -> Result<CudaBuffer> {
2874        let ctx = "wcoj_4cycle_hg_u32_recorded";
2875        validate_binary_u32(ctx, "e1", e1)?;
2876        validate_binary_u32(ctx, "e2", e2)?;
2877        validate_binary_u32(ctx, "e3", e3)?;
2878        validate_binary_u32(ctx, "e4", e4)?;
2879        let plan = self.wcoj_4cycle_hg_work_plan_u32_recorded(
2880            e1,
2881            e2,
2882            e3,
2883            e4,
2884            block_work_unit,
2885            launch_stream,
2886        )?;
2887        let out_schema = Schema::new(vec![
2888            (
2889                "col0".to_string(),
2890                e1.schema().column_type(0).expect("e1.col0 type"),
2891            ),
2892            (
2893                "col1".to_string(),
2894                e1.schema().column_type(1).expect("e1.col1 type"),
2895            ),
2896            (
2897                "col2".to_string(),
2898                e2.schema().column_type(1).expect("e2.col1 type"),
2899            ),
2900            (
2901                "col3".to_string(),
2902                e3.schema().column_type(1).expect("e3.col1 type"),
2903            ),
2904        ]);
2905        if plan.total_work == 0 {
2906            return self.create_empty_buffer(out_schema);
2907        }
2908
2909        let grid = plan.total_work.div_ceil(plan.block_work_unit);
2910        let bytes_count = (grid as usize)
2911            .checked_mul(std::mem::size_of::<u32>())
2912            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: count byte size overflow")))?;
2913        let mut local_counts = None;
2914        let mut local_offsets = None;
2915        if grid > 1024 {
2916            local_counts = Some(self.memory().alloc::<u32>(grid as usize)?);
2917            local_offsets = Some(self.memory().alloc::<u32>(grid as usize)?);
2918        }
2919        let total_rows_device = self.memory().alloc::<u32>(1)?;
2920
2921        let runtime = self.memory().runtime().ok_or_else(|| {
2922            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
2923        })?;
2924        let cu_stream = runtime
2925            .stream_pool()
2926            .resolve(launch_stream)
2927            .ok_or_else(|| {
2928                XlogError::Kernel(format!(
2929                    "{ctx}: launch_stream StreamId({}) does not resolve",
2930                    launch_stream.0
2931                ))
2932            })?;
2933
2934        let e1_col0 = metadata_column_u32(e1, 0)?;
2935        let e1_col1 = metadata_column_u32(e1, 1)?;
2936        let e2_col1 = metadata_column_u32(e2, 1)?;
2937        let e3_col0 = metadata_column_u32(e3, 0)?;
2938        let e3_col1 = metadata_column_u32(e3, 1)?;
2939        let e4_col0 = metadata_column_u32(e4, 0)?;
2940        let e4_col1 = metadata_column_u32(e4, 1)?;
2941        let n_e3 = self.metadata_logical_rows(e3)?;
2942        let n_e4 = self.metadata_logical_rows(e4)?;
2943
2944        let count_u32 = if grid <= 1024 {
2945            &plan.block_counts
2946        } else {
2947            local_counts
2948                .as_ref()
2949                .expect("local HG counts allocated when grid exceeds single-block scan")
2950        };
2951        let mut rec_hg = LaunchRecorder::new_strict(launch_stream);
2952        rec_hg.read(e1.num_rows_device());
2953        rec_hg.read(e2.num_rows_device());
2954        rec_hg.read(e3.num_rows_device());
2955        rec_hg.read(e4.num_rows_device());
2956        rec_hg.read_column(e1.column(0).expect("e1.col0"));
2957        rec_hg.read_column(e1.column(1).expect("e1.col1"));
2958        rec_hg.read_column(e2.column(1).expect("e2.col1"));
2959        rec_hg.read_column(e3.column(0).expect("e3.col0"));
2960        rec_hg.read_column(e3.column(1).expect("e3.col1"));
2961        rec_hg.read_column(e4.column(0).expect("e4.col0"));
2962        rec_hg.read_column(e4.column(1).expect("e4.col1"));
2963        rec_hg.read(&plan.e1_work_prefix);
2964        rec_hg.read(&plan.e2_work_prefix);
2965        rec_hg.read(&plan.e1_e2_start);
2966        rec_hg.read(&plan.e1_e2_end);
2967        rec_hg.read_write(count_u32);
2968        if grid <= 1024 {
2969            rec_hg.read_write(&plan.block_offsets);
2970        } else {
2971            rec_hg.read_write(
2972                local_offsets
2973                    .as_ref()
2974                    .expect("local HG offsets allocated when grid exceeds single-block scan"),
2975            );
2976        }
2977        rec_hg.write(&total_rows_device);
2978        rec_hg
2979            .preflight(runtime)
2980            .map_err(|e| XlogError::Kernel(format!("{ctx}: HG preflight failed: {e}")))?;
2981        {
2982            let kernel = self
2983                .device()
2984                .inner()
2985                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_4CYCLE_COUNT_HG_U32)
2986                .ok_or_else(|| {
2987                    XlogError::Kernel("wcoj_4cycle_count_hg_u32 kernel not found".to_string())
2988                })?;
2989            let mut params: Vec<*mut c_void> = vec![
2990                e1_col0.as_kernel_param(),
2991                e1_col1.as_kernel_param(),
2992                plan.row_count.as_kernel_param(),
2993                e2_col1.as_kernel_param(),
2994                e3_col0.as_kernel_param(),
2995                e3_col1.as_kernel_param(),
2996                n_e3.as_kernel_param(),
2997                e4_col0.as_kernel_param(),
2998                e4_col1.as_kernel_param(),
2999                n_e4.as_kernel_param(),
3000                (&plan.e1_work_prefix).as_kernel_param(),
3001                (&plan.e2_work_prefix).as_kernel_param(),
3002                (&plan.e1_e2_start).as_kernel_param(),
3003                (&plan.e1_e2_end).as_kernel_param(),
3004                plan.total_work.as_kernel_param(),
3005                plan.block_work_unit.as_kernel_param(),
3006                count_u32.as_kernel_param(),
3007            ];
3008            unsafe {
3009                kernel
3010                    .clone()
3011                    .launch_on_stream(
3012                        &cu_stream,
3013                        LaunchConfig {
3014                            grid_dim: (grid, 1, 1),
3015                            block_dim: (BLOCK_SIZE, 1, 1),
3016                            shared_mem_bytes: 0,
3017                        },
3018                        &mut params,
3019                    )
3020                    .map_err(|e| XlogError::Kernel(format!("{ctx}: count launch failed: {e}")))?;
3021            }
3022        }
3023        if grid <= 1024 {
3024            let kernel = self
3025                .device()
3026                .inner()
3027                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_SCAN_HG_BLOCK_COUNTS_U32)
3028                .ok_or_else(|| {
3029                    XlogError::Kernel("wcoj_scan_hg_block_counts_u32 kernel not found".to_string())
3030                })?;
3031            let mut params: Vec<*mut c_void> = vec![
3032                count_u32.as_kernel_param(),
3033                grid.as_kernel_param(),
3034                (&plan.block_offsets).as_kernel_param(),
3035                (&total_rows_device).as_kernel_param(),
3036            ];
3037            unsafe {
3038                kernel
3039                    .clone()
3040                    .launch_on_stream(
3041                        &cu_stream,
3042                        LaunchConfig {
3043                            grid_dim: (1, 1, 1),
3044                            block_dim: (1024, 1, 1),
3045                            shared_mem_bytes: 0,
3046                        },
3047                        &mut params,
3048                    )
3049                    .map_err(|e| XlogError::Kernel(format!("{ctx}: scan failed: {e}")))?;
3050            }
3051        } else {
3052            let offsets_mut = local_offsets
3053                .as_mut()
3054                .expect("local HG offsets allocated when grid exceeds single-block scan");
3055            unsafe {
3056                let res = sys::cuMemcpyDtoDAsync_v2(
3057                    *offsets_mut.device_ptr(),
3058                    *count_u32.device_ptr(),
3059                    bytes_count,
3060                    cu_stream.cu_stream(),
3061                );
3062                if res != sys::cudaError_enum::CUDA_SUCCESS {
3063                    return Err(XlogError::Kernel(format!(
3064                        "{ctx}: DtoD count to offsets failed: {res:?}"
3065                    )));
3066                }
3067            }
3068            self.multiblock_scan_u32_inplace_on_stream(
3069                offsets_mut,
3070                grid,
3071                &cu_stream,
3072                launch_stream,
3073                runtime,
3074            )?;
3075            let total_kernel = self
3076                .device()
3077                .inner()
3078                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_COMPUTE_TOTAL)
3079                .ok_or_else(|| {
3080                    XlogError::Kernel("wcoj_compute_total kernel not found".to_string())
3081                })?;
3082            let mut params: Vec<*mut c_void> = vec![
3083                count_u32.as_kernel_param(),
3084                (&*offsets_mut).as_kernel_param(),
3085                grid.as_kernel_param(),
3086                (&total_rows_device).as_kernel_param(),
3087            ];
3088            unsafe {
3089                total_kernel
3090                    .clone()
3091                    .launch_on_stream(
3092                        &cu_stream,
3093                        LaunchConfig {
3094                            grid_dim: (1, 1, 1),
3095                            block_dim: (1, 1, 1),
3096                            shared_mem_bytes: 0,
3097                        },
3098                        &mut params,
3099                    )
3100                    .map_err(|e| XlogError::Kernel(format!("{ctx}: total failed: {e}")))?;
3101            }
3102        }
3103        rec_hg
3104            .commit(runtime)
3105            .map_err(|e| XlogError::Kernel(format!("{ctx}: count commit failed: {e}")))?;
3106        cu_stream
3107            .synchronize()
3108            .map_err(|e| XlogError::Kernel(format!("{ctx}: count stream sync failed: {e}")))?;
3109        let total_rows = self
3110            .dtoh_scalar_untracked::<u32>(&total_rows_device, 0)
3111            .map_err(|e| XlogError::Kernel(format!("{ctx}: read total rows failed: {e}")))?;
3112        if total_rows == 0 {
3113            return self.create_empty_buffer(out_schema);
3114        }
3115
3116        let bytes_per_col = (total_rows as usize)
3117            .checked_mul(std::mem::size_of::<u32>())
3118            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: output byte size overflow")))?;
3119        let mut out_w = self.memory().alloc::<u8>(bytes_per_col)?;
3120        let mut out_x = self.memory().alloc::<u8>(bytes_per_col)?;
3121        let mut out_y = self.memory().alloc::<u8>(bytes_per_col)?;
3122        let mut out_z = self.memory().alloc::<u8>(bytes_per_col)?;
3123        let materialize_offsets = if grid <= 1024 {
3124            &plan.block_offsets
3125        } else {
3126            local_offsets
3127                .as_ref()
3128                .expect("local HG offsets allocated when grid exceeds single-block scan")
3129        };
3130        let mut rec_mat = LaunchRecorder::new_strict(launch_stream);
3131        rec_mat.read(materialize_offsets);
3132        rec_mat.read(e1.num_rows_device());
3133        rec_mat.read(e2.num_rows_device());
3134        rec_mat.read(e3.num_rows_device());
3135        rec_mat.read(e4.num_rows_device());
3136        rec_mat.read_column(e1.column(0).expect("e1.col0"));
3137        rec_mat.read_column(e1.column(1).expect("e1.col1"));
3138        rec_mat.read_column(e2.column(1).expect("e2.col1"));
3139        rec_mat.read_column(e3.column(0).expect("e3.col0"));
3140        rec_mat.read_column(e3.column(1).expect("e3.col1"));
3141        rec_mat.read_column(e4.column(0).expect("e4.col0"));
3142        rec_mat.read_column(e4.column(1).expect("e4.col1"));
3143        rec_mat.read(&plan.e1_work_prefix);
3144        rec_mat.read(&plan.e2_work_prefix);
3145        rec_mat.read(&plan.e1_e2_start);
3146        rec_mat.read(&plan.e1_e2_end);
3147        rec_mat.write(&out_w);
3148        rec_mat.write(&out_x);
3149        rec_mat.write(&out_y);
3150        rec_mat.write(&out_z);
3151        rec_mat
3152            .preflight(runtime)
3153            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize preflight failed: {e}")))?;
3154        {
3155            let kernel = self
3156                .device()
3157                .inner()
3158                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_4CYCLE_MATERIALIZE_HG_U32)
3159                .ok_or_else(|| {
3160                    XlogError::Kernel("wcoj_4cycle_materialize_hg_u32 kernel not found".to_string())
3161                })?;
3162            let out_w_u32 = unsafe { reinterpret_u8_as_u32(&mut out_w) };
3163            let out_x_u32 = unsafe { reinterpret_u8_as_u32(&mut out_x) };
3164            let out_y_u32 = unsafe { reinterpret_u8_as_u32(&mut out_y) };
3165            let out_z_u32 = unsafe { reinterpret_u8_as_u32(&mut out_z) };
3166            let mut params: Vec<*mut c_void> = vec![
3167                e1_col0.as_kernel_param(),
3168                e1_col1.as_kernel_param(),
3169                plan.row_count.as_kernel_param(),
3170                e2_col1.as_kernel_param(),
3171                e3_col0.as_kernel_param(),
3172                e3_col1.as_kernel_param(),
3173                n_e3.as_kernel_param(),
3174                e4_col0.as_kernel_param(),
3175                e4_col1.as_kernel_param(),
3176                n_e4.as_kernel_param(),
3177                (&plan.e1_work_prefix).as_kernel_param(),
3178                (&plan.e2_work_prefix).as_kernel_param(),
3179                (&plan.e1_e2_start).as_kernel_param(),
3180                (&plan.e1_e2_end).as_kernel_param(),
3181                plan.total_work.as_kernel_param(),
3182                plan.block_work_unit.as_kernel_param(),
3183                materialize_offsets.as_kernel_param(),
3184                total_rows.as_kernel_param(),
3185                out_w_u32.as_kernel_param(),
3186                out_x_u32.as_kernel_param(),
3187                out_y_u32.as_kernel_param(),
3188                out_z_u32.as_kernel_param(),
3189            ];
3190            unsafe {
3191                kernel
3192                    .clone()
3193                    .launch_on_stream(
3194                        &cu_stream,
3195                        LaunchConfig {
3196                            grid_dim: (grid, 1, 1),
3197                            block_dim: (BLOCK_SIZE, 1, 1),
3198                            shared_mem_bytes: 0,
3199                        },
3200                        &mut params,
3201                    )
3202                    .map_err(|e| {
3203                        XlogError::Kernel(format!("{ctx}: materialize launch failed: {e}"))
3204                    })?;
3205            }
3206        }
3207        rec_mat
3208            .commit(runtime)
3209            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize commit failed: {e}")))?;
3210        cu_stream.synchronize().map_err(|e| {
3211            XlogError::Kernel(format!("{ctx}: materialize stream sync failed: {e}"))
3212        })?;
3213
3214        Ok(CudaBuffer::from_columns_with_host_count(
3215            vec![out_w.into(), out_x.into(), out_y.into(), out_z.into()],
3216            total_rows as u64,
3217            total_rows_device,
3218            out_schema,
3219            total_rows,
3220        ))
3221    }
3222
3223    /// Aggregate-fused 4-cycle group-by-root count: evaluate
3224    /// `q(W, count) :- e1(W,X), e2(X,Y), e3(Y,Z), e4(Z,W)` grouped by the
3225    /// variable-order root W, WITHOUT materializing the 4-cycle rows.
3226    ///
3227    /// Pipeline (all recorded; the 4-cycle result never exists as rows):
3228    /// 1. the standard 4-cycle histogram-guided work plan;
3229    /// 2. `wcoj_4cycle_groupby_root_count_hg_u32` accumulates, per e1 row,
3230    ///    a match count (integer atomicAdd — order-insensitive,
3231    ///    deterministic values);
3232    /// 3. a (W, count) staging buffer over the *input* rows is compacted
3233    ///    to count>0 rows (roots with no completion must be absent) and
3234    ///    reduced per W via the recorded groupby Sum.
3235    ///
3236    /// All reduction work is O(n_e1) — input-sized, never join-output-sized.
3237    ///
3238    /// Output schema matches the unfused materialize+groupby baseline:
3239    /// `col0` = W (e1.col0 type, U32/Symbol), `col1` = count (U64).
3240    ///
3241    /// # Errors
3242    /// * `XlogError::Kernel` if the manager has no runtime, the launch
3243    ///   stream does not resolve, an input is not 2-column U32/Symbol, or
3244    ///   any kernel launch fails.
3245    pub fn wcoj_4cycle_groupby_root_count_u32_recorded(
3246        &self,
3247        e1: &CudaBuffer,
3248        e2: &CudaBuffer,
3249        e3: &CudaBuffer,
3250        e4: &CudaBuffer,
3251        block_work_unit: u32,
3252        launch_stream: StreamId,
3253    ) -> Result<CudaBuffer> {
3254        let ctx = "wcoj_4cycle_groupby_root_count_u32_recorded";
3255        // Layout-normalize per dispatch (sorted-fast-path clone when the
3256        // input is already lex-sorted + unique): the fused path must give
3257        // the same guarantee as the unfused pipeline instead of trusting
3258        // store-buffer sortedness — unsorted/duplicated inputs previously
3259        // produced silently wrong (empty) fused results.
3260        let e1 = &self.wcoj_layout_u32_recorded(e1, launch_stream)?;
3261        let e2 = &self.wcoj_layout_u32_recorded(e2, launch_stream)?;
3262        let e3 = &self.wcoj_layout_u32_recorded(e3, launch_stream)?;
3263        let e4 = &self.wcoj_layout_u32_recorded(e4, launch_stream)?;
3264        validate_binary_u32(ctx, "e1", e1)?;
3265        validate_binary_u32(ctx, "e2", e2)?;
3266        validate_binary_u32(ctx, "e3", e3)?;
3267        validate_binary_u32(ctx, "e4", e4)?;
3268        let plan = self.wcoj_4cycle_hg_work_plan_u32_recorded(
3269            e1,
3270            e2,
3271            e3,
3272            e4,
3273            block_work_unit,
3274            launch_stream,
3275        )?;
3276        let n_e1 = plan.row_count;
3277        let w_type = e1.schema().column_type(0).expect("e1.col0 type");
3278        let out_schema = Schema::new(vec![
3279            ("w".to_string(), w_type),
3280            ("count".to_string(), ScalarType::U64),
3281        ]);
3282        if n_e1 == 0 || plan.total_work == 0 {
3283            return self.create_empty_buffer(out_schema);
3284        }
3285
3286        let runtime = self.memory().runtime().ok_or_else(|| {
3287            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
3288        })?;
3289        let cu_stream = runtime
3290            .stream_pool()
3291            .resolve(launch_stream)
3292            .ok_or_else(|| {
3293                XlogError::Kernel(format!(
3294                    "{ctx}: launch_stream StreamId({}) does not resolve",
3295                    launch_stream.0
3296                ))
3297            })?;
3298
3299        let e1_col0 = metadata_column_u32(e1, 0)?;
3300        let e1_col1 = metadata_column_u32(e1, 1)?;
3301        let e2_col1 = metadata_column_u32(e2, 1)?;
3302        let e3_col0 = metadata_column_u32(e3, 0)?;
3303        let e3_col1 = metadata_column_u32(e3, 1)?;
3304        let e4_col0 = metadata_column_u32(e4, 0)?;
3305        let e4_col1 = metadata_column_u32(e4, 1)?;
3306        let n_e3 = self.metadata_logical_rows(e3)?;
3307        let n_e4 = self.metadata_logical_rows(e4)?;
3308
3309        // Per-e1-row match counters, zero-initialized. Allocated as the
3310        // u8-backed column layout so the array doubles as the staging
3311        // buffer's count column after the kernel fills it.
3312        let mut row_counts = self
3313            .memory()
3314            .alloc::<u8>(n_e1 as usize * std::mem::size_of::<u32>())?;
3315        self.device()
3316            .inner()
3317            .memset_zeros(&mut row_counts)
3318            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row counts failed: {e}")))?;
3319
3320        let grid = plan.total_work.div_ceil(plan.block_work_unit);
3321        let mut rec = LaunchRecorder::new_strict(launch_stream);
3322        rec.read(e1.num_rows_device());
3323        rec.read(e2.num_rows_device());
3324        rec.read(e3.num_rows_device());
3325        rec.read(e4.num_rows_device());
3326        rec.read_column(e1.column(0).expect("e1.col0"));
3327        rec.read_column(e1.column(1).expect("e1.col1"));
3328        rec.read_column(e2.column(1).expect("e2.col1"));
3329        rec.read_column(e3.column(0).expect("e3.col0"));
3330        rec.read_column(e3.column(1).expect("e3.col1"));
3331        rec.read_column(e4.column(0).expect("e4.col0"));
3332        rec.read_column(e4.column(1).expect("e4.col1"));
3333        rec.read(&plan.e1_work_prefix);
3334        rec.read(&plan.e2_work_prefix);
3335        rec.read(&plan.e1_e2_start);
3336        rec.read(&plan.e1_e2_end);
3337        rec.write(&row_counts);
3338        rec.preflight(runtime)
3339            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
3340        {
3341            let kernel = self
3342                .device()
3343                .inner()
3344                .get_func(
3345                    WCOJ_MODULE,
3346                    wcoj_kernels::WCOJ_4CYCLE_GROUPBY_ROOT_COUNT_HG_U32,
3347                )
3348                .ok_or_else(|| {
3349                    XlogError::Kernel(
3350                        "wcoj_4cycle_groupby_root_count_hg_u32 kernel not found".to_string(),
3351                    )
3352                })?;
3353            let mut params: Vec<*mut c_void> = vec![
3354                e1_col0.as_kernel_param(),
3355                e1_col1.as_kernel_param(),
3356                plan.row_count.as_kernel_param(),
3357                e2_col1.as_kernel_param(),
3358                e3_col0.as_kernel_param(),
3359                e3_col1.as_kernel_param(),
3360                n_e3.as_kernel_param(),
3361                e4_col0.as_kernel_param(),
3362                e4_col1.as_kernel_param(),
3363                n_e4.as_kernel_param(),
3364                (&plan.e1_work_prefix).as_kernel_param(),
3365                (&plan.e2_work_prefix).as_kernel_param(),
3366                (&plan.e1_e2_start).as_kernel_param(),
3367                (&plan.e1_e2_end).as_kernel_param(),
3368                plan.total_work.as_kernel_param(),
3369                plan.block_work_unit.as_kernel_param(),
3370                (&row_counts).as_kernel_param(),
3371            ];
3372            unsafe {
3373                kernel
3374                    .clone()
3375                    .launch_on_stream(
3376                        &cu_stream,
3377                        LaunchConfig {
3378                            grid_dim: (grid, 1, 1),
3379                            block_dim: (BLOCK_SIZE, 1, 1),
3380                            shared_mem_bytes: 0,
3381                        },
3382                        &mut params,
3383                    )
3384                    .map_err(|e| {
3385                        XlogError::Kernel(format!("{ctx}: groupby-count launch failed: {e}"))
3386                    })?;
3387            }
3388        }
3389        rec.commit(runtime)
3390            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
3391
3392        // Staging buffer (W, count) over the n_e1 input rows: W is a
3393        // device-to-device copy of e1.col0; the count column is the
3394        // kernel-filled array. Rows stay lex-sorted by W.
3395        let w_src = match e1.column(0).expect("e1.col0") {
3396            CudaColumn::Owned(slice) => slice,
3397            _ => {
3398                return Err(XlogError::Kernel(format!(
3399                    "{ctx}: e1.col0 must be an owned CudaColumn"
3400                )))
3401            }
3402        };
3403        let mut w_copy = self
3404            .memory()
3405            .alloc::<u8>(n_e1 as usize * std::mem::size_of::<u32>())?;
3406        self.device()
3407            .inner()
3408            .dtod_copy(w_src, &mut w_copy)
3409            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy W column failed: {e}")))?;
3410        let mut d_num_rows = self.memory().alloc::<u32>(1)?;
3411        self.device()
3412            .inner()
3413            .dtod_copy(e1.num_rows_device(), &mut d_num_rows)
3414            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy row count failed: {e}")))?;
3415        let staging_schema = Schema::new(vec![
3416            ("w".to_string(), w_type),
3417            ("count".to_string(), ScalarType::U32),
3418        ]);
3419        let staging = CudaBuffer::from_columns_with_host_count(
3420            vec![w_copy.into(), row_counts.into()],
3421            n_e1 as u64,
3422            d_num_rows,
3423            staging_schema,
3424            n_e1,
3425        );
3426
3427        // Keep only roots with at least one completed 4-cycle, then reduce
3428        // per W. Both steps run over input-sized data.
3429        let mask = self.compare_const_mask_recorded::<u32>(
3430            &staging,
3431            1,
3432            0u32,
3433            crate::CompareOp::Gt,
3434            launch_stream,
3435        )?;
3436        let compacted =
3437            self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)?;
3438        self.groupby_multi_agg_recorded(
3439            &compacted,
3440            &[0],
3441            &[(1, xlog_core::AggOp::Sum)],
3442            launch_stream,
3443        )
3444    }
3445
3446    /// Aggregate-fused 4-cycle group-by-root sum/min/max: evaluate
3447    /// `q(W, agg(V)) :- e1(W,X), e2(X,Y), e3(Y,Z), e4(Z,W)` with
3448    /// `agg ∈ {Sum, Min, Max}` and `V ∈ {X, Y, Z}` grouped by the
3449    /// variable-order root W, WITHOUT materializing the 4-cycle rows.
3450    ///
3451    /// Pipeline (all recorded; the 4-cycle result never exists as rows):
3452    /// 1. the standard 4-cycle histogram-guided work plan;
3453    /// 2. the per-op fused kernel accumulates, per e1 row, a match count
3454    ///    (compaction mask) and the per-row partial aggregate (integer
3455    ///    atomics — order-insensitive, deterministic values). Sum partials
3456    ///    are u64 (a per-row partial can exceed `u32::MAX`); min partials
3457    ///    start at `u32::MAX`, max partials at 0;
3458    /// 3. a 3-column (W, count, agg) staging buffer over the *input* rows
3459    ///    is compacted to count>0 rows (roots with no completion must be
3460    ///    absent) and reduced per W via the recorded groupby with the same
3461    ///    `AggOp` (Sum over the u64 partials; Min/Max over u32).
3462    ///
3463    /// All reduction work is O(n_e1) — input-sized, never join-output-sized.
3464    ///
3465    /// Output schema matches the unfused materialize+groupby baseline:
3466    /// `col0` = W (e1.col0 type, U32/Symbol), `col1` = U64 for Sum,
3467    /// U32 for Min/Max.
3468    ///
3469    /// Bag semantics: every (X, Y, Z) completion contributes its value,
3470    /// exactly like aggregating the materialized projection.
3471    ///
3472    /// # Errors
3473    /// * `XlogError::Kernel` if `agg_op` is not Sum/Min/Max, the value
3474    ///   column is not plain U32, the manager has no runtime, the launch
3475    ///   stream does not resolve, an input is not 2-column U32/Symbol, or
3476    ///   any kernel launch fails.
3477    #[allow(clippy::too_many_arguments)]
3478    pub fn wcoj_4cycle_groupby_root_agg_u32_recorded(
3479        &self,
3480        e1: &CudaBuffer,
3481        e2: &CudaBuffer,
3482        e3: &CudaBuffer,
3483        e4: &CudaBuffer,
3484        agg_op: AggOp,
3485        value: Wcoj4CycleRootAggValue,
3486        block_work_unit: u32,
3487        launch_stream: StreamId,
3488    ) -> Result<CudaBuffer> {
3489        let ctx = "wcoj_4cycle_groupby_root_agg_u32_recorded";
3490        // Layout-normalize per dispatch (sorted-fast-path clone when the
3491        // input is already lex-sorted + unique): the fused path must give
3492        // the same guarantee as the unfused pipeline instead of trusting
3493        // store-buffer sortedness — unsorted/duplicated inputs previously
3494        // produced silently wrong (empty) fused results.
3495        let e1 = &self.wcoj_layout_u32_recorded(e1, launch_stream)?;
3496        let e2 = &self.wcoj_layout_u32_recorded(e2, launch_stream)?;
3497        let e3 = &self.wcoj_layout_u32_recorded(e3, launch_stream)?;
3498        let e4 = &self.wcoj_layout_u32_recorded(e4, launch_stream)?;
3499        let (kernel_name, agg_elem_size, agg_scalar, agg_name) = match agg_op {
3500            AggOp::Sum => (
3501                wcoj_kernels::WCOJ_4CYCLE_GROUPBY_ROOT_SUM_HG_U32,
3502                std::mem::size_of::<u64>(),
3503                ScalarType::U64,
3504                "sum_0",
3505            ),
3506            AggOp::Min => (
3507                wcoj_kernels::WCOJ_4CYCLE_GROUPBY_ROOT_MIN_HG_U32,
3508                std::mem::size_of::<u32>(),
3509                ScalarType::U32,
3510                "min_0",
3511            ),
3512            AggOp::Max => (
3513                wcoj_kernels::WCOJ_4CYCLE_GROUPBY_ROOT_MAX_HG_U32,
3514                std::mem::size_of::<u32>(),
3515                ScalarType::U32,
3516                "max_0",
3517            ),
3518            other => {
3519                return Err(XlogError::Kernel(format!(
3520                    "{ctx}: unsupported AggOp {other:?} (Sum/Min/Max only; use \
3521                     wcoj_4cycle_groupby_root_count_u32_recorded for Count)"
3522                )))
3523            }
3524        };
3525        validate_binary_u32(ctx, "e1", e1)?;
3526        validate_binary_u32(ctx, "e2", e2)?;
3527        validate_binary_u32(ctx, "e3", e3)?;
3528        validate_binary_u32(ctx, "e4", e4)?;
3529        // The aggregate value is arithmetic: require a plain U32 value
3530        // column (Symbol ids are not summable/orderable data). The column
3531        // checked is exactly the one the kernel reads the value from —
3532        // and the one whose type the materialized (W, X, Y, Z) baseline
3533        // schema carries (`build_4cycle_head_schema`).
3534        let (value_buf, value_label) = match value {
3535            Wcoj4CycleRootAggValue::X => (e1, "e1"),
3536            Wcoj4CycleRootAggValue::Y => (e2, "e2"),
3537            Wcoj4CycleRootAggValue::Z => (e3, "e3"),
3538        };
3539        {
3540            let ty = value_buf.schema().column_type(1).expect("validated 2-col");
3541            if ty != ScalarType::U32 {
3542                return Err(XlogError::Kernel(format!(
3543                    "{ctx}: {value_label}.col1 supplies the aggregate value and must be U32, \
3544                     got {ty:?}"
3545                )));
3546            }
3547        }
3548
3549        let plan = self.wcoj_4cycle_hg_work_plan_u32_recorded(
3550            e1,
3551            e2,
3552            e3,
3553            e4,
3554            block_work_unit,
3555            launch_stream,
3556        )?;
3557        let n_e1 = plan.row_count;
3558        let w_type = e1.schema().column_type(0).expect("e1.col0 type");
3559        let out_schema = Schema::new(vec![
3560            ("w".to_string(), w_type),
3561            (agg_name.to_string(), agg_scalar),
3562        ]);
3563        if n_e1 == 0 || plan.total_work == 0 {
3564            return self.create_empty_buffer(out_schema);
3565        }
3566
3567        let runtime = self.memory().runtime().ok_or_else(|| {
3568            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
3569        })?;
3570        let cu_stream = runtime
3571            .stream_pool()
3572            .resolve(launch_stream)
3573            .ok_or_else(|| {
3574                XlogError::Kernel(format!(
3575                    "{ctx}: launch_stream StreamId({}) does not resolve",
3576                    launch_stream.0
3577                ))
3578            })?;
3579
3580        let e1_col0 = metadata_column_u32(e1, 0)?;
3581        let e1_col1 = metadata_column_u32(e1, 1)?;
3582        let e2_col1 = metadata_column_u32(e2, 1)?;
3583        let e3_col0 = metadata_column_u32(e3, 0)?;
3584        let e3_col1 = metadata_column_u32(e3, 1)?;
3585        let e4_col0 = metadata_column_u32(e4, 0)?;
3586        let e4_col1 = metadata_column_u32(e4, 1)?;
3587        let n_e3 = self.metadata_logical_rows(e3)?;
3588        let n_e4 = self.metadata_logical_rows(e4)?;
3589        let value_sel: u32 = match value {
3590            Wcoj4CycleRootAggValue::X => 0,
3591            Wcoj4CycleRootAggValue::Y => 1,
3592            Wcoj4CycleRootAggValue::Z => 2,
3593        };
3594
3595        // Per-e1-row match counters + aggregate partials, allocated as
3596        // the u8-backed column layout so the arrays double as the staging
3597        // buffer's columns after the kernel fills them.
3598        let mut row_counts = self
3599            .memory()
3600            .alloc::<u8>(n_e1 as usize * std::mem::size_of::<u32>())?;
3601        self.device()
3602            .inner()
3603            .memset_zeros(&mut row_counts)
3604            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row counts failed: {e}")))?;
3605        let mut row_agg = self.memory().alloc::<u8>(n_e1 as usize * agg_elem_size)?;
3606        self.device()
3607            .inner()
3608            .memset_zeros(&mut row_agg)
3609            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row aggregates failed: {e}")))?;
3610
3611        let grid = plan.total_work.div_ceil(plan.block_work_unit);
3612        let mut rec = LaunchRecorder::new_strict(launch_stream);
3613        rec.read(e1.num_rows_device());
3614        rec.read(e2.num_rows_device());
3615        rec.read(e3.num_rows_device());
3616        rec.read(e4.num_rows_device());
3617        rec.read_column(e1.column(0).expect("e1.col0"));
3618        rec.read_column(e1.column(1).expect("e1.col1"));
3619        rec.read_column(e2.column(1).expect("e2.col1"));
3620        rec.read_column(e3.column(0).expect("e3.col0"));
3621        rec.read_column(e3.column(1).expect("e3.col1"));
3622        rec.read_column(e4.column(0).expect("e4.col0"));
3623        rec.read_column(e4.column(1).expect("e4.col1"));
3624        rec.read(&plan.e1_work_prefix);
3625        rec.read(&plan.e2_work_prefix);
3626        rec.read(&plan.e1_e2_start);
3627        rec.read(&plan.e1_e2_end);
3628        rec.write(&row_counts);
3629        rec.write(&row_agg);
3630        rec.preflight(runtime)
3631            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
3632        if matches!(agg_op, AggOp::Min) {
3633            // Min identity: u32::MAX (compaction drops untouched rows).
3634            let fill = self
3635                .device()
3636                .inner()
3637                .get_func(ARITH_MODULE, arith_kernels::ARITH_FILL_CONST_U32)
3638                .ok_or_else(|| {
3639                    XlogError::Kernel("arith_fill_const_u32 kernel not found".to_string())
3640                })?;
3641            let row_agg_u32 = unsafe { reinterpret_u8_as_u32(&mut row_agg) };
3642            // SAFETY: arith_fill_const_u32(value, n, output)
3643            unsafe {
3644                fill.clone()
3645                    .launch_on_stream(
3646                        &cu_stream,
3647                        LaunchConfig::for_num_elems(n_e1),
3648                        (u32::MAX, n_e1, &mut *row_agg_u32),
3649                    )
3650                    .map_err(|e| {
3651                        XlogError::Kernel(format!("{ctx}: min identity fill failed: {e}"))
3652                    })?;
3653            }
3654        }
3655        {
3656            let kernel = self
3657                .device()
3658                .inner()
3659                .get_func(WCOJ_MODULE, kernel_name)
3660                .ok_or_else(|| XlogError::Kernel(format!("{kernel_name} kernel not found")))?;
3661            let mut params: Vec<*mut c_void> = vec![
3662                e1_col0.as_kernel_param(),
3663                e1_col1.as_kernel_param(),
3664                plan.row_count.as_kernel_param(),
3665                e2_col1.as_kernel_param(),
3666                e3_col0.as_kernel_param(),
3667                e3_col1.as_kernel_param(),
3668                n_e3.as_kernel_param(),
3669                e4_col0.as_kernel_param(),
3670                e4_col1.as_kernel_param(),
3671                n_e4.as_kernel_param(),
3672                value_sel.as_kernel_param(),
3673                (&plan.e1_work_prefix).as_kernel_param(),
3674                (&plan.e2_work_prefix).as_kernel_param(),
3675                (&plan.e1_e2_start).as_kernel_param(),
3676                (&plan.e1_e2_end).as_kernel_param(),
3677                plan.total_work.as_kernel_param(),
3678                plan.block_work_unit.as_kernel_param(),
3679                (&row_counts).as_kernel_param(),
3680                (&row_agg).as_kernel_param(),
3681            ];
3682            unsafe {
3683                kernel
3684                    .clone()
3685                    .launch_on_stream(
3686                        &cu_stream,
3687                        LaunchConfig {
3688                            grid_dim: (grid, 1, 1),
3689                            block_dim: (BLOCK_SIZE, 1, 1),
3690                            shared_mem_bytes: 0,
3691                        },
3692                        &mut params,
3693                    )
3694                    .map_err(|e| {
3695                        XlogError::Kernel(format!("{ctx}: groupby-agg launch failed: {e}"))
3696                    })?;
3697            }
3698        }
3699        rec.commit(runtime)
3700            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
3701
3702        // Staging buffer (W, count, agg) over the n_e1 input rows: W is a
3703        // device-to-device copy of e1.col0; count and agg are the
3704        // kernel-filled arrays. Rows stay lex-sorted by W.
3705        let w_src = match e1.column(0).expect("e1.col0") {
3706            CudaColumn::Owned(slice) => slice,
3707            _ => {
3708                return Err(XlogError::Kernel(format!(
3709                    "{ctx}: e1.col0 must be an owned CudaColumn"
3710                )))
3711            }
3712        };
3713        let w_copy = self
3714            .memory()
3715            .alloc::<u8>(n_e1 as usize * std::mem::size_of::<u32>())?;
3716        // Explicit-length copy: layout-normalized columns are allocated at
3717        // capacity, which can exceed the logical n_e1 * 4 bytes a full-slice
3718        // typed copy would assert on.
3719        unsafe {
3720            let res = sys::cuMemcpyDtoD_v2(
3721                *w_copy.device_ptr(),
3722                *w_src.device_ptr(),
3723                n_e1 as usize * std::mem::size_of::<u32>(),
3724            );
3725            if res != sys::cudaError_enum::CUDA_SUCCESS {
3726                return Err(XlogError::Kernel(format!(
3727                    "{ctx}: copy W column failed: {res:?}"
3728                )));
3729            }
3730        }
3731        let mut d_num_rows = self.memory().alloc::<u32>(1)?;
3732        self.device()
3733            .inner()
3734            .dtod_copy(e1.num_rows_device(), &mut d_num_rows)
3735            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy row count failed: {e}")))?;
3736        let staging_schema = Schema::new(vec![
3737            ("w".to_string(), w_type),
3738            ("count".to_string(), ScalarType::U32),
3739            ("agg".to_string(), agg_scalar),
3740        ]);
3741        let staging = CudaBuffer::from_columns_with_host_count(
3742            vec![w_copy.into(), row_counts.into(), row_agg.into()],
3743            n_e1 as u64,
3744            d_num_rows,
3745            staging_schema,
3746            n_e1,
3747        );
3748
3749        // Keep only roots with at least one completed 4-cycle, then reduce
3750        // per W with the same AggOp. Both steps run over input-sized data.
3751        let mask = self.compare_const_mask_recorded::<u32>(
3752            &staging,
3753            1,
3754            0u32,
3755            crate::CompareOp::Gt,
3756            launch_stream,
3757        )?;
3758        let compacted =
3759            self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)?;
3760        self.groupby_multi_agg_recorded(&compacted, &[0], &[(2, agg_op)], launch_stream)
3761    }
3762
3763    /// U64-key aggregate-fused 4-cycle count sibling of
3764    /// [`Self::wcoj_4cycle_groupby_root_count_u32_recorded`]: evaluate
3765    /// `q(W, count) :- e1(W,X), e2(X,Y), e3(Y,Z), e4(Z,W)` grouped by the
3766    /// variable-order root W for U64 relations, WITHOUT materializing the
3767    /// 4-cycle rows.
3768    ///
3769    /// The recorded groupby is U32/Symbol-key only, so the per-W reduction
3770    /// reuses the WCOJ relation metadata instead (mirroring
3771    /// [`Self::wcoj_triangle_groupby_root_count_u64_recorded`]): e1 is
3772    /// lex-sorted, so `wcoj_build_metadata_u64_recorded` yields one
3773    /// (unique W, group start) pair per root, and
3774    /// `wcoj_groupby_root_segment_sum_counts_u32` accumulates the per-row
3775    /// match counts into per-unique-root u64 totals (integer atomicAdd —
3776    /// deterministic). Roots with zero completions are compacted away.
3777    /// All reduction work is O(n_e1).
3778    ///
3779    /// Output schema matches the unfused materialize+groupby baseline:
3780    /// `col0` = W (U64), `col1` = count (U64).
3781    ///
3782    /// # Errors
3783    /// * `XlogError::Kernel` if the manager has no runtime, the launch
3784    ///   stream does not resolve, an input is not 2-column U64, or any
3785    ///   kernel launch fails.
3786    pub fn wcoj_4cycle_groupby_root_count_u64_recorded(
3787        &self,
3788        e1: &CudaBuffer,
3789        e2: &CudaBuffer,
3790        e3: &CudaBuffer,
3791        e4: &CudaBuffer,
3792        block_work_unit: u32,
3793        launch_stream: StreamId,
3794    ) -> Result<CudaBuffer> {
3795        let ctx = "wcoj_4cycle_groupby_root_count_u64_recorded";
3796        // Layout-normalize per dispatch (sorted-fast-path clone when the
3797        // input is already lex-sorted + unique): the fused path must give
3798        // the same guarantee as the unfused pipeline instead of trusting
3799        // store-buffer sortedness — unsorted/duplicated inputs previously
3800        // produced silently wrong (empty) fused results.
3801        let e1 = &self.wcoj_layout_u64_recorded(e1, launch_stream)?;
3802        let e2 = &self.wcoj_layout_u64_recorded(e2, launch_stream)?;
3803        let e3 = &self.wcoj_layout_u64_recorded(e3, launch_stream)?;
3804        let e4 = &self.wcoj_layout_u64_recorded(e4, launch_stream)?;
3805        validate_binary_u64(ctx, "e1", e1)?;
3806        validate_binary_u64(ctx, "e2", e2)?;
3807        validate_binary_u64(ctx, "e3", e3)?;
3808        validate_binary_u64(ctx, "e4", e4)?;
3809        let plan = self.wcoj_4cycle_hg_work_plan_u64_recorded(
3810            e1,
3811            e2,
3812            e3,
3813            e4,
3814            block_work_unit,
3815            launch_stream,
3816        )?;
3817        let n_e1 = plan.row_count;
3818        let out_schema = Schema::new(vec![
3819            ("w".to_string(), ScalarType::U64),
3820            ("count".to_string(), ScalarType::U64),
3821        ]);
3822        if n_e1 == 0 || plan.total_work == 0 {
3823            return self.create_empty_buffer(out_schema);
3824        }
3825
3826        let runtime = self.memory().runtime().ok_or_else(|| {
3827            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
3828        })?;
3829        let cu_stream = runtime
3830            .stream_pool()
3831            .resolve(launch_stream)
3832            .ok_or_else(|| {
3833                XlogError::Kernel(format!(
3834                    "{ctx}: launch_stream StreamId({}) does not resolve",
3835                    launch_stream.0
3836                ))
3837            })?;
3838
3839        let e1_col0 = metadata_column_u64(e1, 0)?;
3840        let e1_col1 = metadata_column_u64(e1, 1)?;
3841        let e2_col1 = metadata_column_u64(e2, 1)?;
3842        let e3_col0 = metadata_column_u64(e3, 0)?;
3843        let e3_col1 = metadata_column_u64(e3, 1)?;
3844        let e4_col0 = metadata_column_u64(e4, 0)?;
3845        let e4_col1 = metadata_column_u64(e4, 1)?;
3846        let n_e3 = self.metadata_logical_rows(e3)?;
3847        let n_e4 = self.metadata_logical_rows(e4)?;
3848
3849        // Per-e1-row match counters, zero-initialized.
3850        let mut row_counts = self.memory().alloc::<u32>(n_e1 as usize)?;
3851        self.device()
3852            .inner()
3853            .memset_zeros(&mut row_counts)
3854            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero row counts failed: {e}")))?;
3855
3856        let grid = plan.total_work.div_ceil(plan.block_work_unit);
3857        let mut rec = LaunchRecorder::new_strict(launch_stream);
3858        rec.read(e1.num_rows_device());
3859        rec.read(e2.num_rows_device());
3860        rec.read(e3.num_rows_device());
3861        rec.read(e4.num_rows_device());
3862        rec.read_column(e1.column(0).expect("e1.col0"));
3863        rec.read_column(e1.column(1).expect("e1.col1"));
3864        rec.read_column(e2.column(1).expect("e2.col1"));
3865        rec.read_column(e3.column(0).expect("e3.col0"));
3866        rec.read_column(e3.column(1).expect("e3.col1"));
3867        rec.read_column(e4.column(0).expect("e4.col0"));
3868        rec.read_column(e4.column(1).expect("e4.col1"));
3869        rec.read(&plan.e1_work_prefix);
3870        rec.read(&plan.e2_work_prefix);
3871        rec.read(&plan.e1_e2_start);
3872        rec.read(&plan.e1_e2_end);
3873        rec.write(&row_counts);
3874        rec.preflight(runtime)
3875            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
3876        {
3877            let kernel = self
3878                .device()
3879                .inner()
3880                .get_func(
3881                    WCOJ_MODULE,
3882                    wcoj_kernels::WCOJ_4CYCLE_GROUPBY_ROOT_COUNT_HG_U64,
3883                )
3884                .ok_or_else(|| {
3885                    XlogError::Kernel(
3886                        "wcoj_4cycle_groupby_root_count_hg_u64 kernel not found".to_string(),
3887                    )
3888                })?;
3889            let mut params: Vec<*mut c_void> = vec![
3890                e1_col0.as_kernel_param(),
3891                e1_col1.as_kernel_param(),
3892                plan.row_count.as_kernel_param(),
3893                e2_col1.as_kernel_param(),
3894                e3_col0.as_kernel_param(),
3895                e3_col1.as_kernel_param(),
3896                n_e3.as_kernel_param(),
3897                e4_col0.as_kernel_param(),
3898                e4_col1.as_kernel_param(),
3899                n_e4.as_kernel_param(),
3900                (&plan.e1_work_prefix).as_kernel_param(),
3901                (&plan.e2_work_prefix).as_kernel_param(),
3902                (&plan.e1_e2_start).as_kernel_param(),
3903                (&plan.e1_e2_end).as_kernel_param(),
3904                plan.total_work.as_kernel_param(),
3905                plan.block_work_unit.as_kernel_param(),
3906                (&row_counts).as_kernel_param(),
3907            ];
3908            unsafe {
3909                kernel
3910                    .clone()
3911                    .launch_on_stream(
3912                        &cu_stream,
3913                        LaunchConfig {
3914                            grid_dim: (grid, 1, 1),
3915                            block_dim: (BLOCK_SIZE, 1, 1),
3916                            shared_mem_bytes: 0,
3917                        },
3918                        &mut params,
3919                    )
3920                    .map_err(|e| {
3921                        XlogError::Kernel(format!("{ctx}: groupby-count launch failed: {e}"))
3922                    })?;
3923            }
3924        }
3925        rec.commit(runtime)
3926            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
3927
3928        // Per-W reduction via the relation metadata: one (unique W, group
3929        // start) pair per root; e1 is lex-sorted by W so group rows are
3930        // contiguous.
3931        let meta = self.wcoj_build_metadata_u64_recorded(e1, 0, launch_stream)?;
3932        let key_count = meta.key_count;
3933        if key_count == 0 {
3934            return self.create_empty_buffer(out_schema);
3935        }
3936        let mut sums = self
3937            .memory()
3938            .alloc::<u8>(key_count as usize * std::mem::size_of::<u64>())?;
3939        self.device()
3940            .inner()
3941            .memset_zeros(&mut sums)
3942            .map_err(|e| XlogError::Kernel(format!("{ctx}: zero group sums failed: {e}")))?;
3943
3944        let mut rec_sum = LaunchRecorder::new_strict(launch_stream);
3945        rec_sum.read(&row_counts);
3946        rec_sum.read(&meta.prefix_sum);
3947        rec_sum.write(&sums);
3948        rec_sum
3949            .preflight(runtime)
3950            .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce preflight failed: {e}")))?;
3951        {
3952            let kernel = self
3953                .device()
3954                .inner()
3955                .get_func(
3956                    WCOJ_MODULE,
3957                    wcoj_kernels::WCOJ_GROUPBY_ROOT_SEGMENT_SUM_COUNTS_U32,
3958                )
3959                .ok_or_else(|| {
3960                    XlogError::Kernel(
3961                        "wcoj_groupby_root_segment_sum_counts_u32 kernel not found".to_string(),
3962                    )
3963                })?;
3964            let reduce_grid = n_e1.div_ceil(BLOCK_SIZE);
3965            let mut params: Vec<*mut c_void> = vec![
3966                (&row_counts).as_kernel_param(),
3967                n_e1.as_kernel_param(),
3968                (&meta.prefix_sum).as_kernel_param(),
3969                key_count.as_kernel_param(),
3970                (&sums).as_kernel_param(),
3971            ];
3972            unsafe {
3973                kernel
3974                    .clone()
3975                    .launch_on_stream(
3976                        &cu_stream,
3977                        LaunchConfig {
3978                            grid_dim: (reduce_grid, 1, 1),
3979                            block_dim: (BLOCK_SIZE, 1, 1),
3980                            shared_mem_bytes: 0,
3981                        },
3982                        &mut params,
3983                    )
3984                    .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce launch failed: {e}")))?;
3985            }
3986        }
3987        rec_sum
3988            .commit(runtime)
3989            .map_err(|e| XlogError::Kernel(format!("{ctx}: reduce commit failed: {e}")))?;
3990
3991        // (unique W, total) buffer over the key_count roots, then drop the
3992        // roots with no completion. The copies run on launch_stream and the
3993        // fresh destination blocks are registered through the strict
3994        // recorder BEFORE the enqueue — a raw async copy into a freshly
3995        // pool-allocated block without recording is a visibility race.
3996        let w_copy = self
3997            .memory()
3998            .alloc::<u8>(key_count as usize * std::mem::size_of::<u64>())?;
3999        let d_num_rows = self.memory().alloc::<u32>(1)?;
4000        let mut rec_copy = LaunchRecorder::new_strict(launch_stream);
4001        rec_copy.read(&meta.unique_keys);
4002        rec_copy.write(&w_copy);
4003        rec_copy.write(&d_num_rows);
4004        rec_copy
4005            .preflight(runtime)
4006            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy preflight failed: {e}")))?;
4007        unsafe {
4008            let res = sys::cuMemcpyDtoDAsync_v2(
4009                *w_copy.device_ptr(),
4010                *meta.unique_keys.device_ptr(),
4011                key_count as usize * std::mem::size_of::<u64>(),
4012                cu_stream.cu_stream(),
4013            );
4014            if res != sys::cudaError_enum::CUDA_SUCCESS {
4015                return Err(XlogError::Kernel(format!(
4016                    "{ctx}: DtoD unique keys copy failed: {res:?}"
4017                )));
4018            }
4019        }
4020        self.htod_launch_metadata_async_copy_one(
4021            &key_count,
4022            &d_num_rows,
4023            &cu_stream,
4024            &format!("{ctx}: d_num_rows"),
4025        )?;
4026        rec_copy
4027            .commit(runtime)
4028            .map_err(|e| XlogError::Kernel(format!("{ctx}: copy commit failed: {e}")))?;
4029        cu_stream
4030            .synchronize()
4031            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
4032        let staging_schema = Schema::new(vec![
4033            ("w".to_string(), ScalarType::U64),
4034            ("count".to_string(), ScalarType::U64),
4035        ]);
4036        let staging = CudaBuffer::from_columns_with_host_count(
4037            vec![w_copy.into(), sums.into()],
4038            u64::from(key_count),
4039            d_num_rows,
4040            staging_schema,
4041            key_count,
4042        );
4043        let mask = self.compare_const_mask_recorded::<u64>(
4044            &staging,
4045            1,
4046            0u64,
4047            crate::CompareOp::Gt,
4048            launch_stream,
4049        )?;
4050        self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)
4051    }
4052
4053    pub fn wcoj_4cycle_hg_work_plan_u64_recorded(
4054        &self,
4055        e1: &CudaBuffer,
4056        e2: &CudaBuffer,
4057        e3: &CudaBuffer,
4058        e4: &CudaBuffer,
4059        block_work_unit: u32,
4060        launch_stream: StreamId,
4061    ) -> Result<WcojCycle4HgWorkPlanU64> {
4062        let ctx = "wcoj_4cycle_hg_work_plan_u64_recorded";
4063        if block_work_unit == 0 {
4064            return Err(XlogError::Kernel(format!(
4065                "{ctx}: block_work_unit must be nonzero"
4066            )));
4067        }
4068        validate_binary_u64(ctx, "e1", e1)?;
4069        validate_binary_u64(ctx, "e2", e2)?;
4070        validate_binary_u64(ctx, "e3", e3)?;
4071        validate_binary_u64(ctx, "e4", e4)?;
4072
4073        let n_e1 = self.metadata_logical_rows(e1)?;
4074        let n_e2 = self.metadata_logical_rows(e2)?;
4075        let n_e3 = self.metadata_logical_rows(e3)?;
4076        let prefix_len = n_e1
4077            .checked_add(1)
4078            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: prefix length overflow")))?;
4079        let e2_prefix_len = n_e2
4080            .checked_add(1)
4081            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: e2 prefix length overflow")))?;
4082        let mut e1_work_prefix = self.memory().alloc::<u32>(prefix_len as usize)?;
4083        let mut e2_work_prefix = self.memory().alloc::<u32>(e2_prefix_len as usize)?;
4084        let mut e1_e2_start = self.memory().alloc::<u32>(n_e1 as usize)?;
4085        let mut e1_e2_end = self.memory().alloc::<u32>(n_e1 as usize)?;
4086
4087        if n_e1 == 0 || n_e2 == 0 || n_e3 == 0 || self.metadata_logical_rows(e4)? == 0 {
4088            let block_counts = self.memory().alloc::<u32>(1)?;
4089            let block_offsets = self.memory().alloc::<u32>(1)?;
4090            return Ok(WcojCycle4HgWorkPlanU64 {
4091                e1_work_prefix,
4092                e2_work_prefix,
4093                e1_e2_start,
4094                e1_e2_end,
4095                block_counts,
4096                block_offsets,
4097                total_work: 0,
4098                block_work_unit,
4099                row_count: n_e1,
4100            });
4101        }
4102
4103        let runtime = self.memory().runtime().ok_or_else(|| {
4104            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
4105        })?;
4106        let cu_stream = runtime
4107            .stream_pool()
4108            .resolve(launch_stream)
4109            .ok_or_else(|| {
4110                XlogError::Kernel(format!(
4111                    "{ctx}: launch_stream StreamId({}) does not resolve",
4112                    launch_stream.0
4113                ))
4114            })?;
4115
4116        let e1_col1 = metadata_column_u64(e1, 1)?;
4117        let e2_col0 = metadata_column_u64(e2, 0)?;
4118        let e2_col1 = metadata_column_u64(e2, 1)?;
4119        let e3_col0 = metadata_column_u64(e3, 0)?;
4120
4121        let mut rec = LaunchRecorder::new_strict(launch_stream);
4122        rec.read(e1.num_rows_device());
4123        rec.read(e2.num_rows_device());
4124        rec.read(e3.num_rows_device());
4125        rec.read_column(e1.column(1).expect("e1.col1"));
4126        rec.read_column(e2.column(0).expect("e2.col0"));
4127        rec.read_column(e2.column(1).expect("e2.col1"));
4128        rec.read_column(e3.column(0).expect("e3.col0"));
4129        rec.read_write(&e2_work_prefix);
4130        rec.write(&e1_work_prefix);
4131        rec.write(&e1_e2_start);
4132        rec.write(&e1_e2_end);
4133        rec.preflight(runtime)
4134            .map_err(|e| XlogError::Kernel(format!("{ctx}: preflight failed: {e}")))?;
4135
4136        let e2_kernel = self
4137            .device()
4138            .inner()
4139            .get_func(
4140                WCOJ_MODULE,
4141                wcoj_kernels::WCOJ_4CYCLE_BUILD_E2_WORK_PREFIX_U64,
4142            )
4143            .ok_or_else(|| {
4144                XlogError::Kernel(
4145                    "wcoj_4cycle_build_e2_work_prefix_u64 kernel not found".to_string(),
4146                )
4147            })?;
4148        let e2_grid = n_e2.div_ceil(BLOCK_SIZE);
4149        unsafe {
4150            e2_kernel
4151                .clone()
4152                .launch_on_stream(
4153                    &cu_stream,
4154                    LaunchConfig {
4155                        grid_dim: (e2_grid, 1, 1),
4156                        block_dim: (BLOCK_SIZE, 1, 1),
4157                        shared_mem_bytes: 0,
4158                    },
4159                    (e2_col1, n_e2, e3_col0, n_e3, &mut e2_work_prefix),
4160                )
4161                .map_err(|e| {
4162                    XlogError::Kernel(format!(
4163                        "wcoj_4cycle_build_e2_work_prefix_u64 launch failed: {e}"
4164                    ))
4165                })?;
4166        }
4167        self.multiblock_scan_u32_inplace_on_stream(
4168            &mut e2_work_prefix,
4169            e2_prefix_len,
4170            &cu_stream,
4171            launch_stream,
4172            runtime,
4173        )?;
4174
4175        let kernel = self
4176            .device()
4177            .inner()
4178            .get_func(
4179                WCOJ_MODULE,
4180                wcoj_kernels::WCOJ_4CYCLE_BUILD_HG_WORK_PLAN_U64,
4181            )
4182            .ok_or_else(|| {
4183                XlogError::Kernel("wcoj_4cycle_build_hg_work_plan_u64 kernel not found".to_string())
4184            })?;
4185        let grid = n_e1.div_ceil(BLOCK_SIZE);
4186        unsafe {
4187            kernel
4188                .clone()
4189                .launch_on_stream(
4190                    &cu_stream,
4191                    LaunchConfig {
4192                        grid_dim: (grid, 1, 1),
4193                        block_dim: (BLOCK_SIZE, 1, 1),
4194                        shared_mem_bytes: 0,
4195                    },
4196                    (
4197                        e1_col1,
4198                        n_e1,
4199                        e2_col0,
4200                        n_e2,
4201                        &e2_work_prefix,
4202                        &mut e1_work_prefix,
4203                        &mut e1_e2_start,
4204                        &mut e1_e2_end,
4205                    ),
4206                )
4207                .map_err(|e| {
4208                    XlogError::Kernel(format!(
4209                        "wcoj_4cycle_build_hg_work_plan_u64 launch failed: {e}"
4210                    ))
4211                })?;
4212        }
4213        self.multiblock_scan_u32_inplace_on_stream(
4214            &mut e1_work_prefix,
4215            prefix_len,
4216            &cu_stream,
4217            launch_stream,
4218            runtime,
4219        )?;
4220        rec.commit(runtime)
4221            .map_err(|e| XlogError::Kernel(format!("{ctx}: commit failed: {e}")))?;
4222        cu_stream
4223            .synchronize()
4224            .map_err(|e| XlogError::Kernel(format!("{ctx}: stream sync failed: {e}")))?;
4225        let total_work = self.dtoh_scalar_untracked::<u32>(&e1_work_prefix, n_e1 as usize)?;
4226        let grid = if total_work == 0 {
4227            1
4228        } else {
4229            total_work.div_ceil(block_work_unit)
4230        };
4231        let block_counts = self.memory().alloc::<u32>(grid as usize)?;
4232        let block_offsets = self.memory().alloc::<u32>(grid as usize)?;
4233
4234        Ok(WcojCycle4HgWorkPlanU64 {
4235            e1_work_prefix,
4236            e2_work_prefix,
4237            e1_e2_start,
4238            e1_e2_end,
4239            block_counts,
4240            block_offsets,
4241            total_work,
4242            block_work_unit,
4243            row_count: n_e1,
4244        })
4245    }
4246
4247    pub fn wcoj_4cycle_hg_u64_recorded(
4248        &self,
4249        e1: &CudaBuffer,
4250        e2: &CudaBuffer,
4251        e3: &CudaBuffer,
4252        e4: &CudaBuffer,
4253        block_work_unit: u32,
4254        launch_stream: StreamId,
4255    ) -> Result<CudaBuffer> {
4256        let ctx = "wcoj_4cycle_hg_u64_recorded";
4257        validate_binary_u64(ctx, "e1", e1)?;
4258        validate_binary_u64(ctx, "e2", e2)?;
4259        validate_binary_u64(ctx, "e3", e3)?;
4260        validate_binary_u64(ctx, "e4", e4)?;
4261        let plan = self.wcoj_4cycle_hg_work_plan_u64_recorded(
4262            e1,
4263            e2,
4264            e3,
4265            e4,
4266            block_work_unit,
4267            launch_stream,
4268        )?;
4269        let out_schema = Schema::new(vec![
4270            ("col0".to_string(), ScalarType::U64),
4271            ("col1".to_string(), ScalarType::U64),
4272            ("col2".to_string(), ScalarType::U64),
4273            ("col3".to_string(), ScalarType::U64),
4274        ]);
4275        if plan.total_work == 0 {
4276            return self.create_empty_buffer(out_schema);
4277        }
4278
4279        let grid = plan.total_work.div_ceil(plan.block_work_unit);
4280        let bytes_count = (grid as usize)
4281            .checked_mul(std::mem::size_of::<u32>())
4282            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: count byte size overflow")))?;
4283        let mut local_counts = None;
4284        let mut local_offsets = None;
4285        if grid > 1024 {
4286            local_counts = Some(self.memory().alloc::<u32>(grid as usize)?);
4287            local_offsets = Some(self.memory().alloc::<u32>(grid as usize)?);
4288        }
4289        let total_rows_device = self.memory().alloc::<u32>(1)?;
4290
4291        let runtime = self.memory().runtime().ok_or_else(|| {
4292            XlogError::Kernel(format!("{ctx} requires a runtime-backed GpuMemoryManager"))
4293        })?;
4294        let cu_stream = runtime
4295            .stream_pool()
4296            .resolve(launch_stream)
4297            .ok_or_else(|| {
4298                XlogError::Kernel(format!(
4299                    "{ctx}: launch_stream StreamId({}) does not resolve",
4300                    launch_stream.0
4301                ))
4302            })?;
4303
4304        let e1_col0 = metadata_column_u64(e1, 0)?;
4305        let e1_col1 = metadata_column_u64(e1, 1)?;
4306        let e2_col1 = metadata_column_u64(e2, 1)?;
4307        let e3_col0 = metadata_column_u64(e3, 0)?;
4308        let e3_col1 = metadata_column_u64(e3, 1)?;
4309        let e4_col0 = metadata_column_u64(e4, 0)?;
4310        let e4_col1 = metadata_column_u64(e4, 1)?;
4311        let n_e3 = self.metadata_logical_rows(e3)?;
4312        let n_e4 = self.metadata_logical_rows(e4)?;
4313
4314        let count_u32 = if grid <= 1024 {
4315            &plan.block_counts
4316        } else {
4317            local_counts
4318                .as_ref()
4319                .expect("local HG counts allocated when grid exceeds single-block scan")
4320        };
4321        let mut rec_hg = LaunchRecorder::new_strict(launch_stream);
4322        rec_hg.read(e1.num_rows_device());
4323        rec_hg.read(e2.num_rows_device());
4324        rec_hg.read(e3.num_rows_device());
4325        rec_hg.read(e4.num_rows_device());
4326        rec_hg.read_column(e1.column(0).expect("e1.col0"));
4327        rec_hg.read_column(e1.column(1).expect("e1.col1"));
4328        rec_hg.read_column(e2.column(1).expect("e2.col1"));
4329        rec_hg.read_column(e3.column(0).expect("e3.col0"));
4330        rec_hg.read_column(e3.column(1).expect("e3.col1"));
4331        rec_hg.read_column(e4.column(0).expect("e4.col0"));
4332        rec_hg.read_column(e4.column(1).expect("e4.col1"));
4333        rec_hg.read(&plan.e1_work_prefix);
4334        rec_hg.read(&plan.e2_work_prefix);
4335        rec_hg.read(&plan.e1_e2_start);
4336        rec_hg.read(&plan.e1_e2_end);
4337        rec_hg.read_write(count_u32);
4338        if grid <= 1024 {
4339            rec_hg.read_write(&plan.block_offsets);
4340        } else {
4341            rec_hg.read_write(
4342                local_offsets
4343                    .as_ref()
4344                    .expect("local HG offsets allocated when grid exceeds single-block scan"),
4345            );
4346        }
4347        rec_hg.write(&total_rows_device);
4348        rec_hg
4349            .preflight(runtime)
4350            .map_err(|e| XlogError::Kernel(format!("{ctx}: HG preflight failed: {e}")))?;
4351        {
4352            let kernel = self
4353                .device()
4354                .inner()
4355                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_4CYCLE_COUNT_HG_U64)
4356                .ok_or_else(|| {
4357                    XlogError::Kernel("wcoj_4cycle_count_hg_u64 kernel not found".to_string())
4358                })?;
4359            let mut params: Vec<*mut c_void> = vec![
4360                e1_col0.as_kernel_param(),
4361                e1_col1.as_kernel_param(),
4362                plan.row_count.as_kernel_param(),
4363                e2_col1.as_kernel_param(),
4364                e3_col0.as_kernel_param(),
4365                e3_col1.as_kernel_param(),
4366                n_e3.as_kernel_param(),
4367                e4_col0.as_kernel_param(),
4368                e4_col1.as_kernel_param(),
4369                n_e4.as_kernel_param(),
4370                (&plan.e1_work_prefix).as_kernel_param(),
4371                (&plan.e2_work_prefix).as_kernel_param(),
4372                (&plan.e1_e2_start).as_kernel_param(),
4373                (&plan.e1_e2_end).as_kernel_param(),
4374                plan.total_work.as_kernel_param(),
4375                plan.block_work_unit.as_kernel_param(),
4376                count_u32.as_kernel_param(),
4377            ];
4378            unsafe {
4379                kernel
4380                    .clone()
4381                    .launch_on_stream(
4382                        &cu_stream,
4383                        LaunchConfig {
4384                            grid_dim: (grid, 1, 1),
4385                            block_dim: (BLOCK_SIZE, 1, 1),
4386                            shared_mem_bytes: 0,
4387                        },
4388                        &mut params,
4389                    )
4390                    .map_err(|e| XlogError::Kernel(format!("{ctx}: count launch failed: {e}")))?;
4391            }
4392        }
4393        if grid <= 1024 {
4394            let kernel = self
4395                .device()
4396                .inner()
4397                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_SCAN_HG_BLOCK_COUNTS_U32)
4398                .ok_or_else(|| {
4399                    XlogError::Kernel("wcoj_scan_hg_block_counts_u32 kernel not found".to_string())
4400                })?;
4401            let mut params: Vec<*mut c_void> = vec![
4402                count_u32.as_kernel_param(),
4403                grid.as_kernel_param(),
4404                (&plan.block_offsets).as_kernel_param(),
4405                (&total_rows_device).as_kernel_param(),
4406            ];
4407            unsafe {
4408                kernel
4409                    .clone()
4410                    .launch_on_stream(
4411                        &cu_stream,
4412                        LaunchConfig {
4413                            grid_dim: (1, 1, 1),
4414                            block_dim: (1024, 1, 1),
4415                            shared_mem_bytes: 0,
4416                        },
4417                        &mut params,
4418                    )
4419                    .map_err(|e| XlogError::Kernel(format!("{ctx}: scan failed: {e}")))?;
4420            }
4421        } else {
4422            let offsets_mut = local_offsets
4423                .as_mut()
4424                .expect("local HG offsets allocated when grid exceeds single-block scan");
4425            unsafe {
4426                let res = sys::cuMemcpyDtoDAsync_v2(
4427                    *offsets_mut.device_ptr(),
4428                    *count_u32.device_ptr(),
4429                    bytes_count,
4430                    cu_stream.cu_stream(),
4431                );
4432                if res != sys::cudaError_enum::CUDA_SUCCESS {
4433                    return Err(XlogError::Kernel(format!(
4434                        "{ctx}: DtoD count to offsets failed: {res:?}"
4435                    )));
4436                }
4437            }
4438            self.multiblock_scan_u32_inplace_on_stream(
4439                offsets_mut,
4440                grid,
4441                &cu_stream,
4442                launch_stream,
4443                runtime,
4444            )?;
4445            let total_kernel = self
4446                .device()
4447                .inner()
4448                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_COMPUTE_TOTAL)
4449                .ok_or_else(|| {
4450                    XlogError::Kernel("wcoj_compute_total kernel not found".to_string())
4451                })?;
4452            let mut params: Vec<*mut c_void> = vec![
4453                count_u32.as_kernel_param(),
4454                (&*offsets_mut).as_kernel_param(),
4455                grid.as_kernel_param(),
4456                (&total_rows_device).as_kernel_param(),
4457            ];
4458            unsafe {
4459                total_kernel
4460                    .clone()
4461                    .launch_on_stream(
4462                        &cu_stream,
4463                        LaunchConfig {
4464                            grid_dim: (1, 1, 1),
4465                            block_dim: (1, 1, 1),
4466                            shared_mem_bytes: 0,
4467                        },
4468                        &mut params,
4469                    )
4470                    .map_err(|e| XlogError::Kernel(format!("{ctx}: total failed: {e}")))?;
4471            }
4472        }
4473        rec_hg
4474            .commit(runtime)
4475            .map_err(|e| XlogError::Kernel(format!("{ctx}: count commit failed: {e}")))?;
4476        cu_stream
4477            .synchronize()
4478            .map_err(|e| XlogError::Kernel(format!("{ctx}: count stream sync failed: {e}")))?;
4479        let total_rows = self
4480            .dtoh_scalar_untracked::<u32>(&total_rows_device, 0)
4481            .map_err(|e| XlogError::Kernel(format!("{ctx}: read total rows failed: {e}")))?;
4482        if total_rows == 0 {
4483            return self.create_empty_buffer(out_schema);
4484        }
4485
4486        let bytes_per_col = (total_rows as usize)
4487            .checked_mul(std::mem::size_of::<u64>())
4488            .ok_or_else(|| XlogError::Kernel(format!("{ctx}: output byte size overflow")))?;
4489        let mut out_w = self.memory().alloc::<u8>(bytes_per_col)?;
4490        let mut out_x = self.memory().alloc::<u8>(bytes_per_col)?;
4491        let mut out_y = self.memory().alloc::<u8>(bytes_per_col)?;
4492        let mut out_z = self.memory().alloc::<u8>(bytes_per_col)?;
4493        let materialize_offsets = if grid <= 1024 {
4494            &plan.block_offsets
4495        } else {
4496            local_offsets
4497                .as_ref()
4498                .expect("local HG offsets allocated when grid exceeds single-block scan")
4499        };
4500        let mut rec_mat = LaunchRecorder::new_strict(launch_stream);
4501        rec_mat.read(materialize_offsets);
4502        rec_mat.read(e1.num_rows_device());
4503        rec_mat.read(e2.num_rows_device());
4504        rec_mat.read(e3.num_rows_device());
4505        rec_mat.read(e4.num_rows_device());
4506        rec_mat.read_column(e1.column(0).expect("e1.col0"));
4507        rec_mat.read_column(e1.column(1).expect("e1.col1"));
4508        rec_mat.read_column(e2.column(1).expect("e2.col1"));
4509        rec_mat.read_column(e3.column(0).expect("e3.col0"));
4510        rec_mat.read_column(e3.column(1).expect("e3.col1"));
4511        rec_mat.read_column(e4.column(0).expect("e4.col0"));
4512        rec_mat.read_column(e4.column(1).expect("e4.col1"));
4513        rec_mat.read(&plan.e1_work_prefix);
4514        rec_mat.read(&plan.e2_work_prefix);
4515        rec_mat.read(&plan.e1_e2_start);
4516        rec_mat.read(&plan.e1_e2_end);
4517        rec_mat.write(&out_w);
4518        rec_mat.write(&out_x);
4519        rec_mat.write(&out_y);
4520        rec_mat.write(&out_z);
4521        rec_mat
4522            .preflight(runtime)
4523            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize preflight failed: {e}")))?;
4524        {
4525            let kernel = self
4526                .device()
4527                .inner()
4528                .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_4CYCLE_MATERIALIZE_HG_U64)
4529                .ok_or_else(|| {
4530                    XlogError::Kernel("wcoj_4cycle_materialize_hg_u64 kernel not found".to_string())
4531                })?;
4532            let out_w_u64 = unsafe { reinterpret_u8_as_u64(&mut out_w) };
4533            let out_x_u64 = unsafe { reinterpret_u8_as_u64(&mut out_x) };
4534            let out_y_u64 = unsafe { reinterpret_u8_as_u64(&mut out_y) };
4535            let out_z_u64 = unsafe { reinterpret_u8_as_u64(&mut out_z) };
4536            let mut params: Vec<*mut c_void> = vec![
4537                e1_col0.as_kernel_param(),
4538                e1_col1.as_kernel_param(),
4539                plan.row_count.as_kernel_param(),
4540                e2_col1.as_kernel_param(),
4541                e3_col0.as_kernel_param(),
4542                e3_col1.as_kernel_param(),
4543                n_e3.as_kernel_param(),
4544                e4_col0.as_kernel_param(),
4545                e4_col1.as_kernel_param(),
4546                n_e4.as_kernel_param(),
4547                (&plan.e1_work_prefix).as_kernel_param(),
4548                (&plan.e2_work_prefix).as_kernel_param(),
4549                (&plan.e1_e2_start).as_kernel_param(),
4550                (&plan.e1_e2_end).as_kernel_param(),
4551                plan.total_work.as_kernel_param(),
4552                plan.block_work_unit.as_kernel_param(),
4553                materialize_offsets.as_kernel_param(),
4554                total_rows.as_kernel_param(),
4555                out_w_u64.as_kernel_param(),
4556                out_x_u64.as_kernel_param(),
4557                out_y_u64.as_kernel_param(),
4558                out_z_u64.as_kernel_param(),
4559            ];
4560            unsafe {
4561                kernel
4562                    .clone()
4563                    .launch_on_stream(
4564                        &cu_stream,
4565                        LaunchConfig {
4566                            grid_dim: (grid, 1, 1),
4567                            block_dim: (BLOCK_SIZE, 1, 1),
4568                            shared_mem_bytes: 0,
4569                        },
4570                        &mut params,
4571                    )
4572                    .map_err(|e| {
4573                        XlogError::Kernel(format!("{ctx}: materialize launch failed: {e}"))
4574                    })?;
4575            }
4576        }
4577        rec_mat
4578            .commit(runtime)
4579            .map_err(|e| XlogError::Kernel(format!("{ctx}: materialize commit failed: {e}")))?;
4580        cu_stream.synchronize().map_err(|e| {
4581            XlogError::Kernel(format!("{ctx}: materialize stream sync failed: {e}"))
4582        })?;
4583
4584        Ok(CudaBuffer::from_columns_with_host_count(
4585            vec![out_w.into(), out_x.into(), out_y.into(), out_z.into()],
4586            total_rows as u64,
4587            total_rows_device,
4588            out_schema,
4589            total_rows,
4590        ))
4591    }
4592
4593    fn validate_metadata_column(
4594        &self,
4595        input: &CudaBuffer,
4596        key_col_idx: usize,
4597        width: MetadataWidth,
4598    ) -> Result<()> {
4599        if key_col_idx >= input.arity() {
4600            return Err(XlogError::Kernel(format!(
4601                "wcoj_build_metadata: key column {} out of range for arity {}",
4602                key_col_idx,
4603                input.arity()
4604            )));
4605        }
4606        let ty = input.schema().column_type(key_col_idx).ok_or_else(|| {
4607            XlogError::Kernel(format!(
4608                "wcoj_build_metadata: column {} type missing",
4609                key_col_idx
4610            ))
4611        })?;
4612        match (width, ty) {
4613            (MetadataWidth::U32, ScalarType::U32 | ScalarType::Symbol)
4614            | (MetadataWidth::U64, ScalarType::U64) => Ok(()),
4615            (MetadataWidth::U32, other) => Err(XlogError::Kernel(format!(
4616                "wcoj_build_metadata_u32_recorded: column {} must be U32 or Symbol, got {:?}",
4617                key_col_idx, other
4618            ))),
4619            (MetadataWidth::U64, other) => Err(XlogError::Kernel(format!(
4620                "wcoj_build_metadata_u64_recorded: column {} must be U64, got {:?}",
4621                key_col_idx, other
4622            ))),
4623        }
4624    }
4625
4626    fn build_metadata_u32_from_column(
4627        &self,
4628        input: &CudaBuffer,
4629        key_col_idx: usize,
4630        keys: &TrackedCudaSlice<u32>,
4631        launch_stream: StreamId,
4632    ) -> Result<WcojRelationMetadata<u32>> {
4633        let n = self.metadata_logical_rows(input)?;
4634        if n == 0 {
4635            return Ok(WcojRelationMetadata {
4636                unique_keys: self.memory().alloc::<u32>(0)?,
4637                fan_out: self.memory().alloc::<u32>(0)?,
4638                prefix_sum: self.memory().alloc::<u32>(0)?,
4639                per_candidate_root: BTreeMap::new(),
4640                total: 0,
4641                key_count: 0,
4642                row_count: 0,
4643            });
4644        }
4645        let mut boundary_mask = self.memory().alloc::<u32>(n as usize)?;
4646        let mut boundary_prefix = self.memory().alloc::<u32>(n as usize)?;
4647        self.mark_metadata_boundaries_u32(
4648            input,
4649            key_col_idx,
4650            keys,
4651            n,
4652            &mut boundary_mask,
4653            &mut boundary_prefix,
4654            launch_stream,
4655        )?;
4656        let key_count = self.metadata_scanned_count(&boundary_mask, &boundary_prefix, n)?;
4657
4658        let mut unique_keys = self.memory().alloc::<u32>(key_count as usize)?;
4659        let mut fan_out = self.memory().alloc::<u32>(key_count as usize)?;
4660        let mut prefix_sum = self.memory().alloc::<u32>(key_count as usize)?;
4661        self.scatter_metadata_u32(
4662            input,
4663            key_col_idx,
4664            keys,
4665            n,
4666            &boundary_mask,
4667            &boundary_prefix,
4668            &mut unique_keys,
4669            &mut fan_out,
4670            &mut prefix_sum,
4671            launch_stream,
4672        )?;
4673
4674        Ok(WcojRelationMetadata {
4675            unique_keys,
4676            fan_out,
4677            prefix_sum,
4678            per_candidate_root: BTreeMap::new(),
4679            total: u64::from(n),
4680            key_count,
4681            row_count: n,
4682        })
4683    }
4684
4685    fn build_metadata_u64_from_column(
4686        &self,
4687        input: &CudaBuffer,
4688        key_col_idx: usize,
4689        keys: &TrackedCudaSlice<u64>,
4690        launch_stream: StreamId,
4691    ) -> Result<WcojRelationMetadata<u64>> {
4692        let n = self.metadata_logical_rows(input)?;
4693        if n == 0 {
4694            return Ok(WcojRelationMetadata {
4695                unique_keys: self.memory().alloc::<u64>(0)?,
4696                fan_out: self.memory().alloc::<u32>(0)?,
4697                prefix_sum: self.memory().alloc::<u32>(0)?,
4698                per_candidate_root: BTreeMap::new(),
4699                total: 0,
4700                key_count: 0,
4701                row_count: 0,
4702            });
4703        }
4704        let mut boundary_mask = self.memory().alloc::<u32>(n as usize)?;
4705        let mut boundary_prefix = self.memory().alloc::<u32>(n as usize)?;
4706        self.mark_metadata_boundaries_u64(
4707            input,
4708            key_col_idx,
4709            keys,
4710            n,
4711            &mut boundary_mask,
4712            &mut boundary_prefix,
4713            launch_stream,
4714        )?;
4715        let key_count = self.metadata_scanned_count(&boundary_mask, &boundary_prefix, n)?;
4716
4717        let mut unique_keys = self.memory().alloc::<u64>(key_count as usize)?;
4718        let mut fan_out = self.memory().alloc::<u32>(key_count as usize)?;
4719        let mut prefix_sum = self.memory().alloc::<u32>(key_count as usize)?;
4720        self.scatter_metadata_u64(
4721            input,
4722            key_col_idx,
4723            keys,
4724            n,
4725            &boundary_mask,
4726            &boundary_prefix,
4727            &mut unique_keys,
4728            &mut fan_out,
4729            &mut prefix_sum,
4730            launch_stream,
4731        )?;
4732
4733        Ok(WcojRelationMetadata {
4734            unique_keys,
4735            fan_out,
4736            prefix_sum,
4737            per_candidate_root: BTreeMap::new(),
4738            total: u64::from(n),
4739            key_count,
4740            row_count: n,
4741        })
4742    }
4743
4744    #[allow(clippy::too_many_arguments)]
4745    fn mark_metadata_boundaries_u32(
4746        &self,
4747        input: &CudaBuffer,
4748        key_col_idx: usize,
4749        keys: &TrackedCudaSlice<u32>,
4750        n: u32,
4751        boundary_mask: &mut TrackedCudaSlice<u32>,
4752        boundary_prefix: &mut TrackedCudaSlice<u32>,
4753        launch_stream: StreamId,
4754    ) -> Result<()> {
4755        let runtime = self.memory().runtime().ok_or_else(|| {
4756            XlogError::Kernel(
4757                "wcoj_build_metadata_u32_recorded requires a runtime-backed GpuMemoryManager"
4758                    .to_string(),
4759            )
4760        })?;
4761        let cu_stream = runtime
4762            .stream_pool()
4763            .resolve(launch_stream)
4764            .ok_or_else(|| {
4765                XlogError::Kernel(format!(
4766                    "wcoj_build_metadata_u32_recorded: launch_stream StreamId({}) does not resolve",
4767                    launch_stream.0
4768                ))
4769            })?;
4770        let mut rec = LaunchRecorder::new_strict(launch_stream);
4771        rec.read(input.num_rows_device());
4772        rec.read_column(input.column(key_col_idx).expect("metadata key column"));
4773        rec.write(boundary_mask);
4774        rec.write(boundary_prefix);
4775        rec.preflight(runtime).map_err(|e| {
4776            XlogError::Kernel(format!(
4777                "wcoj_build_metadata_u32_recorded: mark preflight failed: {e}"
4778            ))
4779        })?;
4780
4781        let kernel = self
4782            .device()
4783            .inner()
4784            .get_func(
4785                WCOJ_MODULE,
4786                wcoj_kernels::WCOJ_BUILD_METADATA_MARK_BOUNDARIES_U32,
4787            )
4788            .ok_or_else(|| {
4789                XlogError::Kernel(
4790                    "wcoj_build_metadata_mark_boundaries_u32 kernel not found".to_string(),
4791                )
4792            })?;
4793        let grid = n.div_ceil(BLOCK_SIZE);
4794        unsafe {
4795            kernel
4796                .clone()
4797                .launch_on_stream(
4798                    &cu_stream,
4799                    LaunchConfig {
4800                        grid_dim: (grid, 1, 1),
4801                        block_dim: (BLOCK_SIZE, 1, 1),
4802                        shared_mem_bytes: 0,
4803                    },
4804                    (keys, n, &mut *boundary_mask, &mut *boundary_prefix),
4805                )
4806                .map_err(|e| {
4807                    XlogError::Kernel(format!(
4808                        "wcoj_build_metadata_mark_boundaries_u32 launch failed: {e}"
4809                    ))
4810                })?;
4811        }
4812        self.multiblock_scan_u32_inplace_on_stream(
4813            boundary_prefix,
4814            n,
4815            &cu_stream,
4816            launch_stream,
4817            runtime,
4818        )?;
4819        rec.commit(runtime).map_err(|e| {
4820            XlogError::Kernel(format!(
4821                "wcoj_build_metadata_u32_recorded: mark commit failed: {e}"
4822            ))
4823        })?;
4824        cu_stream.synchronize().map_err(|e| {
4825            XlogError::Kernel(format!(
4826                "wcoj_build_metadata_u32_recorded: mark stream sync failed: {e}"
4827            ))
4828        })?;
4829        Ok(())
4830    }
4831
4832    #[allow(clippy::too_many_arguments)]
4833    fn mark_metadata_boundaries_u64(
4834        &self,
4835        input: &CudaBuffer,
4836        key_col_idx: usize,
4837        keys: &TrackedCudaSlice<u64>,
4838        n: u32,
4839        boundary_mask: &mut TrackedCudaSlice<u32>,
4840        boundary_prefix: &mut TrackedCudaSlice<u32>,
4841        launch_stream: StreamId,
4842    ) -> Result<()> {
4843        let runtime = self.memory().runtime().ok_or_else(|| {
4844            XlogError::Kernel(
4845                "wcoj_build_metadata_u64_recorded requires a runtime-backed GpuMemoryManager"
4846                    .to_string(),
4847            )
4848        })?;
4849        let cu_stream = runtime
4850            .stream_pool()
4851            .resolve(launch_stream)
4852            .ok_or_else(|| {
4853                XlogError::Kernel(format!(
4854                    "wcoj_build_metadata_u64_recorded: launch_stream StreamId({}) does not resolve",
4855                    launch_stream.0
4856                ))
4857            })?;
4858        let mut rec = LaunchRecorder::new_strict(launch_stream);
4859        rec.read(input.num_rows_device());
4860        rec.read_column(input.column(key_col_idx).expect("metadata key column"));
4861        rec.write(boundary_mask);
4862        rec.write(boundary_prefix);
4863        rec.preflight(runtime).map_err(|e| {
4864            XlogError::Kernel(format!(
4865                "wcoj_build_metadata_u64_recorded: mark preflight failed: {e}"
4866            ))
4867        })?;
4868
4869        let kernel = self
4870            .device()
4871            .inner()
4872            .get_func(
4873                WCOJ_MODULE,
4874                wcoj_kernels::WCOJ_BUILD_METADATA_MARK_BOUNDARIES_U64,
4875            )
4876            .ok_or_else(|| {
4877                XlogError::Kernel(
4878                    "wcoj_build_metadata_mark_boundaries_u64 kernel not found".to_string(),
4879                )
4880            })?;
4881        let grid = n.div_ceil(BLOCK_SIZE);
4882        unsafe {
4883            kernel
4884                .clone()
4885                .launch_on_stream(
4886                    &cu_stream,
4887                    LaunchConfig {
4888                        grid_dim: (grid, 1, 1),
4889                        block_dim: (BLOCK_SIZE, 1, 1),
4890                        shared_mem_bytes: 0,
4891                    },
4892                    (keys, n, &mut *boundary_mask, &mut *boundary_prefix),
4893                )
4894                .map_err(|e| {
4895                    XlogError::Kernel(format!(
4896                        "wcoj_build_metadata_mark_boundaries_u64 launch failed: {e}"
4897                    ))
4898                })?;
4899        }
4900        self.multiblock_scan_u32_inplace_on_stream(
4901            boundary_prefix,
4902            n,
4903            &cu_stream,
4904            launch_stream,
4905            runtime,
4906        )?;
4907        rec.commit(runtime).map_err(|e| {
4908            XlogError::Kernel(format!(
4909                "wcoj_build_metadata_u64_recorded: mark commit failed: {e}"
4910            ))
4911        })?;
4912        cu_stream.synchronize().map_err(|e| {
4913            XlogError::Kernel(format!(
4914                "wcoj_build_metadata_u64_recorded: mark stream sync failed: {e}"
4915            ))
4916        })?;
4917        Ok(())
4918    }
4919
4920    #[allow(clippy::too_many_arguments)]
4921    fn scatter_metadata_u32(
4922        &self,
4923        input: &CudaBuffer,
4924        key_col_idx: usize,
4925        keys: &TrackedCudaSlice<u32>,
4926        n: u32,
4927        boundary_mask: &TrackedCudaSlice<u32>,
4928        boundary_prefix: &TrackedCudaSlice<u32>,
4929        unique_keys: &mut TrackedCudaSlice<u32>,
4930        fan_out: &mut TrackedCudaSlice<u32>,
4931        prefix_sum: &mut TrackedCudaSlice<u32>,
4932        launch_stream: StreamId,
4933    ) -> Result<()> {
4934        let runtime = self.memory().runtime().ok_or_else(|| {
4935            XlogError::Kernel(
4936                "wcoj_build_metadata_u32_recorded requires a runtime-backed GpuMemoryManager"
4937                    .to_string(),
4938            )
4939        })?;
4940        let cu_stream = runtime
4941            .stream_pool()
4942            .resolve(launch_stream)
4943            .ok_or_else(|| {
4944                XlogError::Kernel(format!(
4945                    "wcoj_build_metadata_u32_recorded: launch_stream StreamId({}) does not resolve",
4946                    launch_stream.0
4947                ))
4948            })?;
4949        let mut rec = LaunchRecorder::new_strict(launch_stream);
4950        rec.read(input.num_rows_device());
4951        rec.read_column(input.column(key_col_idx).expect("metadata key column"));
4952        rec.read(boundary_mask);
4953        rec.read(boundary_prefix);
4954        rec.write(unique_keys);
4955        rec.write(fan_out);
4956        rec.write(prefix_sum);
4957        rec.preflight(runtime).map_err(|e| {
4958            XlogError::Kernel(format!(
4959                "wcoj_build_metadata_u32_recorded: scatter preflight failed: {e}"
4960            ))
4961        })?;
4962
4963        let kernel = self
4964            .device()
4965            .inner()
4966            .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_BUILD_METADATA_SCATTER_U32)
4967            .ok_or_else(|| {
4968                XlogError::Kernel("wcoj_build_metadata_scatter_u32 kernel not found".to_string())
4969            })?;
4970        let grid = n.div_ceil(BLOCK_SIZE);
4971        unsafe {
4972            kernel
4973                .clone()
4974                .launch_on_stream(
4975                    &cu_stream,
4976                    LaunchConfig {
4977                        grid_dim: (grid, 1, 1),
4978                        block_dim: (BLOCK_SIZE, 1, 1),
4979                        shared_mem_bytes: 0,
4980                    },
4981                    (
4982                        keys,
4983                        n,
4984                        boundary_mask,
4985                        boundary_prefix,
4986                        &mut *unique_keys,
4987                        &mut *fan_out,
4988                        &mut *prefix_sum,
4989                    ),
4990                )
4991                .map_err(|e| {
4992                    XlogError::Kernel(format!(
4993                        "wcoj_build_metadata_scatter_u32 launch failed: {e}"
4994                    ))
4995                })?;
4996        }
4997        rec.commit(runtime).map_err(|e| {
4998            XlogError::Kernel(format!(
4999                "wcoj_build_metadata_u32_recorded: scatter commit failed: {e}"
5000            ))
5001        })?;
5002        cu_stream.synchronize().map_err(|e| {
5003            XlogError::Kernel(format!(
5004                "wcoj_build_metadata_u32_recorded: scatter stream sync failed: {e}"
5005            ))
5006        })?;
5007        Ok(())
5008    }
5009
5010    #[allow(clippy::too_many_arguments)]
5011    fn scatter_metadata_u64(
5012        &self,
5013        input: &CudaBuffer,
5014        key_col_idx: usize,
5015        keys: &TrackedCudaSlice<u64>,
5016        n: u32,
5017        boundary_mask: &TrackedCudaSlice<u32>,
5018        boundary_prefix: &TrackedCudaSlice<u32>,
5019        unique_keys: &mut TrackedCudaSlice<u64>,
5020        fan_out: &mut TrackedCudaSlice<u32>,
5021        prefix_sum: &mut TrackedCudaSlice<u32>,
5022        launch_stream: StreamId,
5023    ) -> Result<()> {
5024        let runtime = self.memory().runtime().ok_or_else(|| {
5025            XlogError::Kernel(
5026                "wcoj_build_metadata_u64_recorded requires a runtime-backed GpuMemoryManager"
5027                    .to_string(),
5028            )
5029        })?;
5030        let cu_stream = runtime
5031            .stream_pool()
5032            .resolve(launch_stream)
5033            .ok_or_else(|| {
5034                XlogError::Kernel(format!(
5035                    "wcoj_build_metadata_u64_recorded: launch_stream StreamId({}) does not resolve",
5036                    launch_stream.0
5037                ))
5038            })?;
5039        let mut rec = LaunchRecorder::new_strict(launch_stream);
5040        rec.read(input.num_rows_device());
5041        rec.read_column(input.column(key_col_idx).expect("metadata key column"));
5042        rec.read(boundary_mask);
5043        rec.read(boundary_prefix);
5044        rec.write(unique_keys);
5045        rec.write(fan_out);
5046        rec.write(prefix_sum);
5047        rec.preflight(runtime).map_err(|e| {
5048            XlogError::Kernel(format!(
5049                "wcoj_build_metadata_u64_recorded: scatter preflight failed: {e}"
5050            ))
5051        })?;
5052
5053        let kernel = self
5054            .device()
5055            .inner()
5056            .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_BUILD_METADATA_SCATTER_U64)
5057            .ok_or_else(|| {
5058                XlogError::Kernel("wcoj_build_metadata_scatter_u64 kernel not found".to_string())
5059            })?;
5060        let grid = n.div_ceil(BLOCK_SIZE);
5061        unsafe {
5062            kernel
5063                .clone()
5064                .launch_on_stream(
5065                    &cu_stream,
5066                    LaunchConfig {
5067                        grid_dim: (grid, 1, 1),
5068                        block_dim: (BLOCK_SIZE, 1, 1),
5069                        shared_mem_bytes: 0,
5070                    },
5071                    (
5072                        keys,
5073                        n,
5074                        boundary_mask,
5075                        boundary_prefix,
5076                        &mut *unique_keys,
5077                        &mut *fan_out,
5078                        &mut *prefix_sum,
5079                    ),
5080                )
5081                .map_err(|e| {
5082                    XlogError::Kernel(format!(
5083                        "wcoj_build_metadata_scatter_u64 launch failed: {e}"
5084                    ))
5085                })?;
5086        }
5087        rec.commit(runtime).map_err(|e| {
5088            XlogError::Kernel(format!(
5089                "wcoj_build_metadata_u64_recorded: scatter commit failed: {e}"
5090            ))
5091        })?;
5092        cu_stream.synchronize().map_err(|e| {
5093            XlogError::Kernel(format!(
5094                "wcoj_build_metadata_u64_recorded: scatter stream sync failed: {e}"
5095            ))
5096        })?;
5097        Ok(())
5098    }
5099
5100    fn metadata_scanned_count(
5101        &self,
5102        boundary_mask: &TrackedCudaSlice<u32>,
5103        boundary_prefix: &TrackedCudaSlice<u32>,
5104        n: u32,
5105    ) -> Result<u32> {
5106        let last = n - 1;
5107        let prefix_last = self.dtoh_scalar_untracked::<u32>(boundary_prefix, last as usize)?;
5108        let mask_last = self.dtoh_scalar_untracked::<u32>(boundary_mask, last as usize)?;
5109        Ok(prefix_last + mask_last)
5110    }
5111
5112    fn metadata_logical_rows(&self, input: &CudaBuffer) -> Result<u32> {
5113        if let Some(cached) = input.cached_row_count() {
5114            return Ok(cached);
5115        }
5116        self.dtoh_scalar_untracked::<u32>(input.num_rows_device(), 0)
5117    }
5118}
5119
5120#[derive(Clone, Copy)]
5121enum MetadataWidth {
5122    U32,
5123    U64,
5124}
5125
5126fn metadata_column_u32(input: &CudaBuffer, key_col_idx: usize) -> Result<&TrackedCudaSlice<u32>> {
5127    let col = input.column(key_col_idx).ok_or_else(|| {
5128        XlogError::Kernel(format!(
5129            "wcoj_build_metadata_u32_recorded: column {} not found",
5130            key_col_idx
5131        ))
5132    })?;
5133    match col {
5134        CudaColumn::Owned(slice) => unsafe {
5135            Ok(&*(slice as *const TrackedCudaSlice<u8> as *const TrackedCudaSlice<u32>))
5136        },
5137        _ => Err(XlogError::Kernel(
5138            "wcoj_build_metadata_u32_recorded: key column must be an owned CudaColumn".to_string(),
5139        )),
5140    }
5141}
5142
5143fn metadata_column_u64(input: &CudaBuffer, key_col_idx: usize) -> Result<&TrackedCudaSlice<u64>> {
5144    let col = input.column(key_col_idx).ok_or_else(|| {
5145        XlogError::Kernel(format!(
5146            "wcoj_build_metadata_u64_recorded: column {} not found",
5147            key_col_idx
5148        ))
5149    })?;
5150    match col {
5151        CudaColumn::Owned(slice) => unsafe {
5152            Ok(&*(slice as *const TrackedCudaSlice<u8> as *const TrackedCudaSlice<u64>))
5153        },
5154        _ => Err(XlogError::Kernel(
5155            "wcoj_build_metadata_u64_recorded: key column must be an owned CudaColumn".to_string(),
5156        )),
5157    }
5158}
5159
5160fn validate_binary_u32(ctx: &str, label: &str, input: &CudaBuffer) -> Result<()> {
5161    if input.arity() != 2 {
5162        return Err(XlogError::Kernel(format!(
5163            "{ctx}: {label} must be 2-column, got arity {}",
5164            input.arity()
5165        )));
5166    }
5167    for col_idx in 0..2 {
5168        let ty = input.schema().column_type(col_idx).ok_or_else(|| {
5169            XlogError::Kernel(format!("{ctx}: {label}.col{col_idx} type missing"))
5170        })?;
5171        if !matches!(ty, ScalarType::U32 | ScalarType::Symbol) {
5172            return Err(XlogError::Kernel(format!(
5173                "{ctx}: {label}.col{col_idx} must be U32 or Symbol, got {:?}",
5174                ty
5175            )));
5176        }
5177    }
5178    Ok(())
5179}
5180
5181fn validate_binary_u64(ctx: &str, label: &str, input: &CudaBuffer) -> Result<()> {
5182    if input.arity() != 2 {
5183        return Err(XlogError::Kernel(format!(
5184            "{ctx}: {label} must be 2-column, got arity {}",
5185            input.arity()
5186        )));
5187    }
5188    for col_idx in 0..2 {
5189        let ty = input.schema().column_type(col_idx).ok_or_else(|| {
5190            XlogError::Kernel(format!("{ctx}: {label}.col{col_idx} type missing"))
5191        })?;
5192        if !matches!(ty, ScalarType::U64) {
5193            return Err(XlogError::Kernel(format!(
5194                "{ctx}: {label}.col{col_idx} must be U64, got {:?}",
5195                ty
5196            )));
5197        }
5198    }
5199    Ok(())
5200}
5201
5202unsafe fn reinterpret_u8_as_u32(slice: &mut TrackedCudaSlice<u8>) -> &mut TrackedCudaSlice<u32> {
5203    &mut *(slice as *mut TrackedCudaSlice<u8> as *mut TrackedCudaSlice<u32>)
5204}
5205
5206unsafe fn reinterpret_u8_as_u64(slice: &mut TrackedCudaSlice<u8>) -> &mut TrackedCudaSlice<u64> {
5207    &mut *(slice as *mut TrackedCudaSlice<u8> as *mut TrackedCudaSlice<u64>)
5208}
xlog_cuda/provider/wcoj_metadata.rs

xlog_cuda/provider/
wcoj_metadata.rs
