1use std::ffi::c_void;
51use std::time::Instant;
52
53use cudarc::driver::sys;
54use xlog_core::{Result, ScalarType, Schema, XlogError};
55
56use super::{wcoj_kernels, CudaKernelProvider, WCOJ_MODULE};
57use crate::device_runtime::StreamId;
58use crate::launch::LaunchRecorder;
59use crate::memory::{CudaColumn, DeviceMemoryView, DeviceRead, TrackedCudaSlice};
60use crate::wcoj_metadata::WcojRelationMetadata;
61use crate::CudaBuffer;
62use crate::{AsKernelParam, LaunchAsync, LaunchConfig};
63
64const BLOCK_SIZE: u32 = 256;
65
66fn column_u32(input: &CudaBuffer, col_idx: usize) -> Result<DeviceMemoryView<u32>> {
67 let col = input.column(col_idx).ok_or_else(|| {
68 XlogError::Kernel(format!(
69 "wcoj_layout_u32_recorded: column {col_idx} not found"
70 ))
71 })?;
72 unsafe { col.device_view().cast() }
75 .ok_or_else(|| XlogError::Kernel("invalid u32 device column layout".into()))
76}
77
78fn column_u64(input: &CudaBuffer, col_idx: usize) -> Result<DeviceMemoryView<u64>> {
79 let col = input.column(col_idx).ok_or_else(|| {
80 XlogError::Kernel(format!(
81 "wcoj_layout_u64_recorded: column {col_idx} not found"
82 ))
83 })?;
84 unsafe { col.device_view().cast() }
87 .ok_or_else(|| XlogError::Kernel("invalid u64 device column layout".into()))
88}
89
90impl CudaKernelProvider {
91 pub fn wcoj_layout_u32_recorded(
131 &self,
132 input: &CudaBuffer,
133 launch_stream: StreamId,
134 ) -> Result<CudaBuffer> {
135 if self.memory().runtime().is_none() {
139 return Err(XlogError::Kernel(
140 "wcoj_layout_u32_recorded requires a runtime-backed \
141 GpuMemoryManager (constructed via with_runtime)"
142 .to_string(),
143 ));
144 }
145 if input.arity() != 2 {
150 return Err(XlogError::Kernel(format!(
151 "wcoj_layout_u32_recorded: input must be 2-column, got arity {}",
152 input.arity()
153 )));
154 }
155 for col_idx in 0..2 {
156 let ty = input.schema.column_type(col_idx).ok_or_else(|| {
157 XlogError::Kernel(format!(
158 "wcoj_layout_u32_recorded: column {} type missing",
159 col_idx
160 ))
161 })?;
162 if !matches!(ty, ScalarType::U32 | ScalarType::Symbol) {
163 return Err(XlogError::Kernel(format!(
164 "wcoj_layout_u32_recorded: column {} must be U32 or Symbol, got {:?}",
165 col_idx, ty
166 )));
167 }
168 }
169 match self.try_wcoj_layout_fast_path_u32(input, launch_stream) {
178 Ok(Some(out)) => {
179 self.record_wcoj_layout_fast_path_hit();
180 return Ok(out);
181 }
182 Ok(None) => {
183 }
187 Err(_) => {
188 }
191 }
192 self.dedup_full_row_recorded(input, launch_stream)
198 }
199
200 pub fn wcoj_layout_sort_u32_recorded(
243 &self,
244 input: &CudaBuffer,
245 launch_stream: StreamId,
246 ) -> Result<CudaBuffer> {
247 self.record_wcoj_layout_sort_invocation();
248 if self.memory().runtime().is_none() {
249 return Err(XlogError::Kernel(
250 "wcoj_layout_sort_u32_recorded requires a runtime-backed \
251 GpuMemoryManager (constructed via with_runtime)"
252 .to_string(),
253 ));
254 }
255 if input.arity() < 2 {
256 return Err(XlogError::Kernel(format!(
257 "wcoj_layout_sort_u32_recorded: input must have arity >= 2, got {}",
258 input.arity()
259 )));
260 }
261 for col_idx in 0..input.arity() {
262 let ty = input.schema.column_type(col_idx).ok_or_else(|| {
263 XlogError::Kernel(format!(
264 "wcoj_layout_sort_u32_recorded: column {} type missing",
265 col_idx
266 ))
267 })?;
268 if !matches!(ty, ScalarType::U32 | ScalarType::Symbol) {
269 return Err(XlogError::Kernel(format!(
270 "wcoj_layout_sort_u32_recorded: column {} must be U32 or Symbol \
271 (4-byte width-class), got {:?}",
272 col_idx, ty
273 )));
274 }
275 }
276 self.dedup_full_row_recorded(input, launch_stream)
277 }
278
279 pub fn wcoj_triangle_u32_recorded(
303 &self,
304 e_xy: &CudaBuffer,
305 e_yz: &CudaBuffer,
306 e_xz: &CudaBuffer,
307 launch_stream: StreamId,
308 ) -> Result<CudaBuffer> {
309 self.wcoj_triangle_hg_u32_recorded(
310 e_xy,
311 e_yz,
312 e_xz,
313 crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
314 launch_stream,
315 )
316 }
317
318 pub fn wcoj_4cycle_u32_recorded(
346 &self,
347 e1: &CudaBuffer,
348 e2: &CudaBuffer,
349 e3: &CudaBuffer,
350 e4: &CudaBuffer,
351 launch_stream: StreamId,
352 ) -> Result<CudaBuffer> {
353 self.wcoj_4cycle_hg_u32_recorded(
354 e1,
355 e2,
356 e3,
357 e4,
358 crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
359 launch_stream,
360 )
361 }
362
363 fn logical_row_count_u32(&self, buf: &CudaBuffer) -> Result<u32> {
364 if let Some(c) = buf.cached_row_count() {
365 return Ok(c);
366 }
367 self.dtoh_scalar_untracked::<u32>(buf.num_rows_device(), 0)
368 }
369
370 pub fn wcoj_layout_u64_recorded(
387 &self,
388 input: &CudaBuffer,
389 launch_stream: StreamId,
390 ) -> Result<CudaBuffer> {
391 if self.memory().runtime().is_none() {
392 return Err(XlogError::Kernel(
393 "wcoj_layout_u64_recorded requires a runtime-backed \
394 GpuMemoryManager (constructed via with_runtime)"
395 .to_string(),
396 ));
397 }
398 if input.arity() != 2 {
399 return Err(XlogError::Kernel(format!(
400 "wcoj_layout_u64_recorded: input must be 2-column, got arity {}",
401 input.arity()
402 )));
403 }
404 for col_idx in 0..2 {
405 let ty = input.schema.column_type(col_idx).ok_or_else(|| {
406 XlogError::Kernel(format!(
407 "wcoj_layout_u64_recorded: column {} type missing",
408 col_idx
409 ))
410 })?;
411 if !matches!(ty, ScalarType::U64) {
412 return Err(XlogError::Kernel(format!(
413 "wcoj_layout_u64_recorded: column {} must be U64, got {:?}",
414 col_idx, ty
415 )));
416 }
417 }
418 if let Ok(Some(out)) = self.try_wcoj_layout_fast_path_u64(input, launch_stream) {
422 self.record_wcoj_layout_fast_path_hit();
423 return Ok(out);
424 }
425 self.dedup_full_row_recorded(input, launch_stream)
429 }
430
431 pub fn wcoj_layout_sort_u64_recorded(
463 &self,
464 input: &CudaBuffer,
465 launch_stream: StreamId,
466 ) -> Result<CudaBuffer> {
467 self.record_wcoj_layout_sort_invocation();
468 if self.memory().runtime().is_none() {
469 return Err(XlogError::Kernel(
470 "wcoj_layout_sort_u64_recorded requires a runtime-backed \
471 GpuMemoryManager (constructed via with_runtime)"
472 .to_string(),
473 ));
474 }
475 if input.arity() < 2 {
476 return Err(XlogError::Kernel(format!(
477 "wcoj_layout_sort_u64_recorded: input must have arity >= 2, got {}",
478 input.arity()
479 )));
480 }
481 for col_idx in 0..input.arity() {
482 let ty = input.schema.column_type(col_idx).ok_or_else(|| {
483 XlogError::Kernel(format!(
484 "wcoj_layout_sort_u64_recorded: column {} type missing",
485 col_idx
486 ))
487 })?;
488 if !matches!(ty, ScalarType::U64) {
489 return Err(XlogError::Kernel(format!(
490 "wcoj_layout_sort_u64_recorded: column {} must be U64 \
491 (8-byte width-class), got {:?}",
492 col_idx, ty
493 )));
494 }
495 }
496 self.dedup_full_row_recorded(input, launch_stream)
497 }
498
499 pub fn wcoj_triangle_u64_recorded(
511 &self,
512 e_xy: &CudaBuffer,
513 e_yz: &CudaBuffer,
514 e_xz: &CudaBuffer,
515 launch_stream: StreamId,
516 ) -> Result<CudaBuffer> {
517 self.wcoj_triangle_hg_u64_recorded(
518 e_xy,
519 e_yz,
520 e_xz,
521 crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
522 launch_stream,
523 )
524 }
525
526 pub fn wcoj_4cycle_u64_recorded(
543 &self,
544 e1: &CudaBuffer,
545 e2: &CudaBuffer,
546 e3: &CudaBuffer,
547 e4: &CudaBuffer,
548 launch_stream: StreamId,
549 ) -> Result<CudaBuffer> {
550 self.wcoj_4cycle_hg_u64_recorded(
551 e1,
552 e2,
553 e3,
554 e4,
555 crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT,
556 launch_stream,
557 )
558 }
559}
560
561impl CudaKernelProvider {
591 fn try_wcoj_layout_fast_path_u32(
600 &self,
601 input: &CudaBuffer,
602 launch_stream: StreamId,
603 ) -> Result<Option<CudaBuffer>> {
604 let runtime = self
605 .memory()
606 .runtime()
607 .ok_or_else(|| XlogError::Kernel("wcoj_layout fast-path: no runtime".to_string()))?;
608 let cu_stream = runtime
609 .stream_pool()
610 .resolve(launch_stream)
611 .ok_or_else(|| {
612 XlogError::Kernel("wcoj_layout fast-path: stream resolve".to_string())
613 })?;
614
615 let n = self.logical_row_count_u32(input)?;
618 if n == 0 {
619 return Ok(None);
623 }
624 if n == 1 {
625 return Ok(Some(self.recorded_clone_2col_4byte(
627 input,
628 n,
629 launch_stream,
630 &cu_stream,
631 runtime,
632 )?));
633 }
634
635 let mut flag_buf = self.memory.alloc::<u32>(1)?;
637
638 let col0 = &column_u32(input, 0)?;
639 let col1 = &column_u32(input, 1)?;
640
641 let device = self.device.inner();
645 let kernel = device
646 .get_func(
647 WCOJ_MODULE,
648 wcoj_kernels::WCOJ_LAYOUT_CHECK_SORTED_UNIQUE_U32,
649 )
650 .ok_or_else(|| {
651 XlogError::Kernel("wcoj_layout_check_sorted_unique_u32 kernel not found".into())
652 })?;
653
654 let mut rec = LaunchRecorder::new_strict(launch_stream);
655 rec.read(input.num_rows_device());
656 rec.read_column(input.column(0).expect("col0"));
657 rec.read_column(input.column(1).expect("col1"));
658 rec.write(&flag_buf);
659 rec.preflight(runtime)
660 .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path: preflight {e}")))?;
661
662 let one: u32 = 1;
668 let grid = n.div_ceil(BLOCK_SIZE);
669 let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
670 self.initialize_launch_metadata_u32(
671 one,
672 &flag_buf,
673 stream,
674 "wcoj_layout fast-path flag init",
675 )?;
676
677 unsafe {
681 kernel
682 .clone()
683 .launch_in(
684 stream,
685 LaunchConfig {
686 grid_dim: (grid, 1, 1),
687 block_dim: (BLOCK_SIZE, 1, 1),
688 shared_mem_bytes: 0,
689 },
690 (col0, col1, n, &mut flag_buf),
691 )
692 .map_err(|e| {
693 XlogError::Kernel(format!(
694 "wcoj_layout_check_sorted_unique_u32 launch: {e}"
695 ))
696 })?;
697 }
698 Ok(())
699 };
700 let rec = unsafe { rec.enqueue_prepared_with(&cu_stream, enqueue) }
702 .map_err(|error| error.into_xlog_error())?;
703
704 rec.commit()
705 .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path: commit {e}")))?;
706
707 cu_stream
708 .synchronize()
709 .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path: sync {e}")))?;
710
711 let flag_val = self.dtoh_scalar_untracked::<u32>(&flag_buf, 0)?;
712 if flag_val == 1 {
713 Ok(Some(self.recorded_clone_2col_4byte(
714 input,
715 n,
716 launch_stream,
717 &cu_stream,
718 runtime,
719 )?))
720 } else {
721 Ok(None)
722 }
723 }
724
725 fn try_wcoj_layout_fast_path_u64(
727 &self,
728 input: &CudaBuffer,
729 launch_stream: StreamId,
730 ) -> Result<Option<CudaBuffer>> {
731 let runtime = self
732 .memory()
733 .runtime()
734 .ok_or_else(|| XlogError::Kernel("wcoj_layout fast-path: no runtime".to_string()))?;
735 let cu_stream = runtime
736 .stream_pool()
737 .resolve(launch_stream)
738 .ok_or_else(|| {
739 XlogError::Kernel("wcoj_layout fast-path: stream resolve".to_string())
740 })?;
741
742 let n = self.logical_row_count_u32(input)?;
743 if n == 0 {
744 return Ok(None);
745 }
746 if n == 1 {
747 return Ok(Some(self.recorded_clone_2col_8byte(
748 input,
749 n,
750 launch_stream,
751 &cu_stream,
752 runtime,
753 )?));
754 }
755
756 let mut flag_buf = self.memory.alloc::<u32>(1)?;
757 let col0 = &column_u64(input, 0)?;
758 let col1 = &column_u64(input, 1)?;
759
760 let device = self.device.inner();
761 let kernel = device
762 .get_func(
763 WCOJ_MODULE,
764 wcoj_kernels::WCOJ_LAYOUT_CHECK_SORTED_UNIQUE_U64,
765 )
766 .ok_or_else(|| {
767 XlogError::Kernel("wcoj_layout_check_sorted_unique_u64 kernel not found".into())
768 })?;
769
770 let mut rec = LaunchRecorder::new_strict(launch_stream);
771 rec.read(input.num_rows_device());
772 rec.read_column(input.column(0).expect("col0"));
773 rec.read_column(input.column(1).expect("col1"));
774 rec.write(&flag_buf);
775 rec.preflight(runtime)
776 .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path u64: preflight {e}")))?;
777
778 let one: u32 = 1;
779 let grid = n.div_ceil(BLOCK_SIZE);
780 let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
781 self.initialize_launch_metadata_u32(
782 one,
783 &flag_buf,
784 stream,
785 "wcoj_layout fast-path u64 flag init",
786 )?;
787
788 unsafe {
789 kernel
790 .clone()
791 .launch_in(
792 stream,
793 LaunchConfig {
794 grid_dim: (grid, 1, 1),
795 block_dim: (BLOCK_SIZE, 1, 1),
796 shared_mem_bytes: 0,
797 },
798 (col0, col1, n, &mut flag_buf),
799 )
800 .map_err(|e| {
801 XlogError::Kernel(format!(
802 "wcoj_layout_check_sorted_unique_u64 launch: {e}"
803 ))
804 })?;
805 }
806 Ok(())
807 };
808 let rec = unsafe { rec.enqueue_prepared_with(&cu_stream, enqueue) }
810 .map_err(|error| error.into_xlog_error())?;
811
812 rec.commit()
813 .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path u64: commit {e}")))?;
814
815 cu_stream
816 .synchronize()
817 .map_err(|e| XlogError::Kernel(format!("wcoj_layout fast-path u64: sync {e}")))?;
818
819 let flag_val = self.dtoh_scalar_untracked::<u32>(&flag_buf, 0)?;
820 if flag_val == 1 {
821 Ok(Some(self.recorded_clone_2col_8byte(
822 input,
823 n,
824 launch_stream,
825 &cu_stream,
826 runtime,
827 )?))
828 } else {
829 Ok(None)
830 }
831 }
832
833 fn recorded_clone_2col_4byte(
840 &self,
841 input: &CudaBuffer,
842 n: u32,
843 launch_stream: StreamId,
844 cu_stream: &cudarc::driver::CudaStream,
845 runtime: &std::sync::Arc<crate::device_runtime::XlogDeviceRuntime>,
846 ) -> Result<CudaBuffer> {
847 let bpc = (n as usize) * 4;
848 let out_col0 = self.memory.alloc::<u8>(bpc)?;
849 let out_col1 = self.memory.alloc::<u8>(bpc)?;
850 let out_d_num_rows = self.memory.alloc::<u32>(1)?;
851
852 let src_col0 = input.column(0).expect("col0");
853 let src_col1 = input.column(1).expect("col1");
854
855 let mut rec = LaunchRecorder::new_strict(launch_stream);
856 rec.read(input.num_rows_device());
857 rec.read_column(src_col0);
858 rec.read_column(src_col1);
859 rec.write(&out_col0);
860 rec.write(&out_col1);
861 rec.write(&out_d_num_rows);
862 rec.preflight(runtime)
863 .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 4B: preflight {e}")))?;
864
865 let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
866 unsafe {
867 let r0 = sys::cuMemcpyDtoDAsync_v2(
868 *out_col0.device_ptr(),
869 *src_col0.device_ptr(),
870 bpc,
871 stream.stream().cu_stream(),
872 );
873 if r0 != sys::cudaError_enum::CUDA_SUCCESS {
874 return Err(XlogError::Kernel(format!(
875 "wcoj_layout clone 4B: dtod col0 failed: {r0:?}"
876 )));
877 }
878 let r1 = sys::cuMemcpyDtoDAsync_v2(
879 *out_col1.device_ptr(),
880 *src_col1.device_ptr(),
881 bpc,
882 stream.stream().cu_stream(),
883 );
884 if r1 != sys::cudaError_enum::CUDA_SUCCESS {
885 return Err(XlogError::Kernel(format!(
886 "wcoj_layout clone 4B: dtod col1 failed: {r1:?}"
887 )));
888 }
889 }
890 self.initialize_launch_metadata_u32(
891 n,
892 &out_d_num_rows,
893 stream,
894 "wcoj_layout clone 4B d_num_rows",
895 )?;
896 Ok(())
897 };
898 let rec = unsafe { rec.enqueue_prepared_with(cu_stream, enqueue) }
900 .map_err(|error| error.into_xlog_error())?;
901
902 rec.commit()
903 .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 4B: commit {e}")))?;
904
905 Ok(CudaBuffer::from_columns_with_host_count(
906 vec![out_col0.into(), out_col1.into()],
907 n as u64,
908 out_d_num_rows,
909 input.schema().clone(),
910 n,
911 ))
912 }
913
914 fn recorded_clone_2col_8byte(
916 &self,
917 input: &CudaBuffer,
918 n: u32,
919 launch_stream: StreamId,
920 cu_stream: &cudarc::driver::CudaStream,
921 runtime: &std::sync::Arc<crate::device_runtime::XlogDeviceRuntime>,
922 ) -> Result<CudaBuffer> {
923 let bpc = (n as usize) * 8;
924 let out_col0 = self.memory.alloc::<u8>(bpc)?;
925 let out_col1 = self.memory.alloc::<u8>(bpc)?;
926 let out_d_num_rows = self.memory.alloc::<u32>(1)?;
927
928 let src_col0 = input.column(0).expect("col0");
929 let src_col1 = input.column(1).expect("col1");
930
931 let mut rec = LaunchRecorder::new_strict(launch_stream);
932 rec.read(input.num_rows_device());
933 rec.read_column(src_col0);
934 rec.read_column(src_col1);
935 rec.write(&out_col0);
936 rec.write(&out_col1);
937 rec.write(&out_d_num_rows);
938 rec.preflight(runtime)
939 .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 8B: preflight {e}")))?;
940
941 let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
942 unsafe {
943 let r0 = sys::cuMemcpyDtoDAsync_v2(
944 *out_col0.device_ptr(),
945 *src_col0.device_ptr(),
946 bpc,
947 stream.stream().cu_stream(),
948 );
949 if r0 != sys::cudaError_enum::CUDA_SUCCESS {
950 return Err(XlogError::Kernel(format!(
951 "wcoj_layout clone 8B: dtod col0 failed: {r0:?}"
952 )));
953 }
954 let r1 = sys::cuMemcpyDtoDAsync_v2(
955 *out_col1.device_ptr(),
956 *src_col1.device_ptr(),
957 bpc,
958 stream.stream().cu_stream(),
959 );
960 if r1 != sys::cudaError_enum::CUDA_SUCCESS {
961 return Err(XlogError::Kernel(format!(
962 "wcoj_layout clone 8B: dtod col1 failed: {r1:?}"
963 )));
964 }
965 }
966 self.initialize_launch_metadata_u32(
967 n,
968 &out_d_num_rows,
969 stream,
970 "wcoj_layout clone 8B d_num_rows",
971 )?;
972 Ok(())
973 };
974 let rec = unsafe { rec.enqueue_prepared_with(cu_stream, enqueue) }
976 .map_err(|error| error.into_xlog_error())?;
977
978 rec.commit()
979 .map_err(|e| XlogError::Kernel(format!("wcoj_layout clone 8B: commit {e}")))?;
980
981 Ok(CudaBuffer::from_columns_with_host_count(
982 vec![out_col0.into(), out_col1.into()],
983 n as u64,
984 out_d_num_rows,
985 input.schema().clone(),
986 n,
987 ))
988 }
989}
990
991#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1009enum CliqueWidthClass {
1010 FourByte, EightByte, }
1013
1014impl CliqueWidthClass {
1015 fn elem_bytes(self) -> usize {
1016 match self {
1017 CliqueWidthClass::FourByte => 4,
1018 CliqueWidthClass::EightByte => 8,
1019 }
1020 }
1021 fn label(self) -> &'static str {
1022 match self {
1023 CliqueWidthClass::FourByte => "u32",
1024 CliqueWidthClass::EightByte => "u64",
1025 }
1026 }
1027 fn validate_col_type(self, ty: ScalarType) -> bool {
1028 match self {
1029 CliqueWidthClass::FourByte => matches!(ty, ScalarType::U32 | ScalarType::Symbol),
1030 CliqueWidthClass::EightByte => matches!(ty, ScalarType::U64),
1031 }
1032 }
1033}
1034
1035fn clique_kernel_name(k: usize, materialize: bool, w: CliqueWidthClass) -> &'static str {
1037 match (k, materialize, w) {
1038 (5, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE5_COUNT_HG_U32,
1039 (5, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE5_MATERIALIZE_HG_U32,
1040 (5, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE5_COUNT_HG_U64,
1041 (5, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE5_MATERIALIZE_HG_U64,
1042 (6, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE6_COUNT_HG_U32,
1043 (6, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE6_MATERIALIZE_HG_U32,
1044 (6, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE6_COUNT_HG_U64,
1045 (6, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE6_MATERIALIZE_HG_U64,
1046 (7, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE7_COUNT_HG_U32,
1047 (7, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE7_MATERIALIZE_HG_U32,
1048 (7, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE7_COUNT_HG_U64,
1049 (7, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE7_MATERIALIZE_HG_U64,
1050 (8, false, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE8_COUNT_HG_U32,
1051 (8, true, CliqueWidthClass::FourByte) => wcoj_kernels::WCOJ_CLIQUE8_MATERIALIZE_HG_U32,
1052 (8, false, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE8_COUNT_HG_U64,
1053 (8, true, CliqueWidthClass::EightByte) => wcoj_kernels::WCOJ_CLIQUE8_MATERIALIZE_HG_U64,
1054 _ => panic!("clique_kernel_name: K must be 5..8, got {}", k),
1055 }
1056}
1057
1058enum CliqueLeaderMetadata {
1059 U32(WcojRelationMetadata<u32>),
1060 U64(WcojRelationMetadata<u64>),
1061}
1062
1063impl CliqueLeaderMetadata {
1064 fn total_rows_u32(&self, entry_label: &str) -> Result<u32> {
1065 let total = match self {
1066 CliqueLeaderMetadata::U32(metadata) => metadata.total,
1067 CliqueLeaderMetadata::U64(metadata) => metadata.total,
1068 };
1069 u32::try_from(total).map_err(|_| {
1070 XlogError::Kernel(format!(
1071 "{}: leader metadata total {} exceeds u32 kernel surface",
1072 entry_label, total
1073 ))
1074 })
1075 }
1076
1077 fn key_count(&self) -> u32 {
1078 match self {
1079 CliqueLeaderMetadata::U32(metadata) => metadata.key_count,
1080 CliqueLeaderMetadata::U64(metadata) => metadata.key_count,
1081 }
1082 }
1083}
1084
1085fn validate_clique_metadata_leader<'a>(
1086 k: usize,
1087 edges: &'a [&CudaBuffer],
1088 leader_edge_idx: u32,
1089 width_class: CliqueWidthClass,
1090 entry_label: &str,
1091) -> Result<&'a CudaBuffer> {
1092 if !(5..=8).contains(&k) {
1093 return Err(XlogError::Kernel(format!(
1094 "{}: k must be 5..8, got {}",
1095 entry_label, k
1096 )));
1097 }
1098 let expected_edges = k * (k - 1) / 2;
1099 if edges.len() != expected_edges {
1100 return Err(XlogError::Kernel(format!(
1101 "{}: expected {} edges (= C({}, 2)), got {}",
1102 entry_label,
1103 expected_edges,
1104 k,
1105 edges.len()
1106 )));
1107 }
1108 let leader_slot = usize::try_from(leader_edge_idx)
1109 .ok()
1110 .filter(|idx| *idx < expected_edges)
1111 .ok_or_else(|| {
1112 XlogError::Kernel(format!(
1113 "{}: leader_edge_idx {} out of range for {} edges",
1114 entry_label, leader_edge_idx, expected_edges
1115 ))
1116 })?;
1117 let leader = edges[leader_slot];
1118 if leader.arity() != 2 {
1119 return Err(XlogError::Kernel(format!(
1120 "{}: leader edge must be 2-column, got arity {}",
1121 entry_label,
1122 leader.arity()
1123 )));
1124 }
1125 let ty = leader.schema.column_type(0).ok_or_else(|| {
1126 XlogError::Kernel(format!(
1127 "{}: leader edge column 0 type missing",
1128 entry_label
1129 ))
1130 })?;
1131 if !width_class.validate_col_type(ty) {
1132 return Err(XlogError::Kernel(format!(
1133 "{}: leader edge column 0 type {:?} not in {} width-class",
1134 entry_label,
1135 ty,
1136 width_class.label()
1137 )));
1138 }
1139 Ok(leader)
1140}
1141
1142impl CudaKernelProvider {
1143 fn wcoj_clique_metadata_recorded_u32_inner(
1144 &self,
1145 k: usize,
1146 edges: &[&CudaBuffer],
1147 leader_edge_idx: u32,
1148 launch_stream: StreamId,
1149 entry_label: &str,
1150 ) -> Result<WcojRelationMetadata<u32>> {
1151 let leader = validate_clique_metadata_leader(
1152 k,
1153 edges,
1154 leader_edge_idx,
1155 CliqueWidthClass::FourByte,
1156 entry_label,
1157 )?;
1158 self.wcoj_build_metadata_u32_recorded(leader, 0, launch_stream)
1159 }
1160
1161 fn wcoj_clique_metadata_recorded_u64_inner(
1162 &self,
1163 k: usize,
1164 edges: &[&CudaBuffer],
1165 leader_edge_idx: u32,
1166 launch_stream: StreamId,
1167 entry_label: &str,
1168 ) -> Result<WcojRelationMetadata<u64>> {
1169 let leader = validate_clique_metadata_leader(
1170 k,
1171 edges,
1172 leader_edge_idx,
1173 CliqueWidthClass::EightByte,
1174 entry_label,
1175 )?;
1176 self.wcoj_build_metadata_u64_recorded(leader, 0, launch_stream)
1177 }
1178
1179 pub fn wcoj_clique5_metadata_recorded_u32(
1181 &self,
1182 edges: &[&CudaBuffer; 10],
1183 leader_edge_idx: u32,
1184 launch_stream: StreamId,
1185 ) -> Result<WcojRelationMetadata<u32>> {
1186 self.wcoj_clique_metadata_recorded_u32_inner(
1187 5,
1188 edges,
1189 leader_edge_idx,
1190 launch_stream,
1191 "wcoj_clique5_metadata_recorded_u32",
1192 )
1193 }
1194
1195 pub fn wcoj_clique5_metadata_recorded_u64(
1197 &self,
1198 edges: &[&CudaBuffer; 10],
1199 leader_edge_idx: u32,
1200 launch_stream: StreamId,
1201 ) -> Result<WcojRelationMetadata<u64>> {
1202 self.wcoj_clique_metadata_recorded_u64_inner(
1203 5,
1204 edges,
1205 leader_edge_idx,
1206 launch_stream,
1207 "wcoj_clique5_metadata_recorded_u64",
1208 )
1209 }
1210
1211 pub fn wcoj_clique6_metadata_recorded_u32(
1213 &self,
1214 edges: &[&CudaBuffer; 15],
1215 leader_edge_idx: u32,
1216 launch_stream: StreamId,
1217 ) -> Result<WcojRelationMetadata<u32>> {
1218 self.wcoj_clique_metadata_recorded_u32_inner(
1219 6,
1220 edges,
1221 leader_edge_idx,
1222 launch_stream,
1223 "wcoj_clique6_metadata_recorded_u32",
1224 )
1225 }
1226
1227 pub fn wcoj_clique6_metadata_recorded_u64(
1229 &self,
1230 edges: &[&CudaBuffer; 15],
1231 leader_edge_idx: u32,
1232 launch_stream: StreamId,
1233 ) -> Result<WcojRelationMetadata<u64>> {
1234 self.wcoj_clique_metadata_recorded_u64_inner(
1235 6,
1236 edges,
1237 leader_edge_idx,
1238 launch_stream,
1239 "wcoj_clique6_metadata_recorded_u64",
1240 )
1241 }
1242
1243 #[expect(
1258 clippy::too_many_arguments,
1259 reason = "the clique launch core mirrors the kernel's explicit inputs, variable layout, output projection, limits, and stream"
1260 )]
1261 fn wcoj_clique_recorded_inner(
1262 &self,
1263 k: usize,
1264 edges: &[&CudaBuffer],
1265 leader_edge_idx: u32,
1266 edge_order: Option<&[u8]>,
1267 iteration_order: Option<&[u8]>,
1268 width_class: CliqueWidthClass,
1269 launch_stream: StreamId,
1270 entry_label: &str,
1271 ) -> Result<CudaBuffer> {
1272 let runtime = self.memory().runtime().ok_or_else(|| {
1273 XlogError::Kernel(format!(
1274 "{} requires a runtime-backed GpuMemoryManager (with_runtime)",
1275 entry_label
1276 ))
1277 })?;
1278 let cu_stream = runtime
1279 .stream_pool()
1280 .resolve(launch_stream)
1281 .ok_or_else(|| {
1282 XlogError::Kernel(format!(
1283 "{}: launch_stream StreamId({}) does not resolve",
1284 entry_label, launch_stream.0
1285 ))
1286 })?;
1287 if !(5..=8).contains(&k) {
1288 return Err(XlogError::Kernel(format!(
1289 "{}: k must be 5..8, got {}",
1290 entry_label, k
1291 )));
1292 }
1293 let expected_edges = k * (k - 1) / 2;
1294 if edges.len() != expected_edges {
1295 return Err(XlogError::Kernel(format!(
1296 "{}: expected {} edges (= C({}, 2)), got {}",
1297 entry_label,
1298 expected_edges,
1299 k,
1300 edges.len()
1301 )));
1302 }
1303 if usize::try_from(leader_edge_idx)
1304 .ok()
1305 .is_none_or(|idx| idx >= expected_edges)
1306 {
1307 return Err(XlogError::Kernel(format!(
1308 "{}: leader_edge_idx {} out of range for {} edges",
1309 entry_label, leader_edge_idx, expected_edges
1310 )));
1311 }
1312 match (edge_order, iteration_order) {
1313 (Some(edge_order), Some(iteration_order)) => {
1314 validate_clique_u8_permutation(
1315 edge_order,
1316 expected_edges,
1317 "edge_order",
1318 entry_label,
1319 )?;
1320 validate_clique_u8_permutation(iteration_order, k, "iteration_order", entry_label)?;
1321 }
1322 (None, None) => {}
1323 _ => {
1324 return Err(XlogError::Kernel(format!(
1325 "{}: edge_order and iteration_order must both be present or both be omitted",
1326 entry_label
1327 )));
1328 }
1329 }
1330
1331 for (i, buf) in edges.iter().enumerate() {
1333 if buf.arity() != 2 {
1334 return Err(XlogError::Kernel(format!(
1335 "{}: edge[{}] must be 2-column, got arity {}",
1336 entry_label,
1337 i,
1338 buf.arity()
1339 )));
1340 }
1341 for col_idx in 0..2 {
1342 let ty = buf.schema.column_type(col_idx).ok_or_else(|| {
1343 XlogError::Kernel(format!(
1344 "{}: edge[{}] column {} type missing",
1345 entry_label, i, col_idx
1346 ))
1347 })?;
1348 if !width_class.validate_col_type(ty) {
1349 return Err(XlogError::Kernel(format!(
1350 "{}: edge[{}] column {} type {:?} not in {} width-class",
1351 entry_label,
1352 i,
1353 col_idx,
1354 ty,
1355 width_class.label()
1356 )));
1357 }
1358 }
1359 }
1360
1361 let mut head_types = Vec::with_capacity(k);
1365 let leader_slot = edge_order.map(|order| order[0] as usize).unwrap_or(0);
1366 head_types.push(edges[leader_slot].schema.column_type(0).expect("validated"));
1367 head_types.push(edges[leader_slot].schema.column_type(1).expect("validated"));
1368 for i in 2..k {
1369 let logical_edge = i - 1;
1370 let edge_slot = edge_order
1371 .map(|order| order[logical_edge] as usize)
1372 .unwrap_or(logical_edge);
1373 head_types.push(edges[edge_slot].schema.column_type(1).expect("validated"));
1374 }
1375 let out_schema = Schema::new(
1376 head_types
1377 .iter()
1378 .enumerate()
1379 .map(|(i, t)| (format!("col{}", i), *t))
1380 .collect(),
1381 );
1382
1383 let leader_slot = usize::try_from(leader_edge_idx).expect("validated");
1384 let n_leader = self.logical_row_count_u32(edges[leader_slot])?;
1385 if n_leader == 0 {
1386 return self.create_empty_buffer(out_schema);
1387 }
1388 let metadata_start = Instant::now();
1391 let leader_metadata = match width_class {
1392 CliqueWidthClass::FourByte => {
1393 CliqueLeaderMetadata::U32(self.wcoj_clique_metadata_recorded_u32_inner(
1394 k,
1395 edges,
1396 leader_edge_idx,
1397 launch_stream,
1398 entry_label,
1399 )?)
1400 }
1401 CliqueWidthClass::EightByte => {
1402 CliqueLeaderMetadata::U64(self.wcoj_clique_metadata_recorded_u64_inner(
1403 k,
1404 edges,
1405 leader_edge_idx,
1406 launch_stream,
1407 entry_label,
1408 )?)
1409 }
1410 };
1411 self.record_kclique_metadata_build_nanos(metadata_start.elapsed().as_nanos());
1412 let leader_work_total = leader_metadata.total_rows_u32(entry_label)?;
1413 if leader_work_total != n_leader {
1414 return Err(XlogError::Kernel(format!(
1415 "{}: leader metadata total {} does not match leader row count {}",
1416 entry_label, leader_work_total, n_leader
1417 )));
1418 }
1419 let leader_metadata_key_count = leader_metadata.key_count();
1420 if leader_metadata_key_count == 0 {
1421 return self.create_empty_buffer(out_schema);
1422 }
1423
1424 let mut edge_col0_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
1428 let mut edge_col1_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
1429 let mut edge_n_host: Vec<u32> = Vec::with_capacity(expected_edges);
1430 for buf in edges.iter() {
1431 let col0 = buf.column(0).expect("validated");
1432 let col1 = buf.column(1).expect("validated");
1433 edge_col0_ptrs.push(*col0.device_ptr());
1434 edge_col1_ptrs.push(*col1.device_ptr());
1435 edge_n_host.push(self.logical_row_count_u32(buf)?);
1436 }
1437
1438 let mut d_edge_col0 = self.memory.alloc::<u64>(expected_edges)?;
1440 let mut d_edge_col1 = self.memory.alloc::<u64>(expected_edges)?;
1441 let mut d_edge_n = self.memory.alloc::<u32>(expected_edges)?;
1442 let device = self.device.inner();
1443 self.htod_launch_metadata_sync_copy_into(&edge_col0_ptrs, &mut d_edge_col0)
1444 .map_err(|e| {
1445 XlogError::Kernel(format!(
1446 "{}: htod edge_col0_ptrs failed: {}",
1447 entry_label, e
1448 ))
1449 })?;
1450 self.htod_launch_metadata_sync_copy_into(&edge_col1_ptrs, &mut d_edge_col1)
1451 .map_err(|e| {
1452 XlogError::Kernel(format!(
1453 "{}: htod edge_col1_ptrs failed: {}",
1454 entry_label, e
1455 ))
1456 })?;
1457 self.htod_launch_metadata_sync_copy_into(&edge_n_host, &mut d_edge_n)
1458 .map_err(|e| {
1459 XlogError::Kernel(format!("{}: htod edge_n failed: {}", entry_label, e))
1460 })?;
1461 let d_edge_order = if let Some(edge_order) = edge_order {
1462 let mut buf = self.memory.alloc::<u8>(expected_edges)?;
1463 self.htod_launch_metadata_sync_copy_into(edge_order, &mut buf)
1464 .map_err(|e| {
1465 XlogError::Kernel(format!("{}: htod edge_order failed: {}", entry_label, e))
1466 })?;
1467 Some(buf)
1468 } else {
1469 None
1470 };
1471 let d_iteration_order = if let Some(iteration_order) = iteration_order {
1472 let mut buf = self.memory.alloc::<u8>(k)?;
1473 self.htod_launch_metadata_sync_copy_into(iteration_order, &mut buf)
1474 .map_err(|e| {
1475 XlogError::Kernel(format!(
1476 "{}: htod iteration_order failed: {}",
1477 entry_label, e
1478 ))
1479 })?;
1480 Some(buf)
1481 } else {
1482 None
1483 };
1484
1485 let block_work_unit = crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT;
1487 let grid = leader_work_total.div_ceil(block_work_unit);
1488 let count_buf = self.memory.alloc::<u32>(grid as usize)?;
1489 let thread_counts_buf = self
1490 .memory
1491 .alloc::<u32>((grid as usize) * (BLOCK_SIZE as usize))?;
1492 let mut offsets_buf = self.memory.alloc::<u32>(grid as usize)?;
1493 let d_total = self.memory.alloc::<u32>(1)?;
1494
1495 let mut scan_scratch = self.multiblock_scan_u32_scratch_for_len(grid)?;
1496 let mut rec_count = LaunchRecorder::new_strict(launch_stream);
1497 for buf in edges.iter() {
1498 rec_count.read(buf.num_rows_device());
1499 rec_count.read_column(buf.column(0).expect("validated"));
1500 rec_count.read_column(buf.column(1).expect("validated"));
1501 }
1502 rec_count.read(&d_edge_col0);
1503 rec_count.read(&d_edge_col1);
1504 rec_count.read(&d_edge_n);
1505 if let Some(buf) = d_edge_order.as_ref() {
1506 rec_count.read(buf);
1507 }
1508 if let Some(buf) = d_iteration_order.as_ref() {
1509 rec_count.read(buf);
1510 }
1511 match &leader_metadata {
1512 CliqueLeaderMetadata::U32(leader_metadata) => {
1513 rec_count.read(&leader_metadata.unique_keys);
1514 rec_count.read(&leader_metadata.fan_out);
1515 rec_count.read(&leader_metadata.prefix_sum);
1516 }
1517 CliqueLeaderMetadata::U64(leader_metadata) => {
1518 rec_count.read(&leader_metadata.unique_keys);
1519 rec_count.read(&leader_metadata.fan_out);
1520 rec_count.read(&leader_metadata.prefix_sum);
1521 }
1522 }
1523 rec_count.write(&count_buf);
1524 rec_count.write(&thread_counts_buf);
1525 rec_count.write(&offsets_buf);
1526 rec_count.write(&d_total);
1527 for level in scan_scratch.levels() {
1528 rec_count.read_write(level);
1529 }
1530 rec_count.preflight(runtime).map_err(|e| {
1531 XlogError::Kernel(format!("{}: count preflight failed: {}", entry_label, e))
1532 })?;
1533
1534 let count_kernel = device
1535 .get_func(WCOJ_MODULE, clique_kernel_name(k, false, width_class))
1536 .ok_or_else(|| {
1537 XlogError::Kernel(format!(
1538 "{}: count kernel '{}' not found",
1539 entry_label,
1540 clique_kernel_name(k, false, width_class)
1541 ))
1542 })?;
1543 let count_config = LaunchConfig {
1544 grid_dim: (grid, 1, 1),
1545 block_dim: (BLOCK_SIZE, 1, 1),
1546 shared_mem_bytes: 0,
1547 };
1548
1549 let null_order_ptr = 0_u64;
1569 let edge_order_param = match d_edge_order.as_ref() {
1570 Some(buf) => buf.as_kernel_param(),
1571 None => null_order_ptr.as_kernel_param(),
1572 };
1573 let iteration_order_param = match d_iteration_order.as_ref() {
1574 Some(buf) => buf.as_kernel_param(),
1575 None => null_order_ptr.as_kernel_param(),
1576 };
1577 let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
1578 unsafe {
1579 let mut params: Vec<*mut c_void> = match &leader_metadata {
1580 CliqueLeaderMetadata::U32(leader_metadata) => vec![
1581 (&d_edge_col0).as_kernel_param(),
1582 (&d_edge_col1).as_kernel_param(),
1583 (&d_edge_n).as_kernel_param(),
1584 leader_edge_idx.as_kernel_param(),
1585 edge_order_param,
1586 iteration_order_param,
1587 n_leader.as_kernel_param(),
1588 (&leader_metadata.unique_keys).as_kernel_param(),
1589 (&leader_metadata.fan_out).as_kernel_param(),
1590 (&leader_metadata.prefix_sum).as_kernel_param(),
1591 leader_metadata_key_count.as_kernel_param(),
1592 block_work_unit.as_kernel_param(),
1593 (&count_buf).as_kernel_param(),
1594 (&thread_counts_buf).as_kernel_param(),
1595 ],
1596 CliqueLeaderMetadata::U64(leader_metadata) => vec![
1597 (&d_edge_col0).as_kernel_param(),
1598 (&d_edge_col1).as_kernel_param(),
1599 (&d_edge_n).as_kernel_param(),
1600 leader_edge_idx.as_kernel_param(),
1601 edge_order_param,
1602 iteration_order_param,
1603 n_leader.as_kernel_param(),
1604 (&leader_metadata.unique_keys).as_kernel_param(),
1605 (&leader_metadata.fan_out).as_kernel_param(),
1606 (&leader_metadata.prefix_sum).as_kernel_param(),
1607 leader_metadata_key_count.as_kernel_param(),
1608 block_work_unit.as_kernel_param(),
1609 (&count_buf).as_kernel_param(),
1610 (&thread_counts_buf).as_kernel_param(),
1611 ],
1612 };
1613 count_kernel
1614 .clone()
1615 .launch_in(stream, count_config, &mut params)
1616 .map_err(|e| {
1617 XlogError::Kernel(format!(
1618 "{}: count kernel launch failed: {}",
1619 entry_label, e
1620 ))
1621 })?;
1622 }
1623
1624 let bytes_count = (grid as usize) * std::mem::size_of::<u32>();
1626 unsafe {
1627 let res = sys::cuMemcpyDtoDAsync_v2(
1628 *offsets_buf.device_ptr(),
1629 *count_buf.device_ptr(),
1630 bytes_count,
1631 stream.stream().cu_stream(),
1632 );
1633 if res != sys::cudaError_enum::CUDA_SUCCESS {
1634 return Err(XlogError::Kernel(format!(
1635 "{}: dtod count → offsets failed: {:?}",
1636 entry_label, res
1637 )));
1638 }
1639 }
1640
1641 self.multiblock_scan_u32_inplace_on_stream(
1643 &mut offsets_buf,
1644 grid,
1645 stream,
1646 &mut scan_scratch,
1647 )?;
1648
1649 let total_kernel = device
1651 .get_func(WCOJ_MODULE, wcoj_kernels::WCOJ_COMPUTE_TOTAL)
1652 .ok_or_else(|| {
1653 XlogError::Kernel("wcoj_compute_total kernel not found".to_string())
1654 })?;
1655 unsafe {
1656 total_kernel
1657 .clone()
1658 .launch_in(
1659 stream,
1660 LaunchConfig {
1661 grid_dim: (1, 1, 1),
1662 block_dim: (1, 1, 1),
1663 shared_mem_bytes: 0,
1664 },
1665 (&count_buf, &offsets_buf, grid, &d_total),
1666 )
1667 .map_err(|e| {
1668 XlogError::Kernel(format!("wcoj_compute_total launch failed: {}", e))
1669 })?;
1670 }
1671
1672 Ok(())
1673 };
1674 let rec_count = unsafe { rec_count.enqueue_prepared_with(&cu_stream, enqueue) }
1676 .map_err(|error| error.into_xlog_error())?;
1677
1678 rec_count.commit().map_err(|e| {
1679 XlogError::Kernel(format!(
1680 "{}: count+scan+total commit failed: {}",
1681 entry_label, e
1682 ))
1683 })?;
1684
1685 cu_stream.synchronize().map_err(|e| {
1686 XlogError::Kernel(format!(
1687 "{}: stream sync after total failed: {}",
1688 entry_label, e
1689 ))
1690 })?;
1691 let total_rows = self
1692 .dtoh_scalar_untracked::<u32>(&d_total, 0)
1693 .map_err(|e| {
1694 XlogError::Kernel(format!("{}: read d_total failed: {}", entry_label, e))
1695 })?;
1696
1697 if total_rows == 0 {
1698 return self.create_empty_buffer(out_schema);
1699 }
1700
1701 let elem_bytes = width_class.elem_bytes();
1703 let bytes_per_col = (total_rows as usize) * elem_bytes;
1704 let mut out_col_bufs: Vec<TrackedCudaSlice<u8>> = Vec::with_capacity(k);
1705 let mut out_col_ptrs: Vec<u64> = Vec::with_capacity(k);
1706 for _ in 0..k {
1707 let buf = self.memory.alloc::<u8>(bytes_per_col)?;
1708 out_col_ptrs.push(*buf.device_ptr());
1709 out_col_bufs.push(buf);
1710 }
1711 let mut d_out_cols = self.memory.alloc::<u64>(k)?;
1712 self.htod_launch_metadata_sync_copy_into(&out_col_ptrs, &mut d_out_cols)
1713 .map_err(|e| {
1714 XlogError::Kernel(format!("{}: htod out_col_ptrs failed: {}", entry_label, e))
1715 })?;
1716 let out_d_num_rows = self.memory.alloc::<u32>(1)?;
1717
1718 let mut rec_mat = LaunchRecorder::new_strict(launch_stream);
1719 for buf in edges.iter() {
1720 rec_mat.read(buf.num_rows_device());
1721 rec_mat.read_column(buf.column(0).expect("validated"));
1722 rec_mat.read_column(buf.column(1).expect("validated"));
1723 }
1724 rec_mat.read(&d_edge_col0);
1725 rec_mat.read(&d_edge_col1);
1726 rec_mat.read(&d_edge_n);
1727 if let Some(buf) = d_edge_order.as_ref() {
1728 rec_mat.read(buf);
1729 }
1730 if let Some(buf) = d_iteration_order.as_ref() {
1731 rec_mat.read(buf);
1732 }
1733 match &leader_metadata {
1734 CliqueLeaderMetadata::U32(leader_metadata) => {
1735 rec_mat.read(&leader_metadata.unique_keys);
1736 rec_mat.read(&leader_metadata.fan_out);
1737 rec_mat.read(&leader_metadata.prefix_sum);
1738 }
1739 CliqueLeaderMetadata::U64(leader_metadata) => {
1740 rec_mat.read(&leader_metadata.unique_keys);
1741 rec_mat.read(&leader_metadata.fan_out);
1742 rec_mat.read(&leader_metadata.prefix_sum);
1743 }
1744 }
1745 rec_mat.read(&thread_counts_buf);
1746 rec_mat.read(&offsets_buf);
1747 rec_mat.read(&d_out_cols);
1748 for buf in out_col_bufs.iter() {
1749 rec_mat.write(buf);
1750 }
1751 rec_mat.write(&out_d_num_rows);
1752 rec_mat.preflight(runtime).map_err(|e| {
1753 XlogError::Kernel(format!(
1754 "{}: materialize preflight failed: {}",
1755 entry_label, e
1756 ))
1757 })?;
1758
1759 let materialize_kernel = device
1760 .get_func(WCOJ_MODULE, clique_kernel_name(k, true, width_class))
1761 .ok_or_else(|| {
1762 XlogError::Kernel(format!(
1763 "{}: materialize kernel '{}' not found",
1764 entry_label,
1765 clique_kernel_name(k, true, width_class)
1766 ))
1767 })?;
1768
1769 let mat_config = LaunchConfig {
1790 grid_dim: (grid, 1, 1),
1791 block_dim: (BLOCK_SIZE, 1, 1),
1792 shared_mem_bytes: 0,
1793 };
1794 let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
1795 self.initialize_launch_metadata_u32(
1796 total_rows,
1797 &out_d_num_rows,
1798 stream,
1799 &format!("{entry_label}: out_d_num_rows"),
1800 )?;
1801 unsafe {
1802 let mut params: Vec<*mut c_void> = match &leader_metadata {
1803 CliqueLeaderMetadata::U32(leader_metadata) => vec![
1804 (&d_edge_col0).as_kernel_param(),
1805 (&d_edge_col1).as_kernel_param(),
1806 (&d_edge_n).as_kernel_param(),
1807 leader_edge_idx.as_kernel_param(),
1808 edge_order_param,
1809 iteration_order_param,
1810 n_leader.as_kernel_param(),
1811 (&leader_metadata.unique_keys).as_kernel_param(),
1812 (&leader_metadata.fan_out).as_kernel_param(),
1813 (&leader_metadata.prefix_sum).as_kernel_param(),
1814 leader_metadata_key_count.as_kernel_param(),
1815 block_work_unit.as_kernel_param(),
1816 (&thread_counts_buf).as_kernel_param(),
1817 (&offsets_buf).as_kernel_param(),
1818 total_rows.as_kernel_param(),
1819 (&d_out_cols).as_kernel_param(),
1820 ],
1821 CliqueLeaderMetadata::U64(leader_metadata) => vec![
1822 (&d_edge_col0).as_kernel_param(),
1823 (&d_edge_col1).as_kernel_param(),
1824 (&d_edge_n).as_kernel_param(),
1825 leader_edge_idx.as_kernel_param(),
1826 edge_order_param,
1827 iteration_order_param,
1828 n_leader.as_kernel_param(),
1829 (&leader_metadata.unique_keys).as_kernel_param(),
1830 (&leader_metadata.fan_out).as_kernel_param(),
1831 (&leader_metadata.prefix_sum).as_kernel_param(),
1832 leader_metadata_key_count.as_kernel_param(),
1833 block_work_unit.as_kernel_param(),
1834 (&thread_counts_buf).as_kernel_param(),
1835 (&offsets_buf).as_kernel_param(),
1836 total_rows.as_kernel_param(),
1837 (&d_out_cols).as_kernel_param(),
1838 ],
1839 };
1840 materialize_kernel
1841 .clone()
1842 .launch_in(stream, mat_config, &mut params)
1843 .map_err(|e| {
1844 XlogError::Kernel(format!(
1845 "{}: materialize launch failed: {}",
1846 entry_label, e
1847 ))
1848 })?;
1849 }
1850 Ok(())
1851 };
1852 let rec_mat = unsafe { rec_mat.enqueue_prepared_with(&cu_stream, enqueue) }
1854 .map_err(|error| error.into_xlog_error())?;
1855
1856 rec_mat.commit().map_err(|e| {
1857 XlogError::Kernel(format!("{}: materialize commit failed: {}", entry_label, e))
1858 })?;
1859
1860 let columns: Vec<CudaColumn> = out_col_bufs.into_iter().map(|b| b.into()).collect();
1861 Ok(CudaBuffer::from_columns_with_host_count(
1862 columns,
1863 total_rows as u64,
1864 out_d_num_rows,
1865 out_schema,
1866 total_rows,
1867 ))
1868 }
1869
1870 pub fn wcoj_clique5_u32_recorded(
1880 &self,
1881 edges: &[&CudaBuffer; 10],
1882 launch_stream: StreamId,
1883 ) -> Result<CudaBuffer> {
1884 self.wcoj_clique_recorded_inner(
1885 5,
1886 edges,
1887 0,
1888 None,
1889 None,
1890 CliqueWidthClass::FourByte,
1891 launch_stream,
1892 "wcoj_clique5_u32_recorded",
1893 )
1894 }
1895
1896 pub fn wcoj_clique5_u32_recorded_planned(
1898 &self,
1899 edges: &[&CudaBuffer; 10],
1900 leader_edge_idx: u32,
1901 edge_order: &[u8],
1902 iteration_order: &[u8],
1903 launch_stream: StreamId,
1904 ) -> Result<CudaBuffer> {
1905 self.wcoj_clique_recorded_inner(
1906 5,
1907 edges,
1908 leader_edge_idx,
1909 Some(edge_order),
1910 Some(iteration_order),
1911 CliqueWidthClass::FourByte,
1912 launch_stream,
1913 "wcoj_clique5_u32_recorded_planned",
1914 )
1915 }
1916
1917 pub fn wcoj_clique5_u64_recorded(
1919 &self,
1920 edges: &[&CudaBuffer; 10],
1921 launch_stream: StreamId,
1922 ) -> Result<CudaBuffer> {
1923 self.wcoj_clique_recorded_inner(
1924 5,
1925 edges,
1926 0,
1927 None,
1928 None,
1929 CliqueWidthClass::EightByte,
1930 launch_stream,
1931 "wcoj_clique5_u64_recorded",
1932 )
1933 }
1934
1935 pub fn wcoj_clique5_u64_recorded_planned(
1937 &self,
1938 edges: &[&CudaBuffer; 10],
1939 leader_edge_idx: u32,
1940 edge_order: &[u8],
1941 iteration_order: &[u8],
1942 launch_stream: StreamId,
1943 ) -> Result<CudaBuffer> {
1944 self.wcoj_clique_recorded_inner(
1945 5,
1946 edges,
1947 leader_edge_idx,
1948 Some(edge_order),
1949 Some(iteration_order),
1950 CliqueWidthClass::EightByte,
1951 launch_stream,
1952 "wcoj_clique5_u64_recorded_planned",
1953 )
1954 }
1955
1956 pub fn wcoj_clique6_u32_recorded(
1962 &self,
1963 edges: &[&CudaBuffer; 15],
1964 launch_stream: StreamId,
1965 ) -> Result<CudaBuffer> {
1966 self.wcoj_clique_recorded_inner(
1967 6,
1968 edges,
1969 0,
1970 None,
1971 None,
1972 CliqueWidthClass::FourByte,
1973 launch_stream,
1974 "wcoj_clique6_u32_recorded",
1975 )
1976 }
1977
1978 pub fn wcoj_clique6_u32_recorded_planned(
1980 &self,
1981 edges: &[&CudaBuffer; 15],
1982 leader_edge_idx: u32,
1983 edge_order: &[u8],
1984 iteration_order: &[u8],
1985 launch_stream: StreamId,
1986 ) -> Result<CudaBuffer> {
1987 self.wcoj_clique_recorded_inner(
1988 6,
1989 edges,
1990 leader_edge_idx,
1991 Some(edge_order),
1992 Some(iteration_order),
1993 CliqueWidthClass::FourByte,
1994 launch_stream,
1995 "wcoj_clique6_u32_recorded_planned",
1996 )
1997 }
1998
1999 pub fn wcoj_clique6_u64_recorded(
2001 &self,
2002 edges: &[&CudaBuffer; 15],
2003 launch_stream: StreamId,
2004 ) -> Result<CudaBuffer> {
2005 self.wcoj_clique_recorded_inner(
2006 6,
2007 edges,
2008 0,
2009 None,
2010 None,
2011 CliqueWidthClass::EightByte,
2012 launch_stream,
2013 "wcoj_clique6_u64_recorded",
2014 )
2015 }
2016
2017 pub fn wcoj_clique6_u64_recorded_planned(
2019 &self,
2020 edges: &[&CudaBuffer; 15],
2021 leader_edge_idx: u32,
2022 edge_order: &[u8],
2023 iteration_order: &[u8],
2024 launch_stream: StreamId,
2025 ) -> Result<CudaBuffer> {
2026 self.wcoj_clique_recorded_inner(
2027 6,
2028 edges,
2029 leader_edge_idx,
2030 Some(edge_order),
2031 Some(iteration_order),
2032 CliqueWidthClass::EightByte,
2033 launch_stream,
2034 "wcoj_clique6_u64_recorded_planned",
2035 )
2036 }
2037
2038 pub fn wcoj_clique7_u32_recorded(
2040 &self,
2041 edges: &[&CudaBuffer; 21],
2042 launch_stream: StreamId,
2043 ) -> Result<CudaBuffer> {
2044 self.wcoj_clique_recorded_inner(
2045 7,
2046 edges,
2047 0,
2048 None,
2049 None,
2050 CliqueWidthClass::FourByte,
2051 launch_stream,
2052 "wcoj_clique7_u32_recorded",
2053 )
2054 }
2055
2056 pub fn wcoj_clique7_u32_recorded_planned(
2058 &self,
2059 edges: &[&CudaBuffer; 21],
2060 leader_edge_idx: u32,
2061 edge_order: &[u8],
2062 iteration_order: &[u8],
2063 launch_stream: StreamId,
2064 ) -> Result<CudaBuffer> {
2065 self.wcoj_clique_recorded_inner(
2066 7,
2067 edges,
2068 leader_edge_idx,
2069 Some(edge_order),
2070 Some(iteration_order),
2071 CliqueWidthClass::FourByte,
2072 launch_stream,
2073 "wcoj_clique7_u32_recorded_planned",
2074 )
2075 }
2076
2077 pub fn wcoj_clique7_u64_recorded(
2079 &self,
2080 edges: &[&CudaBuffer; 21],
2081 launch_stream: StreamId,
2082 ) -> Result<CudaBuffer> {
2083 self.wcoj_clique_recorded_inner(
2084 7,
2085 edges,
2086 0,
2087 None,
2088 None,
2089 CliqueWidthClass::EightByte,
2090 launch_stream,
2091 "wcoj_clique7_u64_recorded",
2092 )
2093 }
2094
2095 pub fn wcoj_clique7_u64_recorded_planned(
2097 &self,
2098 edges: &[&CudaBuffer; 21],
2099 leader_edge_idx: u32,
2100 edge_order: &[u8],
2101 iteration_order: &[u8],
2102 launch_stream: StreamId,
2103 ) -> Result<CudaBuffer> {
2104 self.wcoj_clique_recorded_inner(
2105 7,
2106 edges,
2107 leader_edge_idx,
2108 Some(edge_order),
2109 Some(iteration_order),
2110 CliqueWidthClass::EightByte,
2111 launch_stream,
2112 "wcoj_clique7_u64_recorded_planned",
2113 )
2114 }
2115
2116 pub fn wcoj_clique8_u32_recorded(
2118 &self,
2119 edges: &[&CudaBuffer; 28],
2120 launch_stream: StreamId,
2121 ) -> Result<CudaBuffer> {
2122 self.wcoj_clique_recorded_inner(
2123 8,
2124 edges,
2125 0,
2126 None,
2127 None,
2128 CliqueWidthClass::FourByte,
2129 launch_stream,
2130 "wcoj_clique8_u32_recorded",
2131 )
2132 }
2133
2134 pub fn wcoj_clique8_u32_recorded_planned(
2136 &self,
2137 edges: &[&CudaBuffer; 28],
2138 leader_edge_idx: u32,
2139 edge_order: &[u8],
2140 iteration_order: &[u8],
2141 launch_stream: StreamId,
2142 ) -> Result<CudaBuffer> {
2143 self.wcoj_clique_recorded_inner(
2144 8,
2145 edges,
2146 leader_edge_idx,
2147 Some(edge_order),
2148 Some(iteration_order),
2149 CliqueWidthClass::FourByte,
2150 launch_stream,
2151 "wcoj_clique8_u32_recorded_planned",
2152 )
2153 }
2154
2155 pub fn wcoj_clique8_u64_recorded(
2157 &self,
2158 edges: &[&CudaBuffer; 28],
2159 launch_stream: StreamId,
2160 ) -> Result<CudaBuffer> {
2161 self.wcoj_clique_recorded_inner(
2162 8,
2163 edges,
2164 0,
2165 None,
2166 None,
2167 CliqueWidthClass::EightByte,
2168 launch_stream,
2169 "wcoj_clique8_u64_recorded",
2170 )
2171 }
2172
2173 pub fn wcoj_clique8_u64_recorded_planned(
2175 &self,
2176 edges: &[&CudaBuffer; 28],
2177 leader_edge_idx: u32,
2178 edge_order: &[u8],
2179 iteration_order: &[u8],
2180 launch_stream: StreamId,
2181 ) -> Result<CudaBuffer> {
2182 self.wcoj_clique_recorded_inner(
2183 8,
2184 edges,
2185 leader_edge_idx,
2186 Some(edge_order),
2187 Some(iteration_order),
2188 CliqueWidthClass::EightByte,
2189 launch_stream,
2190 "wcoj_clique8_u64_recorded_planned",
2191 )
2192 }
2193
2194 #[expect(
2219 clippy::too_many_arguments,
2220 reason = "fused clique aggregation keeps every relation, variable layout, aggregate selector, output bound, and stream explicit"
2221 )]
2222 fn wcoj_clique_groupby_root_count_recorded_inner(
2223 &self,
2224 k: usize,
2225 edges: &[&CudaBuffer],
2226 leader_edge_idx: u32,
2227 edge_order: &[u8],
2228 iteration_order: &[u8],
2229 launch_stream: StreamId,
2230 entry_label: &str,
2231 ) -> Result<CudaBuffer> {
2232 let runtime = self.memory().runtime().ok_or_else(|| {
2233 XlogError::Kernel(format!(
2234 "{} requires a runtime-backed GpuMemoryManager (with_runtime)",
2235 entry_label
2236 ))
2237 })?;
2238 let cu_stream = runtime
2239 .stream_pool()
2240 .resolve(launch_stream)
2241 .ok_or_else(|| {
2242 XlogError::Kernel(format!(
2243 "{}: launch_stream StreamId({}) does not resolve",
2244 entry_label, launch_stream.0
2245 ))
2246 })?;
2247 if !matches!(k, 5 | 6) {
2248 return Err(XlogError::Kernel(format!(
2249 "{}: fused count-by-root supports k 5..6, got {}",
2250 entry_label, k
2251 )));
2252 }
2253 let expected_edges = k * (k - 1) / 2;
2254 if edges.len() != expected_edges {
2255 return Err(XlogError::Kernel(format!(
2256 "{}: expected {} edges (= C({}, 2)), got {}",
2257 entry_label,
2258 expected_edges,
2259 k,
2260 edges.len()
2261 )));
2262 }
2263 let leader_slot = usize::try_from(leader_edge_idx)
2264 .ok()
2265 .filter(|idx| *idx < expected_edges)
2266 .ok_or_else(|| {
2267 XlogError::Kernel(format!(
2268 "{}: leader_edge_idx {} out of range for {} edges",
2269 entry_label, leader_edge_idx, expected_edges
2270 ))
2271 })?;
2272 validate_clique_u8_permutation(edge_order, expected_edges, "edge_order", entry_label)?;
2273 validate_clique_u8_permutation(iteration_order, k, "iteration_order", entry_label)?;
2274
2275 let mut laid_out: Vec<CudaBuffer> = Vec::with_capacity(expected_edges);
2279 for buf in edges {
2280 laid_out.push(self.wcoj_layout_u32_recorded(buf, launch_stream)?);
2281 }
2282 let edges: Vec<&CudaBuffer> = laid_out.iter().collect();
2283 let leader = edges[leader_slot];
2284
2285 let w_type = leader.schema.column_type(0).ok_or_else(|| {
2286 XlogError::Kernel(format!(
2287 "{}: leader edge column 0 type missing",
2288 entry_label
2289 ))
2290 })?;
2291 let out_schema = Schema::new(vec![
2292 ("root".to_string(), w_type),
2293 ("count".to_string(), ScalarType::U64),
2294 ]);
2295 let n_leader = self.logical_row_count_u32(leader)?;
2296 if n_leader == 0 {
2297 return self.create_empty_buffer(out_schema);
2298 }
2299
2300 let leader_metadata = self.wcoj_clique_metadata_recorded_u32_inner(
2301 k,
2302 &edges,
2303 leader_edge_idx,
2304 launch_stream,
2305 entry_label,
2306 )?;
2307 let leader_work_total = u32::try_from(leader_metadata.total).map_err(|_| {
2308 XlogError::Kernel(format!(
2309 "{}: leader metadata total {} exceeds u32 kernel surface",
2310 entry_label, leader_metadata.total
2311 ))
2312 })?;
2313 if leader_work_total != n_leader {
2314 return Err(XlogError::Kernel(format!(
2315 "{}: leader metadata total {} does not match leader row count {}",
2316 entry_label, leader_work_total, n_leader
2317 )));
2318 }
2319 if leader_metadata.key_count == 0 {
2320 return self.create_empty_buffer(out_schema);
2321 }
2322
2323 let mut edge_col0_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
2327 let mut edge_col1_ptrs: Vec<u64> = Vec::with_capacity(expected_edges);
2328 let mut edge_n_host: Vec<u32> = Vec::with_capacity(expected_edges);
2329 for buf in edges.iter() {
2330 let col0 = buf.column(0).ok_or_else(|| {
2331 XlogError::Kernel(format!("{}: edge column 0 missing", entry_label))
2332 })?;
2333 let col1 = buf.column(1).ok_or_else(|| {
2334 XlogError::Kernel(format!("{}: edge column 1 missing", entry_label))
2335 })?;
2336 edge_col0_ptrs.push(*col0.device_ptr());
2337 edge_col1_ptrs.push(*col1.device_ptr());
2338 edge_n_host.push(self.logical_row_count_u32(buf)?);
2339 }
2340 let mut d_edge_col0 = self.memory.alloc::<u64>(expected_edges)?;
2341 let mut d_edge_col1 = self.memory.alloc::<u64>(expected_edges)?;
2342 let mut d_edge_n = self.memory.alloc::<u32>(expected_edges)?;
2343 self.htod_launch_metadata_sync_copy_into(&edge_col0_ptrs, &mut d_edge_col0)
2344 .map_err(|e| {
2345 XlogError::Kernel(format!(
2346 "{}: htod edge_col0_ptrs failed: {}",
2347 entry_label, e
2348 ))
2349 })?;
2350 self.htod_launch_metadata_sync_copy_into(&edge_col1_ptrs, &mut d_edge_col1)
2351 .map_err(|e| {
2352 XlogError::Kernel(format!(
2353 "{}: htod edge_col1_ptrs failed: {}",
2354 entry_label, e
2355 ))
2356 })?;
2357 self.htod_launch_metadata_sync_copy_into(&edge_n_host, &mut d_edge_n)
2358 .map_err(|e| {
2359 XlogError::Kernel(format!("{}: htod edge_n failed: {}", entry_label, e))
2360 })?;
2361 let mut d_edge_order = self.memory.alloc::<u8>(expected_edges)?;
2362 self.htod_launch_metadata_sync_copy_into(edge_order, &mut d_edge_order)
2363 .map_err(|e| {
2364 XlogError::Kernel(format!("{}: htod edge_order failed: {}", entry_label, e))
2365 })?;
2366 let mut d_iteration_order = self.memory.alloc::<u8>(k)?;
2367 self.htod_launch_metadata_sync_copy_into(iteration_order, &mut d_iteration_order)
2368 .map_err(|e| {
2369 XlogError::Kernel(format!(
2370 "{}: htod iteration_order failed: {}",
2371 entry_label, e
2372 ))
2373 })?;
2374
2375 let mut row_counts = self
2379 .memory()
2380 .alloc::<u8>(n_leader as usize * std::mem::size_of::<u32>())?;
2381 self.device()
2382 .inner()
2383 .memset_zeros(&mut row_counts)
2384 .map_err(|e| {
2385 XlogError::Kernel(format!("{}: zero row counts failed: {}", entry_label, e))
2386 })?;
2387
2388 let block_work_unit = crate::wcoj_metadata::WCOJ_HG_BLOCK_WORK_UNIT_DEFAULT;
2389 let grid = leader_work_total.div_ceil(block_work_unit);
2390 let mut rec = LaunchRecorder::new_strict(launch_stream);
2391 for buf in edges.iter() {
2392 rec.read(buf.num_rows_device());
2393 rec.read_column(buf.column(0).expect("validated"));
2394 rec.read_column(buf.column(1).expect("validated"));
2395 }
2396 rec.read(&d_edge_col0);
2397 rec.read(&d_edge_col1);
2398 rec.read(&d_edge_n);
2399 rec.read(&d_edge_order);
2400 rec.read(&d_iteration_order);
2401 rec.read(&leader_metadata.unique_keys);
2402 rec.read(&leader_metadata.fan_out);
2403 rec.read(&leader_metadata.prefix_sum);
2404 rec.write(&row_counts);
2405 rec.preflight(runtime)
2406 .map_err(|e| XlogError::Kernel(format!("{}: preflight failed: {}", entry_label, e)))?;
2407
2408 let kernel_name = match k {
2409 5 => wcoj_kernels::WCOJ_CLIQUE5_GROUPBY_ROOT_COUNT_HG_U32,
2410 _ => wcoj_kernels::WCOJ_CLIQUE6_GROUPBY_ROOT_COUNT_HG_U32,
2411 };
2412 let kernel = self
2413 .device()
2414 .inner()
2415 .get_func(WCOJ_MODULE, kernel_name)
2416 .ok_or_else(|| {
2417 XlogError::Kernel(format!(
2418 "{}: kernel '{}' not found",
2419 entry_label, kernel_name
2420 ))
2421 })?;
2422 let enqueue = |stream: &crate::launch::CudaEnqueue<'_>| -> Result<()> {
2441 unsafe {
2442 let mut params: Vec<*mut c_void> = vec![
2443 (&d_edge_col0).as_kernel_param(),
2444 (&d_edge_col1).as_kernel_param(),
2445 (&d_edge_n).as_kernel_param(),
2446 leader_edge_idx.as_kernel_param(),
2447 (&d_edge_order).as_kernel_param(),
2448 (&d_iteration_order).as_kernel_param(),
2449 n_leader.as_kernel_param(),
2450 (&leader_metadata.unique_keys).as_kernel_param(),
2451 (&leader_metadata.fan_out).as_kernel_param(),
2452 (&leader_metadata.prefix_sum).as_kernel_param(),
2453 leader_metadata.key_count.as_kernel_param(),
2454 block_work_unit.as_kernel_param(),
2455 (&row_counts).as_kernel_param(),
2456 ];
2457 kernel
2458 .clone()
2459 .launch_in(
2460 stream,
2461 LaunchConfig {
2462 grid_dim: (grid, 1, 1),
2463 block_dim: (BLOCK_SIZE, 1, 1),
2464 shared_mem_bytes: 0,
2465 },
2466 &mut params,
2467 )
2468 .map_err(|e| {
2469 XlogError::Kernel(format!(
2470 "{}: groupby-count launch failed: {}",
2471 entry_label, e
2472 ))
2473 })?;
2474 }
2475 Ok(())
2476 };
2477 let rec = unsafe { rec.enqueue_prepared_with(&cu_stream, enqueue) }
2479 .map_err(|error| error.into_xlog_error())?;
2480 rec.commit()
2481 .map_err(|e| XlogError::Kernel(format!("{}: commit failed: {}", entry_label, e)))?;
2482
2483 let root_src = match leader.column(0).expect("validated") {
2488 CudaColumn::Owned(slice) => slice,
2489 _ => {
2490 return Err(XlogError::Kernel(format!(
2491 "{}: leader.col0 must be an owned CudaColumn",
2492 entry_label
2493 )))
2494 }
2495 };
2496 let root_copy = self
2497 .memory()
2498 .alloc::<u8>(n_leader as usize * std::mem::size_of::<u32>())?;
2499 unsafe {
2503 let res = sys::cuMemcpyDtoD_v2(
2504 *root_copy.device_ptr(),
2505 *root_src.device_ptr(),
2506 n_leader as usize * std::mem::size_of::<u32>(),
2507 );
2508 if res != sys::cudaError_enum::CUDA_SUCCESS {
2509 return Err(XlogError::Kernel(format!(
2510 "{}: copy root column failed: {:?}",
2511 entry_label, res
2512 )));
2513 }
2514 }
2515 let mut d_num_rows = self.memory().alloc::<u32>(1)?;
2516 self.device()
2517 .inner()
2518 .dtod_copy(leader.num_rows_device(), &mut d_num_rows)
2519 .map_err(|e| {
2520 XlogError::Kernel(format!("{}: copy row count failed: {}", entry_label, e))
2521 })?;
2522 let staging_schema = Schema::new(vec![
2523 ("root".to_string(), w_type),
2524 ("count".to_string(), ScalarType::U32),
2525 ]);
2526 let staging = CudaBuffer::from_columns_with_host_count(
2527 vec![root_copy.into(), row_counts.into()],
2528 n_leader as u64,
2529 d_num_rows,
2530 staging_schema,
2531 n_leader,
2532 );
2533
2534 let mask = self.compare_const_mask_recorded::<u32>(
2537 &staging,
2538 1,
2539 0u32,
2540 crate::CompareOp::Gt,
2541 launch_stream,
2542 )?;
2543 let compacted =
2544 self.compact_buffer_by_device_mask_counted_recorded(&staging, &mask, launch_stream)?;
2545 self.groupby_multi_agg_recorded(
2546 &compacted,
2547 &[0],
2548 &[(1, xlog_core::AggOp::Sum)],
2549 launch_stream,
2550 )
2551 }
2552
2553 pub fn wcoj_clique5_groupby_root_count_u32_recorded_planned(
2557 &self,
2558 edges: &[&CudaBuffer; 10],
2559 leader_edge_idx: u32,
2560 edge_order: &[u8],
2561 iteration_order: &[u8],
2562 launch_stream: StreamId,
2563 ) -> Result<CudaBuffer> {
2564 self.wcoj_clique_groupby_root_count_recorded_inner(
2565 5,
2566 edges,
2567 leader_edge_idx,
2568 edge_order,
2569 iteration_order,
2570 launch_stream,
2571 "wcoj_clique5_groupby_root_count_u32_recorded_planned",
2572 )
2573 }
2574
2575 pub fn wcoj_clique6_groupby_root_count_u32_recorded_planned(
2579 &self,
2580 edges: &[&CudaBuffer; 15],
2581 leader_edge_idx: u32,
2582 edge_order: &[u8],
2583 iteration_order: &[u8],
2584 launch_stream: StreamId,
2585 ) -> Result<CudaBuffer> {
2586 self.wcoj_clique_groupby_root_count_recorded_inner(
2587 6,
2588 edges,
2589 leader_edge_idx,
2590 edge_order,
2591 iteration_order,
2592 launch_stream,
2593 "wcoj_clique6_groupby_root_count_u32_recorded_planned",
2594 )
2595 }
2596}
2597
2598fn validate_clique_u8_permutation(
2599 values: &[u8],
2600 len: usize,
2601 label: &str,
2602 entry_label: &str,
2603) -> Result<()> {
2604 if values.len() != len {
2605 return Err(XlogError::Kernel(format!(
2606 "{}: {} length {} must equal {}",
2607 entry_label,
2608 label,
2609 values.len(),
2610 len
2611 )));
2612 }
2613 let mut seen = vec![false; len];
2614 for &value in values {
2615 let idx = usize::from(value);
2616 if idx >= len {
2617 return Err(XlogError::Kernel(format!(
2618 "{}: {} value {} out of range 0..{}",
2619 entry_label, label, value, len
2620 )));
2621 }
2622 if seen[idx] {
2623 return Err(XlogError::Kernel(format!(
2624 "{}: {} duplicates value {}",
2625 entry_label, label, value
2626 )));
2627 }
2628 seen[idx] = true;
2629 }
2630 Ok(())
2631}