Fix host-source lifetime races

cpp/examples/parquet_inspect/parquet_inspect_utils.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -121,6 +121,7 @@ auto make_index_column(cudf::size_type num_rows, rmm::cuda_stream_view stream)
   std::vector<cudf::size_type> data(num_rows);
   std::iota(data.begin(), data.end(), 0);
   auto buffer = rmm::device_buffer(data.data(), num_rows * sizeof(int64_t), stream);
+  stream.synchronize();
   return std::make_unique<cudf::column>(cudf::data_type{cudf::type_to_id<cudf::size_type>()},
                                         num_rows,
                                         std::move(buffer),
@@ -141,6 +142,7 @@ template <typename T>
 auto make_column(cudf::host_span<T const> host_data, rmm::cuda_stream_view stream)
 {
   auto device_buffer = rmm::device_buffer(host_data.data(), host_data.size() * sizeof(T), stream);
+  stream.synchronize();
   return std::make_unique<cudf::column>(cudf::data_type{cudf::type_to_id<T>()},
                                         host_data.size(),
                                         std::move(device_buffer),
@@ -174,6 +176,7 @@ auto make_page_data_list_column(cudf::host_span<T const> data,
 
   auto page_data_buffer =
     rmm::device_buffer(data.data(), num_pages_this_column * sizeof(int64_t), stream);
+  stream.synchronize();
 
   auto page_data_column =
     std::make_unique<cudf::column>(cudf::data_type{cudf::type_to_id<int64_t>()},
@@ -285,6 +288,7 @@ void write_rowgroup_metadata(cudf::io::parquet::FileMetaData const& metadata,
   auto byte_offsets_buffer =
     rmm::device_buffer(row_group_byte_offsets.data(), num_row_groups * sizeof(int64_t), stream);
 
+  stream.synchronize();
   columns.emplace_back(std::make_unique<cudf::column>(cudf::data_type{cudf::type_to_id<int64_t>()},
                                                       num_row_groups,
                                                       std::move(row_offsets_buffer),

cpp/libcudf_streaming/tests/streaming/test_channel_metadata.cpp

@@ -167,6 +167,7 @@ class StreamingChannelMetadataGPU : public ::testing::Test {
   std::shared_ptr<table_chunk> make_chunk(std::vector<int32_t> vals)
   {
     rmm::device_buffer buf(vals.data(), vals.size() * sizeof(int32_t), stream);
+    stream.synchronize();
     auto col = std::make_unique<cudf::column>(cudf::data_type{cudf::type_id::INT32},
                                               static_cast<cudf::size_type>(vals.size()),
                                               std::move(buf),

cpp/src/copying/concatenate.cu

@@ -89,6 +89,7 @@ auto create_device_views(host_span<column_view const> views, cuda::stream_ref st
   auto d_offsets =
     make_device_uvector_async(offsets, stream, cudf::get_current_device_resource_ref());
   auto const output_size = offsets.back();
+  stream.sync();
 
   return std::make_tuple(
     std::move(device_view_owners), std::move(d_views), std::move(d_offsets), output_size);

cpp/src/copying/contiguous_split.cu

@@ -1657,6 +1657,7 @@ std::unique_ptr<chunk_iteration_state> chunk_iteration_state::create(
           d_batched_dst_buf_info[i].dst_offset -= *prior_iteration_size;
         });
     }
+    stream.sync();
     return std::make_unique<chunk_iteration_state>(std::move(d_batched_dst_buf_info),
                                                    std::move(d_batch_offsets),
                                                    std::move(num_batches_per_iteration),

cpp/src/groupby/sort/sort_helper.cu

@@ -136,6 +136,7 @@ sort_groupby_helper::index_vector const& sort_groupby_helper::group_offsets(cuda
 
   group_offsets->set_element_async(num_groups, size, stream);
   group_offsets->resize(num_groups + 1, stream);
+  stream.sync();
 
   _group_offsets = std::move(group_offsets);
   return *_group_offsets;

cpp/src/groupby/streaming_groupby/common.cuh

@@ -239,7 +239,9 @@ auto build_cross_comparators(
     h_eqs.push_back(adapter.comparator);
   }
 
-  return cudf::detail::make_device_uvector_async(h_eqs, stream, temp_mr);
+  auto result = cudf::detail::make_device_uvector_async(h_eqs, stream, temp_mr);
+  stream.sync();
+  return result;
 }
 
 /// The impl struct for streaming_groupby. Defined in impl.cu.

cpp/src/interop/from_arrow_device.cu

@@ -197,6 +197,7 @@ dispatch_tuple_t dispatch_from_arrow_device::operator()<cudf::string_view>(
     auto out_col =
       cudf::strings::detail::make_strings_column(d_indices.begin(), d_indices.end(), stream, mr);
     owned.emplace_back(std::move(out_col));
+    stream.synchronize();
     return std::make_tuple<column_view, owned_columns_t>(owned.front()->view(), std::move(owned));
   }
 

cpp/src/interop/from_arrow_host.cu

@@ -445,6 +445,7 @@ std::tuple<std::unique_ptr<column>, int64_t, int64_t> get_offsets_column(
   offsets_array.offset                         = 0;  // already accounted for by the above transform
   auto result = dispatch_copy_from_arrow_host{stream, mr}.template operator()<int32_t>(
     schema, &offsets_array, data_type(type_id::INT32), true);
+  stream.synchronize();
   return std::tuple{std::move(result), offset, length};
 }
 
@@ -489,6 +490,7 @@ std::unique_ptr<table> from_arrow_host(ArrowSchema const* schema,
       std::overflow_error);
     return std::make_unique<table>(std::move(columns), static_cast<size_type>(input->array.length));
   }
+  stream.synchronize();
   return std::make_unique<table>(std::move(columns));
 }
 
@@ -507,8 +509,10 @@ std::unique_ptr<column> from_arrow_host_column(ArrowSchema const* schema,
   ArrowSchemaView view;
   NANOARROW_THROW_NOT_OK(ArrowSchemaViewInit(&view, schema, nullptr));
 
-  auto type = arrow_to_cudf_type(&view);
-  return get_column_copy(&view, &input->array, type, false, stream, mr);
+  auto type   = arrow_to_cudf_type(&view);
+  auto result = get_column_copy(&view, &input->array, type, false, stream, mr);
+  stream.synchronize();
+  return result;
 }
 
 std::unique_ptr<column> get_column_from_host_copy(ArrowSchemaView const* schema,

cpp/src/interop/from_arrow_host_strings.cu

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -113,6 +113,7 @@ std::unique_ptr<column> from_arrow_stringview(ArrowSchemaView const* schema,
       return {data, size};
     });
 
+  stream.synchronize();
   return cudf::strings::detail::make_strings_column(d_indices.begin(), d_indices.end(), stream, mr);
 }
 

cpp/src/io/orc/stripe_enc.cu

@@ -1426,6 +1426,7 @@ void decimal_sizes_to_offsets(device_2dspan<rowgroup_rows const> rg_bounds,
   decimal_sizes_to_offsets_kernel<block_size>
     <<<num_blocks, block_size, 0, stream.get()>>>(rg_bounds, d_sizes);
   CUDF_CUDA_TRY(cudaGetLastError());
+  stream.sync();
 }
 
 }  // namespace cudf::io::orc::detail

cpp/src/io/orc/writer_impl.cu

@@ -1513,6 +1513,7 @@ encoded_footer_statistics finish_statistic_blobs(Footer const& footer,
       file_blobs[i].assign(stat_begin, stat_end);
     }
 
+    stream.sync();
     return {{}, std::move(file_blobs)};
   }
 
@@ -1595,6 +1596,7 @@ encoded_footer_statistics finish_statistic_blobs(Footer const& footer,
     file_blobs[i].assign(stat_begin, stat_end);
   }
 
+  stream.sync();
   return {std::move(stripe_blobs), std::move(file_blobs)};
 }
 
@@ -2040,6 +2042,7 @@ orc_table_view make_orc_table_view(table_view const& table,
     },
     stream);
 
+  stream.sync();
   return {std::move(orc_columns),
           std::move(d_orc_columns),
           str_col_indexes,

cpp/src/io/parquet/experimental/deletion_vectors_helpers.cu

@@ -132,6 +132,7 @@ std::unique_ptr<cudf::column> compute_row_index_column(
     row_indices_iter,
     row_indices_iter);
 
+  stream.sync();
   return std::make_unique<cudf::column>(cudf::data_type{cudf::type_id::UINT64},
                                         num_rows,
                                         std::move(row_indices),

cpp/src/io/parquet/experimental/dictionary_page_filter.cu

@@ -1162,6 +1162,7 @@ struct dictionary_caster {
                                                                   physical_type);
     CUDF_CUDA_TRY(cudaGetLastError());
 
+    stream.sync();
     // Build the BOOL8 columns from the results buffers
     return build_columns(results_buffers, stream, mr);
   }

cpp/src/io/parquet/experimental/page_index_filter.cu

@@ -417,6 +417,7 @@ struct page_stats_caster : public stats_caster_base {
       // Construct a row indices mapping based on page row offsets.
       auto const page_indices = compute_page_indices_async(
         page_row_offsets, total_rows, stream, cudf::get_current_device_resource_ref());
+      stream.sync();
 
       // For non-strings columns, directly gather the page-level column data and bitmask to the
       // row-level.
@@ -583,6 +584,7 @@ struct page_stats_to_row_mask_converter : public page_stats_caster {
               stream)
           : cudf::detail::make_empty_host_vector<bitmask_type>(0, stream);
 
+      stream.sync();
       auto [row_mask_data, row_mask_bitmask] =
         build_data_and_nullmask<bool>(page_mask->mutable_view(),
                                       page_mask_nullmask.data(),

cpp/src/io/parquet/reader_impl.cpp

@@ -1182,6 +1182,7 @@ void reader_impl::update_output_nullmasks_for_pruned_pages(cudf::host_span<bool
     std::fill(pinned_valids.begin(), pinned_valids.end(), false);
     cudf::set_null_masks_safe(
       pinned_null_masks, pinned_begin_bits, pinned_end_bits, pinned_valids, _stream);
+    _stream.sync();
   }
   // Otherwise, update the nullmasks in a loop
   else {

cpp/src/merge/merge.cu

@@ -277,6 +277,7 @@ index_vector generate_merged_indices(table_view const& left_table,
 
   CUDF_CHECK_CUDA(stream.value());
 
+  stream.synchronize();
   return merged_indices;
 }
 

cpp/src/row_operator/row_operators.cu

@@ -648,6 +648,7 @@ std::shared_ptr<preprocessed_table> preprocessed_table::create(
     null_precedence, stream, cudf::get_current_device_resource_ref());
   auto d_depths = detail::make_device_uvector_async(
     verticalized_col_depths, stream, cudf::get_current_device_resource_ref());
+  stream.synchronize();
 
   if (detail::has_nested_columns(preprocessed_input)) {
     auto [dremel_data, d_dremel_device_view] = list_lex_preprocess(preprocessed_input, stream);

cpp/src/transform/row_bit_count.cu

@@ -553,6 +553,7 @@ std::unique_ptr<column> segmented_row_bit_count(table_view const& t,
     h_info.max_branch_depth);
   CUDF_CUDA_TRY(cudaGetLastError());
 
+  stream.sync();
   return output;
 }
 

cpp/tests/io/experimental/hybrid_scan_common.cpp

@@ -347,6 +347,7 @@ std::pair<std::unique_ptr<cudf::table>, std::vector<char>> create_parquet_with_s
       auto [null_mask, null_count] = cudf::test::detail::make_null_mask_vector(begin, end);
       auto d_mask                  = rmm::device_buffer{
         null_mask.data(), cudf::bitmask_allocation_size_bytes(cudf::distance(begin, end)), stream};
+      stream.sync();
       return std::pair{std::move(d_mask), null_count};
     };
 

cpp/tests/io/json/json_quote_normalization_test.cpp

@@ -27,6 +27,7 @@ void run_test(std::string const& host_input,
 {
   auto stream_view  = cudf::test::get_default_stream();
   auto device_input = rmm::device_buffer(host_input.c_str(), host_input.size(), stream_view);
+  stream_view.synchronize();
 
   // Preprocessing FST
   cudf::io::datasource::owning_buffer<rmm::device_buffer> device_data(std::move(device_input));

cpp/tests/io/parquet_deletion_vectors_test.cpp

@@ -76,7 +76,7 @@ auto build_column_from_host_data(cudf::host_span<T const> host_data,
 
   auto const num_rows = host_data.size();
   rmm::device_buffer buffer{num_rows * sizeof(T), stream, mr};
-  cudf::detail::cuda_memcpy_async<T>(
+  cudf::detail::cuda_memcpy<T>(
     cudf::device_span<T>{static_cast<T*>(buffer.data()), num_rows}, host_data, stream);
   return std::make_unique<cudf::column>(
     cudf::data_type{data_type}, num_rows, std::move(buffer), rmm::device_buffer{}, 0);

cpp/tests/utilities/identify_stream_usage.cpp

@@ -218,6 +218,10 @@ void sanitizer_subscriber::callback(Sanitizer_CallbackDomain domain,
           CHECK_STREAM_ARG(cudaMemcpy3DPeerAsync_ptsz, 7000, stream);
           CHECK_STREAM_ARG(cudaMemcpyAsync, 3020, stream);
           CHECK_STREAM_ARG(cudaMemcpyAsync_ptsz, 7000, stream);
+#if CUDART_VERSION >= 13000
+          CHECK_STREAM_ARG(cudaMemcpyBatchAsync, 13000, stream);
+          CHECK_STREAM_ARG(cudaMemcpyBatchAsync_ptsz, 13000, stream);
+#endif
           CHECK_STREAM_ARG(cudaMemcpyFromSymbolAsync, 3020, stream);
           CHECK_STREAM_ARG(cudaMemcpyFromSymbolAsync_ptsz, 7000, stream);
           CHECK_STREAM_ARG(cudaMemcpyToSymbolAsync, 3020, stream);

python/pylibcudf/pylibcudf/column.pyx

@@ -188,10 +188,12 @@ cdef gpumemoryview _copy_array_to_device(object buf, object stream: CudaStreamLi
     cdef size_t nbytes = len(mv) * mv.itemsize
     cdef Stream _stream = _get_stream(stream)
 
-    return gpumemoryview(DeviceBuffer.to_device(
+    cdef DeviceBuffer dbuf = DeviceBuffer.to_device(
         <const unsigned char[:nbytes:1]><const unsigned char*>ptr,
         _stream
-    ))
+    )
+    _stream.synchronize()
+    return gpumemoryview(dbuf)
 
 
 def _infer_list_depth_and_dtype(obj: list) -> tuple[int, type]:
@@ -1034,6 +1036,7 @@ cdef class Column:
             ptr = <const unsigned char*><uintptr_t>data_ptr
             view = (<const unsigned char[:nbytes]> ptr)[:nbytes]
             dbuf = DeviceBuffer.to_device(view, _stream)
+            _stream.synchronize()
         else:
             dbuf = DeviceBuffer(size=0, stream=_stream)
 

python/pylibcudf/tests/io/test_experimental_hybrid_scan.py

@@ -348,7 +348,7 @@ def test_hybrid_scan_materialize_columns(
     filter_data = [
         plc.gpumemoryview(
             rmm.DeviceBuffer.to_device(
-                simple_parquet_bytes[r.offset : r.offset + r.size],
+                memoryview(simple_parquet_bytes)[r.offset : r.offset + r.size],
                 plc.utils._get_stream(stream),
             )
         )
@@ -383,7 +383,7 @@ def test_hybrid_scan_materialize_columns(
     payload_data = [
         plc.gpumemoryview(
             rmm.DeviceBuffer.to_device(
-                simple_parquet_bytes[r.offset : r.offset + r.size],
+                memoryview(simple_parquet_bytes)[r.offset : r.offset + r.size],
                 plc.utils._get_stream(stream),
             )
         )
@@ -474,7 +474,7 @@ def test_hybrid_scan_single_step_materialize(
     all_columns_data = [
         plc.gpumemoryview(
             rmm.DeviceBuffer.to_device(
-                simple_parquet_bytes[r.offset : r.offset + r.size],
+                memoryview(simple_parquet_bytes)[r.offset : r.offset + r.size],
                 plc.utils._get_stream(stream),
             )
         )
@@ -556,7 +556,7 @@ def test_hybrid_scan_has_next_table_chunk(
     filter_data = [
         plc.gpumemoryview(
             rmm.DeviceBuffer.to_device(
-                simple_parquet_bytes[r.offset : r.offset + r.size],
+                memoryview(simple_parquet_bytes)[r.offset : r.offset + r.size],
                 plc.utils._get_stream(),
             )
         )
@@ -626,7 +626,7 @@ def test_hybrid_scan_chunked_reading(
     filter_data = [
         plc.gpumemoryview(
             rmm.DeviceBuffer.to_device(
-                simple_parquet_bytes[r.offset : r.offset + r.size],
+                memoryview(simple_parquet_bytes)[r.offset : r.offset + r.size],
                 plc.utils._get_stream(stream),
             )
         )