experimental/row_operators.cuh

/*
 * Copyright (c) 2022-2024, NVIDIA CORPORATION.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#pragma once
 
#include <cudf/column/column_device_view.cuh>
#include <cudf/detail/iterator.cuh>
#include <cudf/detail/utilities/algorithm.cuh>
#include <cudf/detail/utilities/assert.cuh>
#include <cudf/hashing/detail/default_hash.cuh>
#include <cudf/hashing/detail/hashing.hpp>
#include <cudf/lists/detail/dremel.hpp>
#include <cudf/lists/list_device_view.cuh>
#include <cudf/lists/lists_column_device_view.cuh>
#include <cudf/sorting.hpp>
#include <cudf/structs/structs_column_device_view.cuh>
#include <cudf/table/row_operators.cuh>
#include <cudf/table/table_device_view.cuh>
#include <cudf/utilities/default_stream.hpp>
#include <cudf/utilities/traits.hpp>
#include <cudf/utilities/type_dispatcher.hpp>
 
#include <cuda/std/tuple>
#include <cuda/std/utility>
#include <thrust/detail/use_default.h>
#include <thrust/equal.h>
#include <thrust/execution_policy.h>
#include <thrust/functional.h>
#include <thrust/iterator/counting_iterator.h>
#include <thrust/iterator/iterator_adaptor.h>
#include <thrust/iterator/iterator_categories.h>
#include <thrust/iterator/iterator_facade.h>
#include <thrust/iterator/transform_iterator.h>
#include <thrust/logical.h>
#include <thrust/swap.h>
#include <thrust/transform_reduce.h>
 
#include <limits>
#include <memory>
#include <optional>
#include <type_traits>
#include <utility>
 
namespace cudf {
 
namespace experimental {
 
template <cudf::type_id t>
struct dispatch_void_if_nested {
  using type = std::conditional_t<t == type_id::STRUCT or t == type_id::LIST, void, id_to_type<t>>;
};
 
namespace row {
 
enum class lhs_index_type : size_type {};
enum class rhs_index_type : size_type {};
 
template <typename Index, typename Underlying = std::underlying_type_t<Index>>
struct strong_index_iterator : public thrust::iterator_facade<strong_index_iterator<Index>,
                                                              Index,
                                                              thrust::use_default,
                                                              thrust::random_access_traversal_tag,
                                                              Index,
                                                              Underlying> {
  using super_t =
    thrust::iterator_adaptor<strong_index_iterator<Index>, Index>;  
 
  explicit constexpr strong_index_iterator(Underlying n) : begin{n} {}
 
  friend class thrust::iterator_core_access;  
 
 private:
  __device__ constexpr void increment() { ++begin; }
  __device__ constexpr void decrement() { --begin; }
 
  __device__ constexpr void advance(Underlying n) { begin += n; }
 
  __device__ constexpr bool equal(strong_index_iterator<Index> const& other) const noexcept
  {
    return begin == other.begin;
  }
 
  __device__ constexpr Index dereference() const noexcept { return static_cast<Index>(begin); }
 
  __device__ constexpr Underlying distance_to(
    strong_index_iterator<Index> const& other) const noexcept
  {
    return other.begin - begin;
  }
 
  Underlying begin{};
};
 
using lhs_iterator = strong_index_iterator<lhs_index_type>;
 
using rhs_iterator = strong_index_iterator<rhs_index_type>;
 
namespace lexicographic {
 
struct physical_element_comparator {
  template <typename Element>
  __device__ constexpr weak_ordering operator()(Element const lhs, Element const rhs) const noexcept
  {
    return detail::compare_elements(lhs, rhs);
  }
};
 
struct sorting_physical_element_comparator {
  template <typename Element, CUDF_ENABLE_IF(not std::is_floating_point_v<Element>)>
  __device__ constexpr weak_ordering operator()(Element const lhs, Element const rhs) const noexcept
  {
    return detail::compare_elements(lhs, rhs);
  }
 
  template <typename Element, CUDF_ENABLE_IF(std::is_floating_point_v<Element>)>
  __device__ constexpr weak_ordering operator()(Element const lhs, Element const rhs) const noexcept
  {
    if (isnan(lhs)) {
      return isnan(rhs) ? weak_ordering::EQUIVALENT : weak_ordering::GREATER;
    } else if (isnan(rhs)) {
      return weak_ordering::LESS;
    }
 
    return detail::compare_elements(lhs, rhs);
  }
};
 
using optional_dremel_view = thrust::optional<detail::dremel_device_view const>;
 
// The has_nested_columns template parameter of the device_row_comparator is
// necessary to help the compiler optimize our code. Without it, the list and
// struct view specializations are present in the code paths used for primitive
// types, and the compiler fails to inline this nearly as well resulting in a
// significant performance drop.  As a result, there is some minor tension in
// the current design between the presence of this parameter and the way that
// the Dremel data is passed around, first as a
// std::optional<device_span<dremel_device_view>> in the
// preprocessed_table/device_row_comparator (which is always valid when
// has_nested_columns and is otherwise invalid) that is then unpacked to a
// thrust::optional<dremel_device_view> at the element_comparator level (which
// is always valid for a list column and otherwise invalid).  We cannot use an
// additional template parameter for the element_comparator on a per-column
// basis because we cannot conditionally define dremel_device_view member
// variables without jumping through extra hoops with inheritance, so the
// thrust::optional<dremel_device_view> member must be an optional rather than
// a raw dremel_device_view.
template <bool has_nested_columns,
          typename Nullate,
          typename PhysicalElementComparator = sorting_physical_element_comparator>
class device_row_comparator {
 public:
  friend class self_comparator;       
  friend class two_table_comparator;  
 
  device_row_comparator(Nullate check_nulls,
                        table_device_view lhs,
                        table_device_view rhs,
                        device_span<detail::dremel_device_view const> l_dremel_device_views,
                        device_span<detail::dremel_device_view const> r_dremel_device_views,
                        std::optional<device_span<int const>> depth                  = std::nullopt,
                        std::optional<device_span<order const>> column_order         = std::nullopt,
                        std::optional<device_span<null_order const>> null_precedence = std::nullopt,
                        PhysicalElementComparator comparator                         = {}) noexcept
    : _lhs{lhs},
      _rhs{rhs},
      _l_dremel(l_dremel_device_views),
      _r_dremel(r_dremel_device_views),
      _check_nulls{check_nulls},
      _depth{depth},
      _column_order{column_order},
      _null_precedence{null_precedence},
      _comparator{comparator}
  {
  }
 
  template <bool nested_disable = not has_nested_columns, CUDF_ENABLE_IF(nested_disable)>
  __device__ device_row_comparator(
    Nullate check_nulls,
    table_device_view lhs,
    table_device_view rhs,
    std::optional<device_span<order const>> column_order         = std::nullopt,
    std::optional<device_span<null_order const>> null_precedence = std::nullopt,
    PhysicalElementComparator comparator                         = {}) noexcept
    : _lhs{lhs},
      _rhs{rhs},
      _l_dremel{},
      _r_dremel{},
      _check_nulls{check_nulls},
      _depth{},
      _column_order{column_order},
      _null_precedence{null_precedence},
      _comparator{comparator}
  {
  }
 
  class element_comparator {
   public:
    __device__ element_comparator(Nullate check_nulls,
                                  column_device_view lhs,
                                  column_device_view rhs,
                                  null_order null_precedence                = null_order::BEFORE,
                                  int depth                                 = 0,
                                  PhysicalElementComparator comparator      = {},
                                  optional_dremel_view l_dremel_device_view = {},
                                  optional_dremel_view r_dremel_device_view = {})
      : _lhs{lhs},
        _rhs{rhs},
        _check_nulls{check_nulls},
        _null_precedence{null_precedence},
        _depth{depth},
        _l_dremel_device_view{l_dremel_device_view},
        _r_dremel_device_view{r_dremel_device_view},
        _comparator{comparator}
    {
    }
 
    template <typename Element,
              CUDF_ENABLE_IF(cudf::is_relationally_comparable<Element, Element>())>
    __device__ cuda::std::pair<weak_ordering, int> operator()(
      size_type const lhs_element_index, size_type const rhs_element_index) const noexcept
    {
      if (_check_nulls) {
        bool const lhs_is_null{_lhs.is_null(lhs_element_index)};
        bool const rhs_is_null{_rhs.is_null(rhs_element_index)};
 
        if (lhs_is_null or rhs_is_null) {  // at least one is null
          return cuda::std::pair(null_compare(lhs_is_null, rhs_is_null, _null_precedence), _depth);
        }
      }
 
      return cuda::std::pair(_comparator(_lhs.element<Element>(lhs_element_index),
                                         _rhs.element<Element>(rhs_element_index)),
                             std::numeric_limits<int>::max());
    }
 
    template <typename Element,
              CUDF_ENABLE_IF(not cudf::is_relationally_comparable<Element, Element>() and
                             (not has_nested_columns or not cudf::is_nested<Element>()))>
    __device__ cuda::std::pair<weak_ordering, int> operator()(size_type const,
                                                              size_type const) const noexcept
    {
      CUDF_UNREACHABLE("Attempted to compare elements of uncomparable types.");
    }
 
    template <typename Element,
              CUDF_ENABLE_IF(has_nested_columns and std::is_same_v<Element, cudf::struct_view>)>
    __device__ cuda::std::pair<weak_ordering, int> operator()(
      size_type const lhs_element_index, size_type const rhs_element_index) const noexcept
    {
      column_device_view lcol = _lhs;
      column_device_view rcol = _rhs;
      int depth               = _depth;
      while (lcol.type().id() == type_id::STRUCT) {
        bool const lhs_is_null{lcol.is_null(lhs_element_index)};
        bool const rhs_is_null{rcol.is_null(rhs_element_index)};
 
        if (lhs_is_null or rhs_is_null) {  // at least one is null
          weak_ordering state = null_compare(lhs_is_null, rhs_is_null, _null_precedence);
          return cuda::std::pair(state, depth);
        }
 
        if (lcol.num_child_columns() == 0) {
          return cuda::std::pair(weak_ordering::EQUIVALENT, std::numeric_limits<int>::max());
        }
 
        // Non-empty structs have been modified to only have 1 child when using this.
        lcol = detail::structs_column_device_view(lcol).get_sliced_child(0);
        rcol = detail::structs_column_device_view(rcol).get_sliced_child(0);
        ++depth;
      }
 
      return cudf::type_dispatcher<dispatch_void_if_nested>(
        lcol.type(),
        element_comparator{_check_nulls, lcol, rcol, _null_precedence, depth, _comparator},
        lhs_element_index,
        rhs_element_index);
    }
 
    template <typename Element,
              CUDF_ENABLE_IF(has_nested_columns and std::is_same_v<Element, cudf::list_view>)>
    __device__ cuda::std::pair<weak_ordering, int> operator()(size_type lhs_element_index,
                                                              size_type rhs_element_index)
    {
      // only order top-NULLs according to null_order
      auto const is_l_row_null = _lhs.is_null(lhs_element_index);
      auto const is_r_row_null = _rhs.is_null(rhs_element_index);
      if (is_l_row_null || is_r_row_null) {
        return cuda::std::pair(null_compare(is_l_row_null, is_r_row_null, _null_precedence),
                               _depth);
      }
 
      // These are all the values from the Dremel encoding.
      auto const l_max_def_level = _l_dremel_device_view->max_def_level;
      auto const r_max_def_level = _r_dremel_device_view->max_def_level;
      auto const l_def_levels    = _l_dremel_device_view->def_levels;
      auto const r_def_levels    = _r_dremel_device_view->def_levels;
      auto const l_rep_levels    = _l_dremel_device_view->rep_levels;
      auto const r_rep_levels    = _r_dremel_device_view->rep_levels;
 
      // Traverse the nested list hierarchy to get a column device view
      // pointing to the underlying child data.
      column_device_view lcol = _lhs.slice(lhs_element_index, 1);
      column_device_view rcol = _rhs.slice(rhs_element_index, 1);
 
      while (lcol.type().id() == type_id::LIST) {
        lcol = detail::lists_column_device_view(lcol).get_sliced_child();
        rcol = detail::lists_column_device_view(rcol).get_sliced_child();
      }
 
      // These start and end values indicate the start and end points of all
      // the elements of the lists in the current list element
      // (`[lhs|rhs]_element_index`) that we are comparing.
      auto const l_offsets = _l_dremel_device_view->offsets;
      auto const r_offsets = _r_dremel_device_view->offsets;
      auto l_start         = l_offsets[lhs_element_index];
      auto l_end           = l_offsets[lhs_element_index + 1];
      auto r_start         = r_offsets[rhs_element_index];
      auto r_end           = r_offsets[rhs_element_index + 1];
 
      // This comparator will be used to compare leaf (non-nested) data types.
      auto comparator =
        element_comparator{_check_nulls, lcol, rcol, _null_precedence, _depth, _comparator};
 
      // Loop over each element in the encoding. Note that this includes nulls
      // and empty lists, so not every index corresponds to an actual element
      // in the child column. The element_index is used to keep track of the current
      // child element that we're actually comparing.
      for (int l_dremel_index = l_start, r_dremel_index = r_start, element_index = 0;
           l_dremel_index < l_end and r_dremel_index < r_end;
           ++l_dremel_index, ++r_dremel_index) {
        auto const l_rep_level = l_rep_levels[l_dremel_index];
        auto const r_rep_level = r_rep_levels[r_dremel_index];
 
        // early exit for smaller sub-list
        if (l_rep_level != r_rep_level) {
          // the lower repetition level is a smaller sub-list
          return l_rep_level < r_rep_level ? cuda::std::pair(weak_ordering::LESS, _depth)
                                           : cuda::std::pair(weak_ordering::GREATER, _depth);
        }
 
        // only compare if left and right are at same nesting level
        auto const l_def_level = l_def_levels[l_dremel_index];
        auto const r_def_level = r_def_levels[r_dremel_index];
 
        // either left or right are empty or NULLs of arbitrary nesting
        if (l_def_level < l_max_def_level || r_def_level < r_max_def_level) {
          // in the fully unraveled version of the list column, only the
          // most nested NULLs and leafs are present
          // In this rare condition that we get to the most nested NULL, we increment
          // element_index because either both rows have a deeply nested NULL at the
          // same position, and we'll "continue" in our iteration, or we will early
          // exit if only one of the rows has a deeply nested NULL
          if ((lcol.nullable() and l_def_levels[l_dremel_index] == l_max_def_level - 1) or
              (rcol.nullable() and r_def_levels[r_dremel_index] == r_max_def_level - 1)) {
            ++element_index;
          }
          if (l_def_level == r_def_level) { continue; }
          // We require [] < [NULL] < [leaf] for nested nulls.
          // The null_precedence only affects top level nulls.
          return l_def_level < r_def_level ? cuda::std::pair(weak_ordering::LESS, _depth)
                                           : cuda::std::pair(weak_ordering::GREATER, _depth);
        }
 
        // finally, compare leaf to leaf
        weak_ordering state{weak_ordering::EQUIVALENT};
        int last_null_depth                    = _depth;
        cuda::std::tie(state, last_null_depth) = cudf::type_dispatcher<dispatch_void_if_nested>(
          lcol.type(), comparator, element_index, element_index);
        if (state != weak_ordering::EQUIVALENT) { return cuda::std::pair(state, _depth); }
        ++element_index;
      }
 
      // If we have reached this stage, we know that definition levels,
      // repetition levels, and actual elements are identical in both list
      // columns up to the `min(l_end - l_start, r_end - r_start)` element of
      // the Dremel encoding. However, two lists can only compare equivalent if
      // they are of the same length. Otherwise, the shorter of the two is less
      // than the longer. This final check determines the appropriate resulting
      // ordering by checking how many total elements each list is composed of.
      return cuda::std::pair(detail::compare_elements(l_end - l_start, r_end - r_start), _depth);
    }
 
   private:
    column_device_view const _lhs;
    column_device_view const _rhs;
    Nullate const _check_nulls;
    null_order const _null_precedence;
    int const _depth;
    optional_dremel_view _l_dremel_device_view;
    optional_dremel_view _r_dremel_device_view;
    PhysicalElementComparator const _comparator;
  };
 
 public:
  __device__ constexpr weak_ordering operator()(size_type const lhs_index,
                                                size_type const rhs_index) const noexcept
  {
    int last_null_depth = std::numeric_limits<int>::max();
    size_type list_column_index{-1};
    for (size_type i = 0; i < _lhs.num_columns(); ++i) {
      if (_lhs.column(i).type().id() == type_id::LIST) { ++list_column_index; }
 
      int const depth = _depth.has_value() ? (*_depth)[i] : 0;
      if (depth > last_null_depth) { continue; }
 
      bool const ascending =
        _column_order.has_value() ? (*_column_order)[i] == order::ASCENDING : true;
 
      null_order const null_precedence =
        _null_precedence.has_value() ? (*_null_precedence)[i] : null_order::BEFORE;
 
      // TODO: At what point do we verify that the columns of lhs and rhs are
      // all of the same types? I assume that it's already happened before
      // here, otherwise the current code would be failing.
      auto const [l_dremel_i, r_dremel_i] =
        _lhs.column(i).type().id() == type_id::LIST
          ? std::make_tuple(optional_dremel_view(_l_dremel[list_column_index]),
                            optional_dremel_view(_r_dremel[list_column_index]))
          : std::make_tuple(optional_dremel_view{}, optional_dremel_view{});
 
      auto element_comp = element_comparator{_check_nulls,
                                             _lhs.column(i),
                                             _rhs.column(i),
                                             null_precedence,
                                             depth,
                                             _comparator,
                                             l_dremel_i,
                                             r_dremel_i};
 
      weak_ordering state;
      cuda::std::tie(state, last_null_depth) =
        cudf::type_dispatcher(_lhs.column(i).type(), element_comp, lhs_index, rhs_index);
 
      if (state == weak_ordering::EQUIVALENT) { continue; }
 
      return ascending
               ? state
               : (state == weak_ordering::GREATER ? weak_ordering::LESS : weak_ordering::GREATER);
    }
    return weak_ordering::EQUIVALENT;
  }
 
 private:
  table_device_view const _lhs;
  table_device_view const _rhs;
  device_span<detail::dremel_device_view const> const _l_dremel;
  device_span<detail::dremel_device_view const> const _r_dremel;
  Nullate const _check_nulls;
  std::optional<device_span<int const>> const _depth;
  std::optional<device_span<order const>> const _column_order;
  std::optional<device_span<null_order const>> const _null_precedence;
  PhysicalElementComparator const _comparator;
};  // class device_row_comparator
 
template <typename Comparator, weak_ordering... values>
struct weak_ordering_comparator_impl {
  static_assert(not((weak_ordering::EQUIVALENT == values) && ...),
                "weak_ordering_comparator should not be used for pure equality comparisons. The "
                "`row_equality_comparator` should be used instead");
 
  template <typename LhsType, typename RhsType>
  __device__ constexpr bool operator()(LhsType const lhs_index,
                                       RhsType const rhs_index) const noexcept
  {
    weak_ordering const result = comparator(lhs_index, rhs_index);
    return ((result == values) || ...);
  }
  Comparator const comparator;
};
 
template <typename Comparator>
struct less_comparator : weak_ordering_comparator_impl<Comparator, weak_ordering::LESS> {
  less_comparator(Comparator const& comparator)
    : weak_ordering_comparator_impl<Comparator, weak_ordering::LESS>{comparator}
  {
  }
};
 
template <typename Comparator>
struct less_equivalent_comparator
  : weak_ordering_comparator_impl<Comparator, weak_ordering::LESS, weak_ordering::EQUIVALENT> {
  less_equivalent_comparator(Comparator const& comparator)
    : weak_ordering_comparator_impl<Comparator, weak_ordering::LESS, weak_ordering::EQUIVALENT>{
        comparator}
  {
  }
};
 
struct preprocessed_table {
  using table_device_view_owner =
    std::invoke_result_t<decltype(table_device_view::create), table_view, rmm::cuda_stream_view>;
 
  static std::shared_ptr<preprocessed_table> create(table_view const& table,
                                                    host_span<order const> column_order,
                                                    host_span<null_order const> null_precedence,
                                                    rmm::cuda_stream_view stream);
 
  static std::pair<std::shared_ptr<preprocessed_table>, std::shared_ptr<preprocessed_table>> create(
    table_view const& lhs,
    table_view const& rhs,
    host_span<order const> column_order,
    host_span<null_order const> null_precedence,
    rmm::cuda_stream_view stream);
 
 private:
  friend class self_comparator;       
  friend class two_table_comparator;  
 
  static std::shared_ptr<preprocessed_table> create(
    table_view const& preprocessed_input,
    std::vector<int>&& verticalized_col_depths,
    std::vector<std::unique_ptr<column>>&& transformed_columns,
    host_span<order const> column_order,
    host_span<null_order const> null_precedence,
    bool has_ranked_children,
    rmm::cuda_stream_view stream);
 
  preprocessed_table(table_device_view_owner&& table,
                     rmm::device_uvector<order>&& column_order,
                     rmm::device_uvector<null_order>&& null_precedence,
                     rmm::device_uvector<size_type>&& depths,
                     std::vector<detail::dremel_data>&& dremel_data,
                     rmm::device_uvector<detail::dremel_device_view>&& dremel_device_views,
                     std::vector<std::unique_ptr<column>>&& transformed_columns,
                     bool has_ranked_children);
 
  preprocessed_table(table_device_view_owner&& table,
                     rmm::device_uvector<order>&& column_order,
                     rmm::device_uvector<null_order>&& null_precedence,
                     rmm::device_uvector<size_type>&& depths,
                     std::vector<std::unique_ptr<column>>&& transformed_columns,
                     bool has_ranked_children);
 
  operator table_device_view() { return *_t; }
 
  [[nodiscard]] std::optional<device_span<order const>> column_order() const
  {
    return _column_order.size() ? std::optional<device_span<order const>>(_column_order)
                                : std::nullopt;
  }
 
  [[nodiscard]] std::optional<device_span<null_order const>> null_precedence() const
  {
    return _null_precedence.size() ? std::optional<device_span<null_order const>>(_null_precedence)
                                   : std::nullopt;
  }
 
  [[nodiscard]] std::optional<device_span<int const>> depths() const
  {
    return _depths.size() ? std::optional<device_span<int const>>(_depths) : std::nullopt;
  }
 
  [[nodiscard]] device_span<detail::dremel_device_view const> dremel_device_views() const
  {
    if (_dremel_device_views.has_value()) {
      return device_span<detail::dremel_device_view const>(*_dremel_device_views);
    } else {
      return {};
    }
  }
 
  template <typename PhysicalElementComparator>
  void check_physical_element_comparator()
  {
    if constexpr (!std::is_same_v<PhysicalElementComparator, sorting_physical_element_comparator>) {
      CUDF_EXPECTS(!_has_ranked_children,
                   "The input table has nested type children and they were transformed using a "
                   "different type of physical element comparator.");
    }
  }
 
 private:
  table_device_view_owner const _t;
  rmm::device_uvector<order> const _column_order;
  rmm::device_uvector<null_order> const _null_precedence;
  rmm::device_uvector<size_type> const _depths;
 
  // Dremel encoding of list columns used for the comparison algorithm
  std::optional<std::vector<detail::dremel_data>> _dremel_data;
  std::optional<rmm::device_uvector<detail::dremel_device_view>> _dremel_device_views;
 
  // Intermediate columns generated from transforming nested children columns into
  // integers columns using `cudf::rank()`, need to be kept alive.
  std::vector<std::unique_ptr<column>> _transformed_columns;
 
  // Flag to record if the input table was preprocessed to transform any nested children column(s)
  // into integer column(s) using `cudf::rank`.
  bool const _has_ranked_children;
};
 
class self_comparator {
 public:
  self_comparator(table_view const& t,
                  host_span<order const> column_order         = {},
                  host_span<null_order const> null_precedence = {},
                  rmm::cuda_stream_view stream                = cudf::get_default_stream())
    : d_t{preprocessed_table::create(t, column_order, null_precedence, stream)}
  {
  }
 
  self_comparator(std::shared_ptr<preprocessed_table> t) : d_t{std::move(t)} {}
 
  template <bool has_nested_columns,
            typename Nullate,
            typename PhysicalElementComparator = sorting_physical_element_comparator>
  auto less(Nullate nullate = {}, PhysicalElementComparator comparator = {}) const
  {
    d_t->check_physical_element_comparator<PhysicalElementComparator>();
 
    return less_comparator{
      device_row_comparator<has_nested_columns, Nullate, PhysicalElementComparator>{
        nullate,
        *d_t,
        *d_t,
        d_t->dremel_device_views(),
        d_t->dremel_device_views(),
        d_t->depths(),
        d_t->column_order(),
        d_t->null_precedence(),
        comparator}};
  }
 
  template <bool has_nested_columns,
            typename Nullate,
            typename PhysicalElementComparator = sorting_physical_element_comparator>
  auto less_equivalent(Nullate nullate = {}, PhysicalElementComparator comparator = {}) const
  {
    d_t->check_physical_element_comparator<PhysicalElementComparator>();
 
    return less_equivalent_comparator{
      device_row_comparator<has_nested_columns, Nullate, PhysicalElementComparator>{
        nullate,
        *d_t,
        *d_t,
        d_t->dremel_device_views(),
        d_t->dremel_device_views(),
        d_t->depths(),
        d_t->column_order(),
        d_t->null_precedence(),
        comparator}};
  }
 
 private:
  std::shared_ptr<preprocessed_table> d_t;
};
 
// @cond
template <typename Comparator>
struct strong_index_comparator_adapter {
  strong_index_comparator_adapter(Comparator const& comparator) : comparator{comparator} {}
 
  __device__ constexpr weak_ordering operator()(lhs_index_type const lhs_index,
                                                rhs_index_type const rhs_index) const noexcept
  {
    return comparator(static_cast<cudf::size_type>(lhs_index),
                      static_cast<cudf::size_type>(rhs_index));
  }
 
  __device__ constexpr weak_ordering operator()(rhs_index_type const rhs_index,
                                                lhs_index_type const lhs_index) const noexcept
  {
    auto const left_right_ordering =
      comparator(static_cast<cudf::size_type>(lhs_index), static_cast<cudf::size_type>(rhs_index));
 
    // Invert less/greater values to reflect right to left ordering
    if (left_right_ordering == weak_ordering::LESS) {
      return weak_ordering::GREATER;
    } else if (left_right_ordering == weak_ordering::GREATER) {
      return weak_ordering::LESS;
    }
    return weak_ordering::EQUIVALENT;
  }
 
  Comparator const comparator;
};
// @endcond
 
class two_table_comparator {
 public:
  two_table_comparator(table_view const& left,
                       table_view const& right,
                       host_span<order const> column_order         = {},
                       host_span<null_order const> null_precedence = {},
                       rmm::cuda_stream_view stream                = cudf::get_default_stream());
 
  two_table_comparator(std::shared_ptr<preprocessed_table> left,
                       std::shared_ptr<preprocessed_table> right)
    : d_left_table{std::move(left)}, d_right_table{std::move(right)}
  {
  }
 
  template <bool has_nested_columns,
            typename Nullate,
            typename PhysicalElementComparator = sorting_physical_element_comparator>
  auto less(Nullate nullate = {}, PhysicalElementComparator comparator = {}) const
  {
    d_left_table->check_physical_element_comparator<PhysicalElementComparator>();
    d_right_table->check_physical_element_comparator<PhysicalElementComparator>();
 
    return less_comparator{strong_index_comparator_adapter{
      device_row_comparator<has_nested_columns, Nullate, PhysicalElementComparator>{
        nullate,
        *d_left_table,
        *d_right_table,
        d_left_table->dremel_device_views(),
        d_right_table->dremel_device_views(),
        d_left_table->depths(),
        d_left_table->column_order(),
        d_left_table->null_precedence(),
        comparator}}};
  }
 
  template <bool has_nested_columns,
            typename Nullate,
            typename PhysicalElementComparator = sorting_physical_element_comparator>
  auto less_equivalent(Nullate nullate = {}, PhysicalElementComparator comparator = {}) const
  {
    d_left_table->check_physical_element_comparator<PhysicalElementComparator>();
    d_right_table->check_physical_element_comparator<PhysicalElementComparator>();
 
    return less_equivalent_comparator{strong_index_comparator_adapter{
      device_row_comparator<has_nested_columns, Nullate, PhysicalElementComparator>{
        nullate,
        *d_left_table,
        *d_right_table,
        d_left_table->dremel_device_views(),
        d_right_table->dremel_device_views(),
        d_left_table->depths(),
        d_left_table->column_order(),
        d_left_table->null_precedence(),
        comparator}}};
  }
 
 private:
  std::shared_ptr<preprocessed_table> d_left_table;
  std::shared_ptr<preprocessed_table> d_right_table;
};
 
}  // namespace lexicographic
 
namespace hash {
class row_hasher;
}  // namespace hash
 
namespace equality {
 
struct physical_equality_comparator {
  template <typename Element>
  __device__ constexpr bool operator()(Element const lhs, Element const rhs) const noexcept
  {
    return lhs == rhs;
  }
};
 
struct nan_equal_physical_equality_comparator {
  template <typename Element, CUDF_ENABLE_IF(not std::is_floating_point_v<Element>)>
  __device__ constexpr bool operator()(Element const lhs, Element const rhs) const noexcept
  {
    return lhs == rhs;
  }
 
  template <typename Element, CUDF_ENABLE_IF(std::is_floating_point_v<Element>)>
  __device__ constexpr bool operator()(Element const lhs, Element const rhs) const noexcept
  {
    return isnan(lhs) and isnan(rhs) ? true : lhs == rhs;
  }
};
 
template <bool has_nested_columns,
          typename Nullate,
          typename PhysicalEqualityComparator = nan_equal_physical_equality_comparator>
class device_row_comparator {
  friend class self_comparator;       
  friend class two_table_comparator;  
 
 public:
  __device__ constexpr bool operator()(size_type const lhs_index,
                                       size_type const rhs_index) const noexcept
  {
    auto equal_elements = [=](column_device_view l, column_device_view r) {
      return cudf::type_dispatcher(
        l.type(),
        element_comparator{check_nulls, l, r, nulls_are_equal, comparator},
        lhs_index,
        rhs_index);
    };
 
    return thrust::equal(thrust::seq, lhs.begin(), lhs.end(), rhs.begin(), equal_elements);
  }
 
 private:
  device_row_comparator(Nullate check_nulls,
                        table_device_view lhs,
                        table_device_view rhs,
                        null_equality nulls_are_equal         = null_equality::EQUAL,
                        PhysicalEqualityComparator comparator = {}) noexcept
    : lhs{lhs},
      rhs{rhs},
      check_nulls{check_nulls},
      nulls_are_equal{nulls_are_equal},
      comparator{comparator}
  {
  }
 
  class element_comparator {
   public:
    __device__ element_comparator(Nullate check_nulls,
                                  column_device_view lhs,
                                  column_device_view rhs,
                                  null_equality nulls_are_equal         = null_equality::EQUAL,
                                  PhysicalEqualityComparator comparator = {}) noexcept
      : lhs{lhs},
        rhs{rhs},
        check_nulls{check_nulls},
        nulls_are_equal{nulls_are_equal},
        comparator{comparator}
    {
    }
 
    template <typename Element, CUDF_ENABLE_IF(cudf::is_equality_comparable<Element, Element>())>
    __device__ bool operator()(size_type const lhs_element_index,
                               size_type const rhs_element_index) const noexcept
    {
      if (check_nulls) {
        bool const lhs_is_null{lhs.is_null(lhs_element_index)};
        bool const rhs_is_null{rhs.is_null(rhs_element_index)};
        if (lhs_is_null and rhs_is_null) {
          return nulls_are_equal == null_equality::EQUAL;
        } else if (lhs_is_null != rhs_is_null) {
          return false;
        }
      }
 
      return comparator(lhs.element<Element>(lhs_element_index),
                        rhs.element<Element>(rhs_element_index));
    }
 
    template <typename Element,
              CUDF_ENABLE_IF(not cudf::is_equality_comparable<Element, Element>() and
                             (not has_nested_columns or not cudf::is_nested<Element>())),
              typename... Args>
    __device__ bool operator()(Args...)
    {
      CUDF_UNREACHABLE("Attempted to compare elements of uncomparable types.");
    }
 
    template <typename Element, CUDF_ENABLE_IF(has_nested_columns and cudf::is_nested<Element>())>
    __device__ bool operator()(size_type const lhs_element_index,
                               size_type const rhs_element_index) const noexcept
    {
      column_device_view lcol = lhs.slice(lhs_element_index, 1);
      column_device_view rcol = rhs.slice(rhs_element_index, 1);
      while (lcol.type().id() == type_id::STRUCT || lcol.type().id() == type_id::LIST) {
        if (check_nulls) {
          auto lvalid = detail::make_validity_iterator<true>(lcol);
          auto rvalid = detail::make_validity_iterator<true>(rcol);
          if (nulls_are_equal == null_equality::UNEQUAL) {
            if (thrust::any_of(
                  thrust::seq, lvalid, lvalid + lcol.size(), thrust::logical_not<bool>()) or
                thrust::any_of(
                  thrust::seq, rvalid, rvalid + rcol.size(), thrust::logical_not<bool>())) {
              return false;
            }
          } else {
            if (not thrust::equal(thrust::seq, lvalid, lvalid + lcol.size(), rvalid)) {
              return false;
            }
          }
        }
        if (lcol.type().id() == type_id::STRUCT) {
          if (lcol.num_child_columns() == 0) { return true; }
          // Non-empty structs are assumed to be decomposed and contain only one child
          lcol = detail::structs_column_device_view(lcol).get_sliced_child(0);
          rcol = detail::structs_column_device_view(rcol).get_sliced_child(0);
        } else if (lcol.type().id() == type_id::LIST) {
          auto l_list_col = detail::lists_column_device_view(lcol);
          auto r_list_col = detail::lists_column_device_view(rcol);
 
          auto lsizes = make_list_size_iterator(l_list_col);
          auto rsizes = make_list_size_iterator(r_list_col);
          if (not thrust::equal(thrust::seq, lsizes, lsizes + lcol.size(), rsizes)) {
            return false;
          }
 
          lcol = l_list_col.get_sliced_child();
          rcol = r_list_col.get_sliced_child();
          if (lcol.size() != rcol.size()) { return false; }
        }
      }
 
      auto comp = column_comparator{
        element_comparator{check_nulls, lcol, rcol, nulls_are_equal, comparator}, lcol.size()};
      return type_dispatcher<dispatch_void_if_nested>(lcol.type(), comp);
    }
 
   private:
    struct column_comparator {
      element_comparator const comp;
      size_type const size;
 
      template <typename Element, CUDF_ENABLE_IF(cudf::is_equality_comparable<Element, Element>())>
      __device__ bool operator()() const noexcept
      {
        return thrust::all_of(thrust::seq,
                              thrust::make_counting_iterator(0),
                              thrust::make_counting_iterator(0) + size,
                              [=](auto i) { return comp.template operator()<Element>(i, i); });
      }
 
      template <typename Element,
                CUDF_ENABLE_IF(not cudf::is_equality_comparable<Element, Element>()),
                typename... Args>
      __device__ bool operator()(Args...) const noexcept
      {
        CUDF_UNREACHABLE("Attempted to compare elements of uncomparable types.");
      }
    };
 
    column_device_view const lhs;
    column_device_view const rhs;
    Nullate const check_nulls;
    null_equality const nulls_are_equal;
    PhysicalEqualityComparator const comparator;
  };
 
  table_device_view const lhs;
  table_device_view const rhs;
  Nullate const check_nulls;
  null_equality const nulls_are_equal;
  PhysicalEqualityComparator const comparator;
};
 
struct preprocessed_table {
  static std::shared_ptr<preprocessed_table> create(table_view const& table,
                                                    rmm::cuda_stream_view stream);
 
 private:
  friend class self_comparator;       
  friend class two_table_comparator;  
  friend class hash::row_hasher;      
 
  using table_device_view_owner =
    std::invoke_result_t<decltype(table_device_view::create), table_view, rmm::cuda_stream_view>;
 
  preprocessed_table(table_device_view_owner&& table,
                     std::vector<rmm::device_buffer>&& null_buffers,
                     std::vector<std::unique_ptr<column>>&& tmp_columns)
    : _t(std::move(table)),
      _null_buffers(std::move(null_buffers)),
      _tmp_columns(std::move(tmp_columns))
  {
  }
 
  operator table_device_view() { return *_t; }
 
  table_device_view_owner _t;
  std::vector<rmm::device_buffer> _null_buffers;
  std::vector<std::unique_ptr<column>> _tmp_columns;
};
 
class self_comparator {
 public:
  self_comparator(table_view const& t, rmm::cuda_stream_view stream)
    : d_t(preprocessed_table::create(t, stream))
  {
  }
 
  self_comparator(std::shared_ptr<preprocessed_table> t) : d_t{std::move(t)} {}
 
  template <bool has_nested_columns,
            typename Nullate,
            typename PhysicalEqualityComparator = nan_equal_physical_equality_comparator>
  auto equal_to(Nullate nullate                       = {},
                null_equality nulls_are_equal         = null_equality::EQUAL,
                PhysicalEqualityComparator comparator = {}) const noexcept
  {
    return device_row_comparator<has_nested_columns, Nullate, PhysicalEqualityComparator>{
      nullate, *d_t, *d_t, nulls_are_equal, comparator};
  }
 
 private:
  std::shared_ptr<preprocessed_table> d_t;
};
 
// @cond
template <typename Comparator>
struct strong_index_comparator_adapter {
  strong_index_comparator_adapter(Comparator const& comparator) : comparator{comparator} {}
 
  __device__ constexpr bool operator()(lhs_index_type const lhs_index,
                                       rhs_index_type const rhs_index) const noexcept
  {
    return comparator(static_cast<cudf::size_type>(lhs_index),
                      static_cast<cudf::size_type>(rhs_index));
  }
 
  __device__ constexpr bool operator()(rhs_index_type const rhs_index,
                                       lhs_index_type const lhs_index) const noexcept
  {
    return this->operator()(lhs_index, rhs_index);
  }
 
  Comparator const comparator;
};
// @endcond
 
class two_table_comparator {
 public:
  two_table_comparator(table_view const& left,
                       table_view const& right,
                       rmm::cuda_stream_view stream);
 
  two_table_comparator(std::shared_ptr<preprocessed_table> left,
                       std::shared_ptr<preprocessed_table> right)
    : d_left_table{std::move(left)}, d_right_table{std::move(right)}
  {
  }
 
  template <bool has_nested_columns,
            typename Nullate,
            typename PhysicalEqualityComparator = nan_equal_physical_equality_comparator>
  auto equal_to(Nullate nullate                       = {},
                null_equality nulls_are_equal         = null_equality::EQUAL,
                PhysicalEqualityComparator comparator = {}) const noexcept
  {
    return strong_index_comparator_adapter{
      device_row_comparator<has_nested_columns, Nullate, PhysicalEqualityComparator>(
        nullate, *d_left_table, *d_right_table, nulls_are_equal, comparator)};
  }
 
 private:
  std::shared_ptr<preprocessed_table> d_left_table;
  std::shared_ptr<preprocessed_table> d_right_table;
};
 
}  // namespace equality
 
namespace hash {
 
template <template <typename> class hash_function, typename Nullate>
class element_hasher {
 public:
  __device__ element_hasher(
    Nullate nulls,
    uint32_t seed             = DEFAULT_HASH_SEED,
    hash_value_type null_hash = std::numeric_limits<hash_value_type>::max()) noexcept
    : _check_nulls(nulls), _seed(seed), _null_hash(null_hash)
  {
  }
 
  template <typename T, CUDF_ENABLE_IF(column_device_view::has_element_accessor<T>())>
  __device__ hash_value_type operator()(column_device_view const& col,
                                        size_type row_index) const noexcept
  {
    if (_check_nulls && col.is_null(row_index)) { return _null_hash; }
    return hash_function<T>{_seed}(col.element<T>(row_index));
  }
 
  template <typename T, CUDF_ENABLE_IF(not column_device_view::has_element_accessor<T>())>
  __device__ hash_value_type operator()(column_device_view const& col,
                                        size_type row_index) const noexcept
  {
    CUDF_UNREACHABLE("Unsupported type in hash.");
  }
 
  Nullate _check_nulls;        
  uint32_t _seed;              
  hash_value_type _null_hash;  
};
 
template <template <typename> class hash_function, typename Nullate>
class device_row_hasher {
  friend class row_hasher;  
 
 public:
  __device__ auto operator()(size_type row_index) const noexcept
  {
    auto it = thrust::make_transform_iterator(_table.begin(), [=](auto const& column) {
      return cudf::type_dispatcher<dispatch_storage_type>(
        column.type(),
        element_hasher_adapter<hash_function>{_check_nulls, _seed},
        column,
        row_index);
    });
 
    // Hash each element and combine all the hash values together
    return detail::accumulate(it, it + _table.num_columns(), _seed, [](auto hash, auto h) {
      return cudf::hashing::detail::hash_combine(hash, h);
    });
  }
 
 private:
  template <template <typename> class hash_fn>
  class element_hasher_adapter {
    static constexpr hash_value_type NULL_HASH     = std::numeric_limits<hash_value_type>::max();
    static constexpr hash_value_type NON_NULL_HASH = 0;
 
   public:
    __device__ element_hasher_adapter(Nullate check_nulls, uint32_t seed) noexcept
      : _element_hasher(check_nulls, seed), _check_nulls(check_nulls)
    {
    }
 
    template <typename T, CUDF_ENABLE_IF(not cudf::is_nested<T>())>
    __device__ hash_value_type operator()(column_device_view const& col,
                                          size_type row_index) const noexcept
    {
      return _element_hasher.template operator()<T>(col, row_index);
    }
 
    template <typename T, CUDF_ENABLE_IF(cudf::is_nested<T>())>
    __device__ hash_value_type operator()(column_device_view const& col,
                                          size_type row_index) const noexcept
    {
      auto hash                   = hash_value_type{0};
      column_device_view curr_col = col.slice(row_index, 1);
      while (curr_col.type().id() == type_id::STRUCT || curr_col.type().id() == type_id::LIST) {
        if (_check_nulls) {
          auto validity_it = detail::make_validity_iterator<true>(curr_col);
          hash             = detail::accumulate(
            validity_it, validity_it + curr_col.size(), hash, [](auto hash, auto is_valid) {
              return cudf::hashing::detail::hash_combine(hash,
                                                         is_valid ? NON_NULL_HASH : NULL_HASH);
            });
        }
        if (curr_col.type().id() == type_id::STRUCT) {
          if (curr_col.num_child_columns() == 0) { return hash; }
          // Non-empty structs are assumed to be decomposed and contain only one child
          curr_col = detail::structs_column_device_view(curr_col).get_sliced_child(0);
        } else if (curr_col.type().id() == type_id::LIST) {
          auto list_col   = detail::lists_column_device_view(curr_col);
          auto list_sizes = make_list_size_iterator(list_col);
          hash            = detail::accumulate(
            list_sizes, list_sizes + list_col.size(), hash, [](auto hash, auto size) {
              return cudf::hashing::detail::hash_combine(hash, hash_fn<size_type>{}(size));
            });
          curr_col = list_col.get_sliced_child();
        }
      }
      for (int i = 0; i < curr_col.size(); ++i) {
        hash = cudf::hashing::detail::hash_combine(
          hash,
          type_dispatcher<dispatch_void_if_nested>(curr_col.type(), _element_hasher, curr_col, i));
      }
      return hash;
    }
 
    element_hasher<hash_fn, Nullate> const _element_hasher;
    Nullate const _check_nulls;
  };
 
  CUDF_HOST_DEVICE device_row_hasher(Nullate check_nulls,
                                     table_device_view t,
                                     uint32_t seed = DEFAULT_HASH_SEED) noexcept
    : _check_nulls{check_nulls}, _table{t}, _seed(seed)
  {
  }
 
  Nullate const _check_nulls;
  table_device_view const _table;
  uint32_t const _seed;
};
 
// Inject row::equality::preprocessed_table into the row::hash namespace
// As a result, row::equality::preprocessed_table and row::hash::preprocessed table are the same
// type and are interchangeable.
using preprocessed_table = row::equality::preprocessed_table;
 
class row_hasher {
 public:
  row_hasher(table_view const& t, rmm::cuda_stream_view stream)
    : d_t(preprocessed_table::create(t, stream))
  {
  }
 
  row_hasher(std::shared_ptr<preprocessed_table> t) : d_t{std::move(t)} {}
 
  template <template <typename> class hash_function = cudf::hashing::detail::default_hash,
            template <template <typename> class, typename>
            class DeviceRowHasher = device_row_hasher,
            typename Nullate>
  DeviceRowHasher<hash_function, Nullate> device_hasher(Nullate nullate = {},
                                                        uint32_t seed   = DEFAULT_HASH_SEED) const
  {
    return DeviceRowHasher<hash_function, Nullate>(nullate, *d_t, seed);
  }
 
 private:
  std::shared_ptr<preprocessed_table> d_t;
};
 
}  // namespace hash
 
}  // namespace row
 
}  // namespace experimental
}  // namespace cudf