Skip to content

Commit e55b447

Browse files
committed
feat: add CUDA cuDF convenience API
Add vortex_cuda.to_cudf and install optional Vortex Array helpers for cuDF conversion and Arrow C Device export. Keep the conversion path CUDA-only by rejecting unsupported fallback policies and routing cuDF ingestion through fresh Arrow C Device capsules. Expand CUDA Python tests for the convenience API, installed Array methods, Arrow Device export smoke coverage, and capsule ownership paths. Signed-off-by: Alexander Droste <alexander.droste@protonmail.com>
1 parent 4a90e13 commit e55b447

7 files changed

Lines changed: 673 additions & 16 deletions

File tree

vortex-cuda/src/arrow/canonical.rs

Lines changed: 85 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -764,7 +764,7 @@ fn gather_binary_values(
764764
///
765765
/// Returns `None` for the buffer when Arrow can omit validity because all rows are valid.
766766
///
767-
/// Returned buffers use zeroed 4-byte padding so cuDF's word-sized mask reads stay in bounds.
767+
/// Returned buffers use zeroed cuDF-sized padding so mask reads stay in bounds.
768768
/// Bits at positions `>= len + arrow_offset` within the final data byte are unspecified, as
769769
/// Arrow permits.
770770
pub(super) async fn export_arrow_validity_buffer(
@@ -798,21 +798,26 @@ pub(super) async fn export_arrow_validity_buffer(
798798
let bitmap = ctx.ensure_on_device(bits).await?;
799799
// ArrowDeviceArray uses ArrowArray layout with its buffers being device pointers.
800800
//
801-
// Validity is one bit per row, addressed via the Arrow array offset. Reuse the bitmap
802-
// when Vortex's validity offset already matches Arrow's; otherwise repack on the GPU
803-
// so row i is at Arrow bit `arrow_offset + i`.
804-
let bitmap = if meta.offset() == arrow_offset {
805-
bitmap
806-
} else {
807-
repack_arrow_validity_buffer(&bitmap, meta.offset(), len, arrow_offset, ctx)?
808-
};
801+
// Validity is one bit per row, addressed via the Arrow array offset. Repack on the GPU
802+
// so row i is at Arrow bit `arrow_offset + i` and the backing allocation has the
803+
// zeroed cuDF-sized padding expected by Arrow Device consumers.
804+
let bitmap =
805+
repack_arrow_validity_buffer(&bitmap, meta.offset(), len, arrow_offset, ctx)?;
809806
// Keep nullable exports self-describing for consumers that require exact null counts.
810807
let null_count = count_arrow_validity_nulls(&bitmap, len, arrow_offset, ctx)?;
811808
Ok((Some(bitmap), null_count))
812809
}
813810
}
814811
}
815812

813+
/// Minimum backing allocation quantum for Arrow validity buffers handed to cuDF.
814+
///
815+
/// Arrow exposes only the logical bitmap byte length, but cuDF imports null masks into 64-byte
816+
/// padded buffers and its kernels may read through that padded extent. Vortex therefore keeps the
817+
/// exported `BufferHandle` sliced to Arrow's logical length while zero-padding the underlying CUDA
818+
/// allocation to this boundary.
819+
const CUDF_VALIDITY_BUFFER_PADDING: usize = 64;
820+
816821
/// Return the byte length needed for `len` validity bits at the given bit offset.
817822
fn validity_bitmap_byte_len(len: usize, arrow_offset: usize) -> VortexResult<usize> {
818823
Ok(len
@@ -821,12 +826,25 @@ fn validity_bitmap_byte_len(len: usize, arrow_offset: usize) -> VortexResult<usi
821826
.div_ceil(8))
822827
}
823828

