Address partitioning review suggestions

Remove redundant histogram-only and reordered-output block barriers while
retaining synchronization before shared storage is reused. Reuse the warp
histogram helper for local ranks, declare identity-copy dependencies, and
stage trivial partition boundaries through pinned host memory.

Dense-null and wide-column exact-result coverage was added in eac9bee.
Validation: 2140 partition tests, 7 stream tests, stream interposition,
15 memcheck cases, and 5 synccheck/racecheck cases passed. Fresh 348-case
before/after matrices and three paired repeat rounds found no consistent
slowdown above 5%; tracked GPU allocation peaks were unchanged.

Author: res-life

cpp/src/partitioning/partition_identity.cuh

@@ -9,13 +9,16 @@
 #include <cudf/detail/utilities/buffer_factories.hpp>
 #include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/null_mask.hpp>
+#include <cudf/strings/detail/gather.cuh>
 #include <cudf/strings/detail/utilities.hpp>
 #include <cudf/strings/utilities.hpp>
 #include <cudf/table/table.hpp>
 #include <cudf/utilities/traits.hpp>
 
 #include <rmm/device_buffer.hpp>
+#include <rmm/exec_policy.hpp>
 
+#include <cuda/iterator>
 #include <thrust/transform.h>
 
 namespace cudf::detail::partition_identity {

cpp/src/partitioning/partition_map_plan.cuh

@@ -79,8 +79,8 @@ CUDF_KERNEL void count_tiles(MapType const* map, int n, int p, uint16_t* saved,
       }
       total += sum;
     }
-    __syncthreads();
     if constexpr (SaveRanks) {
+      __syncthreads();
 #pragma unroll
       for (int j = 0; j < 4; ++j) {
         int i = start + (threadIdx.x / 32) * 128 + j * 32 + threadIdx.x % 32;
@@ -155,7 +155,7 @@ __device__ void emit_reordered(int const (&keys)[4],
     int position = threadIdx.x + j * threads;
     if (position < min(tile_rows, n - start)) sink(source_rows[position], destinations[position]);
   }
-  __syncthreads();
+  // The caller synchronizes before reusing the shared storage.
 }
 
 template <int Tiles, bool Reorder = false, typename Sink, typename MapType, typename Offsets>

cpp/src/partitioning/partitioning.cu

@@ -841,34 +841,16 @@ CUDF_KERNEL void partition_map_local_ranks(MapType const* partition_map,
   __shared__ size_type histogram[num_warps][MAP_PARTITION_MAX_BINS];
   auto const lane = threadIdx.x % warp_size;
   auto const warp = threadIdx.x / warp_size;
-  for (auto p = lane; p < num_partitions; p += warp_size) {
-    histogram[warp][p] = 0;
-  }
-  __syncwarp();
-
   size_type ranks[MAP_PARTITION_ITEMS_PER_THREAD];
   size_type partitions[MAP_PARTITION_ITEMS_PER_THREAD];
-#pragma unroll
-  for (size_type item = 0; item < MAP_PARTITION_ITEMS_PER_THREAD; ++item) {
-    auto const row = static_cast<cuda::std::size_t>(blockIdx.x) * MAP_PARTITION_TILE_SIZE +
-                     warp * warp_size * MAP_PARTITION_ITEMS_PER_THREAD + item * warp_size + lane;
-    auto const valid  = row < num_rows;
-    auto const active = __ballot_sync(0xffffffffu, valid);
-    if (valid) {
-      auto const partition = static_cast<size_type>(partition_map[row]);
-      partitions[item]     = partition;
-      auto const peers     = __match_any_sync(active, partition);
-      auto const base      = histogram[warp][partition];
-      ranks[item]          = base + cuda::std::popcount(peers & ((1u << lane) - 1u));
-      // All readers of a bin finish before its elected writer updates it.
-      __syncwarp(active);
-      if (lane == __ffs(peers) - 1) {
-        histogram[warp][partition] = base + cuda::std::popcount(peers);
-      }
-    }
-    __syncwarp();
-  }
-  __syncthreads();
+  partition_map_warp_histogram<true>(
+    partition_map,
+    num_rows,
+    num_partitions,
+    static_cast<cuda::std::size_t>(blockIdx.x) * MAP_PARTITION_TILE_SIZE,
+    histogram,
+    ranks,
+    partitions);
 
   for (auto partition = threadIdx.x; partition < num_partitions;
        partition += MAP_PARTITION_BLOCK_SIZE) {
@@ -1198,7 +1180,8 @@ std::vector<size_type> make_map_partition_offsets(column_view const& partition_m
                          offsets.begin(),
                          offsets.end(),
                          offsets.begin());
-  return cudf::detail::make_std_vector(offsets, stream);
+  auto const pinned_offsets = cudf::detail::make_pinned_vector(offsets, stream);
+  return {pinned_offsets.begin(), pinned_offsets.end()};
 }
 
 /**