Merge remote-tracking branch 'upstream/branch-21.12' into async-alloc…

…ator-size

cpp/CMakeLists.txt

@@ -576,7 +576,7 @@ add_library(cudftestutil STATIC
             tests/utilities/column_utilities.cu
             tests/utilities/table_utilities.cu
             tests/io/metadata_utilities.cpp
-            tests/strings/utilities.cu)
+            tests/strings/utilities.cpp)
 
 set_target_properties(cudftestutil
     PROPERTIES BUILD_RPATH                         "\$ORIGIN"

cpp/benchmarks/common/generate_benchmark_input.cpp

@@ -297,12 +297,21 @@ std::unique_ptr<cudf::column> create_random_column(data_profile const& profile,
     }
   }
 
+  // cudf expects the null mask buffer to be padded up to 64 bytes. so allocate the proper size and
+  // copy what we have.
+  rmm::device_buffer result_bitmask{cudf::bitmask_allocation_size_bytes(num_rows),
+                                    rmm::cuda_stream_default};
+  cudaMemcpyAsync(result_bitmask.data(),
+                  null_mask.data(),
+                  null_mask.size() * sizeof(cudf::bitmask_type),
+                  cudaMemcpyHostToDevice,
+                  rmm::cuda_stream_default);
+
   return std::make_unique<cudf::column>(
     cudf::data_type{cudf::type_to_id<T>()},
     num_rows,
     rmm::device_buffer(data.data(), num_rows * sizeof(stored_Type), rmm::cuda_stream_default),
-    rmm::device_buffer(
-      null_mask.data(), null_mask.size() * sizeof(cudf::bitmask_type), rmm::cuda_stream_default));
+    std::move(result_bitmask));
 }
 
 /**

cpp/include/cudf/aggregation.hpp

@@ -503,20 +503,25 @@ std::unique_ptr<Base> make_merge_m2_aggregation();
  *
  * Compute covariance between two columns.
  * The input columns are child columns of a non-nullable struct columns.
+ * @param min_periods Minimum number of non-null observations required to produce a result.
+ * @param ddof Delta Degrees of Freedom. The divisor used in calculations is N - ddof, where N is
+ *        the number of non-null observations.
  */
 template <typename Base = aggregation>
-std::unique_ptr<Base> make_covariance_aggregation();
+std::unique_ptr<Base> make_covariance_aggregation(size_type min_periods = 1, size_type ddof = 1);
 
 /**
  * @brief Factory to create a CORRELATION aggregation
  *
  * Compute correlation coefficient between two columns.
  * The input columns are child columns of a non-nullable struct columns.
  *
- * @param[in] type: correlation_type
+ * @param type correlation_type
+ * @param min_periods Minimum number of non-null observations required to produce a result.
  */
 template <typename Base = aggregation>
