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

@@ -145,7 +145,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 +211,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);

cpp/tests/scalar/scalar_test.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2024, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -10,13 +10,69 @@
 #include <cudf_test/type_lists.hpp>
 
 #include <cudf/scalar/scalar.hpp>
+#include <cudf/utilities/error.hpp>
+
+#include <rmm/cuda_stream.hpp>
+
+#include <cuda_runtime_api.h>
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <mutex>
+
+namespace {
+
+class host_func_gate {
+ public:
+  ~host_func_gate() { release(); }
+
+  void wait()
+  {
+    std::unique_lock<std::mutex> lock{mutex_};
+    EXPECT_TRUE(condition_.wait_for(lock, std::chrono::seconds{10}, [this] { return released_; }));
+    complete_.store(true);
+  }
+
+  void release()
+  {
+    {
+      std::lock_guard<std::mutex> lock{mutex_};
+      released_ = true;
+    }
+    condition_.notify_one();
+  }
+
+  bool complete() const { return complete_.load(); }
+
+ private:
+  std::mutex mutex_;
+  std::condition_variable condition_;
+  bool released_{false};
+  std::atomic<bool> complete_{false};
+};
+
+class lifetime_test_scalar : public cudf::numeric_scalar<int32_t> {
+ public:
+  using numeric_scalar::numeric_scalar;
+
+  void set_data_async(int32_t const& value, cuda::stream_ref stream)
+  {
+    this->_data.set_value_async(value, stream);
+  }
+};
+
+}  // namespace
 
 template <typename T>
 struct TypedScalarTest : public cudf::test::BaseFixture {};
 
 template <typename T>
 struct TypedScalarTestWithoutFixedPoint : public cudf::test::BaseFixture {};
 
+struct ScalarTest : public cudf::test::BaseFixture {};
+
 TYPED_TEST_SUITE(TypedScalarTest, cudf::test::FixedWidthTypes);
 TYPED_TEST_SUITE(TypedScalarTestWithoutFixedPoint, cudf::test::FixedWidthTypesWithoutFixedPoint);
 
@@ -49,6 +105,24 @@ TYPED_TEST(TypedScalarTestWithoutFixedPoint, SetValue)
   EXPECT_EQ(value, s.value());
 }
 
+TEST_F(ScalarTest, AsyncSetValueOwnsHostSource)
+{
+  rmm::cuda_stream stream;
+  auto const stream_ref = cuda::stream_ref{stream.value()};
+  int32_t source        = 42;
+  lifetime_test_scalar scalar{0, true, stream_ref};
+  host_func_gate gate;
+  CUDF_CUDA_TRY(cudaLaunchHostFunc(
+    stream.value(), [](void* data) { static_cast<host_func_gate*>(data)->wait(); }, &gate));
+
+  scalar.set_data_async(source, stream_ref);
+  source = -1;
+  EXPECT_FALSE(gate.complete());
+
+  gate.release();
+  EXPECT_EQ(42, scalar.value(stream_ref));
+}
+
 TYPED_TEST(TypedScalarTestWithoutFixedPoint, SetNull)
 {
   TypeParam value = cudf::test::make_type_param_scalar<TypeParam>(6);