From 333c065e86732b08dd7ccd5d41d43432a62a6b7e Mon Sep 17 00:00:00 2001 From: Bradley Dice Date: Tue, 1 Sep 2026 17:16:44 -0500 Subject: [PATCH 1/3] Adapt stream accessors to cuda::stream_ref --- .../cudf_streaming/partition_utils.pyx | 12 ++++++------ python/cudf_streaming/cudf_streaming/table_chunk.pyx | 2 +- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/python/cudf_streaming/cudf_streaming/partition_utils.pyx b/python/cudf_streaming/cudf_streaming/partition_utils.pyx index 5646bec5b83b..a7a143324a37 100644 --- a/python/cudf_streaming/cudf_streaming/partition_utils.pyx +++ b/python/cudf_streaming/cudf_streaming/partition_utils.pyx @@ -119,7 +119,7 @@ cpdef size_t partition_and_pack_cost( -------- cudf_streaming.partition_utils.partition_and_pack """ - cdef stream_ref _stream = stream_ref(stream.view().get()) + cdef stream_ref _stream = stream.view() cdef cpp_BufferResource* _br = br.ptr() cdef table_view tbl = table.view() cdef size_t ret @@ -177,7 +177,7 @@ cpdef object partition_and_pack( pylibcudf.contiguous_split.pack cudf_streaming.partition_utils.split_and_pack """ - cdef stream_ref _stream = stream_ref(stream.view().get()) + cdef stream_ref _stream = stream.view() cdef cpp_BufferResource* _br = br.ptr() cdef vector[size_type] _columns_to_hash = tuple(columns_to_hash) cdef unordered_map[uint32_t, cpp_PackedData] _ret @@ -244,7 +244,7 @@ cpdef size_t split_and_pack_cost( -------- cudf_streaming.partition_utils.split_and_pack """ - cdef stream_ref _stream = stream_ref(stream.view().get()) + cdef stream_ref _stream = stream.view() cdef cpp_BufferResource* _br = br.ptr() cdef table_view tbl = table.view() cdef size_t ret @@ -299,7 +299,7 @@ cpdef object split_and_pack( pylibcudf.copying.split cudf_streaming.partition_utils.partition_and_pack """ - cdef stream_ref _stream = stream_ref(stream.view().get()) + cdef stream_ref _stream = stream.view() cdef cpp_BufferResource* _br = br.ptr() cdef vector[size_type] _splits = tuple(splits) cdef unordered_map[uint32_t, cpp_PackedData] _ret @@ -463,7 +463,7 @@ cpdef object unpack_and_concat( cudf_streaming.partition_utils.unpack_and_concat_cost cudf_streaming.partition_utils.partition_and_pack """ - cdef stream_ref _stream = stream_ref(stream.view().get()) + cdef stream_ref _stream = stream.view() cdef cpp_BufferResource* _br = br.ptr() cdef vector[cpp_PackedData] _partitions = _partitions_py_to_cpp(partitions) cdef unique_ptr[cpp_table] _ret @@ -550,7 +550,7 @@ cpdef object packed_data_from_cudf_packed_columns( """ if packed_columns is None or stream is None or br is None: raise TypeError("Arguments must not be None") - cdef stream_ref _stream = stream_ref(stream.view().get()) + cdef stream_ref _stream = stream.view() cdef cpp_BufferResource* _br = br.ptr() cdef PackedData ret = PackedData.__new__(PackedData) with nogil: diff --git a/python/cudf_streaming/cudf_streaming/table_chunk.pyx b/python/cudf_streaming/cudf_streaming/table_chunk.pyx index ed6a41674e49..735fb78d8622 100644 --- a/python/cudf_streaming/cudf_streaming/table_chunk.pyx +++ b/python/cudf_streaming/cudf_streaming/table_chunk.pyx @@ -183,7 +183,7 @@ cdef class TableChunk: persists even when the chunk is transferred through Channels. """ - cdef stream_ref _stream = stream_ref(stream.view().get()) + cdef stream_ref _stream = stream.view() cdef cpp_table_view view = table.view() return TableChunk.from_handle( cpp_from_table_view_with_owner( From 5347e1c6fd11335527c6c65a2d60371018ac7ebb Mon Sep 17 00:00:00 2001 From: Bradley Dice Date: Wed, 2 Sep 2026 17:32:59 -0500 Subject: [PATCH 2/3] Use cuda::stream_ref in reader_impl_dict_transcode.cu --- cpp/src/io/parquet/reader_impl_dict_transcode.cu | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cpp/src/io/parquet/reader_impl_dict_transcode.cu b/cpp/src/io/parquet/reader_impl_dict_transcode.cu index 6c45678d3803..4ee6af6460d8 100644 --- a/cpp/src/io/parquet/reader_impl_dict_transcode.cu +++ b/cpp/src/io/parquet/reader_impl_dict_transcode.cu @@ -194,7 +194,7 @@ void remap_dict_indices_by_chunk(cudf::device_span indices, cudf::device_span row_offsets, cudf::device_span key_counts_prefix, cudf::device_span stacked_to_unique, - rmm::cuda_stream_view stream) + cuda::stream_ref stream) { thrust::for_each( rmm::exec_policy_nosync(stream, get_current_device_resource_ref()), From 1e03a9cba1ecd15bc7318de2d5a08c70dcb06a88 Mon Sep 17 00:00:00 2001 From: Bradley Dice Date: Wed, 2 Sep 2026 21:29:59 -0500 Subject: [PATCH 3/3] Fix remaining cuda::stream_ref build errors --- cpp/libcudf_streaming/src/bloom_filter.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cpp/libcudf_streaming/src/bloom_filter.cpp b/cpp/libcudf_streaming/src/bloom_filter.cpp index da857b99773b..7f77697cb399 100644 --- a/cpp/libcudf_streaming/src/bloom_filter.cpp +++ b/cpp/libcudf_streaming/src/bloom_filter.cpp @@ -118,7 +118,7 @@ rapidsmpf::streaming::Actor bloom_filter::apply( auto storage = (co_await bloom_filter->receive()).release(); RAPIDSMPF_EXPECTS((co_await bloom_filter->receive()).empty(), "Bloom filter channel contained more than one message"); - auto stream = cuda::stream_ref{storage.stream().get()}; + auto stream = storage.stream(); rapidsmpf::CudaEvent event; auto filter = cudf_streaming::detail::device_bloom_filter(filter_size_, seed_, storage.data()); auto meta = co_await ch_in->receive_metadata();