GH-40280: [C++] Specialize ResolvedChunk::Value on value-specific types instead of entire class

cpp/src/arrow/chunk_resolver.cc

@@ -56,14 +56,33 @@ inline std::vector<int64_t> MakeChunksOffsets(const std::vector<T>& chunks) {
 }
 }  // namespace
 
-ChunkResolver::ChunkResolver(const ArrayVector& chunks)
+ChunkResolver::ChunkResolver(const ArrayVector& chunks) noexcept
     : offsets_(MakeChunksOffsets(chunks)), cached_chunk_(0) {}
 
-ChunkResolver::ChunkResolver(const std::vector<const Array*>& chunks)
+ChunkResolver::ChunkResolver(const std::vector<const Array*>& chunks) noexcept
     : offsets_(MakeChunksOffsets(chunks)), cached_chunk_(0) {}
 
-ChunkResolver::ChunkResolver(const RecordBatchVector& batches)
+ChunkResolver::ChunkResolver(const RecordBatchVector& batches) noexcept
     : offsets_(MakeChunksOffsets(batches)), cached_chunk_(0) {}
 
+ChunkResolver::ChunkResolver(ChunkResolver&& other) noexcept
+    : offsets_(std::move(other.offsets_)),
+      cached_chunk_(other.cached_chunk_.load(std::memory_order_relaxed)) {}
+
+ChunkResolver& ChunkResolver::operator=(ChunkResolver&& other) noexcept {
+  offsets_ = std::move(other.offsets_);
+  cached_chunk_.store(other.cached_chunk_.load(std::memory_order_relaxed));
+  return *this;
+}
+
+ChunkResolver::ChunkResolver(const ChunkResolver& other) noexcept
+    : offsets_(other.offsets_), cached_chunk_(0) {}
+
+ChunkResolver& ChunkResolver::operator=(const ChunkResolver& other) noexcept {
+  offsets_ = other.offsets_;
+  cached_chunk_.store(0, std::memory_order_relaxed);
+  return *this;
+}
+
 }  // namespace internal
 }  // namespace arrow

cpp/src/arrow/chunk_resolver.h

@@ -32,12 +32,12 @@ struct ChunkLocation {
   ///
   /// The value is always in the range `[0, chunks.size()]`. `chunks.size()` is used
   /// to represent out-of-bounds locations.
-  int64_t chunk_index;
+  int64_t chunk_index = 0;
 
   /// \brief Index of the value in the chunk
   ///
   /// The value is undefined if chunk_index >= chunks.size()
-  int64_t index_in_chunk;
+  int64_t index_in_chunk = 0;
 };
 
 /// \brief An utility that incrementally resolves logical indices into