-std::unique_ptr<Base> make_correlation_aggregation(correlation_type type);
+std::unique_ptr<Base> make_correlation_aggregation(correlation_type type,
+                                                   size_type min_periods = 1);
 
 /**
  * @brief Factory to create a TDIGEST aggregation

cpp/include/cudf/ast/detail/operators.hpp

@@ -192,6 +192,15 @@ CUDA_HOST_DEVICE_CALLABLE constexpr void ast_operator_dispatcher(ast_operator op
     case ast_operator::NOT:
       f.template operator()<ast_operator::NOT>(std::forward<Ts>(args)...);
       break;
+    case ast_operator::CAST_TO_INT64:
+      f.template operator()<ast_operator::CAST_TO_INT64>(std::forward<Ts>(args)...);
+      break;
+    case ast_operator::CAST_TO_UINT64:
+      f.template operator()<ast_operator::CAST_TO_UINT64>(std::forward<Ts>(args)...);
+      break;
+    case ast_operator::CAST_TO_FLOAT64:
+      f.template operator()<ast_operator::CAST_TO_FLOAT64>(std::forward<Ts>(args)...);
+      break;
     default:
 #ifndef __CUDA_ARCH__
       CUDF_FAIL("Invalid operator.");
@@ -780,6 +789,26 @@ struct operator_functor<ast_operator::NOT, false> {
   }
 };
 
+template <typename To>
+struct cast {
+  static constexpr auto arity{1};
+  template <typename From>
+  CUDA_DEVICE_CALLABLE auto operator()(From f) -> decltype(static_cast<To>(f))
+  {
+    return static_cast<To>(f);
+  }
+};
+
+template <>
+struct operator_functor<ast_operator::CAST_TO_INT64, false> : cast<int64_t> {
+};
+template <>
+struct operator_functor<ast_operator::CAST_TO_UINT64, false> : cast<uint64_t> {
+};
+template <>
+struct operator_functor<ast_operator::CAST_TO_FLOAT64, false> : cast<double> {
+};
+
 /*
  * The default specialization of nullable operators is to fall back to the non-nullable
  * implementation

cpp/include/cudf/ast/expressions.hpp

@@ -88,29 +88,32 @@ enum class ast_operator {
                      ///< NULL_LOGICAL_OR(null, false) is null, and NULL_LOGICAL_OR(valid, valid) ==
                      ///< LOGICAL_OR(valid, valid)
   // Unary operators
-  IDENTITY,    ///< Identity function
-  SIN,         ///< Trigonometric sine
-  COS,         ///< Trigonometric cosine
-  TAN,         ///< Trigonometric tangent
-  ARCSIN,      ///< Trigonometric sine inverse
-  ARCCOS,      ///< Trigonometric cosine inverse
-  ARCTAN,      ///< Trigonometric tangent inverse
-  SINH,        ///< Hyperbolic sine
-  COSH,        ///< Hyperbolic cosine
-  TANH,        ///< Hyperbolic tangent
-  ARCSINH,     ///< Hyperbolic sine inverse
-  ARCCOSH,     ///< Hyperbolic cosine inverse
-  ARCTANH,     ///< Hyperbolic tangent inverse
-  EXP,         ///< Exponential (base e, Euler number)
-  LOG,         ///< Natural Logarithm (base e)
-  SQRT,        ///< Square-root (x^0.5)
-  CBRT,        ///< Cube-root (x^(1.0/3))
-  CEIL,        ///< Smallest integer value not less than arg
-  FLOOR,       ///< largest integer value not greater than arg
-  ABS,         ///< Absolute value
-  RINT,        ///< Rounds the floating-point argument arg to an integer value
-  BIT_INVERT,  ///< Bitwise Not (~)
-  NOT          ///< Logical Not (!)
+  IDENTITY,        ///< Identity function
+  SIN,             ///< Trigonometric sine
+  COS,             ///< Trigonometric cosine
+  TAN,             ///< Trigonometric tangent
+  ARCSIN,          ///< Trigonometric sine inverse
+  ARCCOS,          ///< Trigonometric cosine inverse
+  ARCTAN,          ///< Trigonometric tangent inverse
+  SINH,            ///< Hyperbolic sine
+  COSH,            ///< Hyperbolic cosine
+  TANH,            ///< Hyperbolic tangent
+  ARCSINH,         ///< Hyperbolic sine inverse
+  ARCCOSH,         ///< Hyperbolic cosine inverse
+  ARCTANH,         ///< Hyperbolic tangent inverse
+  EXP,             ///< Exponential (base e, Euler number)
+  LOG,             ///< Natural Logarithm (base e)
+  SQRT,            ///< Square-root (x^0.5)
+  CBRT,            ///< Cube-root (x^(1.0/3))
+  CEIL,            ///< Smallest integer value not less than arg
+  FLOOR,           ///< largest integer value not greater than arg
+  ABS,             ///< Absolute value
+  RINT,            ///< Rounds the floating-point argument arg to an integer value
+  BIT_INVERT,      ///< Bitwise Not (~)
+  NOT,             ///< Logical Not (!)
+  CAST_TO_INT64,   ///< Cast value to int64_t
+  CAST_TO_UINT64,  ///< Cast value to uint64_t
+  CAST_TO_FLOAT64  ///< Cast value to double
 };
 
 /**

cpp/include/cudf/detail/aggregation/aggregation.hpp

@@ -901,7 +901,14 @@ class merge_m2_aggregation final : public groupby_aggregation {
  */
 class covariance_aggregation final : public groupby_aggregation {
  public:
-  explicit covariance_aggregation() : aggregation{COVARIANCE} {}
+  explicit covariance_aggregation(size_type min_periods, size_type ddof)
+    : aggregation{COVARIANCE}, _min_periods{min_periods}, _ddof(ddof)
+  {
+  }
+  size_type _min_periods;
+  size_type _ddof;
+
+  size_t do_hash() const override { return this->aggregation::do_hash() ^ hash_impl(); }
 
   std::unique_ptr<aggregation> clone() const override
   {
@@ -913,15 +920,25 @@ class covariance_aggregation final : public groupby_aggregation {
     return collector.visit(col_type, *this);
   }
   void finalize(aggregation_finalizer& finalizer) const override { finalizer.visit(*this); }
+
+ protected:
+  size_t hash_impl() const
+  {
+    return std::hash<size_type>{}(_min_periods) ^ std::hash<size_type>{}(_ddof);
+  }
 };
 
 /**
  * @brief Derived aggregation class for specifying CORRELATION aggregation
  */
 class correlation_aggregation final : public groupby_aggregation {
  public:
-  explicit correlation_aggregation(correlation_type type) : aggregation{CORRELATION}, _type{type} {}
+  explicit correlation_aggregation(correlation_type type, size_type min_periods)
+    : aggregation{CORRELATION}, _type{type}, _min_periods{min_periods}
+  {
+  }
   correlation_type _type;
+  size_type _min_periods;
 
   bool is_equal(aggregation const& _other) const override
   {
@@ -944,7 +961,10 @@ class correlation_aggregation final : public groupby_aggregation {
   void finalize(aggregation_finalizer& finalizer) const override { finalizer.visit(*this); }
 
  protected:
-  size_t hash_impl() const { return std::hash<int>{}(static_cast<int>(_type)); }
+  size_t hash_impl() const
+  {
+    return std::hash<int>{}(static_cast<int>(_type)) ^ std::hash<size_type>{}(_min_periods);
+  }
 };
 
 /**

cpp/include/cudf/io/detail/csv.hpp

@@ -24,55 +24,21 @@ namespace cudf {
 namespace io {
 namespace detail {
 namespace csv {
+
 /**
- * @brief Class to read CSV dataset data into columns.
+ * @brief Reads the entire dataset.
+ *
+ * @param sources Input `datasource` object to read the dataset from
+ * @param options Settings for controlling reading behavior
+ * @param stream CUDA stream used for device memory operations and kernel launches
+ * @param mr Device memory resource to use for device memory allocation
+ *
+ * @return The set of columns along with table metadata
  */
-class reader {
- private:
-  class impl;
-  std::unique_ptr<impl> _impl;
-
- public:
-  /**
-   * @brief Constructor from an array of file paths
-   *
-   * @param filepaths Paths to the files containing the input dataset
-   * @param options Settings for controlling reading behavior
-   * @param stream CUDA stream used for device memory operations and kernel launches
-   * @param mr Device memory resource to use for device memory allocation
-   */
-  explicit reader(std::vector<std::string> const& filepaths,
-                  csv_reader_options const& options,
-                  rmm::cuda_stream_view stream,
-                  rmm::mr::device_memory_resource* mr);
-
-  /**
-   * @brief Constructor from an array of datasources
-   *
-   * @param sources Input `datasource` objects to read the dataset from
-   * @param options Settings for controlling reading behavior
-   * @param stream CUDA stream used for device memory operations and kernel launches
-   * @param mr Device memory resource to use for device memory allocation
-   */
-  explicit reader(std::vector<std::unique_ptr<cudf::io::datasource>>&& sources,
-                  csv_reader_options const& options,
-                  rmm::cuda_stream_view stream,
-                  rmm::mr::device_memory_resource* mr);
-
-  /**
-   * @brief Destructor explicitly-declared to avoid inlined in header
-   */
-  ~reader();
-
-  /**
-   * @brief Reads the entire dataset.
-   *
-   * @param stream CUDA stream used for device memory operations and kernel launches.
-   *
-   * @return The set of columns along with table metadata
-   */
-  table_with_metadata read(rmm::cuda_stream_view stream = rmm::cuda_stream_default);
-};
+table_with_metadata read_csv(std::unique_ptr<cudf::io::datasource>&& source,
+                             csv_reader_options const& options,
+                             rmm::cuda_stream_view stream,
+                             rmm::mr::device_memory_resource* mr);
 
 class writer {
  public:

cpp/include/cudf_test/base_fixture.hpp

@@ -24,7 +24,9 @@
 #include <cudf_test/cxxopts.hpp>
 #include <cudf_test/file_utilities.hpp>
 
+#include <rmm/mr/device/arena_memory_resource.hpp>
 #include <rmm/mr/device/binning_memory_resource.hpp>
+#include <rmm/mr/device/cuda_async_memory_resource.hpp>
 #include <rmm/mr/device/cuda_memory_resource.hpp>
 #include <rmm/mr/device/managed_memory_resource.hpp>
 #include <rmm/mr/device/owning_wrapper.hpp>
@@ -217,13 +219,20 @@ class TempDirTestEnvironment : public ::testing::Environment {
 /// MR factory functions
 inline auto make_cuda() { return std::make_shared<rmm::mr::cuda_memory_resource>(); }
 
+inline auto make_async() { return std::make_shared<rmm::mr::cuda_async_memory_resource>(); }
+
 inline auto make_managed() { return std::make_shared<rmm::mr::managed_memory_resource>(); }
 
 inline auto make_pool()
 {
   return rmm::mr::make_owning_wrapper<rmm::mr::pool_memory_resource>(make_cuda());
 }
 
+inline auto make_arena()
+{
+  return rmm::mr::make_owning_wrapper<rmm::mr::arena_memory_resource>(make_cuda());
+}
+
 inline auto make_binning()
 {
   auto pool = make_pool();
@@ -253,7 +262,9 @@ inline std::shared_ptr<rmm::mr::device_memory_resource> create_memory_resource(
 {
   if (allocation_mode == "binning") return make_binning();
   if (allocation_mode == "cuda") return make_cuda();
+  if (allocation_mode == "async") return make_async();
   if (allocation_mode == "pool") return make_pool();
+  if (allocation_mode == "arena") return make_arena();
   if (allocation_mode == "managed") return make_managed();
   CUDF_FAIL("Invalid RMM allocation mode: " + allocation_mode);
 }

cpp/src/aggregation/aggregation.cpp

@@ -713,23 +713,25 @@ template std::unique_ptr<groupby_aggregation> make_merge_m2_aggregation<groupby_
 
 /// Factory to create a COVARIANCE aggregation
 template <typename Base>
-std::unique_ptr<Base> make_covariance_aggregation()
+std::unique_ptr<Base> make_covariance_aggregation(size_type min_periods, size_type ddof)
 {
-  return std::make_unique<detail::covariance_aggregation>();
+  return std::make_unique<detail::covariance_aggregation>(min_periods, ddof);
 }
-template std::unique_ptr<aggregation> make_covariance_aggregation<aggregation>();
-template std::unique_ptr<groupby_aggregation> make_covariance_aggregation<groupby_aggregation>();
+template std::unique_ptr<aggregation> make_covariance_aggregation<aggregation>(
+  size_type min_periods, size_type ddof);
+template std::unique_ptr<groupby_aggregation> make_covariance_aggregation<groupby_aggregation>(
+  size_type min_periods, size_type ddof);
 
 /// Factory to create a CORRELATION aggregation
 template <typename Base>
-std::unique_ptr<Base> make_correlation_aggregation(correlation_type type)
+std::unique_ptr<Base> make_correlation_aggregation(correlation_type type, size_type min_periods)
 {
-  return std::make_unique<detail::correlation_aggregation>(type);
+  return std::make_unique<detail::correlation_aggregation>(type, min_periods);
 }
 template std::unique_ptr<aggregation> make_correlation_aggregation<aggregation>(
-  correlation_type type);
+  correlation_type type, size_type min_periods);
 template std::unique_ptr<groupby_aggregation> make_correlation_aggregation<groupby_aggregation>(
-  correlation_type type);
+  correlation_type type, size_type min_periods);
 
 template <typename Base>
 std::unique_ptr<Base> make_tdigest_aggregation(int max_centroids)

cpp/src/aggregation/result_cache.cpp

@@ -30,10 +30,10 @@ void result_cache::add_result(column_view const& input,
 {
   // We can't guarantee that agg will outlive the cache, so we need to take ownership of a copy.
   // To allow lookup by reference, make the key a reference and keep the owner in the value pair.
-  auto owned_agg       = agg.clone();
-  auto const& key      = *owned_agg;
-  auto value           = std::make_pair(std::move(owned_agg), std::move(col));
-  _cache[{input, key}] = std::move(value);
+  auto owned_agg  = agg.clone();
+  auto const& key = *owned_agg;
+  // try_emplace doesn't update/insert if already present
+  _cache.try_emplace({input, key}, std::move(owned_agg), std::move(col));
 }
 
 column_view result_cache::get_result(column_view const& input, aggregation const& agg) const