diff --git a/benchmarks/compress-bench/README.md b/benchmarks/compress-bench/README.md index 01b3e6e7d82..9f4b0d46624 100644 --- a/benchmarks/compress-bench/README.md +++ b/benchmarks/compress-bench/README.md @@ -117,7 +117,11 @@ rendered before the failure summary, so a dataset the GPU cannot decode still le the matrix with numbers — the process exits non-zero either way. The dataset list in `src/main.rs` therefore holds only datasets a `--gpu-verify` run has confirmed. -Several others are waiting on `vortex-cuda` kernel gaps (`u16` in `date_time_parts`, a -`vortex.masked` kernel, and a CPU fallback reached with device-resident buffers); they are listed -with their reasons next to `gpu_datasets`. Add one there once its gap is closed and verification -passes. +It now covers the whole compress suite: the kernel gaps that kept `taxi`, `Arade`, `CMSprovider`, +`Euro2016`, `HashTags` and the `StructListOfInts` wide tables off it — `u16` components in +`date_time_parts`, per-element `RunEnd` validity, and missing `vortex.masked` and `vortex.list` +kernels — have since been closed. Add a new dataset there once verification passes. + +`airquality` and `rplace` download from pcodec's public bucket, which the CPU suite skips to avoid +creating egress charges for pcodec. The GPU suite runs every entry on its explicit list, so both +are fetched on each GPU run. diff --git a/benchmarks/compress-bench/src/main.rs b/benchmarks/compress-bench/src/main.rs index cbf0378f344..d8ef76afc93 100644 --- a/benchmarks/compress-bench/src/main.rs +++ b/benchmarks/compress-bench/src/main.rs @@ -241,24 +241,23 @@ async fn run_compress( // Datasets run in GPU mode. Add one only after a `--gpu-verify` run has confirmed its CUDA // decode end to end; a dataset here that cannot decode fails the benchmark job. Between them - // these cover FSST strings, bit-packed numerics and columns with nulls. - // - // Not yet listed, each blocked on a `vortex-cuda` gap rather than on the benchmark: - // - // - `taxi` and `Arade`: `Unsupported ptype u16`. The CUDA `date_time_parts` kernel dispatches - // with `match_each_signed_integer_ptype!` where the CPU canonicaliser uses - // `match_each_integer_ptype!`. Widening the fused kernel takes 4³ = 64 PTX instantiations - // to 8³ = 512, so it is not a free change. - // - `Euro2016` and `HashTags`: `No CUDA kernel for encoding vortex.masked`. - // - `CMSprovider`: `expected host buffer` — a CPU fallback is reached with device-resident - // buffers, which `CudaArrayExt::execute_cuda` refuses. - // - `StructListOfInts`: its list layouts have no verified CUDA decode path. - let gpu_datasets: [&dyn Dataset; 4] = [ + // these cover FSST strings, bit-packed numerics, timestamps, columns with nulls, and lists. + let gpu_datasets: Vec<&dyn Dataset> = [ &TPCHLCommentCanonical as &dyn Dataset, &TPCHLCommentChunked, + &TaxiData, + PBI_DATASETS.get(Arade), PBI_DATASETS.get(Bimbo), + PBI_DATASETS.get(CMSprovider), + PBI_DATASETS.get(Euro2016), PBI_DATASETS.get(Food), - ]; + PBI_DATASETS.get(HashTags), + &DownloadableDataset::AirQuality, + &DownloadableDataset::RPlace, + ] + .into_iter() + .chain(structlistofints.iter().map(|d| d as &dyn Dataset)) + .collect(); let all_datasets: Vec<&dyn Dataset> = [ &TaxiData as &dyn Dataset, @@ -283,7 +282,7 @@ async fn run_compress( .collect(); let datasets: Vec<&dyn Dataset> = if mode.is_gpu() { - gpu_datasets.to_vec() + gpu_datasets } else { all_datasets } @@ -291,6 +290,9 @@ async fn run_compress( .filter(|d| { if let Some(filter) = datasets_filter.as_ref() { filter.is_match(d.name()) + } else if mode.is_gpu() { + // The GPU suite is an explicit list, so every entry on it runs. + true } else { // These download data from pcodec's public bucket, presumably creating egress charges // for pcodec. As such, we do not run in CI. diff --git a/encodings/onpair/src/array.rs b/encodings/onpair/src/array.rs index a25de45eafd..912fc27c278 100644 --- a/encodings/onpair/src/array.rs +++ b/encodings/onpair/src/array.rs @@ -232,6 +232,27 @@ fn build_dictionary( .map_err(|e| vortex_err!(InvalidArgument: "Unsafe OnPair dictionary: {e}")) } +/// A safety-validated dictionary built from host-materialized parts, owned +/// independently of any array. +/// +/// [`dict_view`] reaches the same parts through host-only accessors, which panics when +/// the array's buffers are device-resident. Callers that have already copied the +/// dictionary blob and its widened offsets to the host — the CUDA executor, which stages +/// the dictionary on the host regardless — build through this instead. +pub struct OnPairDictionary(CompactDictionary); + +impl OnPairDictionary { + /// Validates `(bytes, offsets)` and seals them into a dictionary. + pub fn try_new(bytes: ByteBuffer, offsets: Buffer) -> VortexResult { + Ok(Self(build_dictionary(bytes, offsets)?)) + } + + /// Borrows the dictionary as a view. + pub fn as_view(&self) -> CompactDictionaryView<'_> { + self.0.as_view() + } +} + /// A safety-validated [`CompactDictionaryView`] over `array`'s dictionary. /// /// The first successful initialization widens the `dict_offsets` child and diff --git a/vortex-btrblocks/src/builder.rs b/vortex-btrblocks/src/builder.rs index cc1477e4de8..fe8072d5e66 100644 --- a/vortex-btrblocks/src/builder.rs +++ b/vortex-btrblocks/src/builder.rs @@ -175,6 +175,7 @@ impl BtrBlocksCompressorBuilder { float::ALPRDScheme.id(), float::FloatRLEScheme.id(), float::NullDominatedSparseScheme.id(), + string::NullDominatedSparseScheme.id(), string::StringDictScheme.id(), binary::BinaryDictScheme.id(), ]; @@ -270,6 +271,22 @@ mod tests { ); } + /// `vortex.sparse` has no CUDA decode kernel, so no sparse scheme may survive this preset. + #[test] + fn cuda_compatible_excludes_every_sparse_scheme() { + let builder = BtrBlocksCompressorBuilder::default().only_cuda_compatible(); + for excluded in [ + integer::SparseScheme.id(), + float::NullDominatedSparseScheme.id(), + string::NullDominatedSparseScheme.id(), + ] { + assert!( + !builder.schemes.iter().any(|s| s.id() == excluded), + "{excluded} should be excluded" + ); + } + } + #[test] fn cuda_compatible_uses_fsst_for_strings() { let builder = BtrBlocksCompressorBuilder::default().only_cuda_compatible(); diff --git a/vortex-cuda/kernels/src/date_time_parts.cu b/vortex-cuda/kernels/src/date_time_parts.cu index ccb3e614991..a556ee5c57e 100644 --- a/vortex-cuda/kernels/src/date_time_parts.cu +++ b/vortex-cuda/kernels/src/date_time_parts.cu @@ -43,22 +43,34 @@ __device__ void date_time_parts(const DaysT *__restrict days, } #define EXPAND_DAYS(X) \ + X(u8, uint8_t) \ + X(u16, uint16_t) \ + X(u32, uint32_t) \ + X(u64, uint64_t) \ X(i8, int8_t) \ X(i16, int16_t) \ X(i32, int32_t) \ X(i64, int64_t) #define EXPAND_SUBSECONDS(d, DT, s, ST) \ + GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, u8, uint8_t) \ + GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, u16, uint16_t) \ + GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, u32, uint32_t) \ + GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, u64, uint64_t) \ GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, i8, int8_t) \ GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, i16, int16_t) \ GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, i32, int32_t) \ GENERATE_DATE_TIME_PARTS_KERNEL(d, DT, s, ST, i64, int64_t) #define EXPAND_SECONDS(d, DT) \ + EXPAND_SUBSECONDS(d, DT, u8, uint8_t) \ + EXPAND_SUBSECONDS(d, DT, u16, uint16_t) \ + EXPAND_SUBSECONDS(d, DT, u32, uint32_t) \ + EXPAND_SUBSECONDS(d, DT, u64, uint64_t) \ EXPAND_SUBSECONDS(d, DT, i8, int8_t) \ EXPAND_SUBSECONDS(d, DT, i16, int16_t) \ EXPAND_SUBSECONDS(d, DT, i32, int32_t) \ EXPAND_SUBSECONDS(d, DT, i64, int64_t) -// Generate all 64 kernels (4³) +// Generate all 512 kernels (8³: every signed and unsigned integer width per component) EXPAND_DAYS(EXPAND_SECONDS) diff --git a/vortex-cuda/kernels/src/list.cu b/vortex-cuda/kernels/src/list.cu new file mode 100644 index 00000000000..3e87e267299 --- /dev/null +++ b/vortex-cuda/kernels/src/list.cu @@ -0,0 +1,36 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +#include "config.cuh" +#include "types.cuh" + +// Converts Arrow-style `List` offsets into `ListView` offset/size pairs. +// +// `List` stores `list_len + 1` monotonically increasing offsets; a `ListView` stores one offset +// and one size per list. Both outputs are written by the same thread so the two views of a list +// are always produced together. +template +__device__ void list_views(const OffsetT *const __restrict offsets, + OffsetT *const __restrict out_offsets, + OffsetT *const __restrict out_sizes, + uint64_t list_len) { + const uint32_t elements_per_block = blockDim.x * ELEMENTS_PER_THREAD; + const uint64_t block_start = static_cast(blockIdx.x) * elements_per_block; + const uint64_t block_end = min(block_start + elements_per_block, list_len); + + for (uint64_t idx = block_start + threadIdx.x; idx < block_end; idx += blockDim.x) { + const OffsetT start = offsets[idx]; + out_offsets[idx] = start; + out_sizes[idx] = static_cast(offsets[idx + 1] - start); + } +} + +#define GENERATE_LIST_VIEWS_KERNEL(offset_suffix, OffsetT) \ + extern "C" __global__ void list_views_##offset_suffix(const OffsetT *const __restrict offsets, \ + OffsetT *const __restrict out_offsets, \ + OffsetT *const __restrict out_sizes, \ + uint64_t list_len) { \ + list_views(offsets, out_offsets, out_sizes, list_len); \ + } + +FOR_EACH_INTEGER(GENERATE_LIST_VIEWS_KERNEL) diff --git a/vortex-cuda/kernels/src/runend.cu b/vortex-cuda/kernels/src/runend.cu index a3f1d245dbe..7472e7dafba 100644 --- a/vortex-cuda/kernels/src/runend.cu +++ b/vortex-cuda/kernels/src/runend.cu @@ -127,6 +127,45 @@ __device__ void runend_decode_kernel(const EndsT *const __restrict ends, } } +// Expands run-end encoded validity bits into a packed output bitmap. +// +// Mirrors `runend_decode_kernel`, but each thread owns one complete output byte so that +// threads never race on bits within the same byte. Runs are located with a global binary +// search per element rather than the shared-memory cache: validity expansion runs once per +// array and is not the decode hot path. +template +__device__ void runend_bool_kernel(const EndsT *const __restrict ends, + uint64_t num_runs, + const uint8_t *const __restrict values, + uint64_t values_bit_offset, + uint64_t offset, + uint64_t output_len, + uint8_t *const __restrict output) { + const uint64_t output_bytes = (output_len + 7) / 8; + const uint32_t elements_per_block = blockDim.x * ELEMENTS_PER_THREAD; + const uint64_t block_start = static_cast(blockIdx.x) * elements_per_block; + const uint64_t block_end = min(block_start + elements_per_block, output_bytes); + + for (uint64_t byte_idx = block_start + threadIdx.x; byte_idx < block_end; byte_idx += blockDim.x) { + const uint64_t row_start = byte_idx * 8; + uint8_t packed = 0; +#pragma unroll + for (uint32_t bit = 0; bit < 8; ++bit) { + const uint64_t row = row_start + bit; + if (row < output_len) { + uint64_t run_idx = upper_bound(ends, num_runs, row + offset); + if (run_idx >= num_runs) { + run_idx = num_runs - 1; + } + const uint64_t value_idx = values_bit_offset + run_idx; + const uint8_t value = (values[value_idx / 8] >> (value_idx % 8)) & 1; + packed |= static_cast(value << bit); + } + } + output[byte_idx] = packed; + } +} + #define GENERATE_RUNEND_KERNEL(value_suffix, ValueType, ends_suffix, EndsType) \ extern "C" __global__ void runend_##value_suffix##_##ends_suffix( \ const EndsType *const __restrict ends, \ @@ -155,3 +194,21 @@ GENERATE_RUNEND_KERNELS_FOR_VALUE(i64, int64_t) GENERATE_RUNEND_KERNELS_FOR_VALUE(f16, __half) GENERATE_RUNEND_KERNELS_FOR_VALUE(f32, float) GENERATE_RUNEND_KERNELS_FOR_VALUE(f64, double) + +#define GENERATE_RUNEND_BOOL_KERNEL(ends_suffix, EndsType) \ + extern "C" __global__ void runend_bool_##ends_suffix(const EndsType *const __restrict ends, \ + uint64_t num_runs, \ + const uint8_t *const __restrict values, \ + uint64_t values_bit_offset, \ + uint64_t offset, \ + uint64_t output_len, \ + uint8_t *const __restrict output) { \ + runend_bool_kernel(ends, num_runs, values, values_bit_offset, offset, output_len, output); \ + } + +// Validity bitmaps use a different physical layout and launch unit, but dispatch over the +// same run-end index types. +GENERATE_RUNEND_BOOL_KERNEL(u8, uint8_t) +GENERATE_RUNEND_BOOL_KERNEL(u16, uint16_t) +GENERATE_RUNEND_BOOL_KERNEL(u32, uint32_t) +GENERATE_RUNEND_BOOL_KERNEL(u64, uint64_t) diff --git a/vortex-cuda/src/canonical.rs b/vortex-cuda/src/canonical.rs index fe832c4aa01..5a3d1e224f2 100644 --- a/vortex-cuda/src/canonical.rs +++ b/vortex-cuda/src/canonical.rs @@ -5,18 +5,21 @@ use std::sync::Arc; use async_trait::async_trait; use futures::future::try_join_all; +use vortex::array::ArrayRef; use vortex::array::Canonical; use vortex::array::IntoArray; use vortex::array::VortexSessionExecute; use vortex::array::arrays::BoolArray; use vortex::array::arrays::DecimalArray; use vortex::array::arrays::ExtensionArray; +use vortex::array::arrays::ListViewArray; use vortex::array::arrays::PrimitiveArray; use vortex::array::arrays::StructArray; use vortex::array::arrays::VarBinViewArray; use vortex::array::arrays::bool::BoolDataParts; use vortex::array::arrays::decimal::DecimalDataParts; use vortex::array::arrays::extension::ExtensionArrayExt; +use vortex::array::arrays::listview::ListViewDataParts; use vortex::array::arrays::primitive::PrimitiveDataParts; use vortex::array::arrays::struct_::StructDataParts; use vortex::array::arrays::varbinview::BinaryView; @@ -29,6 +32,19 @@ use vortex::buffer::Buffer; use vortex::buffer::ByteBuffer; use vortex::error::VortexResult; +/// Copy a canonical child array to the host. +async fn child_into_host(child: ArrayRef) -> VortexResult { + #[expect( + clippy::disallowed_methods, + reason = "CanonicalCudaExt threads no session through" + )] + Ok(child + .execute::(&mut legacy_session().create_execution_ctx())? + .into_host() + .await? + .into_array()) +} + /// Move all canonical data from to_host from device. #[async_trait] pub trait CanonicalCudaExt { @@ -102,7 +118,7 @@ impl CanonicalCudaExt for Canonical { n @ Canonical::Null(_) => Ok(n), Canonical::Bool(bool) => { let len = bool.len(); - let validity = bool.validity()?; + let validity = validity_into_host(bool.validity()?).await?; let BoolDataParts { bits, meta } = bool.into_data().into_parts(len); let bits = BitBuffer::new_with_offset( @@ -110,10 +126,7 @@ impl CanonicalCudaExt for Canonical { meta.len(), meta.offset(), ); - Ok(Canonical::Bool(BoolArray::new( - bits, - validity_into_host(validity).await?, - ))) + Ok(Canonical::Bool(BoolArray::new(bits, validity))) } Canonical::Primitive(prim) => { let PrimitiveDataParts { @@ -153,6 +166,7 @@ impl CanonicalCudaExt for Canonical { validity, dtype, } = varbinview.into_data_parts(); + let validity = validity_into_host(validity).await?; // Copy all device views to host let host_views = views.try_into_host()?.await?; @@ -167,11 +181,27 @@ impl CanonicalCudaExt for Canonical { let host_buffers = try_join_all(host_buffers).await?; let host_buffers: Arc<[ByteBuffer]> = Arc::from(host_buffers); - let validity = validity_into_host(validity).await?; Ok(Canonical::VarBinView(unsafe { VarBinViewArray::new_unchecked(host_views, host_buffers, dtype, validity) })) } + Canonical::List(list) => { + let ListViewDataParts { + elements, + offsets, + sizes, + validity, + .. + } = list.into_data_parts(); + let validity = validity_into_host(validity).await?; + + Ok(Canonical::List(ListViewArray::try_new( + child_into_host(elements).await?, + child_into_host(offsets).await?, + child_into_host(sizes).await?, + validity, + )?)) + } Canonical::Extension(ext) => { // Copy the storage array to host and rewrap in ExtensionArray. #[expect( diff --git a/vortex-cuda/src/kernel/arrays/list.rs b/vortex-cuda/src/kernel/arrays/list.rs new file mode 100644 index 00000000000..d6aab480e4e --- /dev/null +++ b/vortex-cuda/src/kernel/arrays/list.rs @@ -0,0 +1,214 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use std::sync::Arc; + +use async_trait::async_trait; +use cudarc::driver::DeviceRepr; +use cudarc::driver::PushKernelArg; +use tracing::instrument; +use vortex::array::ArrayRef; +use vortex::array::Canonical; +use vortex::array::IntoArray; +use vortex::array::arrays::List; +use vortex::array::arrays::ListViewArray; +use vortex::array::arrays::PrimitiveArray; +use vortex::array::arrays::list::ListArrayExt; +use vortex::array::arrays::list::ListArraySlotsExt; +use vortex::array::arrays::primitive::PrimitiveDataParts; +use vortex::array::buffer::BufferHandle; +use vortex::array::match_each_integer_ptype; +use vortex::array::validity::Validity; +use vortex::dtype::NativePType; +use vortex::dtype::Nullability; +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; + +/// CUDA executor for `ListArray`. +/// +/// `List` stores `len + 1` Arrow-style offsets; its canonical form, `ListView`, stores one +/// offset and one size per list. Decode the elements on the GPU and derive the view pair from +/// the offsets with a single kernel. +#[derive(Debug)] +pub(crate) struct ListExecutor; + +#[async_trait] +impl CudaExecute for ListExecutor { + #[instrument(level = "trace", skip_all, fields(executor = ?self))] + async fn execute( + &self, + array: ArrayRef, + ctx: &mut CudaExecutionCtx, + ) -> VortexResult { + let list = array + .try_downcast::() + .map_err(|_| vortex_err!("Expected ListArray"))?; + + let list_len = list.len(); + let validity = execute_validity_cuda(list.list_validity(), list_len, ctx).await?; + let elements = list + .elements() + .clone() + .execute_cuda(ctx) + .await? + .into_array(); + + if list_len == 0 { + let empty = PrimitiveArray::empty::(Nullability::NonNullable); + return Ok(Canonical::List(ListViewArray::try_new( + elements, + empty.clone().into_array(), + empty.into_array(), + validity, + )?)); + } + + let offsets = list + .offsets() + .clone() + .execute_cuda(ctx) + .await? + .into_primitive(); + vortex_ensure!( + offsets.len() == list_len + 1, + "ListArray must have {} offsets, got {}", + list_len + 1, + offsets.len() + ); + + let offsets_ptype = offsets.ptype(); + match_each_integer_ptype!(offsets_ptype, |O| { + list_views_typed::(offsets, elements, validity, list_len, ctx).await + }) + } +} + +async fn list_views_typed( + offsets: PrimitiveArray, + elements: ArrayRef, + validity: Validity, + list_len: usize, + ctx: &mut CudaExecutionCtx, +) -> VortexResult { + let PrimitiveDataParts { + buffer: offsets_buffer, + .. + } = offsets.into_data_parts(); + + let offsets_device = ctx.ensure_on_device(offsets_buffer).await?; + let offsets_view = offsets_device.cuda_view::()?; + + let mut view_offsets = ctx.device_alloc::(list_len)?; + let mut view_sizes = ctx.device_alloc::(list_len)?; + let list_len_u64 = list_len as u64; + + let offsets_ptype = O::PTYPE.to_string(); + let cuda_function = ctx.load_function_with_suffixes("list", &["views", &offsets_ptype])?; + ctx.launch_kernel(&cuda_function, list_len, |args| { + args.arg(&offsets_view) + .arg(&mut view_offsets) + .arg(&mut view_sizes) + .arg(&list_len_u64); + })?; + + let view_offsets = PrimitiveArray::from_buffer_handle( + BufferHandle::new_device(Arc::new(CudaDeviceBuffer::new(view_offsets))), + O::PTYPE, + Validity::NonNullable, + ); + let view_sizes = PrimitiveArray::from_buffer_handle( + BufferHandle::new_device(Arc::new(CudaDeviceBuffer::new(view_sizes))), + O::PTYPE, + Validity::NonNullable, + ); + + Ok(Canonical::List(ListViewArray::try_new( + elements, + view_offsets.into_array(), + view_sizes.into_array(), + validity, + )?)) +} + +#[cfg(test)] +mod tests { + use rstest::rstest; + use vortex::array::arrays::BoolArray; + use vortex::array::arrays::ListArray; + use vortex::array::assert_arrays_eq; + use vortex::buffer::Buffer; + use vortex::buffer::buffer; + use vortex::error::VortexExpect; + use vortex_array::VortexSessionExecute; + + use super::*; + use crate::CanonicalCudaExt; + use crate::session::CudaSession; + + #[rstest] + #[case::single_run(vec![0i32, 2, 5, 9])] + #[case::empty_lists(vec![0i32, 0, 3, 3])] + #[crate::test] + async fn test_cuda_list_decompression(#[case] offsets: Vec) -> 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 element_count = *offsets.last().vortex_expect("offsets are non-empty"); + let elements = PrimitiveArray::new( + (0..element_count).collect::>(), + Validity::NonNullable, + ) + .into_array(); + let offsets_array = + PrimitiveArray::new(Buffer::from(offsets), Validity::NonNullable).into_array(); + let list = ListArray::try_new(elements, offsets_array, Validity::NonNullable)?; + + let gpu_result = ListExecutor + .execute(list.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(list, gpu_result, &mut ctx); + + Ok(()) + } + + #[crate::test] + async fn test_cuda_list_with_nulls() -> 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 elements = + PrimitiveArray::new(buffer![10i32, 20, 30, 40], Validity::NonNullable).into_array(); + let offsets = + PrimitiveArray::new(buffer![0i32, 2, 2, 4], Validity::NonNullable).into_array(); + let validity = + Validity::Array(BoolArray::from_iter([true, false, true].into_iter()).into_array()); + let list = ListArray::try_new(elements, offsets, validity)?; + + let gpu_result = ListExecutor + .execute(list.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(list, gpu_result, &mut ctx); + + Ok(()) + } +} diff --git a/vortex-cuda/src/kernel/arrays/masked.rs b/vortex-cuda/src/kernel/arrays/masked.rs new file mode 100644 index 00000000000..b739ed2bf60 --- /dev/null +++ b/vortex-cuda/src/kernel/arrays/masked.rs @@ -0,0 +1,99 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use async_trait::async_trait; +use tracing::instrument; +use vortex::array::ArrayRef; +use vortex::array::Canonical; +use vortex::array::arrays::Masked; +use vortex::array::arrays::masked::MaskedArrayExt; +use vortex::array::arrays::masked::MaskedArraySlotsExt; +use vortex::array::arrays::masked::mask_validity_canonical; +use vortex::array::validity::Validity; +use vortex::error::VortexResult; +use vortex::error::vortex_bail; +use vortex::error::vortex_err; + +use crate::executor::CudaArrayExt; +use crate::executor::CudaExecute; +use crate::executor::CudaExecutionCtx; +use crate::executor::execute_validity_cuda; + +/// CUDA executor for MaskedArray. +/// +/// A `MaskedArray` is a child array that carries no nulls of its own, plus the validity +/// bitmap that supplies them. Decode the child on the GPU, decode the mask on the GPU, and +/// attach the mask to the result. +#[derive(Debug)] +pub(crate) struct MaskedExecutor; + +#[async_trait] +impl CudaExecute for MaskedExecutor { + #[instrument(level = "trace", skip_all, fields(executor = ?self))] + async fn execute( + &self, + array: ArrayRef, + ctx: &mut CudaExecutionCtx, + ) -> VortexResult { + let masked = array + .try_downcast::() + .map_err(|_| vortex_err!("Expected MaskedArray"))?; + + let len = masked.len(); + let validity = masked.masked_validity(); + + // `MaskedArray` guarantees its child holds no nulls, so the mask alone determines the + // output validity. Combining two device-resident bitmaps would need a CPU compute pass. + if matches!(masked.child().validity()?, Validity::Array(_)) { + vortex_bail!( + "MaskedArray child carries a per-element validity bitmap, which cannot be combined with the mask on the GPU" + ); + } + + let child = masked.child().clone().execute_cuda(ctx).await?; + + let validity = execute_validity_cuda(validity, len, ctx).await?; + mask_validity_canonical(child, validity, ctx.execution_ctx()) + } +} + +#[cfg(test)] +mod tests { + use vortex::array::IntoArray; + use vortex::array::arrays::BoolArray; + use vortex::array::arrays::MaskedArray; + use vortex::array::arrays::PrimitiveArray; + use vortex::array::assert_arrays_eq; + use vortex::buffer::buffer; + use vortex::error::VortexExpect; + use vortex_array::VortexSessionExecute; + + use super::*; + use crate::CanonicalCudaExt; + use crate::session::CudaSession; + + #[crate::test] + async fn test_cuda_masked_applies_validity() -> 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 child = PrimitiveArray::new(buffer![1i32, 2, 3, 4], Validity::NonNullable).into_array(); + let validity = Validity::Array( + BoolArray::from_iter([true, false, true, true].into_iter()).into_array(), + ); + let masked = MaskedArray::try_new(child, validity)?; + + let gpu_result = MaskedExecutor + .execute(masked.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(masked, gpu_result, &mut ctx); + + Ok(()) + } +} diff --git a/vortex-cuda/src/kernel/arrays/mod.rs b/vortex-cuda/src/kernel/arrays/mod.rs index ab81934bb27..56d4c483480 100644 --- a/vortex-cuda/src/kernel/arrays/mod.rs +++ b/vortex-cuda/src/kernel/arrays/mod.rs @@ -3,8 +3,12 @@ mod constant; mod dict; +mod list; +mod masked; mod shared; pub(crate) use constant::ConstantNumericExecutor; pub(crate) use dict::DictExecutor; +pub(crate) use list::ListExecutor; +pub(crate) use masked::MaskedExecutor; pub(crate) use shared::SharedExecutor; diff --git a/vortex-cuda/src/kernel/encodings/date_time_parts.rs b/vortex-cuda/src/kernel/encodings/date_time_parts.rs index bff691e262b..5bd8b019ee2 100644 --- a/vortex-cuda/src/kernel/encodings/date_time_parts.rs +++ b/vortex-cuda/src/kernel/encodings/date_time_parts.rs @@ -15,7 +15,7 @@ use vortex::array::arrays::PrimitiveArray; use vortex::array::arrays::TemporalArray; use vortex::array::arrays::primitive::PrimitiveDataParts; use vortex::array::buffer::BufferHandle; -use vortex::array::match_each_signed_integer_ptype; +use vortex::array::match_each_integer_ptype; use vortex::array::validity::Validity; use vortex::dtype::DType; use vortex::dtype::NativePType; @@ -110,9 +110,9 @@ impl CudaExecute for DateTimePartsExecutor { let seconds_ptype = seconds_prim.ptype(); let subseconds_ptype = subseconds_prim.ptype(); - match_each_signed_integer_ptype!(days_ptype, |DaysT| { - match_each_signed_integer_ptype!(seconds_ptype, |SecondsT| { - match_each_signed_integer_ptype!(subseconds_ptype, |SubsecondsT| { + match_each_integer_ptype!(days_ptype, |DaysT| { + match_each_integer_ptype!(seconds_ptype, |SecondsT| { + match_each_integer_ptype!(subseconds_ptype, |SubsecondsT| { decode_datetimeparts_typed::( days_prim, seconds_prim, @@ -297,6 +297,88 @@ mod tests { Ok(()) } + /// Compression narrows each component to its smallest ptype, choosing unsigned whenever the + /// component is non-negative. Every post-epoch timestamp column therefore decomposes into + /// entirely unsigned components, which is what panicked on `taxi`. + #[crate::test] + async fn test_cuda_datetimeparts_unsigned_components() -> 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 len = 3; + let days_arr = PrimitiveArray::new(buffer![1u32, 2, 3], Validity::NonNullable).into_array(); + let seconds_arr = + PrimitiveArray::new(buffer![3600u16, 0, 60], Validity::NonNullable).into_array(); + let subseconds_arr = + PrimitiveArray::new(buffer![250u8, 0, 99], Validity::NonNullable).into_array(); + + let temporal = TemporalArray::new_timestamp( + PrimitiveArray::new(buffer![0i64; len], Validity::NonNullable).into_array(), + TimeUnit::Milliseconds, + None, + ); + let dtp_array = DateTimeParts::try_new( + temporal.dtype().clone(), + days_arr, + seconds_arr, + subseconds_arr, + )?; + + let gpu_result = DateTimePartsExecutor + .execute(dtp_array.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(dtp_array, gpu_result, &mut ctx); + + Ok(()) + } + + /// Components are narrowed independently, so their signedness can differ within one array. + /// A timestamp column straddling the epoch narrows to unsigned days (all zero) with signed + /// seconds, which is only decoded correctly if each component keeps its own signedness. + #[crate::test] + async fn test_cuda_datetimeparts_mixed_signedness() -> 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"); + + // Milliseconds -32_000, 0 and 31_000 split into these components. + let days_arr = PrimitiveArray::new(buffer![0u8, 0, 0], Validity::NonNullable).into_array(); + let seconds_arr = + PrimitiveArray::new(buffer![-32i8, 0, 31], Validity::NonNullable).into_array(); + let subseconds_arr = + PrimitiveArray::new(buffer![0u8, 0, 0], Validity::NonNullable).into_array(); + + let temporal = TemporalArray::new_timestamp( + PrimitiveArray::new(buffer![0i64; 3], Validity::NonNullable).into_array(), + TimeUnit::Milliseconds, + None, + ); + let dtp_array = DateTimeParts::try_new( + temporal.dtype().clone(), + days_arr, + seconds_arr, + subseconds_arr, + )?; + + let gpu_result = DateTimePartsExecutor + .execute(dtp_array.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(dtp_array, gpu_result, &mut ctx); + + Ok(()) + } + #[crate::test] async fn test_cuda_datetimeparts_large_array() -> VortexResult<()> { let mut ctx = vortex_array::array_session().create_execution_ctx(); diff --git a/vortex-cuda/src/kernel/encodings/onpair.rs b/vortex-cuda/src/kernel/encodings/onpair.rs index 5858889187d..8298048b9d9 100644 --- a/vortex-cuda/src/kernel/encodings/onpair.rs +++ b/vortex-cuda/src/kernel/encodings/onpair.rs @@ -49,6 +49,7 @@ use num_traits::AsPrimitive; use tracing::instrument; use vortex::array::ArrayRef; use vortex::array::Canonical; +use vortex::array::IntoArray; use vortex::array::arrays::PrimitiveArray; use vortex::array::arrays::VarBinViewArray; use vortex::array::arrays::primitive::PrimitiveDataParts; @@ -56,8 +57,10 @@ use vortex::array::arrays::varbinview::build_views::MAX_BUFFER_LEN; use vortex::array::arrays::varbinview::build_views::build_views; use vortex::array::buffer::BufferHandle; use vortex::array::buffer::DeviceBuffer; +use vortex::array::builtins::ArrayBuiltins; use vortex::array::match_each_integer_ptype; use vortex::array::validity::Validity; +use vortex::buffer::Buffer; use vortex::dtype::DType; use vortex::dtype::NativePType; use vortex::dtype::PType; @@ -73,7 +76,7 @@ use vortex_onpair::OnPair; use vortex_onpair::OnPairArray; use vortex_onpair::OnPairArrayExt; use vortex_onpair::OnPairArraySlotsExt; -use vortex_onpair::dict_view; +use vortex_onpair::OnPairDictionary; use crate::CanonicalCudaExt; use crate::CudaBufferExt; @@ -321,7 +324,13 @@ async fn stage_dict( onpair: ArrayView<'_, OnPair>, ctx: &mut CudaExecutionCtx, ) -> VortexResult { - let dict = dict_view(onpair, ctx.execution_ctx())?; + // Both dictionary parts are read here by the host staging loop below, so both must be + // host-resident first. `dict_view` reads them through `BufferHandle::as_host` and a CPU + // cast kernel, which panic once the array's buffers live on the device. + let dict_bytes = onpair.dict_bytes_handle().clone().try_into_host()?.await?; + let dict_offsets = host_dict_offsets(onpair, ctx).await?; + let dict = OnPairDictionary::try_new(dict_bytes, dict_offsets)?; + let dict = dict.as_view(); let dict_size = dict.num_tokens(); let dict_size_u32 = u32::try_from(dict_size)?; let mut dict_padded = vec![0u8; dict_size * MAX_TOKEN_SIZE]; @@ -596,6 +605,28 @@ async fn decode_primitive_child( Ok(child.execute_cuda(ctx).await?.into_primitive()) } +/// The `dict_offsets` child, decoded and widened to the `u32` the dictionary stores. +/// +/// Decoded on the GPU like the other children, then copied back: the dictionary is staged +/// on the host, and the child is `dict_size + 1` elements, so the round trip is bounded by +/// the dictionary size rather than the array length. +async fn host_dict_offsets( + onpair: ArrayView<'_, OnPair>, + ctx: &mut CudaExecutionCtx, +) -> VortexResult> { + let offsets = decode_primitive_child(onpair.dict_offsets().clone(), ctx).await?; + let offsets = Canonical::Primitive(offsets) + .into_host() + .await? + .into_primitive(); + let nullability = offsets.dtype().nullability(); + Ok(offsets + .into_array() + .cast(DType::Primitive(PType::U32, nullability))? + .execute::(ctx.execution_ctx())? + .into_buffer::()) +} + /// Cold path: the window has no codes, so the rows must decode to zero bytes. async fn ensure_zero_lengths(lengths: PrimitiveArray) -> VortexResult<()> { let lengths = Canonical::Primitive(lengths) @@ -734,6 +765,52 @@ mod tests { Ok(()) } + /// Arrays read through the CUDA flat layout arrive with their buffers already on the + /// device. Staging the dictionary must copy it back rather than assuming a host buffer, + /// which is what panicked with "expected host buffer" on `taxi` and TPC-H `l_comment`. + #[crate::test] + async fn test_cuda_onpair_device_resident_dictionary() -> 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 strings = vec![ + Some(&b"the quick brown fox"[..]), + Some(&b"jumps over the lazy dog"[..]), + Some(&b"hello world"[..]), + Some(&b"vortex onpair test string"[..]), + ]; + let dtype = DType::Utf8(Nullability::NonNullable); + let onpair = compress_onpair(strings, dtype.clone(), &mut cuda_ctx)?; + let view = onpair + .as_typed::() + .vortex_expect("expected OnPair array"); + + // Rebuild the array with its dictionary blob resident on the device. + let dict_bytes = view.dict_bytes_handle().as_host().to_vec(); + let dict_device = cuda_ctx.copy_to_device(dict_bytes)?.await?; + let device_onpair = OnPair::try_new( + dtype.clone(), + dict_device, + view.dict_offsets().clone(), + view.codes().clone(), + view.codes_offsets().clone(), + view.uncompressed_lengths().clone(), + view.array_validity(), + )? + .into_array(); + + let gpu_result = OnPairExecutor + .execute(device_onpair, &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed"); + assert_eq!(gpu_result.dtype(), &dtype); + + let host_result = gpu_result.into_host().await?.into_array(); + assert_arrays_eq!(onpair, host_result, &mut ctx); + Ok(()) + } + /// A slice keeps the whole `codes` child and narrows only `codes_offsets`, /// so this exercises the nonzero `code_start` window. #[crate::test] diff --git a/vortex-cuda/src/kernel/encodings/runend.rs b/vortex-cuda/src/kernel/encodings/runend.rs index 36ceb8c7b8b..da6973ff11c 100644 --- a/vortex-cuda/src/kernel/encodings/runend.rs +++ b/vortex-cuda/src/kernel/encodings/runend.rs @@ -10,8 +10,10 @@ use tracing::instrument; use vortex::array::ArrayRef; use vortex::array::Canonical; use vortex::array::IntoArray; +use vortex::array::arrays::BoolArray; use vortex::array::arrays::ConstantArray; use vortex::array::arrays::PrimitiveArray; +use vortex::array::arrays::bool::BoolDataParts; use vortex::array::arrays::primitive::PrimitiveDataParts; use vortex::array::buffer::BufferHandle; use vortex::array::match_each_native_ptype; @@ -149,10 +151,41 @@ async fn decode_runend_typed { unreachable!("AllInvalid should be handled by RunEndExecutor::execute") } - Validity::Array(_) => { - vortex_bail!( - "RunEnd GPU decoding does not yet support per-element validity in values; falling back to CPU" - ); + Validity::Array(validity) => { + // Expand the per-run validity bitmap through the same run mapping as the values. + let validity_bools = validity.execute_cuda(ctx).await?.into_bool(); + let validity_len = validity_bools.len(); + let BoolDataParts { + bits: validity_bits, + meta: validity_meta, + } = validity_bools.into_data().into_parts(validity_len); + let validity_device = ctx.ensure_on_device(validity_bits).await?; + let validity_view = validity_device.cuda_view::()?; + + // Each thread owns a whole output byte, so threads never race on bits in one byte. + let output_bytes = output_len.div_ceil(8); + let mut validity_out = ctx.device_alloc::(output_bytes)?; + let validity_offset_u64 = validity_meta.offset() as u64; + + let ends_ptype = E::PTYPE.to_string(); + let validity_function = + ctx.load_function_with_suffixes("runend", &["bool", &ends_ptype])?; + ctx.launch_kernel(&validity_function, output_bytes, |args| { + args.arg(&ends_view) + .arg(&num_runs_u64) + .arg(&validity_view) + .arg(&validity_offset_u64) + .arg(&offset_u64) + .arg(&output_len_u64) + .arg(&mut validity_out); + })?; + + let validity_buffer = + BufferHandle::new_device(Arc::new(CudaDeviceBuffer::new(validity_out))); + Validity::Array( + BoolArray::new_handle(validity_buffer, 0, output_len, Validity::NonNullable) + .into_array(), + ) } }; @@ -182,7 +215,6 @@ mod tests { use super::*; use crate::CanonicalCudaExt; - use crate::executor::CudaArrayExt; use crate::session::CudaSession; fn make_runend_array(ends: Vec, values: Vec, ctx: &mut ExecutionCtx) -> RunEndArray @@ -303,7 +335,7 @@ mod tests { } #[crate::test] - async fn test_cuda_runend_nullable_values_falls_back_to_cpu() -> VortexResult<()> { + async fn test_cuda_runend_nullable_values() -> 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"); @@ -318,13 +350,48 @@ mod tests { PrimitiveArray::new(Buffer::from(vec![10i32, 0, 30]), validity).into_array(); let runend_array = RunEnd::new(ends_array, values_array, cuda_ctx.execution_ctx()); - // execute_cuda should fall back to CPU and still produce the correct result. - let gpu_result = runend_array - .clone() - .into_array() - .execute_cuda(&mut cuda_ctx) + // The GPU expands the per-run validity bitmap through the run mapping. + // Call the executor directly: `execute_cuda` would silently fall back to CPU for a + // host-resident array, hiding a GPU failure. + let gpu_result = RunEndExecutor + .execute(runend_array.clone().into_array(), &mut cuda_ctx) + .await + .vortex_expect("GPU decompression failed") + .into_host() + .await? + .into_array(); + + assert_arrays_eq!(runend_array, gpu_result, &mut ctx); + + Ok(()) + } + + /// Validity expansion packs bits a byte at a time, so exercise a run layout whose runs + /// straddle output byte boundaries. + #[crate::test] + async fn test_cuda_runend_nullable_values_across_byte_boundaries() -> 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 num_runs = 301; + let ends: Vec = (1..=num_runs).map(|run| run * 3).collect(); + let values: Vec = (0..num_runs as i32).collect(); + let validity = Validity::Array( + BoolArray::from_iter((0..num_runs).map(|run| run % 3 != 0)).into_array(), + ); + + let ends_array = + PrimitiveArray::new(Buffer::from(ends), Validity::NonNullable).into_array(); + let values_array = PrimitiveArray::new(Buffer::from(values), validity).into_array(); + let runend_array = RunEnd::new(ends_array, values_array, cuda_ctx.execution_ctx()); + + // Call the executor directly: `execute_cuda` would silently fall back to CPU for a + // host-resident array, hiding a GPU failure. + let gpu_result = RunEndExecutor + .execute(runend_array.clone().into_array(), &mut cuda_ctx) .await - .vortex_expect("GPU/CPU fallback should succeed") + .vortex_expect("GPU decompression failed") .into_host() .await? .into_array(); diff --git a/vortex-cuda/src/kernel/mod.rs b/vortex-cuda/src/kernel/mod.rs index 36735024c7f..19da5a1f6dd 100644 --- a/vortex-cuda/src/kernel/mod.rs +++ b/vortex-cuda/src/kernel/mod.rs @@ -31,6 +31,8 @@ mod slice; pub(crate) use arrays::ConstantNumericExecutor; pub(crate) use arrays::DictExecutor; +pub(crate) use arrays::ListExecutor; +pub(crate) use arrays::MaskedExecutor; pub(crate) use arrays::SharedExecutor; pub use encodings::ZstdKernelPrep; pub use encodings::zstd_kernel_prepare; diff --git a/vortex-cuda/src/lib.rs b/vortex-cuda/src/lib.rs index 03d524a7f68..927859d724e 100644 --- a/vortex-cuda/src/lib.rs +++ b/vortex-cuda/src/lib.rs @@ -49,6 +49,8 @@ use kernel::FSSTExecutor; use kernel::FilterExecutor; use kernel::FoRExecutor; pub use kernel::LaunchStrategy; +use kernel::ListExecutor; +use kernel::MaskedExecutor; use kernel::OnPairExecutor; use kernel::RunEndExecutor; use kernel::SharedExecutor; @@ -75,6 +77,8 @@ use vortex::array::ArrayVTable; use vortex::array::arrays::Constant; use vortex::array::arrays::Dict; use vortex::array::arrays::Filter; +use vortex::array::arrays::List; +use vortex::array::arrays::Masked; use vortex::array::arrays::Shared; use vortex::array::arrays::Slice; use vortex::encodings::alp::ALP; @@ -119,6 +123,8 @@ pub fn initialize_cuda(session: &CudaSession) { session.register_kernel(DateTimeParts.id(), &DateTimePartsExecutor); session.register_kernel(DecimalByteParts.id(), &DecimalBytePartsExecutor); session.register_kernel(Dict.id(), &DictExecutor); + session.register_kernel(List.id(), &ListExecutor); + session.register_kernel(Masked.id(), &MaskedExecutor); session.register_kernel(Shared.id(), &SharedExecutor); session.register_kernel(Delta.id(), &DeltaExecutor); session.register_kernel(FoR.id(), &FoRExecutor); diff --git a/vortex/src/editions/preview/v2026_06.rs b/vortex/src/editions/preview/v2026_06.rs index fee5ee4da4d..3a888457c1f 100644 --- a/vortex/src/editions/preview/v2026_06.rs +++ b/vortex/src/editions/preview/v2026_06.rs @@ -17,5 +17,11 @@ pub static DECLARATION: EditionDeclaration = EditionDeclaration { id: PREVIEW_2026_06_0, min_vortex_version: None, }, - added: &[EditionMember::layout(&"vortex.list")], + added: &[ + EditionMember::layout(&"vortex.list"), + // Written only by CUDA-enabled sessions, which register the layout through + // `vortex_cuda::layout::register_cuda_layout`. A writer resolves layouts against the + // enabled editions, so the GPU flat layout has to be a member to be written at all. + EditionMember::layout(&"vortex.cuda_flat"), + ], };