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/include/cudf/scalar/scalar.hpp

@@ -190,7 +190,7 @@ class fixed_width_scalar : public scalar {
   [[nodiscard]] T const* data() const;
 
  protected:
-  rmm::device_scalar<T> _data;  ///< device memory containing the value
+  cudf::detail::device_scalar<T> _data;  ///< device memory containing the value
 
   /**
    * @brief Construct a new fixed width scalar object.
@@ -402,7 +402,7 @@ class fixed_point_scalar : public scalar {
   [[nodiscard]] rep_type const* data() const;
 
  protected:
-  rmm::device_scalar<rep_type> _data;  ///< device memory containing the value
+  cudf::detail::device_scalar<rep_type> _data;  ///< device memory containing the value
 };
 
 /**

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,7 @@ auto build_cross_comparators(
     h_eqs.push_back(adapter.comparator);
   }
 
-  return cudf::detail::make_device_uvector_async(h_eqs, stream, temp_mr);
+  return cudf::detail::make_device_uvector(h_eqs, stream, temp_mr);
 }
 
 /// 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

@@ -1425,7 +1425,7 @@ void decimal_sizes_to_offsets(device_2dspan<rowgroup_rows const> rg_bounds,
   auto const num_blocks = elem_sizes.size() * rg_bounds.size().first;
   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

@@ -275,8 +275,7 @@ index_vector generate_merged_indices(table_view const& left_table,
                   ineq_op);
   }
 
-  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/scalar/scalar.cpp

@@ -6,6 +6,7 @@
 #include <cudf/column/column.hpp>
 #include <cudf/detail/null_mask.hpp>
 #include <cudf/detail/structs/utilities.hpp>
+#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/fixed_point/fixed_point.hpp>
 #include <cudf/scalar/scalar.hpp>
 #include <cudf/strings/string_view.hpp>
@@ -16,10 +17,20 @@
 
 #include <cuda/stream>
 
+#include <algorithm>
 #include <string>
 
 namespace cudf {
 
+static rmm::device_buffer make_string_device_buffer(std::string_view string,
+                                                    cuda::stream_ref stream,
+                                                    rmm::device_async_resource_ref mr)
+{
+  auto host_data = cudf::detail::make_pinned_vector<char>(string.size(), stream);
+  std::copy(string.begin(), string.end(), host_data.begin());
+  return rmm::device_buffer(host_data.data(), host_data.size(), stream, mr);
+}
+
 scalar::scalar(data_type type,
                bool is_valid,
                cuda::stream_ref stream,
@@ -51,7 +62,7 @@ string_scalar::string_scalar(std::string_view string,
                              cuda::stream_ref stream,
                              rmm::device_async_resource_ref mr)
   : scalar(data_type(type_id::STRING), is_valid, stream, mr),
-    _data(string.data(), string.size(), stream, mr)
+    _data(make_string_device_buffer(string, stream, mr))
 {
   CUDF_EXPECTS(
     string.size() <= static_cast<std::size_t>(std::numeric_limits<cudf::size_type>::max()),
@@ -145,7 +156,8 @@ fixed_point_scalar<T>::fixed_point_scalar(rmm::device_scalar<rep_type>&& data,
                                           bool is_valid,
                                           cuda::stream_ref stream,
                                           rmm::device_async_resource_ref mr)
-  : scalar{data_type{type_to_id<T>(), scale}, is_valid, stream, mr}, _data{std::move(data)}
+  : scalar{data_type{type_to_id<T>(), scale}, is_valid, stream, mr},
+    _data{data.value(stream), stream, mr}
 {
 }
 
@@ -210,7 +222,7 @@ fixed_width_scalar<T>::fixed_width_scalar(rmm::device_scalar<T>&& data,
                                           bool is_valid,
                                           cuda::stream_ref stream,
                                           rmm::device_async_resource_ref mr)
-  : scalar(data_type(type_to_id<T>()), is_valid, stream, mr), _data{std::move(data)}
+  : scalar(data_type(type_to_id<T>()), is_valid, stream, mr), _data{data.value(stream), stream, mr}
 {
 }
 

cpp/src/transform/row_bit_count.cu

@@ -551,8 +551,8 @@ std::unique_ptr<column> segmented_row_bit_count(table_view const& t,
     {mcv.data<size_type>(), static_cast<std::size_t>(mcv.size())},
     segment_length,
     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);