@@ -56,19 +56,15 @@ struct ARROW_EXPORT ChunkResolver {
   mutable std::atomic<int64_t> cached_chunk_;
 
  public:
-  explicit ChunkResolver(const ArrayVector& chunks);
-  explicit ChunkResolver(const std::vector<const Array*>& chunks);
-  explicit ChunkResolver(const RecordBatchVector& batches);
-
-  ChunkResolver(ChunkResolver&& other) noexcept
-      : offsets_(std::move(other.offsets_)),
-        cached_chunk_(other.cached_chunk_.load(std::memory_order_relaxed)) {}
-
-  ChunkResolver& operator=(ChunkResolver&& other) {
-    offsets_ = std::move(other.offsets_);
-    cached_chunk_.store(other.cached_chunk_.load(std::memory_order_relaxed));
-    return *this;
-  }
+  explicit ChunkResolver(const ArrayVector& chunks) noexcept;
+  explicit ChunkResolver(const std::vector<const Array*>& chunks) noexcept;
+  explicit ChunkResolver(const RecordBatchVector& batches) noexcept;
+
+  ChunkResolver(ChunkResolver&& other) noexcept;
+  ChunkResolver& operator=(ChunkResolver&& other) noexcept;
+
+  ChunkResolver(const ChunkResolver& other) noexcept;
+  ChunkResolver& operator=(const ChunkResolver& other) noexcept;
 
   /// \brief Resolve a logical index to a ChunkLocation.
   ///
@@ -96,16 +92,16 @@ struct ARROW_EXPORT ChunkResolver {
   /// \pre index >= 0
   /// \post location.chunk_index in [0, chunks.size()]
   /// \param index The logical index to resolve
-  /// \param cached_chunk_index 0 or the chunk_index of the last ChunkLocation
-  /// returned by this ChunkResolver.
+  /// \param hint ChunkLocation{} or the last ChunkLocation returned by
+  ///             this ChunkResolver.
   /// \return ChunkLocation with a valid chunk_index if index is within
   ///         bounds, or with chunk_index == chunks.size() if logical index is
   ///         `>= chunked_array.length()`.
   inline ChunkLocation ResolveWithChunkIndexHint(int64_t index,
-                                                 int64_t cached_chunk_index) const {
-    assert(cached_chunk_index < static_cast<int64_t>(offsets_.size()));
+                                                 ChunkLocation hint) const {
+    assert(hint.chunk_index < static_cast<int64_t>(offsets_.size()));
     const auto chunk_index =
-        ResolveChunkIndex</*StoreCachedChunk=*/false>(index, cached_chunk_index);
+        ResolveChunkIndex</*StoreCachedChunk=*/false>(index, hint.chunk_index);
     return {chunk_index, index - offsets_[chunk_index]};
   }
 

cpp/src/arrow/compute/kernels/chunked_internal.h

@@ -31,51 +31,44 @@ namespace compute {
 namespace internal {
 
 // The target chunk in a chunked array.
-template <typename ArrayType>
 struct ResolvedChunk {
-  using ViewType = GetViewType<typename ArrayType::TypeClass>;
-  using LogicalValueType = typename ViewType::T;
-
-  // The target array in chunked array.
-  const ArrayType* array;
-  // The index in the target array.
-  const int64_t index;
-
-  ResolvedChunk(const ArrayType* array, int64_t index) : array(array), index(index) {}
-
-  bool IsNull() const { return array->IsNull(index); }
-
-  LogicalValueType Value() const { return ViewType::LogicalValue(array->GetView(index)); }
-};
-
-// ResolvedChunk specialization for untyped arrays when all is needed is null lookup
-template <>
-struct ResolvedChunk<Array> {
   // The target array in chunked array.
   const Array* array;
   // The index in the target array.
   const int64_t index;
 
   ResolvedChunk(const Array* array, int64_t index) : array(array), index(index) {}
 
+ public:
   bool IsNull() const { return array->IsNull(index); }
+
+  template <typename ArrowType, typename ViewType = GetViewType<ArrowType>>
+  typename ViewType::T Value() const {
+    using LogicalArrayType = typename TypeTraits<ArrowType>::ArrayType;
+    auto* typed_array = checked_cast<const LogicalArrayType*>(array);
+    return ViewType::LogicalValue(typed_array->GetView(index));
+  }
 };
 
-struct ChunkedArrayResolver : protected ::arrow::internal::ChunkResolver {
-  ChunkedArrayResolver(const ChunkedArrayResolver& other)
-      : ::arrow::internal::ChunkResolver(other.chunks_), chunks_(other.chunks_) {}
+class ChunkedArrayResolver {
+ private:
+  ::arrow::internal::ChunkResolver resolver_;
+  std::vector<const Array*> chunks_;
 
+ public:
   explicit ChunkedArrayResolver(const std::vector<const Array*>& chunks)
-      : ::arrow::internal::ChunkResolver(chunks), chunks_(chunks) {}
+      : resolver_(chunks), chunks_(chunks) {}
 
-  template <typename ArrayType>
-  ResolvedChunk<ArrayType> Resolve(int64_t index) const {
-    const auto loc = ::arrow::internal::ChunkResolver::Resolve(index);
-    return {checked_cast<const ArrayType*>(chunks_[loc.chunk_index]), loc.index_in_chunk};
-  }
+  ChunkedArrayResolver(ChunkedArrayResolver&& other) = default;
+  ChunkedArrayResolver& operator=(ChunkedArrayResolver&& other) = default;
+
+  ChunkedArrayResolver(const ChunkedArrayResolver& other) = default;
+  ChunkedArrayResolver& operator=(const ChunkedArrayResolver& other) = default;
 
- protected:
-  const std::vector<const Array*> chunks_;
+  ResolvedChunk Resolve(int64_t index) const {
+    const auto loc = resolver_.Resolve(index);
+    return {chunks_[loc.chunk_index], loc.index_in_chunk};
+  }
 };
 
 inline std::vector<const Array*> GetArrayPointers(const ArrayVector& arrays) {

cpp/src/arrow/compute/kernels/vector_rank.cc

@@ -227,8 +227,6 @@ class Ranker<ChunkedArray> : public RankerMixin<ChunkedArray, Ranker<ChunkedArra
 
   template <typename InType>
   Status RankInternal() {
-    using ArrayType = typename TypeTraits<InType>::ArrayType;
-
     if (physical_chunks_.empty()) {
       return Status::OK();
     }
@@ -240,7 +238,7 @@ class Ranker<ChunkedArray> : public RankerMixin<ChunkedArray, Ranker<ChunkedArra
 
     const auto arrays = GetArrayPointers(physical_chunks_);
     auto value_selector = [resolver = ChunkedArrayResolver(arrays)](int64_t index) {
-      return resolver.Resolve<ArrayType>(index).Value();
+      return resolver.Resolve(index).Value<InType>();
     };
     ARROW_ASSIGN_OR_RAISE(*output_, CreateRankings(ctx_, sorted, null_placement_,
                                                    tiebreaker_, value_selector));

cpp/src/arrow/compute/kernels/vector_select_k.cc

@@ -406,10 +406,7 @@ class TableSelector : public TypeVisitor {
 
     // Find the target chunk and index in the target chunk from an
     // index in chunked array.
-    template <typename ArrayType>
-    ResolvedChunk<ArrayType> GetChunk(int64_t index) const {
-      return resolver.Resolve<ArrayType>(index);
-    }
+    ResolvedChunk GetChunk(int64_t index) const { return resolver.Resolve(index); }
 
     const SortOrder order;
     const std::shared_ptr<DataType> type;
@@ -495,7 +492,6 @@ class TableSelector : public TypeVisitor {
 
   template <typename InType, SortOrder sort_order>
   Status SelectKthInternal() {
-    using ArrayType = typename TypeTraits<InType>::ArrayType;
     auto& comparator = comparator_;
     const auto& first_sort_key = sort_keys_[0];
 
@@ -509,10 +505,10 @@ class TableSelector : public TypeVisitor {
     std::function<bool(const uint64_t&, const uint64_t&)> cmp;
     SelectKComparator<sort_order> select_k_comparator;
     cmp = [&](const uint64_t& left, const uint64_t& right) -> bool {
-      auto chunk_left = first_sort_key.template GetChunk<ArrayType>(left);
-      auto chunk_right = first_sort_key.template GetChunk<ArrayType>(right);
-      auto value_left = chunk_left.Value();
-      auto value_right = chunk_right.Value();
+      auto chunk_left = first_sort_key.GetChunk(left);
+      auto chunk_right = first_sort_key.GetChunk(right);
+      auto value_left = chunk_left.Value<InType>();
+      auto value_right = chunk_right.Value<InType>();
       if (value_left == value_right) {
         return comparator.Compare(left, right, 1);
       }

cpp/src/arrow/compute/kernels/vector_sort.cc

@@ -156,25 +156,26 @@ class ChunkedArraySorter : public TypeVisitor {
   template <typename ArrayType>
   void MergeNonNulls(uint64_t* range_begin, uint64_t* range_middle, uint64_t* range_end,
                      const std::vector<const Array*>& arrays, uint64_t* temp_indices) {
+    using ArrowType = typename ArrayType::TypeClass;
     const ChunkedArrayResolver left_resolver(arrays);
     const ChunkedArrayResolver right_resolver(arrays);
 
     if (order_ == SortOrder::Ascending) {
       std::merge(range_begin, range_middle, range_middle, range_end, temp_indices,
                  [&](uint64_t left, uint64_t right) {
-                   const auto chunk_left = left_resolver.Resolve<ArrayType>(left);
-                   const auto chunk_right = right_resolver.Resolve<ArrayType>(right);
-                   return chunk_left.Value() < chunk_right.Value();
+                   const auto chunk_left = left_resolver.Resolve(left);
+                   const auto chunk_right = right_resolver.Resolve(right);
+                   return chunk_left.Value<ArrowType>() < chunk_right.Value<ArrowType>();
                  });
     } else {
       std::merge(range_begin, range_middle, range_middle, range_end, temp_indices,
                  [&](uint64_t left, uint64_t right) {
-                   const auto chunk_left = left_resolver.Resolve<ArrayType>(left);
-                   const auto chunk_right = right_resolver.Resolve<ArrayType>(right);
+                   const auto chunk_left = left_resolver.Resolve(left);
+                   const auto chunk_right = right_resolver.Resolve(right);
                    // We don't use 'left > right' here to reduce required
                    // operator. If we use 'right < left' here, '<' is only
                    // required.
-                   return chunk_right.Value() < chunk_left.Value();
+                   return chunk_right.Value<ArrowType>() < chunk_left.Value<ArrowType>();
                  });
     }
     // Copy back temp area into main buffer
@@ -744,8 +745,6 @@ class TableSorter {
                                                             uint64_t* nulls_end,
                                                             uint64_t* temp_indices,
                                                             int64_t null_count) {
-    using ArrayType = typename TypeTraits<Type>::ArrayType;
-
     auto& comparator = comparator_;
     const auto& first_sort_key = sort_keys_[0];
 
@@ -755,11 +754,11 @@ class TableSorter {
                [&](uint64_t left, uint64_t right) {
                  // First column is either null or nan
                  left_loc =
-                     left_resolver_.ResolveWithChunkIndexHint(left, left_loc.chunk_index);
-                 right_loc = right_resolver_.ResolveWithChunkIndexHint(
-                     right, right_loc.chunk_index);
-                 auto chunk_left = first_sort_key.GetChunk<ArrayType>(left_loc);
-                 auto chunk_right = first_sort_key.GetChunk<ArrayType>(right_loc);
+                     left_resolver_.ResolveWithChunkIndexHint(left, /*hint=*/left_loc);
+                 right_loc =
+                     right_resolver_.ResolveWithChunkIndexHint(right, /*hint=*/right_loc);
+                 auto chunk_left = first_sort_key.GetChunk(left_loc);
+                 auto chunk_right = first_sort_key.GetChunk(right_loc);
                  const auto left_is_null = chunk_left.IsNull();
                  const auto right_is_null = chunk_right.IsNull();
                  if (left_is_null == right_is_null) {
@@ -794,9 +793,9 @@ class TableSorter {
                [&](uint64_t left, uint64_t right) {
                  // First column is always null
                  left_loc =
-                     left_resolver_.ResolveWithChunkIndexHint(left, left_loc.chunk_index);
-                 right_loc = right_resolver_.ResolveWithChunkIndexHint(
-                     right, right_loc.chunk_index);
+                     left_resolver_.ResolveWithChunkIndexHint(left, /*hint=*/left_loc);
+                 right_loc =
+                     right_resolver_.ResolveWithChunkIndexHint(right, /*hint=*/right_loc);
                  return comparator.Compare(left_loc, right_loc, 1);
                });
     // Copy back temp area into main buffer
@@ -811,8 +810,6 @@ class TableSorter {
                                                         uint64_t* range_middle,
                                                         uint64_t* range_end,
                                                         uint64_t* temp_indices) {
-    using ArrayType = typename TypeTraits<Type>::ArrayType;
-
     auto& comparator = comparator_;
     const auto& first_sort_key = sort_keys_[0];
 
@@ -822,15 +819,15 @@ class TableSorter {
                [&](uint64_t left, uint64_t right) {
                  // Both values are never null nor NaN.
                  left_loc =
-                     left_resolver_.ResolveWithChunkIndexHint(left, left_loc.chunk_index);
-                 right_loc = right_resolver_.ResolveWithChunkIndexHint(
-                     right, right_loc.chunk_index);
-                 auto chunk_left = first_sort_key.GetChunk<ArrayType>(left_loc);
-                 auto chunk_right = first_sort_key.GetChunk<ArrayType>(right_loc);
+                     left_resolver_.ResolveWithChunkIndexHint(left, /*hint=*/left_loc);
+                 right_loc =
+                     right_resolver_.ResolveWithChunkIndexHint(right, /*hint=*/right_loc);
+                 auto chunk_left = first_sort_key.GetChunk(left_loc);
+                 auto chunk_right = first_sort_key.GetChunk(right_loc);
                  DCHECK(!chunk_left.IsNull());
                  DCHECK(!chunk_right.IsNull());
-                 auto value_left = chunk_left.Value();
-                 auto value_right = chunk_right.Value();
+                 auto value_left = chunk_left.Value<Type>();
+                 auto value_right = chunk_right.Value<Type>();
                  if (value_left == value_right) {
                    // If the left value equals to the right value,
                    // we need to compare the second and following