Fix null handling for structs min and arg_min in groupby, groupby scan, reduction, and inclusive_scan

cpp/src/groupby/sort/group_scan_util.cuh

@@ -16,7 +16,7 @@
 
 #pragma once
 
-#include <reductions/arg_minmax_util.cuh>
+#include <reductions/struct_minmax_util.cuh>
 
 #include <cudf/column/column.hpp>
 #include <cudf/column/column_factories.hpp>
@@ -26,8 +26,6 @@
 #include <cudf/detail/gather.hpp>
 #include <cudf/detail/iterator.cuh>
 #include <cudf/detail/null_mask.hpp>
-#include <cudf/detail/structs/utilities.hpp>
-#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/table/table_device_view.cuh>
 #include <cudf/types.hpp>
 #include <cudf/utilities/span.hpp>
@@ -192,43 +190,18 @@ struct group_scan_functor<K,
   {
     if (values.is_empty()) { return cudf::empty_like(values); }
 
-    // When finding MIN, we need to consider nulls as larger than non-null elements.
-    // Thing is opposite when finding MAX.
-    auto const null_precedence  = (K == aggregation::MIN) ? null_order::AFTER : null_order::BEFORE;
-    auto const flattened_values = structs::detail::flatten_nested_columns(
-      table_view{{values}}, {}, std::vector<null_order>{null_precedence});
-    auto const d_flattened_values_ptr = table_device_view::create(flattened_values, stream);
-    auto const flattened_null_precedences =
-      (K == aggregation::MIN)
-        ? cudf::detail::make_device_uvector_async(flattened_values.null_orders(), stream)
-        : rmm::device_uvector<null_order>(0, stream);
+    // Create a gather map containing indices of the prefix min/max elements within each group.
+    auto gather_map = rmm::device_uvector<size_type>(values.size(), stream);
 
-    // Create a gather map contaning indices of the prefix min/max elements.
-    auto gather_map      = rmm::device_uvector<size_type>(values.size(), stream);
-    auto const map_begin = gather_map.begin();
-
-    // Perform segmented scan.
-    auto const do_scan = [&](auto const& inp_iter, auto const& out_iter, auto const& binop) {
-      thrust::inclusive_scan_by_key(rmm::exec_policy(stream),
-                                    group_labels.begin(),
-                                    group_labels.end(),
-                                    inp_iter,
-                                    out_iter,
-                                    thrust::equal_to{},
-                                    binop);
-    };
-
-    // Find the indices of the prefix min/max elements within each group.
-    auto const count_iter = thrust::make_counting_iterator<size_type>(0);
-    auto const binop      = cudf::reduction::detail::row_arg_minmax_fn(values.size(),
-                                                                  *d_flattened_values_ptr,
-                                                                  values.has_nulls(),
-                                                                  flattened_null_precedences.data(),
-                                                                  K == aggregation::MIN);
-    do_scan(count_iter, map_begin, binop);
-
-    auto gather_map_view =
-      column_view(data_type{type_to_id<offset_type>()}, gather_map.size(), gather_map.data());
+    auto const binop_generator =
+      cudf::reduction::detail::comparison_binop_generator::create<K>(values, stream);
+    thrust::inclusive_scan_by_key(rmm::exec_policy(stream),
+                                  group_labels.begin(),
+                                  group_labels.end(),
+                                  thrust::make_counting_iterator<size_type>(0),
+                                  gather_map.begin(),
+                                  thrust::equal_to{},
+                                  binop_generator.binop());
 
     //
     // Gather the children elements of the prefix min/max struct elements first.
@@ -240,7 +213,7 @@ struct group_scan_functor<K,
     auto scanned_children =
       cudf::detail::gather(
         table_view(std::vector<column_view>{values.child_begin(), values.child_end()}),
-        gather_map_view,
+        gather_map,
         cudf::out_of_bounds_policy::DONT_CHECK,
         cudf::detail::negative_index_policy::NOT_ALLOWED,
         stream,

cpp/src/groupby/sort/group_single_pass_reduction_util.cuh

@@ -16,15 +16,13 @@
 
 #pragma once
 
-#include <reductions/arg_minmax_util.cuh>
+#include <reductions/struct_minmax_util.cuh>
 
 #include <cudf/column/column.hpp>
 #include <cudf/column/column_factories.hpp>
 #include <cudf/column/column_view.hpp>
 #include <cudf/detail/aggregation/aggregation.cuh>
 #include <cudf/detail/iterator.cuh>
-#include <cudf/detail/structs/utilities.hpp>
-#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/detail/valid_if.cuh>
 #include <cudf/table/row_operators.cuh>
 #include <cudf/types.hpp>
@@ -244,18 +242,6 @@ struct group_reduction_functor<
 
     if (values.is_empty()) { return result; }
 
-    // When finding ARGMIN, we need to consider nulls as larger than non-null elements.
-    // Thing is opposite for ARGMAX.
-    auto const null_precedence =
-      (K == aggregation::ARGMIN) ? null_order::AFTER : null_order::BEFORE;
-    auto const flattened_values = structs::detail::flatten_nested_columns(
-      table_view{{values}}, {}, std::vector<null_order>{null_precedence});
-    auto const d_flattened_values_ptr = table_device_view::create(flattened_values, stream);
-    auto const flattened_null_precedences =
-      (K == aggregation::ARGMIN)
-        ? cudf::detail::make_device_uvector_async(flattened_values.null_orders(), stream)
-        : rmm::device_uvector<null_order>(0, stream);
-
     // Perform segmented reduction to find ARGMIN/ARGMAX.
     auto const do_reduction = [&](auto const& inp_iter, auto const& out_iter, auto const& binop) {
       thrust::reduce_by_key(rmm::exec_policy(stream),
@@ -270,12 +256,9 @@ struct group_reduction_functor<
 
     auto const count_iter   = thrust::make_counting_iterator<ResultType>(0);
     auto const result_begin = result->mutable_view().template begin<ResultType>();
-    auto const binop        = cudf::reduction::detail::row_arg_minmax_fn(values.size(),
-                                                                  *d_flattened_values_ptr,
-                                                                  values.has_nulls(),
-                                                                  flattened_null_precedences.data(),
-                                                                  K == aggregation::ARGMIN);
-    do_reduction(count_iter, result_begin, binop);
+    auto const binop_generator =
+      cudf::reduction::detail::comparison_binop_generator::create<K>(values, stream);
+    do_reduction(count_iter, result_begin, binop_generator.binop());
 
     if (values.has_nulls()) {
       // Generate bitmask for the output by segmented reduction of the input bitmask.

cpp/src/reductions/scan/scan_inclusive.cu

@@ -14,17 +14,15 @@
  * limitations under the License.
  */
 
-#include <reductions/arg_minmax_util.cuh>
 #include <reductions/scan/scan.cuh>
+#include <reductions/struct_minmax_util.cuh>
 
 #include <cudf/column/column_device_view.cuh>
 #include <cudf/column/column_factories.hpp>
 #include <cudf/detail/copy.hpp>
 #include <cudf/detail/gather.hpp>
 #include <cudf/detail/iterator.cuh>
 #include <cudf/detail/null_mask.hpp>
-#include <cudf/detail/structs/utilities.hpp>
-#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/reduction.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
@@ -159,35 +157,15 @@ struct scan_functor<Op, cudf::struct_view> {
                                         rmm::cuda_stream_view stream,
                                         rmm::mr::device_memory_resource* mr)
   {
-    // Op is used only to determined if we want to find the min or max element.
-    auto constexpr is_min_op = std::is_same_v<Op, DeviceMin>;
-
-    // Build indices of the scan operation results (ARGMIN/ARGMAX).
-    // When finding ARGMIN, we need to consider nulls as larger than non-null elements, and the
-    // opposite for ARGMAX.
-    auto gather_map    = rmm::device_uvector<size_type>(input.size(), stream);
-    auto const do_scan = [&](auto const& binop) {
-      thrust::inclusive_scan(rmm::exec_policy(stream),
-                             thrust::counting_iterator<size_type>(0),
-                             thrust::counting_iterator<size_type>(input.size()),
-                             gather_map.begin(),
-                             binop);
-    };
-
-    auto constexpr null_precedence = is_min_op ? cudf::null_order::AFTER : cudf::null_order::BEFORE;
-    auto const flattened_input     = cudf::structs::detail::flatten_nested_columns(
-      table_view{{input}}, {}, std::vector<null_order>{null_precedence});
-    auto const d_flattened_input_ptr = table_device_view::create(flattened_input, stream);
-    auto const flattened_null_precedences =
-      is_min_op ? cudf::detail::make_device_uvector_async(flattened_input.null_orders(), stream)
-                : rmm::device_uvector<cudf::null_order>(0, stream);
-
-    auto const binop = cudf::reduction::detail::row_arg_minmax_fn(input.size(),
-                                                                  *d_flattened_input_ptr,
-                                                                  input.has_nulls(),
-                                                                  flattened_null_precedences.data(),
-                                                                  is_min_op);
-    do_scan(binop);
+    // Create a gather map contaning indices of the prefix min/max elements.
+    auto gather_map = rmm::device_uvector<size_type>(input.size(), stream);
+    auto const binop_generator =
+      cudf::reduction::detail::comparison_binop_generator::create<Op>(input, stream);
+    thrust::inclusive_scan(rmm::exec_policy(stream),
+                           thrust::counting_iterator<size_type>(0),
+                           thrust::counting_iterator<size_type>(input.size()),
+                           gather_map.begin(),
+                           binop_generator.binop());
 
     // Gather the children columns of the input column. Must use `get_sliced_child` to properly
     // handle input in case it is a sliced view.

cpp/src/reductions/simple.cuh

@@ -16,13 +16,12 @@
 
 #pragma once
 
-#include <reductions/arg_minmax_util.cuh>
+#include <reductions/struct_minmax_util.cuh>
 
 #include <cudf/detail/copy.hpp>
 #include <cudf/detail/reduction.cuh>
 #include <cudf/detail/structs/utilities.hpp>
 #include <cudf/detail/utilities/cuda.cuh>
-#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/dictionary/detail/iterator.cuh>
 #include <cudf/dictionary/dictionary_column_view.hpp>
 #include <cudf/scalar/scalar_device_view.cuh>
@@ -294,37 +293,14 @@ struct same_element_type_dispatcher {
   {
     if (input.is_empty()) { return cudf::make_empty_scalar_like(input, stream, mr); }
 
-    auto constexpr is_min_op = std::is_same_v<Op, cudf::reduction::op::min>;
-
     // We will do reduction to find the ARGMIN/ARGMAX index, then return the element at that index.
-    // When finding ARGMIN, we need to consider nulls as larger than non-null elements, and the
-    // opposite for ARGMAX.
-    auto constexpr null_precedence = is_min_op ? cudf::null_order::AFTER : cudf::null_order::BEFORE;
-    auto const flattened_input     = cudf::structs::detail::flatten_nested_columns(
-      table_view{{input}}, {}, std::vector<null_order>{null_precedence});
-    auto const d_flattened_input_ptr = table_device_view::create(flattened_input, stream);
-    auto const flattened_null_precedences =
-      is_min_op ? cudf::detail::make_device_uvector_async(flattened_input.null_orders(), stream)
-                : rmm::device_uvector<cudf::null_order>(0, stream);
-
-    // Perform reduction to find ARGMIN/ARGMAX.
-    auto const do_reduction = [&](auto const& binop) {
-      return thrust::reduce(rmm::exec_policy(stream),
-                            thrust::make_counting_iterator(0),
-                            thrust::make_counting_iterator(input.size()),
-                            size_type{0},
-                            binop);
-    };
-
-    auto const minmax_idx = [&] {
-      auto const binop =
-        cudf::reduction::detail::row_arg_minmax_fn(input.size(),
-                                                   *d_flattened_input_ptr,
-                                                   input.has_nulls(),
-                                                   flattened_null_precedences.data(),
-                                                   is_min_op);
-      return do_reduction(binop);
-    }();
+    auto const binop_generator =
+      cudf::reduction::detail::comparison_binop_generator::create<Op>(input, stream);
+    auto const minmax_idx = thrust::reduce(rmm::exec_policy(stream),
+                                           thrust::make_counting_iterator(0),
+                                           thrust::make_counting_iterator(input.size()),
+                                           size_type{0},
+                                           binop_generator.binop());
 
     return cudf::detail::get_element(input, minmax_idx, stream, mr);
   }