diff --git a/.github/workflows/codspeed.yml b/.github/workflows/codspeed.yml index 96c61baf094..7e98e37f5bc 100644 --- a/.github/workflows/codspeed.yml +++ b/.github/workflows/codspeed.yml @@ -256,7 +256,7 @@ jobs: include: - { shard: 1, name: "Bitpacked", benches: "bitpacked_cuda" } - { shard: 2, name: "Dynamic dispatch", benches: "dynamic_dispatch_cuda" } - - { shard: 3, name: "Standalone kernels", benches: "alp_cuda date_time_parts_cuda dict_cuda fsst_cuda runend_cuda" } + - { shard: 3, name: "Standalone kernels", benches: "alp_cuda date_time_parts_cuda delta_cuda dict_cuda fsst_cuda runend_cuda" } name: "Benchmark with Codspeed (CUDA Shard #${{ matrix.shard }} - ${{ matrix.name }})" timeout-minutes: 30 runs-on: >- diff --git a/vortex-btrblocks/src/builder.rs b/vortex-btrblocks/src/builder.rs index 6f38e29cd86..cc1477e4de8 100644 --- a/vortex-btrblocks/src/builder.rs +++ b/vortex-btrblocks/src/builder.rs @@ -178,8 +178,10 @@ impl BtrBlocksCompressorBuilder { string::StringDictScheme.id(), binary::BinaryDictScheme.id(), ]; - // Delta has no GPU decode kernel and its prefix-sum decode is inherently sequential, so it - // is incompatible with pure-GPU decompression paths. + // Delta now has a CUDA decode kernel, so arrays that reach the GPU already encoded with + // it — the Delta children OnPair emits, for instance — decode there. It stays excluded + // from this preset until GPU delta decode is benchmarked against the schemes it would + // displace, since the preset picks encodings rather than merely decoding them. #[cfg(feature = "unstable_encodings")] excluded.push(integer::DeltaScheme::default().id()); #[cfg(feature = "pco")] diff --git a/vortex-cuda/Cargo.toml b/vortex-cuda/Cargo.toml index 0364a0c455f..9b231ef39e4 100644 --- a/vortex-cuda/Cargo.toml +++ b/vortex-cuda/Cargo.toml @@ -65,6 +65,10 @@ fastlanes = { workspace = true } name = "for_cuda" harness = false +[[bench]] +name = "delta_cuda" +harness = false + [[bench]] name = "dict_cuda" harness = false diff --git a/vortex-cuda/benches/delta_cuda.rs b/vortex-cuda/benches/delta_cuda.rs new file mode 100644 index 00000000000..1fa72b90d30 --- /dev/null +++ b/vortex-cuda/benches/delta_cuda.rs @@ -0,0 +1,114 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! CUDA benchmarks for FastLanes delta decompression. +//! +//! Each element width is its own case: the lane count is `1024 / bit-width`, so the width sets +//! both how a chunk is split (128 lanes of 8 rows for `u8` through 16 lanes of 64 rows for +//! `u64`) and how much of a block is busy during the scan. + +#![expect(clippy::unwrap_used)] + +mod bench_config; +mod timed_launch_strategy; + +use std::mem::size_of; +use std::sync::Arc; +use std::sync::atomic::Ordering; +use std::time::Duration; + +use criterion::BenchmarkId; +use criterion::Criterion; +use criterion::Throughput; +use cudarc::driver::DeviceRepr; +use futures::executor::block_on; +use vortex::array::IntoArray; +use vortex::array::VortexSessionExecute; +use vortex::array::array_session; +use vortex::array::arrays::PrimitiveArray; +use vortex::array::validity::Validity; +use vortex::buffer::Buffer; +use vortex::dtype::NativePType; +use vortex::encodings::fastlanes::Delta; +use vortex::encodings::fastlanes::DeltaArray; +use vortex::error::VortexExpect; +use vortex_cuda::CudaDispatchMode; +use vortex_cuda::CudaSession; +use vortex_cuda::executor::CudaArrayExt; +use vortex_cuda_macros::cuda_available; +use vortex_cuda_macros::cuda_not_available; + +use crate::bench_config::BENCH_SIZES; +use crate::timed_launch_strategy::TimedLaunchStrategy; + +/// Builds a delta-encoded array of `len` values that stay inside `T`. +fn make_delta_array(len: usize) -> DeltaArray +where + T: NativePType + From, +{ + // A small repeating step keeps every delta narrow, which is the shape delta is chosen for. + let data: Vec = (0..len) + .map(|i| >::from(u8::try_from(i % 251).vortex_expect("modulo fits u8"))) + .collect(); + let primitive = PrimitiveArray::new(Buffer::from(data), Validity::NonNullable); + + let mut ctx = array_session().create_execution_ctx(); + Delta::try_from_primitive_array(&primitive, &mut ctx).vortex_expect("failed to delta encode") +} + +fn benchmark_delta_typed(c: &mut Criterion, type_name: &str) +where + T: NativePType + DeviceRepr + From, +{ + let mut group = c.benchmark_group("cuda"); + + for &(len, len_str) in BENCH_SIZES { + group.throughput(Throughput::Bytes((len * size_of::()) as u64)); + + let delta = make_delta_array::(len); + + group.bench_with_input( + BenchmarkId::new(format!("cuda/delta/{type_name}"), len_str), + &delta, + |b, delta| { + b.iter_custom(|iters| { + let timed = TimedLaunchStrategy::default(); + let timer = timed.timer(); + + let mut cuda_ctx = + CudaSession::create_execution_ctx(&vortex_cuda::cuda_session()) + .vortex_expect("failed to create execution context") + .with_dispatch_mode(CudaDispatchMode::StandaloneOnly) + .with_launch_strategy(Arc::new(timed)); + + for _ in 0..iters { + block_on(delta.clone().into_array().execute_cuda(&mut cuda_ctx)).unwrap(); + } + + Duration::from_nanos(timer.load(Ordering::Relaxed)) + }); + }, + ); + } + + group.finish(); +} + +fn benchmark_delta(c: &mut Criterion) { + benchmark_delta_typed::(c, "u8"); + benchmark_delta_typed::(c, "u16"); + benchmark_delta_typed::(c, "u32"); + benchmark_delta_typed::(c, "u64"); +} + +criterion::criterion_group! { + name = benches; + config = bench_config::cuda_bench_config(); + targets = benchmark_delta +} + +#[cuda_available] +criterion::criterion_main!(benches); + +#[cuda_not_available] +fn main() {} diff --git a/vortex-cuda/kernels/src/delta.cu b/vortex-cuda/kernels/src/delta.cu new file mode 100644 index 00000000000..9e7cd4e74aa --- /dev/null +++ b/vortex-cuda/kernels/src/delta.cu @@ -0,0 +1,78 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +#include "fastlanes_common.cuh" +#include "types.cuh" + +// FastLanes delta decode. +// +// Delta is stored in the FastLanes transposed layout, where a 1024-element chunk is +// FL_LANES independent columns of (1024 / FL_LANES) rows. Each column carries its own +// running total seeded from that lane's base, so the chunk decodes as FL_LANES independent +// sequential scans rather than one scan over 1024 elements. One thread owns one lane. +// +// This mirrors `fastlanes::Delta::undelta` followed by `Transpose::untranspose`, which is what +// the CPU decoder (`delta_decompress`) runs. Both steps must stay in step with that crate. + +/// Maps a position in the transposed layout to its position in natural order. +/// +/// Mirrors `fastlanes::transpose`; `untranspose` is `output[transpose(i)] = input[i]`. +__device__ inline uint32_t fl_transpose_index(uint32_t idx) { + const uint32_t lane = idx % 16; + const uint32_t order = (idx / 16) % 8; + const uint32_t row = idx / 128; + return (lane * 64) + (FL_ORDER[order] * 8) + row; +} + +/// Decodes `num_chunks` full 1024-element delta chunks. +/// +/// `deltas` and `output` hold `num_chunks * FL_CHUNK` elements; `bases` holds +/// `num_chunks * FL_LANES`. Only unsigned types are instantiated: the CPU decoder +/// reinterprets signed input through its unsigned counterpart so that the wrapping add here +/// inverts the wrapping subtract done at compress time. +template +__device__ void delta_decode_kernel(const T *const __restrict deltas, + const T *const __restrict bases, + T *const __restrict output, + uint64_t num_chunks) { + constexpr uint32_t LANES = FL_LANES; + constexpr uint32_t ROWS = FL_CHUNK / LANES; + + // The undelta pass writes the chunk in transposed order; the untranspose pass then reads it + // back in a different order, so the whole chunk is staged in shared memory between them. + __shared__ T transposed[FL_CHUNK]; + + for (uint64_t chunk = blockIdx.x; chunk < num_chunks; chunk += gridDim.x) { + const T *const in = deltas + chunk * FL_CHUNK; + const T *const base = bases + chunk * LANES; + + // Each lane accumulates down its own column, seeded by that lane's base. + for (uint32_t lane = threadIdx.x; lane < LANES; lane += blockDim.x) { + T running = base[lane]; + for (uint32_t row = 0; row < ROWS; ++row) { + const uint32_t idx = INDEX(row, lane); + running = static_cast(running + in[idx]); + transposed[idx] = running; + } + } + __syncthreads(); + + T *const out = output + chunk * FL_CHUNK; + for (uint32_t i = threadIdx.x; i < FL_CHUNK; i += blockDim.x) { + out[fl_transpose_index(i)] = transposed[i]; + } + // Guard the staging buffer before the next chunk overwrites it. + __syncthreads(); + } +} + +#define GENERATE_DELTA_KERNEL(suffix, Type) \ + extern "C" __global__ void delta_##suffix(const Type *const __restrict deltas, \ + const Type *const __restrict bases, \ + Type *const __restrict output, \ + uint64_t num_chunks) { \ + delta_decode_kernel(deltas, bases, output, num_chunks); \ + } + +// Signed input is reinterpreted to its unsigned counterpart before the launch. +FOR_EACH_UNSIGNED_INT(GENERATE_DELTA_KERNEL) diff --git a/vortex-cuda/src/kernel/encodings/delta.rs b/vortex-cuda/src/kernel/encodings/delta.rs new file mode 100644 index 00000000000..5dde85cfd65 --- /dev/null +++ b/vortex-cuda/src/kernel/encodings/delta.rs @@ -0,0 +1,285 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! CUDA executor for FastLanes delta. +//! +//! Delta stores its values in the FastLanes transposed layout: a 1024-element chunk is +//! `LANES` independent columns, each carrying its own running total seeded from that lane's +//! base. A chunk therefore decodes as `LANES` independent scans, not one scan of 1024 +//! elements, so the whole array is data-parallel across both chunks and lanes. +//! +//! This mirrors the CPU decoder in `vortex-fastlanes`: `undelta` into the transposed layout, +//! then untranspose into natural order, then apply the array's logical slice. + +use std::fmt::Debug; +use std::sync::Arc; + +use async_trait::async_trait; +use cudarc::driver::LaunchConfig; +use cudarc::driver::PushKernelArg; +use tracing::instrument; +use vortex::array::ArrayRef; +use vortex::array::Canonical; +use vortex::array::arrays::PrimitiveArray; +use vortex::array::arrays::primitive::PrimitiveDataParts; +use vortex::array::buffer::BufferHandle; +use vortex::array::match_each_unsigned_integer_ptype; +use vortex::dtype::NativePType; +use vortex::encodings::fastlanes::Delta; +use vortex::encodings::fastlanes::DeltaArray; +use vortex::encodings::fastlanes::DeltaArrayExt; +use vortex::encodings::fastlanes::DeltaArraySlotsExt; +use vortex::error::VortexResult; +use vortex::error::vortex_ensure; +use vortex::error::vortex_err; + +use crate::CudaBufferExt; +use crate::CudaDeviceBuffer; +use crate::executor::CudaArrayExt; +use crate::executor::CudaExecute; +use crate::executor::CudaExecutionCtx; +use crate::executor::execute_validity_cuda; + +/// Elements per FastLanes chunk. Must match `FL_CHUNK` in `kernels/src/fastlanes_common.cuh`. +const FL_CHUNK: usize = 1024; +/// Threads per block: covers the widest lane count (128, for 8-bit values) in a single pass. +const BLOCK_THREADS: u32 = 128; + +/// CUDA decoder for FastLanes delta. +#[derive(Debug)] +pub(crate) struct DeltaExecutor; + +#[async_trait] +impl CudaExecute for DeltaExecutor { + #[instrument(level = "trace", skip_all, fields(executor = ?self))] + async fn execute( + &self, + array: ArrayRef, + ctx: &mut CudaExecutionCtx, + ) -> VortexResult { + let delta = array + .try_downcast::() + .map_err(|_| vortex_err!("Expected DeltaArray"))?; + decode_delta(delta, ctx).await + } +} + +#[instrument(skip_all)] +async fn decode_delta(array: DeltaArray, ctx: &mut CudaExecutionCtx) -> VortexResult { + let dtype = array.dtype().clone(); + let len = array.len(); + if len == 0 { + return Ok(Canonical::empty(&dtype)); + } + + // The vtable already narrows validity to the logical slice. + let validity = execute_validity_cuda(array.validity()?, len, ctx).await?; + + let deltas = array + .deltas() + .clone() + .execute_cuda(ctx) + .await? + .into_primitive(); + let bases = array + .bases() + .clone() + .execute_cuda(ctx) + .await? + .into_primitive(); + + // Signed values decode through their unsigned counterpart: the kernel's wrapping add + // inverts the wrapping subtract applied at compress time regardless of signedness. The + // buffer is untyped, so only the kernel and the device view need the unsigned type. + let ptype = deltas.ptype(); + let deltas_len = deltas.len(); + let offset = array.offset(); + vortex_ensure!( + deltas_len % FL_CHUNK == 0, + "Delta deltas child must be padded to a multiple of {FL_CHUNK}, got {deltas_len}" + ); + vortex_ensure!( + offset + len <= deltas_len, + "Delta slice {offset}..{} exceeds its {deltas_len} decoded values", + offset + len + ); + let num_chunks = deltas_len / FL_CHUNK; + let lanes = FL_CHUNK / (ptype.byte_width() * 8); + let required_bases = num_chunks * lanes; + vortex_ensure!( + bases.len() >= required_bases, + "Delta needs {required_bases} bases for {num_chunks} chunks, got {}", + bases.len() + ); + + let PrimitiveDataParts { + buffer: deltas_buffer, + .. + } = deltas.into_data_parts(); + let PrimitiveDataParts { + buffer: bases_buffer, + .. + } = bases.into_data_parts(); + let deltas_device = ctx.ensure_on_device(deltas_buffer).await?; + let bases_device = ctx.ensure_on_device(bases_buffer).await?; + + let num_chunks_u64 = num_chunks as u64; + let config = LaunchConfig { + grid_dim: (u32::try_from(num_chunks)?, 1, 1), + block_dim: (BLOCK_THREADS, 1, 1), + shared_mem_bytes: 0, + }; + + let decoded: BufferHandle = match_each_unsigned_integer_ptype!(ptype.to_unsigned(), |U| { + let deltas_view = deltas_device.cuda_view::()?; + let bases_view = bases_device.cuda_view::()?; + let mut output = ctx.device_alloc::(deltas_len)?; + let function = ctx.load_function("delta", &[U::PTYPE])?; + ctx.launch_kernel_config(&function, config, deltas_len, |args| { + args.arg(&deltas_view) + .arg(&bases_view) + .arg(&mut output) + .arg(&num_chunks_u64); + })?; + BufferHandle::new_device(Arc::new(CudaDeviceBuffer::new(output))) + }); + + // Chunks are decoded whole; the logical slice is applied to the result. + let width = ptype.byte_width(); + let sliced = decoded.slice(offset * width..(offset + len) * width); + + Ok(Canonical::Primitive(PrimitiveArray::from_buffer_handle( + sliced, ptype, validity, + ))) +} + +#[cfg(test)] +mod tests { + use rstest::rstest; + use vortex::array::IntoArray; + use vortex::array::assert_arrays_eq; + use vortex::array::validity::Validity; + use vortex::buffer::Buffer; + use vortex::error::VortexExpect; + use vortex_array::VortexSessionExecute; + + use super::*; + use crate::CanonicalCudaExt; + use crate::session::CudaSession; + + /// Decodes `array` on the GPU and asserts it matches the CPU canonical form. + async fn assert_gpu_matches_cpu(delta: DeltaArray) -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let mut cuda_ctx = CudaSession::create_execution_ctx(&crate::cuda_session()) + .vortex_expect("failed to create execution context"); + + let gpu = DeltaExecutor + .execute(delta.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(delta, gpu, &mut ctx); + Ok(()) + } + + /// Every element width is worth covering: lane count is `1024 / bit-width`, so each width + /// splits the 1024-element chunk differently — 128 lanes of 8 rows for `u8` through 16 + /// lanes of 64 rows for `u64` — and each has its own kernel instantiation. Every case + /// spans several chunks. + #[crate::test] + async fn test_cuda_delta_u8() -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let primitive = PrimitiveArray::new( + Buffer::from_iter((0u8..=255).cycle().take(3000)), + Validity::NonNullable, + ); + + let delta = Delta::try_from_primitive_array(&primitive, &mut ctx)?; + assert_gpu_matches_cpu(delta).await + } + + #[crate::test] + async fn test_cuda_delta_u16() -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let primitive = PrimitiveArray::new(Buffer::from_iter(0u16..3000), Validity::NonNullable); + + let delta = Delta::try_from_primitive_array(&primitive, &mut ctx)?; + assert_gpu_matches_cpu(delta).await + } + + #[crate::test] + async fn test_cuda_delta_u32() -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let primitive = PrimitiveArray::new( + Buffer::from_iter((0u32..3000).map(|i| i * 7)), + Validity::NonNullable, + ); + + let delta = Delta::try_from_primitive_array(&primitive, &mut ctx)?; + assert_gpu_matches_cpu(delta).await + } + + #[crate::test] + async fn test_cuda_delta_u64() -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let primitive = PrimitiveArray::new( + Buffer::from_iter((0u64..3000).map(|i| i * 1_000_003)), + Validity::NonNullable, + ); + + let delta = Delta::try_from_primitive_array(&primitive, &mut ctx)?; + assert_gpu_matches_cpu(delta).await + } + + /// Deltas across negative values wrap at compress time, so signed input must decode + /// through the unsigned kernel unchanged. + #[crate::test] + async fn test_cuda_delta_signed_values() -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let primitive = PrimitiveArray::new( + Buffer::from_iter((0..3000i32).map(|i| i - 1500)), + Validity::NonNullable, + ); + + let delta = Delta::try_from_primitive_array(&primitive, &mut ctx)?; + assert_gpu_matches_cpu(delta).await + } + + /// A sliced Delta keeps whole chunks and carries a nonzero offset, which the decode + /// applies only after the chunks are decoded. + #[rstest] + #[case::within_first_chunk(5, 100)] + #[case::across_chunks(1000, 1500)] + #[crate::test] + async fn test_cuda_delta_sliced(#[case] start: usize, #[case] end: usize) -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let primitive = PrimitiveArray::new( + Buffer::from_iter((0..3000u32).map(|i| i * 3)), + Validity::NonNullable, + ); + + let delta = Delta::try_from_primitive_array(&primitive, &mut ctx)? + .into_array() + .slice(start..end)?; + let Ok(delta) = delta.try_downcast::() else { + // A slice that the encoding chose to canonicalise is not this test's concern. + return Ok(()); + }; + assert_gpu_matches_cpu(delta).await + } + + /// Nulls ride alongside the values, and the validity is narrowed to the logical slice. + #[crate::test] + async fn test_cuda_delta_nullable() -> VortexResult<()> { + let mut ctx = vortex_array::array_session().create_execution_ctx(); + let primitive = PrimitiveArray::from_option_iter( + (0..3000u32).map(|value| (value % 7 != 0).then_some(value)), + ); + + let delta = Delta::try_from_primitive_array(&primitive, &mut ctx)?; + assert_gpu_matches_cpu(delta).await + } +} diff --git a/vortex-cuda/src/kernel/encodings/mod.rs b/vortex-cuda/src/kernel/encodings/mod.rs index 8d33433a2db..ba4ab51217e 100644 --- a/vortex-cuda/src/kernel/encodings/mod.rs +++ b/vortex-cuda/src/kernel/encodings/mod.rs @@ -5,6 +5,7 @@ mod alp; mod bitpacked; mod date_time_parts; mod decimal_byte_parts; +mod delta; mod for_; mod fsst; mod onpair; @@ -20,6 +21,7 @@ pub(crate) use bitpacked::BitPackedExecutor; pub(crate) use bitpacked::bitpacked_slice_view; pub(crate) use date_time_parts::DateTimePartsExecutor; pub(crate) use decimal_byte_parts::DecimalBytePartsExecutor; +pub(crate) use delta::DeltaExecutor; pub(crate) use for_::FoRExecutor; pub(crate) use fsst::DecodedVarBin; pub(crate) use fsst::FSSTExecutor; diff --git a/vortex-cuda/src/lib.rs b/vortex-cuda/src/lib.rs index d2b3ea3d22a..03d524a7f68 100644 --- a/vortex-cuda/src/lib.rs +++ b/vortex-cuda/src/lib.rs @@ -43,6 +43,7 @@ use kernel::ConstantNumericExecutor; use kernel::DateTimePartsExecutor; use kernel::DecimalBytePartsExecutor; pub use kernel::DefaultLaunchStrategy; +use kernel::DeltaExecutor; use kernel::DictExecutor; use kernel::FSSTExecutor; use kernel::FilterExecutor; @@ -80,6 +81,7 @@ use vortex::encodings::alp::ALP; use vortex::encodings::datetime_parts::DateTimeParts; use vortex::encodings::decimal_byte_parts::DecimalByteParts; use vortex::encodings::fastlanes::BitPacked; +use vortex::encodings::fastlanes::Delta; use vortex::encodings::fastlanes::FoR; use vortex::encodings::fsst::FSST; use vortex::encodings::runend::RunEnd; @@ -118,6 +120,7 @@ pub fn initialize_cuda(session: &CudaSession) { session.register_kernel(DecimalByteParts.id(), &DecimalBytePartsExecutor); session.register_kernel(Dict.id(), &DictExecutor); session.register_kernel(Shared.id(), &SharedExecutor); + session.register_kernel(Delta.id(), &DeltaExecutor); session.register_kernel(FoR.id(), &FoRExecutor); session.register_kernel(FSST.id(), &FSSTExecutor); session.register_kernel(OnPair.id(), &OnPairExecutor);