Replace default stream for scalars and column factories usages (because of defaulted arguments)

cpp/src/column/column_factories.cpp

@@ -178,7 +178,7 @@ std::unique_ptr<column> make_dictionary_from_scalar(scalar const& s,
   CUDF_EXPECTS(s.is_valid(stream), "cannot create a dictionary with a null key");
   return make_dictionary_column(
     make_column_from_scalar(s, 1, stream, mr),
-    make_column_from_scalar(numeric_scalar<uint32_t>(0), size, stream, mr),
+    make_column_from_scalar(numeric_scalar<uint32_t>(0, true, stream), size, stream, mr),
     rmm::device_buffer{0, stream, mr},
     0);
 }

cpp/src/filling/fill.cu

@@ -79,7 +79,7 @@ struct in_place_fill_range_dispatch {
   {
     auto unscaled = static_cast<cudf::fixed_point_scalar<T> const&>(value).value(stream);
     using RepType = typename T::rep;
-    auto s        = cudf::numeric_scalar<RepType>(unscaled, value.is_valid(stream));
+    auto s        = cudf::numeric_scalar<RepType>(unscaled, value.is_valid(stream), stream);
     in_place_fill<RepType>(destination, begin, end, s, stream);
   }
 

cpp/src/groupby/sort/aggregate.cpp

@@ -569,11 +569,14 @@ void aggregate_result_functor::operator()<aggregation::MERGE_HISTOGRAM>(aggregat
  *
  * @param column_0 The first column
  * @param column_1 The second column
+ * @param stream CUDA stream used for device memory operations and kernel launches
  * @return tuple with new null mask (if null masks of input differ) and new column views
  */
-auto column_view_with_common_nulls(column_view const& column_0, column_view const& column_1)
+auto column_view_with_common_nulls(column_view const& column_0,
+                                   column_view const& column_1,
+                                   rmm::cuda_stream_view stream)
 {
-  auto [new_nullmask, null_count] = cudf::bitmask_and(table_view{{column_0, column_1}});
+  auto [new_nullmask, null_count] = cudf::bitmask_and(table_view{{column_0, column_1}}, stream);
   if (null_count == 0) { return std::make_tuple(std::move(new_nullmask), column_0, column_1); }
   auto column_view_with_new_nullmask = [](auto const& col, void* nullmask, auto null_count) {
     return column_view(col.type(),
@@ -610,7 +613,7 @@ void aggregate_result_functor::operator()<aggregation::COVARIANCE>(aggregation c
   // Covariance only for valid values in both columns.
   // in non-identical null mask cases, this prevents caching of the results - STD, MEAN, COUNT.
   auto [_, values_child0, values_child1] =
-    column_view_with_common_nulls(values.child(0), values.child(1));
+    column_view_with_common_nulls(values.child(0), values.child(1), stream);
 
   auto mean_agg = make_mean_aggregation();
   aggregate_result_functor(values_child0, helper, cache, stream, mr).operator()<aggregation::MEAN>(*mean_agg);
@@ -659,7 +662,7 @@ void aggregate_result_functor::operator()<aggregation::CORRELATION>(aggregation
   // Correlation only for valid values in both columns.
   // in non-identical null mask cases, this prevents caching of the results - STD, MEAN, COUNT
   auto [_, values_child0, values_child1] =
-    column_view_with_common_nulls(values.child(0), values.child(1));
+    column_view_with_common_nulls(values.child(0), values.child(1), stream);
 
   auto std_agg = make_std_aggregation();
   aggregate_result_functor(values_child0, helper, cache, stream, mr).operator()<aggregation::STD>(*std_agg);

cpp/src/groupby/sort/sort_helper.cu

@@ -96,8 +96,8 @@ column_view sort_groupby_helper::key_sort_order(rmm::cuda_stream_view stream)
 
   if (_keys_pre_sorted == sorted::YES) {
     _key_sorted_order = cudf::detail::sequence(_keys.num_rows(),
-                                               numeric_scalar<size_type>(0),
-                                               numeric_scalar<size_type>(1),
+                                               numeric_scalar<size_type>(0, true, stream),
+                                               numeric_scalar<size_type>(1, true, stream),
                                                stream,
                                                rmm::mr::get_current_device_resource());
     return sliced_key_sorted_order();
@@ -236,7 +236,7 @@ column_view sort_groupby_helper::keys_bitmask_column(rmm::cuda_stream_view strea
   auto [row_bitmask, null_count] =
     cudf::detail::bitmask_and(_keys, stream, rmm::mr::get_current_device_resource());
 
-  auto const zero = numeric_scalar<int8_t>(0);
+  auto const zero = numeric_scalar<int8_t>(0, true, stream);
   // Create a temporary variable and only set _keys_bitmask_column right before the return.
   // This way, a 2nd (parallel) call to this will not be given a partially created object.
   auto keys_bitmask_column = cudf::detail::sequence(

cpp/src/hash/md5_hash.cu

@@ -311,7 +311,7 @@ std::unique_ptr<column> md5(table_view const& input,
 {
   if (input.num_columns() == 0 || input.num_rows() == 0) {
     // Return the MD5 hash of a zero-length input.
-    string_scalar const string_128bit("d41d8cd98f00b204e9orig98ecf8427e");
+    string_scalar const string_128bit("d41d8cd98f00b204e9orig98ecf8427e", true, stream);
     return make_column_from_scalar(string_128bit, input.num_rows(), stream, mr);
   }
 

cpp/src/io/csv/reader_impl.cu

@@ -868,8 +868,8 @@ table_with_metadata read_csv(cudf::io::datasource* source,
         // quotechars in quoted fields results in reduction to a single quotechar
         // TODO: Would be much more efficient to perform this operation in-place
         // during the conversion stage
-        std::string const quotechar(1, parse_opts.quotechar);
-        std::string const dblquotechar(2, parse_opts.quotechar);
+        cudf::string_scalar const quotechar(std::string(1, parse_opts.quotechar), true, stream);
+        cudf::string_scalar const dblquotechar(std::string(2, parse_opts.quotechar), true, stream);
         std::unique_ptr<column> col = cudf::make_strings_column(*out_buffers[i]._strings, stream);
         out_columns.emplace_back(
           cudf::strings::detail::replace(col->view(), dblquotechar, quotechar, -1, stream, mr));

cpp/src/io/csv/writer_impl.cu

@@ -148,11 +148,6 @@ struct column_to_strings_fn {
 
   // Note: `null` replacement with `na_rep` deferred to `concatenate()`
   // instead of column-wise; might be faster
-  //
-  // Note: Cannot pass `stream` to detail::<fname> version of <fname> calls below, because they are
-  // not exposed in header (see, for example, detail::concatenate(tbl_view, separator, na_rep,
-  // stream, mr) is declared and defined in combine.cu); Possible solution: declare `extern`, or
-  // just declare a prototype inside `namespace cudf::strings::detail`;
 
   // bools:
   //
@@ -161,7 +156,11 @@ struct column_to_strings_fn {
     column_view const& column) const
   {
     return cudf::strings::detail::from_booleans(
-      column, options_.get_true_value(), options_.get_false_value(), stream_, mr_);
+      column,
+      cudf::string_scalar(options_.get_true_value(), true, stream_),
+      cudf::string_scalar(options_.get_false_value(), true, stream_),
+      stream_,
+      mr_);
   }
 
   // strings:
@@ -367,10 +366,10 @@ void write_chunked(data_sink* out_sink,
 
   CUDF_EXPECTS(str_column_view.size() > 0, "Unexpected empty strings column.");
 
-  cudf::string_scalar newline{options.get_line_terminator()};
+  cudf::string_scalar newline{options.get_line_terminator(), true, stream};
   auto p_str_col_w_nl = cudf::strings::detail::join_strings(str_column_view,
                                                             newline,
-                                                            string_scalar("", false),
+                                                            string_scalar("", false, stream),
                                                             stream,
                                                             rmm::mr::get_current_device_resource());
   strings_column_view strings_column{p_str_col_w_nl->view()};
@@ -472,14 +471,15 @@ void write_csv(data_sink* out_sink,
       //
       std::string delimiter_str{options.get_inter_column_delimiter()};
       auto str_concat_col = [&] {
+        cudf::string_scalar narep{options.get_na_rep(), true, stream};
         if (str_table_view.num_columns() > 1)
-          return cudf::strings::detail::concatenate(str_table_view,
-                                                    delimiter_str,
-                                                    options.get_na_rep(),
-                                                    strings::separator_on_nulls::YES,
-                                                    stream,
-                                                    rmm::mr::get_current_device_resource());
-        cudf::string_scalar narep{options.get_na_rep()};
+          return cudf::strings::detail::concatenate(
+            str_table_view,
+            cudf::string_scalar(delimiter_str, true, stream),
+            narep,
+            strings::separator_on_nulls::YES,
+            stream,
+            rmm::mr::get_current_device_resource());
         return cudf::strings::detail::replace_nulls(
           str_table_view.column(0), narep, stream, rmm::mr::get_current_device_resource());
       }();

cpp/src/io/json/write_json.cu

@@ -527,7 +527,11 @@ struct column_to_strings_fn {
     column_view const& column) const
   {
     return cudf::strings::detail::from_booleans(
-      column, options_.get_true_value(), options_.get_false_value(), stream_, mr_);
+      column,
+      cudf::string_scalar(options_.get_true_value(), true, stream_),
+      cudf::string_scalar(options_.get_false_value(), true, stream_),
+      stream_,
+      mr_);
   }
 
   // strings:

cpp/src/io/parquet/writer_impl.cu

@@ -228,7 +228,7 @@ size_t column_size(column_view const& column, rmm::cuda_stream_view stream)
     auto const scol = structs_column_view(column);
     size_t ret      = 0;
     for (int i = 0; i < scol.num_children(); i++) {
-      ret += column_size(scol.get_sliced_child(i), stream);
+      ret += column_size(scol.get_sliced_child(i, stream), stream);
     }
     return ret;
   } else if (column.type().id() == type_id::LIST) {

cpp/src/io/utilities/column_buffer.cpp

@@ -194,7 +194,8 @@ std::unique_ptr<column> make_column(column_buffer_base<string_policy>& buffer,
           std::move(col_content.children[strings_column_view::offsets_column_index]),
           std::move(uint8_col),
           null_count,
-          std::move(*col_content.null_mask));
+          std::move(*col_content.null_mask),
+          stream);
       }
 
     case type_id::LIST: {

cpp/src/lists/dremel.cu

@@ -127,7 +127,8 @@ dremel_data get_encoding(column_view h_col,
   }
   std::unique_ptr<column> empty_list_offset_col;
   if (has_empty_list_offsets) {
-    empty_list_offset_col = make_fixed_width_column(data_type(type_id::INT32), 1);
+    empty_list_offset_col = make_fixed_width_column(
+      data_type(type_to_id<size_type>()), 1, mask_state::UNALLOCATED, stream);
     CUDF_CUDA_TRY(cudaMemsetAsync(
       empty_list_offset_col->mutable_view().head(), 0, sizeof(size_type), stream.value()));
     std::function<column_view(column_view const&)> normalize_col = [&](column_view const& col) {

cpp/src/lists/interleave_columns.cu

@@ -310,7 +310,7 @@ struct interleave_list_entries_impl<T, std::enable_if_t<cudf::is_fixed_width<T>(
     if (data_has_null_mask) {
       auto [null_mask, null_count] = cudf::detail::valid_if(
         validities.begin(), validities.end(), thrust::identity{}, stream, mr);
-      if (null_count > 0) { output->set_null_mask(null_mask, null_count); }
+      if (null_count > 0) { output->set_null_mask(std::move(null_mask), null_count); }
     }
 
     return output;

cpp/src/reductions/scan/scan_exclusive.cu

@@ -97,7 +97,7 @@ std::unique_ptr<column> scan_exclusive(column_view const& input,
 
   auto output = scan_agg_dispatch<scan_dispatcher>(
     input, agg, static_cast<bitmask_type*>(mask.data()), stream, mr);
-  output->set_null_mask(mask, null_count);
+  output->set_null_mask(std::move(mask), null_count);
 
   return output;
 }

cpp/src/reductions/scan/scan_inclusive.cu

@@ -177,7 +177,7 @@ std::unique_ptr<column> scan_inclusive(column_view const& input,
 
   auto output = scan_agg_dispatch<scan_dispatcher>(
     input, agg, static_cast<bitmask_type*>(mask.data()), stream, mr);
-  output->set_null_mask(mask, null_count);
+  output->set_null_mask(std::move(mask), null_count);
 
   // If the input is a structs column, we also need to push down nulls from the parent output column
   // into the children columns.

cpp/src/reductions/segmented/simple.cuh

@@ -80,7 +80,7 @@ std::unique_ptr<column> simple_segmented_reduction(
 
   // Cast initial value
   ResultType initial_value = [&] {
-    if (init.has_value() && init.value().get().is_valid()) {
+    if (init.has_value() && init.value().get().is_valid(stream)) {
       using ScalarType = cudf::scalar_type_t<InputType>;
       auto input_value = static_cast<ScalarType const*>(&init.value().get())->value(stream);
       return static_cast<ResultType>(input_value);

cpp/src/reductions/segmented/update_validity.cu

@@ -38,7 +38,7 @@ void segmented_update_validity(column& result,
     offsets.end() - 1,
     offsets.begin() + 1,
     null_handling,
-    init.has_value() ? std::optional(init.value().get().is_valid()) : std::nullopt,
+    init.has_value() ? std::optional(init.value().get().is_valid(stream)) : std::nullopt,
     stream,
     mr);
   result.set_null_mask(std::move(output_null_mask), output_null_count);

cpp/src/reductions/simple.cuh

@@ -66,7 +66,7 @@ std::unique_ptr<scalar> simple_reduction(column_view const& col,
 
   // Cast initial value
   std::optional<ResultType> const initial_value = [&] {
-    if (init.has_value() && init.value().get().is_valid()) {
+    if (init.has_value() && init.value().get().is_valid(stream)) {
       using ScalarType = cudf::scalar_type_t<ElementType>;
       auto input_value = static_cast<ScalarType const*>(&init.value().get())->value(stream);
       return std::optional<ResultType>(static_cast<ResultType>(input_value));
@@ -89,7 +89,8 @@ std::unique_ptr<scalar> simple_reduction(column_view const& col,
 
   // set scalar is valid
   result->set_valid_async(
-    col.null_count() < col.size() && (!init.has_value() || init.value().get().is_valid()), stream);
+    col.null_count() < col.size() && (!init.has_value() || init.value().get().is_valid(stream)),
+    stream);
   return result;
 }
 
@@ -130,7 +131,8 @@ std::unique_ptr<scalar> fixed_point_reduction(
   auto result_scalar =
     cudf::make_fixed_point_scalar<DecimalXX>(val->value(stream), scale, stream, mr);
   result_scalar->set_valid_async(
-    col.null_count() < col.size() && (!init.has_value() || init.value().get().is_valid()), stream);
+    col.null_count() < col.size() && (!init.has_value() || init.value().get().is_valid(stream)),
+    stream);
   return result_scalar;
 }
 
@@ -169,7 +171,8 @@ std::unique_ptr<scalar> dictionary_reduction(
 
   // set scalar is valid
   result->set_valid_async(
-    col.null_count() < col.size() && (!init.has_value() || init.value().get().is_valid()), stream);
+    col.null_count() < col.size() && (!init.has_value() || init.value().get().is_valid(stream)),
+    stream);
   return result;
 }
 

cpp/src/scalar/scalar.cpp

@@ -421,7 +421,7 @@ duration_scalar<T>::duration_scalar(duration_scalar<T> const& other,
 template <typename T>
 typename duration_scalar<T>::rep_type duration_scalar<T>::count()
 {
-  return this->value().count();
+  return this->value(cudf::get_default_stream()).count();
 }
 
 /**
@@ -441,7 +441,7 @@ template class duration_scalar<duration_ns>;
 template <typename T>
 typename timestamp_scalar<T>::rep_type timestamp_scalar<T>::ticks_since_epoch()
 {
-  return this->value().time_since_epoch().count();
+  return this->value(cudf::get_default_stream()).time_since_epoch().count();
 }
 
 /**
@@ -541,7 +541,7 @@ struct_scalar::struct_scalar(table_view const& data,
                              rmm::cuda_stream_view stream,
                              rmm::mr::device_memory_resource* mr)
   : scalar(data_type(type_id::STRUCT), is_valid, stream, mr),
-    _data{init_data(table{data}, is_valid, stream, mr)}
+    _data{init_data(table{data, stream, mr}, is_valid, stream, mr)}
 {
   assert_valid_size();
 }
@@ -551,8 +551,11 @@ struct_scalar::struct_scalar(host_span<column_view const> data,
                              rmm::cuda_stream_view stream,
                              rmm::mr::device_memory_resource* mr)
   : scalar(data_type(type_id::STRUCT), is_valid, stream, mr),
-    _data{init_data(
-      table{table_view{std::vector<column_view>{data.begin(), data.end()}}}, is_valid, stream, mr)}
+    _data{
+      init_data(table{table_view{std::vector<column_view>{data.begin(), data.end()}}, stream, mr},
+                is_valid,
+                stream,
+                mr)}
 {
   assert_valid_size();
 }

cpp/src/sort/sort_impl.cuh

@@ -38,7 +38,8 @@ std::unique_ptr<column> sorted_order(table_view input,
                                      rmm::mr::device_memory_resource* mr)
 {
   if (input.num_rows() == 0 or input.num_columns() == 0) {
-    return cudf::make_numeric_column(data_type(type_to_id<size_type>()), 0);
+    return cudf::make_numeric_column(
+      data_type(type_to_id<size_type>()), 0, mask_state::UNALLOCATED, stream);
   }
 
   if (not column_order.empty()) {

cpp/src/strings/convert/convert_booleans.cu

@@ -45,7 +45,8 @@ std::unique_ptr<column> to_booleans(strings_column_view const& input,
                                     rmm::mr::device_memory_resource* mr)
 {
   size_type strings_count = input.size();
-  if (strings_count == 0) return make_numeric_column(data_type{type_id::BOOL8}, 0);
+  if (strings_count == 0)
+    return make_numeric_column(data_type{type_id::BOOL8}, 0, mask_state::UNALLOCATED, stream);
 
   CUDF_EXPECTS(true_string.is_valid(stream) && true_string.size() > 0,
                "Parameter true_string must not be empty.");

cpp/src/strings/convert/convert_durations.cu

@@ -697,7 +697,8 @@ std::unique_ptr<column> to_durations(strings_column_view const& input,
                                      rmm::mr::device_memory_resource* mr)
 {
   size_type strings_count = input.size();
-  if (strings_count == 0) return make_duration_column(duration_type, 0);
+  if (strings_count == 0)
+    return make_duration_column(duration_type, 0, mask_state::UNALLOCATED, stream);
 
   CUDF_EXPECTS(!format.empty(), "Format parameter must not be empty.");
 

cpp/src/strings/convert/convert_floats.cu

@@ -97,7 +97,8 @@ std::unique_ptr<column> to_floats(strings_column_view const& input,
                                   rmm::mr::device_memory_resource* mr)
 {
   size_type strings_count = input.size();
-  if (strings_count == 0) return make_numeric_column(output_type, 0);
+  if (strings_count == 0)
+    return make_numeric_column(output_type, 0, mask_state::UNALLOCATED, stream);
   auto strings_column = column_device_view::create(input.parent(), stream);
   auto d_strings      = *strings_column;
   // create float output column copying the strings null-mask

cpp/src/strings/convert/convert_integers.cu

@@ -274,7 +274,8 @@ std::unique_ptr<column> to_integers(strings_column_view const& input,
                                     rmm::mr::device_memory_resource* mr)
 {
   size_type strings_count = input.size();
-  if (strings_count == 0) return make_numeric_column(output_type, 0);
+  if (strings_count == 0)
+    return make_numeric_column(output_type, 0, mask_state::UNALLOCATED, stream);
 
   // Create integer output column copying the strings null-mask
   auto results = make_numeric_column(output_type,

cpp/src/strings/convert/convert_ipv4.cu

@@ -77,7 +77,8 @@ std::unique_ptr<column> ipv4_to_integers(strings_column_view const& input,
                                          rmm::mr::device_memory_resource* mr)
 {
   size_type strings_count = input.size();
-  if (strings_count == 0) return make_numeric_column(data_type{type_id::INT64}, 0);
+  if (strings_count == 0)
+    return make_numeric_column(data_type{type_id::INT64}, 0, mask_state::UNALLOCATED, stream);
 
   auto strings_column = column_device_view::create(input.parent(), stream);
   // create output column copying the strings' null-mask

cpp/src/strings/filling/fill.cu

@@ -56,7 +56,7 @@ std::unique_ptr<column> fill(strings_column_view const& strings,
   auto d_strings      = *strings_column;
 
   // create resulting null mask
-  auto valid_mask = [begin, end, d_value, value, d_strings, stream, mr] {
+  auto valid_mask = [begin, end, d_value, &value, d_strings, stream, mr] {
     if (begin == 0 and end == d_strings.size() and value.is_valid(stream))
       return std::pair(rmm::device_buffer{}, 0);
     return cudf::detail::valid_if(

cpp/src/strings/replace/backref_re.cu

@@ -122,7 +122,7 @@ std::unique_ptr<column> replace_with_backrefs(strings_column_view const& input,
   rmm::device_uvector<backref_type> backrefs = cudf::detail::make_device_uvector_async(
     parse_result.second, stream, rmm::mr::get_current_device_resource());
   string_scalar repl_scalar(parse_result.first, true, stream);
-  string_view const d_repl_template = repl_scalar.value();
+  string_view const d_repl_template = repl_scalar.value(stream);
 
   auto const d_strings = column_device_view::create(input.parent(), stream);