bench: run the full compress suite under --gpu-decompress

.github/workflows/pr-bench-gpu-compress.yml

@@ -32,6 +32,13 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
           enable-sccache: "true"
+      # The Public BI datasets are converted from CSV with the DuckDB CLI.
+      - name: Install DuckDB
+        run: |
+          wget -qO- https://github.com/duckdb/duckdb/releases/download/v1.5.5/duckdb_cli-linux-amd64.zip | funzip > duckdb
+          chmod +x duckdb
+          echo "$PWD" >> "$GITHUB_PATH"
+
       - uses: ./.github/actions/system-info
       - name: Display NVIDIA GPU details
         run: |

benchmarks/compress-bench/README.md

@@ -15,8 +15,9 @@ See [`src/main.rs`](./src/main.rs) for the dataset list and CLI flags (`--format
 cargo run -p compress-bench --profile release_debug
 ```
 
-GPU decompression is opt-in and runs only the existing benchmark names allow-listed in
-`src/main.rs`:
+GPU decompression is opt-in and runs the full compress suite, including the
+pcodec-hosted `airquality` and `rplace` datasets. Use `--datasets` to narrow it to a
+subset:
 
 ```bash
 cargo run -p compress-bench --profile release_debug \

benchmarks/compress-bench/src/main.rs

@@ -178,14 +178,6 @@ async fn run_compress(
         // ),
     ];
 
-    // Add an existing benchmark name here only after its CUDA-compatible compression and
-    // decompression kernels have been verified end to end.
-    #[expect(
-        clippy::useless_vec,
-        reason = "this is an intentionally incremental allow-list of benchmark names"
-    )]
-    let gpu_decompress_benchmarks = vec!["TPC-H l_comment canonical"];
-
     let datasets: Vec<&dyn Dataset> = [
         &TaxiData as &dyn Dataset,
         PBI_DATASETS.get(Arade),
@@ -207,11 +199,11 @@ async fn run_compress(
     .into_iter()
     .chain(structlistofints.iter().map(|d| d as &dyn Dataset))
     .filter(|d| {
-        if gpu_decompress && !gpu_decompress_benchmarks.contains(&d.name()) {
-            return false;
-        }
         if let Some(filter) = datasets_filter.as_ref() {
             filter.is_match(d.name())
+        } else if gpu_decompress {
+            // The GPU suite runs every dataset, including the pcodec-hosted ones.
+            true
         } else {
             // These download data from pcodec's public bucket, presumably creating egress charges
             // for pcodec. As such, we do not run in CI.

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();

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)

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 <typename OffsetT>
+__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<uint64_t>(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<OffsetT>(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<OffsetT>(offsets, out_offsets, out_sizes, list_len);                                      \
+    }
+
+FOR_EACH_INTEGER(GENERATE_LIST_VIEWS_KERNEL)

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 <typename EndsT>
+__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<uint64_t>(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<uint8_t>(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<EndsType>(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)

vortex-cuda/src/canonical.rs

@@ -5,29 +5,67 @@ 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::Bool;
 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;
 use vortex::array::arrays::varbinview::VarBinViewDataParts;
 use vortex::array::buffer::BufferHandle;
 use vortex::array::legacy_session;
+use vortex::array::validity::Validity;
 use vortex::buffer::BitBuffer;
 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<ArrayRef> {
+    #[allow(clippy::disallowed_methods)]
+    Ok(child
+        .execute::<Canonical>(&mut legacy_session().create_execution_ctx())?
+        .into_host()
+        .await?
+        .into_array())
+}
+
+/// Copy a `Validity::Array` bitmap back to the host.
+///
+/// Canonical arrays keep their validity as a separate child array, so moving only the data
+/// buffer to the host leaves a device-resident bitmap behind that CPU compute cannot read.
+async fn validity_into_host(validity: Validity) -> VortexResult<Validity> {
+    let Validity::Array(array) = validity else {
+        return Ok(validity);
+    };
+
+    let Ok(bools) = array.clone().try_downcast::<Bool>() else {
+        return Ok(Validity::Array(array));
+    };
+
+    let len = bools.len();
+    let inner_validity = bools.validity()?;
+    let BoolDataParts { bits, meta } = bools.into_data().into_parts(len);
+    let bits = BitBuffer::new_with_offset(bits.try_into_host()?.await?, meta.len(), meta.offset());
+
+    Ok(Validity::Array(
+        BoolArray::new(bits, inner_validity).into_array(),
+    ))
+}
+
 /// Move all canonical data from to_host from device.
 #[async_trait]
 pub trait CanonicalCudaExt {
@@ -50,6 +88,7 @@ impl CanonicalCudaExt for Canonical {
                     validity,
                     ..
                 } = struct_array.into_data_parts();
+                let validity = validity_into_host(validity).await?;
 
                 let mut host_fields = vec![];
                 for field in fields.iter() {
@@ -75,7 +114,7 @@ impl CanonicalCudaExt for Canonical {
                 // NOTE: update to copy to host when adding buffer handle.
                 // Also update other method to copy validity to host.
                 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(
@@ -92,6 +131,7 @@ impl CanonicalCudaExt for Canonical {
                     validity,
                     ..
                 } = prim.into_data_parts();
+                let validity = validity_into_host(validity).await?;
                 Ok(Canonical::Primitive(PrimitiveArray::from_byte_buffer(
                     buffer.try_into_host()?.await?,
                     ptype,
@@ -106,6 +146,7 @@ impl CanonicalCudaExt for Canonical {
                     validity,
                     ..
                 } = decimal.into_data_parts();
+                let validity = validity_into_host(validity).await?;
                 Ok(Canonical::Decimal(unsafe {
                     DecimalArray::new_unchecked_handle(
                         BufferHandle::new_host(values.try_into_host()?.await?),
@@ -122,6 +163,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?;
@@ -140,6 +182,23 @@ impl CanonicalCudaExt for Canonical {
                     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.
                 let host_storage = ext