[cudf] Thread memory resources through equality/hash row preprocessing

cpp/include/cudf/detail/row_operator/equality.cuh

@@ -407,9 +407,10 @@ class self_comparator {
    * @param t The table to compare
    * @param stream The stream to construct this object on. Not the stream that will be used for
    * comparisons using this object.
+   * @param mr Memory resources used for temporary device allocations
    */
-  self_comparator(table_view const& t, rmm::cuda_stream_view stream)
-    : d_t(preprocessed_table::create(t, stream))
+  self_comparator(table_view const& t, rmm::cuda_stream_view stream, cudf::memory_resources mr)
+    : d_t(preprocessed_table::create(t, stream, mr))
   {
   }
 
@@ -515,10 +516,12 @@ class two_table_comparator {
    * @param right The right table to compare.
    * @param stream The stream to construct this object on. Not the stream that will be used for
    * comparisons using this object.
+   * @param mr Memory resources used for temporary device allocations
    */
   two_table_comparator(table_view const& left,
                        table_view const& right,
-                       rmm::cuda_stream_view stream);
+                       rmm::cuda_stream_view stream,
+                       cudf::memory_resources mr);
 
   /**
    * @brief Construct an owning object for performing equality comparisons between two rows from two

cpp/include/cudf/detail/row_operator/hashing.cuh

@@ -240,9 +240,10 @@ class row_hasher {
    * @param t The table containing rows to hash
    * @param stream The stream to construct this object on. Not the stream that will be used for
    * comparisons using this object.
+   * @param mr Memory resources used for temporary device allocations
    */
-  row_hasher(table_view const& t, rmm::cuda_stream_view stream)
-    : d_t(preprocessed_table::create(t, stream))
+  row_hasher(table_view const& t, rmm::cuda_stream_view stream, cudf::memory_resources mr)
+    : d_t(preprocessed_table::create(t, stream, mr))
   {
   }
 

cpp/include/cudf/detail/row_operator/preprocessed_table.cuh

@@ -6,6 +6,7 @@
 #pragma once
 
 #include <cudf/table/table_device_view.cuh>
+#include <cudf/utilities/memory_resource.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
 #include <rmm/device_buffer.hpp>
@@ -47,10 +48,12 @@ struct preprocessed_table {
    *
    * @param table The table to preprocess
    * @param stream The cuda stream to use while preprocessing.
+   * @param mr Memory resources used for temporary device allocations
    * @return A preprocessed table as shared pointer
    */
   static std::shared_ptr<preprocessed_table> create(table_view const& table,
-                                                    rmm::cuda_stream_view stream);
+                                                    rmm::cuda_stream_view stream,
+                                                    cudf::memory_resources mr);
 
   /**
    * @brief Implicit conversion operator to a `table_device_view` of the preprocessed table.

cpp/include/cudf_test/base_fixture.hpp

@@ -1,21 +1,26 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2020-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2020-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
 #pragma once
 
 #include <cudf_test/cudf_gtest.hpp>
+#include <cudf_test/default_stream.hpp>
 #include <cudf_test/file_utilities.hpp>
+#include <cudf_test/memory_resource_utilities.hpp>
 
 #include <cudf/utilities/export.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 #include <cudf/utilities/traits.hpp>
 
+#include <rmm/cuda_stream_view.hpp>
 #include <rmm/resource_ref.hpp>
 
 #include <cuda/memory_resource>
 
+#include <optional>
+
 namespace CUDF_EXPORT cudf {
 namespace test {
 
@@ -39,6 +44,46 @@ class BaseFixture : public ::testing::Test {
   rmm::device_async_resource_ref mr() { return _mr; }
 };
 
+/**
+ * @brief Base fixture that instruments tests with a memory-resource harness.
+ *
+ * Each test instantiates a fresh harness. The failing current-device-resource scope is installed
+ * after `_harness` so accidental fallback to the default MR fails the test. Tests should construct
+ * results with `resources()`. `TearDown` asserts that no output or temporary allocations remain
+ * live; the prior current resource is restored when the optional scope is reset or destroyed.
+ */
+struct BaseFixtureWithHarness : public BaseFixture {
+  /**
+   * @brief Assert that the harness has no live output or temporary allocations.
+   */
+  void TearDown() override { _harness.expect_no_live_allocations(stream()); }
+
+  /**
+   * @brief Return the default stream used by tests inheriting from this fixture.
+   * @return CUDA stream view
+   */
+  [[nodiscard]] rmm::cuda_stream_view stream() const { return cudf::test::get_default_stream(); }
+
+  /**
+   * @brief Return the harness output and temporary memory resources.
+   * @return Explicit output and temporary resources that do not consult the current resource
+   */
+  cudf::memory_resources resources() { return _harness.resources(); }
+
+  /**
+   * @brief Clear the failing current-device-resource scope for the remainder of the test.
+   *
+   * After this call, APIs may allocate from the restored current device resource without failing
+   * the test. Prefer removing the need for this once those APIs accept explicit resources.
+   */
+  void disable_current_device_resource_use() { _fail_on_current.reset(); }
+
+ protected:
+  memory_resource_test_harness _harness{mr()};
+  std::optional<scoped_current_device_resource> _fail_on_current{
+    _harness.fail_on_current_device_resource_use()};
+};
+
 /**
  * @brief Base test fixture that takes a parameter.
  *