829+
/// Return the CUDA allocation size for a logical Arrow validity bitmap byte length.
830+
///
831+
/// The returned allocation length may be larger than `byte_len`; callers slice the exported
832+
/// `BufferHandle` back to `byte_len` while retaining the padded backing allocation. A zero-length
833+
/// bitmap still gets one byte so we never request a zero-sized CUDA allocation.
834+
fn validity_bitmap_allocation_byte_len(byte_len: usize) -> usize {
835+
if byte_len == 0 {
836+
1
837+
} else {
838+
byte_len.next_multiple_of(CUDF_VALIDITY_BUFFER_PADDING)
839+
}
840+
}
841+
824842
/// Allocate a zeroed device buffer with cuDF-safe padding for Arrow validity masks.
825843
fn device_zeroed_byte_buffer(
826844
byte_len: usize,
827845
ctx: &mut CudaExecutionCtx,
828846
) -> VortexResult<BufferHandle> {
829-
let allocation_len = byte_len.next_multiple_of(size_of::<u32>()).max(1);
847+
let allocation_len = validity_bitmap_allocation_byte_len(byte_len);
830848
let mut buffer = ctx.device_alloc::<u8>(allocation_len)?;
831849
ctx.stream()
832850
.memset_zeros(&mut buffer)
@@ -894,8 +912,8 @@ pub fn count_arrow_validity_nulls(
894912
///
895913
/// Vortex bitmaps may start at any bit offset. Arrow exposes only a byte-addressed validity buffer
896914
/// plus an array offset, so sliced compact exports need a GPU rewrite when either side has a
897-
/// bit-level offset. The kernel writes the output one 64-bit word at a time, funnel-shifting two
898-
/// adjacent input words, so the allocation is padded to whole words (zeroed by the edge masks).
915+
/// bit-level offset. The output handle keeps Arrow's logical byte length, while the backing
916+
/// allocation is zero-padded to cuDF's mask allocation size for consumers that read full masks.
899917
pub fn repack_arrow_validity_buffer(
900918
input_buffer: &BufferHandle,
901919
input_offset: usize,
@@ -904,7 +922,13 @@ pub fn repack_arrow_validity_buffer(
904922
ctx: &mut CudaExecutionCtx,
905923
) -> VortexResult<BufferHandle> {
906924
let output_bytes = validity_bitmap_byte_len(len, arrow_offset)?;
925+
// The CUDA kernel writes the bitmap as u64 words, so round the logical byte length up to the
926+
// number of words that cover the exported Arrow bytes.
907927
let output_words = output_bytes.div_ceil(size_of::<u64>());
928+
// `device_alloc::<u64>` takes a word count, while the padding policy is expressed in bytes.
929+
// Round up so the padded byte allocation is fully represented by whole u64 words.
930+
let allocation_words =
931+
validity_bitmap_allocation_byte_len(output_bytes).div_ceil(size_of::<u64>());
908932

909933
// The kernel loads the input bitmap as 64-bit words.
910934
if !input_buffer
@@ -914,7 +938,12 @@ pub fn repack_arrow_validity_buffer(
914938
vortex_bail!("Arrow validity repack requires an 8-byte aligned device buffer");
915939
}
916940

917-
let output = ctx.device_alloc::<u64>(output_words.max(1))?;
941+
let mut output = ctx.device_alloc::<u64>(allocation_words.max(1))?;
942+
// The repack kernel writes only the logical bitmap words. Zero the whole backing allocation so
943+
// cuDF's padded mask reads see invalid rows, not uninitialized CUDA memory.
944+
ctx.stream()
945+
.memset_zeros(&mut output)
946+
.map_err(|err| vortex_err!("Failed to zero Arrow validity buffer padding: {err}"))?;
918947
let output_device = CudaDeviceBuffer::new(output);
919948

920949
if output_words > 0 {
@@ -1337,6 +1366,7 @@ mod tests {
13371366
use crate::arrow::ArrowDeviceArray;
13381367
use crate::arrow::DeviceArrayExt;
13391368
use crate::arrow::PrivateData;
1369+
use crate::arrow::canonical::CUDF_VALIDITY_BUFFER_PADDING;
13401370
use crate::arrow::canonical::export_arrow_validity_buffer;
13411371
use crate::arrow::canonical::repack_arrow_validity_buffer;
13421372
use crate::device_buffer::cuda_backing_allocation;
@@ -2955,7 +2985,43 @@ mod tests {
29552985
let backing_bytes = backing.to_host_sync();
29562986
assert_eq!(
29572987
backing_bytes.len(),
2958-
output_bytes.next_multiple_of(size_of::<u64>())
2988+
output_bytes.next_multiple_of(CUDF_VALIDITY_BUFFER_PADDING)
2989+
);
2990+
assert!(backing_bytes[output_bytes..].iter().all(|byte| *byte == 0));
2991+
2992+
Ok(())
2993+
}
2994+
2995+
#[crate::test]
2996+
async fn test_export_validity_buffer_pads_matching_offset() -> VortexResult<()> {
2997+
let mut ctx = CudaSession::create_execution_ctx(&crate::cuda_session())
2998+
.vortex_expect("failed to create execution context");
2999+
3000+
let len = 3;
3001+
let arrow_offset = 0;
3002+
let (buffer, null_count) = export_arrow_validity_buffer(
3003+
Validity::from(BitBuffer::from_iter([true, false, true])),
3004+
len,
3005+
arrow_offset,
3006+
&mut ctx,
3007+
)
3008+
.await?;
3009+
ctx.synchronize_stream()?;
3010+
3011+
assert_eq!(null_count, 1);
3012+
let buffer = buffer.vortex_expect("nullable validity should export a null buffer");
3013+
let output_bytes = (len + arrow_offset).div_ceil(8);
3014+
assert_eq!(buffer.len(), output_bytes);
3015+
let actual = BitBuffer::new(buffer.to_host_sync(), len + arrow_offset)
3016+
.iter()
3017+
.collect::<Vec<_>>();
3018+
assert_eq!(actual, [true, false, true]);
3019+
3020+
let backing = cuda_backing_allocation(&buffer)?;
3021+
let backing_bytes = backing.to_host_sync();
3022+
assert_eq!(
3023+
backing_bytes.len(),
3024+
output_bytes.next_multiple_of(CUDF_VALIDITY_BUFFER_PADDING)
29593025
);
29603026
assert!(backing_bytes[output_bytes..].iter().all(|byte| *byte == 0));
29613027

@@ -2983,6 +3049,11 @@ mod tests {
29833049
let bytes = buffer.to_host_sync();
29843050
assert_eq!(bytes.len(), (len + arrow_offset).div_ceil(8));
29853051
assert!(bytes.iter().all(|byte| *byte == 0));
3052+
let backing = cuda_backing_allocation(&buffer)?;
3053+
assert_eq!(
3054+
backing.len(),
3055+
bytes.len().next_multiple_of(CUDF_VALIDITY_BUFFER_PADDING)
3056+
);
29863057

29873058
Ok(())
29883059
}

vortex-python-cuda/README.md

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
# vortex-data-cuda
2+
3+
CUDA extension for [Vortex](https://vortex.dev). Exports a `vortex.Array` to
4+
[RAPIDS cuDF](https://docs.rapids.ai/api/cudf/stable/) or any
5+
[Arrow C Device](https://arrow.apache.org/docs/format/CDeviceDataInterface.html) consumer, on the
6+
GPU. Imported as `vortex_cuda`.
7+
8+
## Install
9+
10+
```bash
11+
pip install vortex-data vortex-data-cuda # versions must match; CUDA device required
12+
```
13+
14+
`to_cudf` also needs RAPIDS `cudf` and `pylibcudf` in the environment.
15+
16+
## Export to cuDF
17+
18+
`to_cudf` converts via the Arrow C Device interface: struct arrays become a `cudf.DataFrame`,
19+
everything else a `cudf.Series`. Importing `vortex_cuda` installs it as `vortex.Array.to_cudf`.
20+
21+
```python
22+
import vortex, vortex_cuda
23+
import pyarrow as pa
24+
25+
s = vortex.array([1, None, 3]).to_cudf() # -> cudf.Series
26+
df = vortex_cuda.to_cudf( # struct -> cudf.DataFrame
27+
vortex.Array.from_arrow(pa.table({"x": [1, None, 3], "y": [4.0, 5.0, 6.0]}))
28+
)
29+
```
30+
31+
Buffers are imported zero-copy; host arrays are moved to the GPU as part of the export. cuDF keeps
32+
shared ownership for the lifetime of the result and any view derived from it, so no extra
33+
bookkeeping is needed.
34+
35+
Signature: `to_cudf(obj, *, fallback="error")`. Only `fallback="error"` is supported
36+
(`NotImplementedError` otherwise); raises `TypeError` for a non-`vortex.Array`, `RuntimeError`
37+
without a CUDA device, `ImportError` if cuDF/pylibcudf are missing.
38+
39+
## Export an Arrow C Device array
40+
41+
`vortex.Array` exposes the standard `__arrow_c_device_array__` protocol (installed when CUDA is
42+
available), so any Arrow-C-Device consumer can ingest it zero-copy:
43+
44+
```python
45+
import vortex, vortex_cuda, pylibcudf
46+
47+
array = vortex.array([1, None, 3])
48+
column = pylibcudf.Column.from_arrow(array) # via the protocol
49+
50+
schema_capsule, device_array_capsule = vortex_cuda.export_device_array(array) # raw capsules
51+
```
52+
53+
`export_device_array` returns `PyCapsule`s named `"arrow_schema"` and `"arrow_device_array"`. The
54+
consumer owns the exported structs and runs the Arrow release callbacks when done (libcudf does
55+
this automatically); Vortex's device buffers stay alive until then.
56+
57+
## Notes
58+
59+
- Integer, float, bool, and string arrays (incl. nullable) and structs are supported; nulls are
60+
preserved.
61+
- A CUDA device is required; there is no CPU fallback.

vortex-python-cuda/python/vortex_cuda/__init__.py

Lines changed: 97 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,107 @@
22
# SPDX-FileCopyrightText: Copyright the Vortex contributors
33
# pyright: reportMissingModuleSource=false, reportPrivateUsage=false
44

5+
import importlib
6+
from typing import Protocol, cast
7+
58
from . import _lib
69

10+
# Private debug hooks used by CUDA bridge tests.
711
_debug_array_metadata_dtype = _lib._debug_array_metadata_dtype
812
_debug_array_metadata_display_values = _lib._debug_array_metadata_display_values
13+
_debug_arrow_device_array_capsule_summary = _lib._debug_arrow_device_array_capsule_summary
14+
_debug_consume_arrow_device_array_capsules = _lib._debug_consume_arrow_device_array_capsules
915
cuda_available = _lib.cuda_available
1016
export_device_array = _lib.export_device_array
1117

12-
__all__ = ["cuda_available", "export_device_array"]
18+
19+
class _FromArrow(Protocol):
20+
def from_arrow(self, obj: object) -> object: ...
21+
22+
23+
class _FromPylibcudf(Protocol):
24+
def from_pylibcudf(self, obj: object) -> object: ...
25+
26+
27+
class _DataFrameFromPylibcudf(Protocol):
28+
def from_pylibcudf(self, obj: object, metadata: dict[str, object] | None = None) -> object: ...
29+
30+
31+
class _PylibcudfModule(Protocol):
32+
Column: _FromArrow
33+
Table: _FromArrow
34+
35+
36+
class _CudfModule(Protocol):
37+
Series: _FromPylibcudf
38+
DataFrame: _DataFrameFromPylibcudf
39+
40+
41+
_SUPPORTED_FALLBACKS = frozenset({"error"})
42+
43+
44+
def _Array_to_cudf(self: object, *, fallback: str = "error") -> object:
45+
return to_cudf(self, fallback=fallback)
46+
47+
48+
def _Array___arrow_c_device_array__(
49+
self: object,
50+
requested_schema: object | None = None,
51+
**kwargs: object,
52+
) -> tuple[object, object]:
53+
return export_device_array(self, requested_schema, **kwargs)
54+
55+
56+
def _install_vortex_array_methods() -> None:
57+
import vortex
58+
59+
setattr(vortex.Array, "to_cudf", _Array_to_cudf)
60+
if cuda_available():
61+
setattr(vortex.Array, "__arrow_c_device_array__", _Array___arrow_c_device_array__)
62+
63+
64+
def _import_cudf_modules() -> tuple[_CudfModule, _PylibcudfModule]:
65+
try:
66+
cudf = importlib.import_module("cudf")
67+
pylibcudf = importlib.import_module("pylibcudf")
68+
except ImportError as err:
69+
raise ImportError("vortex_cuda.to_cudf requires RAPIDS cuDF and pylibcudf to be installed") from err
70+
return cast(_CudfModule, cast(object, cudf)), cast(_PylibcudfModule, cast(object, pylibcudf))
71+
72+
73+
def to_cudf(obj: object, *, fallback: str = "error") -> object:
74+
"""Convert a Vortex array to a cuDF object through the Arrow Device interface.
75+
76+
pylibcudf imports the exported Arrow Device array zero-copy and keeps shared ownership of
77+
Vortex's device buffers (via libcudf's ``arrow_column``) for the lifetime of the returned
78+
cuDF object and any view derived from it, so no extra keepalive is required here.
79+
80+
``fallback`` is reserved for future policy choices. The initial implementation
81+
supports only ``fallback="error"`` and never falls back to host Arrow conversion.
82+
"""
83+
if fallback not in _SUPPORTED_FALLBACKS:
84+
raise NotImplementedError("vortex_cuda.to_cudf currently supports only fallback='error'")
85+
86+
import vortex
87+
88+
if not isinstance(obj, vortex.Array):
89+
raise TypeError(f"vortex_cuda.to_cudf expected a vortex.Array, got {type(obj).__name__}")
90+
91+
if not cuda_available():
92+
raise RuntimeError("CUDA is not available; vortex_cuda.to_cudf requires a CUDA device")
93+
94+
cudf, pylibcudf = _import_cudf_modules()
95+
96+
dtype = obj.dtype
97+
if isinstance(dtype, vortex.StructDType):
98+
table = pylibcudf.Table.from_arrow(obj)
99+
return cudf.DataFrame.from_pylibcudf(table, metadata={"columns": dtype.names()})
100+
101+
column = pylibcudf.Column.from_arrow(obj)
102+
return cudf.Series.from_pylibcudf(column)
103+
104+
105+
_install_vortex_array_methods()
106+
107+
108+
__all__ = ["cuda_available", "export_device_array", "to_cudf"]

vortex-python-cuda/python/vortex_cuda/_lib.pyi

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,10 @@
33

44
def _debug_array_metadata_dtype(array: object) -> str: ...
55
def _debug_array_metadata_display_values(array: object) -> str: ...
6+
def _debug_arrow_device_array_capsule_summary(schema: object, device_array: object) -> dict[str, object]: ...
7+
def _debug_consume_arrow_device_array_capsules(
8+
schema: object, device_array: object
9+
) -> tuple[bool, bool, bool, bool, bool, bool]: ...
610
def cuda_available() -> bool: ...
711
def export_device_array(
812
array: object, requested_schema: object | None = None, **kwargs: object

0 commit comments

Comments
 (0)