From 27f8a2cb80400c55725a484182ebe0cf12c72726 Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Wed, 10 Jun 2026 14:02:56 -0700
Subject: [PATCH 1/8] WIP

---
 ...tcher_decimal_aggregation_refactor.plan.md | 392 ++++++++++++++++++
 .../cudf/exec/DecimalAggregationDevice.cu     | 200 +++++----
 .../cudf/exec/DecimalAggregationDevice.h      |  36 +-
 .../cudf/exec/DecimalAggregationState.cpp     | 159 +++++--
 4 files changed, 619 insertions(+), 168 deletions(-)
 create mode 100644 .cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md

diff --git a/.cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md b/.cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md
new file mode 100644
index 00000000000..1ee0e840345
--- /dev/null
+++ b/.cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md
@@ -0,0 +1,392 @@
+---
+name: type_dispatcher decimal aggregation refactor
+overview: Replace all manual type branches and void* casts in decimal sum state pack/unpack/average paths with cudf::type_dispatcher (and double_type_dispatcher where two dimensions apply), using typed device helpers and dispatch_storage_type for decimal columns.
+todos:
+  - id: retype-unpack-api
+    content: Change unpackDecimalSumState to template<OffsetT>(const OffsetT* offsets, ...) in DecimalAggregationDevice.h/.cu with explicit int32_t/int64_t instantiations
+    status: pending
+  - id: retype-pack-fill-api
+    content: Change fillOffsetsForDecimalSumState to template<OffsetT> and packDecimalSumState to template<SumT, OffsetT> in DecimalAggregationDevice.h/.cu with explicit instantiations for all 4 combinations
+    status: pending
+  - id: retype-average-api
+    content: Change averageRoundDecimalSum to template<SumT>(const SumT* sums, SumT* out, ...) in DecimalAggregationDevice.h/.cu with explicit int64_t/__int128_t instantiations
+    status: pending
+  - id: dispatch-deserialize-call-site
+    content: Add type_dispatcher.hpp include and replace bool/void* branch at line 109 in DecimalAggregationState.cpp with UnpackDecimalSumStateDispatcher functor
+    status: pending
+  - id: dispatch-serialize-call-sites
+    content: Replace bool/void* branches at lines 187 and 207 with FillOffsetsDispatcher and nested/double_type_dispatcher for pack (sum + offset types)
+    status: pending
+  - id: dispatch-average-call-site
+    content: Replace sumType/void* branch at lines 257-265 in computeDecimalAverage with AverageRoundDecimalSumDispatcher using dispatch_storage_type
+    status: pending
+  - id: verify-tests
+    content: Run DecimalAggregationTest sum-state and average tests covering DECIMAL64/DECIMAL128 and INT32/INT64 offsets
+    status: pending
+isProject: true
+---
+
+# Refactor decimal aggregation device calls to use type_dispatcher
+
+## Current state
+
+### Deserialize path (line 109)
+
+In [`DecimalAggregationState.cpp`](velox/experimental/cudf/exec/DecimalAggregationState.cpp) (lines 83–117), deserialize manually inspects the strings offsets type, sets `offsets64`, and casts `offsetsView.data<int64_t>()` / `data<int32_t>()` to `void*` before calling the device helper.
+
+The device implementation in [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu) already has fully typed kernels (`launchUnpackState<OffsetT>`); the `bool` + `void*` wrapper (lines 309–337) is redundant type erasure.
+
+### Serialize path (lines 187 and 207)
+
+The serialize path uses offset type-erasure for both `fillOffsetsForDecimalSumState` and `packDecimalSumState`, and additionally erases sum types via `sumType` + `sumPtr`:
+
+```187:215:velox/experimental/cudf/exec/DecimalAggregationState.cpp
+  detail::fillOffsetsForDecimalSumState(
+      useLargeOffsets,
+      useLargeOffsets ? static_cast<void*>(offsetsView.data<int64_t>())
+                      : static_cast<void*>(offsetsView.data<int32_t>()),
+      rowCount,
+      stream);
+  ...
+  const void* sumPtr = sumType == cudf::type_id::DECIMAL64
+      ? static_cast<const void*>(sumCol.data<int64_t>())
+      : static_cast<const void*>(sumCol.data<__int128_t>());
+  detail::packDecimalSumState(
+      sumType,
+      useLargeOffsets,
+      sumPtr,
+      countCol.data<int64_t>(),
+      offsetsPtr,
+      charsPtr,
+      rowCount,
+      stream);
+```
+
+`useLargeOffsets` is still needed to **choose** `offsetsType` when creating the offsets column (lines 172–179), but it should no longer be threaded into device calls or used for pointer casts.
+
+The device wrappers `fillOffsetsForDecimalSumState` (lines 237–253) and `packDecimalSumState` (lines 255–307) mirror the unpack problem: typed kernels (`launchFillOffsets<OffsetT>`, `launchPackState<SumT, OffsetT>`) exist, but the public API erases types via `bool`/`type_id` + `void*`.
+
+### Average path (lines 257–265)
+
+`computeDecimalAverage` applies the same sum-type erasure pattern:
+
+```257:265:velox/experimental/cudf/exec/DecimalAggregationState.cpp
+    const auto sumType = sumCol.type().id();
+    const void* sumsPtr = sumType == cudf::type_id::DECIMAL64
+        ? static_cast<const void*>(sumCol.data<int64_t>())
+        : static_cast<const void*>(sumCol.data<__int128_t>());
+    void* outPtr = sumType == cudf::type_id::DECIMAL64
+        ? static_cast<void*>(out->mutable_view().data<int64_t>())
+        : static_cast<void*>(out->mutable_view().data<__int128_t>());
+    detail::averageRoundDecimalSum(
+        sumType, sumsPtr, countCol.data<int64_t>(), outPtr, rowCount, stream);
+```
+
+The device wrapper `averageRoundDecimalSum` (lines 339–362) branches on `sumType` and casts to typed spans for `launchAvgRound<SumT>`.
+
+There is **no existing `type_dispatcher` usage** in `velox/experimental/cudf/` today — this will be the first adoption in this module.
+
+## Target design
+
+```mermaid
+flowchart TB
+  subgraph deserialize [deserializeDecimalSumState]
+    dOffsets["strings.offsets()"]
+    dDispatch["type_dispatcher(offsetsView.type())"]
+    dUnpack["unpackDecimalSumState&lt;OffsetT&gt;"]
+    dKernel["launchUnpackState&lt;OffsetT&gt;"]
+    dOffsets --> dDispatch --> dUnpack --> dKernel
+  end
+
+  subgraph serialize [serializeDecimalSumState]
+    sOffsets["offsetsCol mutable_view"]
+    sFillDispatch["type_dispatcher(offsetsView.type())"]
+    sPackDispatch["double dispatch sum + offset types"]
+    sFill["fillOffsetsForDecimalSumState&lt;OffsetT&gt;"]
+    sPack["packDecimalSumState&lt;SumT,OffsetT&gt;"]
+    sFillKernel["launchFillOffsets&lt;OffsetT&gt;"]
+    sPackKernel["launchPackState&lt;SumT,OffsetT&gt;"]
+    sOffsets --> sFillDispatch --> sFill --> sFillKernel
+    sOffsets --> sPackDispatch --> sPack --> sPackKernel
+  end
+
+  subgraph average [computeDecimalAverage]
+    aSum["sumCol.type()"]
+    aDispatch["type_dispatcher&lt;dispatch_storage_type&gt;"]
+    aAvg["averageRoundDecimalSum&lt;SumT&gt;"]
+    aKernel["launchAvgRound&lt;SumT&gt;"]
+    aSum --> aDispatch --> aAvg --> aKernel
+  end
+```
+
+### Dispatcher conventions
+
+- **Offset columns (INT32 / INT64):** use default `cudf::type_dispatcher` — `id_to_type` maps to `int32_t` / `int64_t`, matching `column_view::data<T>()`.
+- **Decimal sum columns (DECIMAL64 / DECIMAL128):** use `cudf::type_dispatcher<cudf::dispatch_storage_type>` so the dispatched `T` is the **device storage type** (`int64_t` / `__int128_t`), not the logical `numeric::decimal64` / `numeric::decimal128` type. This matches how `.data<int64_t>()` / `.data<__int128_t>()` are used today and is the libcudf-recommended pattern when operating on raw column buffers.
+- **Pack (two dimensions):** use **nested dispatch** — outer `type_dispatcher<dispatch_storage_type>` on `sumCol.type()`, inner `type_dispatcher` on `offsetsView.type()` — because `double_type_dispatcher` accepts only one `IdTypeMap` and we need `dispatch_storage_type` for sums but default `id_to_type` semantics also work for offsets via `dispatch_storage_type` (non-decimal types pass through unchanged). Alternatively, a single `double_type_dispatcher<cudf::dispatch_storage_type>` on `(sumCol.type(), offsetsView.type())` is equivalent and slightly cleaner if preferred during implementation.
+
+Add `#include <cudf/utilities/type_dispatcher.hpp>` **before** [`CudfNoDefaults.h`](velox/experimental/cudf/CudfNoDefaults.h) in [`DecimalAggregationState.cpp`](velox/experimental/cudf/exec/DecimalAggregationState.cpp).
+
+All call-site functors live in an anonymous namespace in that file.
+
+---
+
+## Part A: Unpack refactor
+
+### A1. Retype `detail::unpackDecimalSumState` (header + .cu)
+
+In [`DecimalAggregationDevice.h`](velox/experimental/cudf/exec/DecimalAggregationDevice.h):
+
+```cpp
+template <typename OffsetT>
+void unpackDecimalSumState(
+    const OffsetT* offsets,
+    const uint8_t* chars,
+    __int128_t* sums,
+    int64_t* counts,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+```
+
+In [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu):
+
+- Collapse the `if (offsets64)` branch into the template body calling `launchUnpackState` with `cuda::std::span<const OffsetT>{offsets, n}`.
+- Explicit instantiations: `int32_t`, `int64_t`.
+
+### A2. Dispatch at deserialize call site (line 109)
+
+```cpp
+struct UnpackDecimalSumStateDispatcher {
+  const cudf::column_view& offsetsView;
+  const uint8_t* charsPtr;
+  __int128_t* sums;
+  int64_t* counts;
+  cudf::size_type numRows;
+  rmm::cuda_stream_view stream;
+
+  template <typename OffsetT>
+  void operator()() {
+    detail::unpackDecimalSumState(
+        offsetsView.data<OffsetT>(),
+        charsPtr,
+        sums,
+        counts,
+        numRows,
+        stream);
+  }
+};
+
+auto const offsetsType = offsetsView.type().id();
+VELOX_CHECK(
+    offsetsType == cudf::type_id::INT32 || offsetsType == cudf::type_id::INT64,
+    "Decimal sum state requires INT32 or INT64 offsets (offset type is {})",
+    cudf::type_to_name(offsetsView.type()));
+
+cudf::type_dispatcher(
+    offsetsView.type(),
+    UnpackDecimalSumStateDispatcher{...});
+```
+
+---
+
+## Part B: Pack + fill refactor
+
+### B1. Retype device helpers (header + .cu)
+
+**`fillOffsetsForDecimalSumState`:**
+
+```cpp
+template <typename OffsetT>
+void fillOffsetsForDecimalSumState(
+    OffsetT* offsetsMutable,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+```
+
+**`packDecimalSumState`** — fully typed on both sum and offset dimensions; remove `cudf::type_id sumType`, `const void* sumPtr`, `bool use64BitOffsets`, and `const void* offsetsPtr`:
+
+```cpp
+template <typename SumT, typename OffsetT>
+void packDecimalSumState(
+    const SumT* sums,
+    const int64_t* counts,
+    const OffsetT* offsets,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+```
+
+In [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu):
+
+- `fillOffsetsForDecimalSumState`: call `launchFillOffsets` with `cuda::std::span<OffsetT>{offsetsMutable, n}`.
+- `packDecimalSumState`: call `launchPackState` directly with typed spans — no inner `sumType` or `use64BitOffsets` branches.
+- Explicit instantiations for all four `(SumT, OffsetT)` pairs: `(int64_t, int32_t)`, `(int64_t, int64_t)`, `(__int128_t, int32_t)`, `(__int128_t, int64_t)`.
+
+### B2. Dispatch at serialize call sites (lines 187 and 207)
+
+**Fill** — single dispatch on offset type:
+
+```cpp
+struct FillOffsetsForDecimalSumStateDispatcher {
+  cudf::column_view offsetsView;
+  cudf::size_type rowCount;
+  rmm::cuda_stream_view stream;
+
+  template <typename OffsetT>
+  void operator()() {
+    detail::fillOffsetsForDecimalSumState(
+        offsetsView.data<OffsetT>(), rowCount, stream);
+  }
+};
+
+cudf::type_dispatcher(
+    offsetsView.type(),
+    FillOffsetsForDecimalSumStateDispatcher{offsetsView, rowCount, stream});
+```
+
+**Pack** — nested dispatch on sum then offset types (eliminates `sumType`, `sumPtr`, `offsetsPtr`):
+
+```cpp
+struct PackDecimalSumStateDispatcher {
+  const cudf::column_view& sumCol;
+  const int64_t* countPtr;
+  cudf::column_view offsetsView;
+  uint8_t* charsPtr;
+  cudf::size_type rowCount;
+  rmm::cuda_stream_view stream;
+
+  template <typename SumT>
+  void operator()() {
+    struct PackWithOffset {
+      const cudf::column_view& sumCol;
+      const int64_t* countPtr;
+      cudf::column_view offsetsView;
+      uint8_t* charsPtr;
+      cudf::size_type rowCount;
+      rmm::cuda_stream_view stream;
+
+      template <typename OffsetT>
+      void operator()() {
+        detail::packDecimalSumState(
+            sumCol.data<SumT>(),
+            countPtr,
+            offsetsView.data<OffsetT>(),
+            charsPtr,
+            rowCount,
+            stream);
+      }
+    };
+    cudf::type_dispatcher(
+        offsetsView.type(),
+        PackWithOffset{sumCol, countPtr, offsetsView, charsPtr, rowCount, stream});
+  }
+};
+
+auto const sumType = sumCol.type().id();
+VELOX_CHECK(
+    sumType == cudf::type_id::DECIMAL64 ||
+        sumType == cudf::type_id::DECIMAL128,
+    "Unsupported decimal sum column type (type is {})",
+    cudf::type_to_name(sumCol.type()));
+
+cudf::type_dispatcher<cudf::dispatch_storage_type>(
+    sumCol.type(),
+    PackDecimalSumStateDispatcher{
+        sumCol,
+        countCol.data<int64_t>(),
+        offsetsView,
+        charsPtr,
+        rowCount,
+        stream});
+```
+
+**Alternative (equivalent):** replace the nested struct with a single call:
+
+```cpp
+cudf::double_type_dispatcher<cudf::dispatch_storage_type>(
+    sumCol.type(),
+    offsetsView.type(),
+    PackDecimalSumStateDoubleDispatcher{...});  // operator()<SumT, OffsetT>()
+```
+
+Pick whichever reads cleaner during implementation; both eliminate all `void*` casts in the serialize path.
+
+---
+
+## Part C: Average refactor
+
+### C1. Retype `detail::averageRoundDecimalSum` (header + .cu)
+
+Replace `cudf::type_id sumType, const void* sums, void* out` with:
+
+```cpp
+template <typename SumT>
+void averageRoundDecimalSum(
+    const SumT* sums,
+    const int64_t* counts,
+    SumT* out,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+```
+
+In [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu):
+
+- Template body calls `launchAvgRound` with `cuda::std::span<const SumT>{sums, n}` and `cuda::std::span<SumT>{out, n}`.
+- Explicit instantiations: `int64_t`, `__int128_t`.
+
+### C2. Dispatch at average call site (lines 257–265)
+
+Replace the `sumType` / `sumsPtr` / `outPtr` setup with:
+
+```cpp
+struct AverageRoundDecimalSumDispatcher {
+  const cudf::column_view& sumCol;
+  const int64_t* countPtr;
+  cudf::mutable_column_view outView;
+  cudf::size_type rowCount;
+  rmm::cuda_stream_view stream;
+
+  template <typename SumT>
+  void operator()() {
+    detail::averageRoundDecimalSum(
+        sumCol.data<SumT>(),
+        countPtr,
+        outView.data<SumT>(),
+        rowCount,
+        stream);
+  }
+};
+
+cudf::type_dispatcher<cudf::dispatch_storage_type>(
+    sumCol.type(),
+    AverageRoundDecimalSumDispatcher{
+        sumCol,
+        countCol.data<int64_t>(),
+        out->mutable_view(),
+        rowCount,
+        stream});
+```
+
+The existing `VELOX_CHECK` for DECIMAL64/DECIMAL128 at the top of `computeDecimalAverage` (lines 235–239) remains; it guards dispatch the same way as today.
+
+---
+
+## Verification
+
+Run existing coverage in [`DecimalAggregationTest.cpp`](velox/experimental/cudf/tests/DecimalAggregationTest.cpp):
+
+**Sum state (Parts A + B):**
+- `decimalDeserializeSumState*` (INT32 offsets)
+- `decimalSerializeSumState*` / `decimalSumStateRoundTrip*`
+- `decimalSumStateRoundTripUsesInt64Offsets` / `decimalSerializeSumStateUsesInt64OffsetsWhenEnabled` (INT64 offsets)
+
+**Average (Part C):**
+- All `computeDecimalAverage` tests (lines ~1245–1448): DECIMAL64 and DECIMAL128 sums, null/zero-count edge cases, rounding behavior
+
+No new tests are strictly required unless you want explicit invalid-type regression tests.
+
+Target test command (adjust to your build setup):
+
+```bash
+./velox/experimental/cudf/tests/cudf_test --gtest_filter='*decimal*SumState*:*decimal*Average*:*Decimal*'
+```
diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
index a24c2b71171..e1a68622b09 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
@@ -234,133 +234,131 @@ std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMaskImpl(
 
 namespace detail {
 
+template <typename OffsetT>
 void fillOffsetsForDecimalSumState(
-    bool use64BitOffsets,
-    void* offsetsMutable,
-    int32_t numRows,
+    OffsetT* offsetsMutable,
+    cudf::size_type numRows,
     rmm::cuda_stream_view stream) {
   // The offsets buffer holds numRows + 1 entries.
   auto const n = static_cast<size_t>(numRows) + 1;
-  if (use64BitOffsets) {
-    launchFillOffsets(
-        cuda::std::span<int64_t>{static_cast<int64_t*>(offsetsMutable), n},
-        stream);
-  } else {
-    launchFillOffsets(
-        cuda::std::span<int32_t>{static_cast<int32_t*>(offsetsMutable), n},
-        stream);
-  }
+  launchFillOffsets(cuda::std::span<OffsetT>{offsetsMutable, n}, stream);
 }
 
+template <typename SumT, typename OffsetT>
 void packDecimalSumState(
-    cudf::type_id sumType,
-    bool use64BitOffsets,
-    const void* sumPtr,
-    const int64_t* countPtr,
-    const void* offsetsPtr,
+    const SumT* sums,
+    const int64_t* counts,
+    const OffsetT* offsets,
     uint8_t* chars,
-    int32_t numRows,
+    cudf::size_type numRows,
     rmm::cuda_stream_view stream) {
   auto const n = static_cast<size_t>(numRows);
-  cuda::std::span<const int64_t> counts{countPtr, n};
-  if (use64BitOffsets) {
-    cuda::std::span<const int64_t> offsets{
-        static_cast<const int64_t*>(offsetsPtr), n};
-    if (sumType == cudf::type_id::DECIMAL64) {
-      launchPackState(
-          cuda::std::span<const int64_t>{
-              static_cast<const int64_t*>(sumPtr), n},
-          counts,
-          offsets,
-          chars,
-          stream);
-    } else {
-      launchPackState(
-          cuda::std::span<const __int128_t>{
-              static_cast<const __int128_t*>(sumPtr), n},
-          counts,
-          offsets,
-          chars,
-          stream);
-    }
-  } else {
-    cuda::std::span<const int32_t> offsets{
-        static_cast<const int32_t*>(offsetsPtr), n};
-    if (sumType == cudf::type_id::DECIMAL64) {
-      launchPackState(
-          cuda::std::span<const int64_t>{
-              static_cast<const int64_t*>(sumPtr), n},
-          counts,
-          offsets,
-          chars,
-          stream);
-    } else {
-      launchPackState(
-          cuda::std::span<const __int128_t>{
-              static_cast<const __int128_t*>(sumPtr), n},
-          counts,
-          offsets,
-          chars,
-          stream);
-    }
-  }
+  launchPackState(
+      cuda::std::span<const SumT>{sums, n},
+      cuda::std::span<const int64_t>{counts, n},
+      cuda::std::span<const OffsetT>{offsets, n},
+      chars,
+      stream);
 }
 
+template <typename OffsetT>
 void unpackDecimalSumState(
-    bool offsets64,
-    const void* offsetsPtr,
+    const OffsetT* offsets,
     const uint8_t* chars,
     __int128_t* sums,
     int64_t* counts,
-    int32_t numRows,
+    cudf::size_type numRows,
     rmm::cuda_stream_view stream) {
   auto const n = static_cast<size_t>(numRows);
-  cuda::std::span<__int128_t> sumsSpan{sums, n};
-  cuda::std::span<int64_t> countsSpan{counts, n};
-  if (offsets64) {
-    launchUnpackState(
-        cuda::std::span<const int64_t>{
-            static_cast<const int64_t*>(offsetsPtr), n},
-        chars,
-        sumsSpan,
-        countsSpan,
-        stream);
-  } else {
-    launchUnpackState(
-        cuda::std::span<const int32_t>{
-            static_cast<const int32_t*>(offsetsPtr), n},
-        chars,
-        sumsSpan,
-        countsSpan,
-        stream);
-  }
+  launchUnpackState(
+      cuda::std::span<const OffsetT>{offsets, n},
+      chars,
+      cuda::std::span<__int128_t>{sums, n},
+      cuda::std::span<int64_t>{counts, n},
+      stream);
 }
 
+template <typename SumT>
 void averageRoundDecimalSum(
-    cudf::type_id sumType,
-    const void* sums,
+    const SumT* sums,
     const int64_t* counts,
-    void* out,
-    int32_t numRows,
+    SumT* out,
+    cudf::size_type numRows,
     rmm::cuda_stream_view stream) {
   auto const n = static_cast<size_t>(numRows);
-  cuda::std::span<const int64_t> countsSpan{counts, n};
-  if (sumType == cudf::type_id::DECIMAL64) {
-    launchAvgRound(
-        cuda::std::span<const int64_t>{static_cast<const int64_t*>(sums), n},
-        countsSpan,
-        cuda::std::span<int64_t>{static_cast<int64_t*>(out), n},
-        stream);
-  } else {
-    launchAvgRound(
-        cuda::std::span<const __int128_t>{
-            static_cast<const __int128_t*>(sums), n},
-        countsSpan,
-        cuda::std::span<__int128_t>{static_cast<__int128_t*>(out), n},
-        stream);
-  }
+  launchAvgRound(
+      cuda::std::span<const SumT>{sums, n},
+      cuda::std::span<const int64_t>{counts, n},
+      cuda::std::span<SumT>{out, n},
+      stream);
 }
 
+template void fillOffsetsForDecimalSumState<int32_t>(
+    int32_t* offsetsMutable,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+template void fillOffsetsForDecimalSumState<int64_t>(
+    int64_t* offsetsMutable,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+
+template void packDecimalSumState<int64_t, int32_t>(
+    const int64_t* sums,
+    const int64_t* counts,
+    const int32_t* offsets,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+template void packDecimalSumState<int64_t, int64_t>(
+    const int64_t* sums,
+    const int64_t* counts,
+    const int64_t* offsets,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+template void packDecimalSumState<__int128_t, int32_t>(
+    const __int128_t* sums,
+    const int64_t* counts,
+    const int32_t* offsets,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+template void packDecimalSumState<__int128_t, int64_t>(
+    const __int128_t* sums,
+    const int64_t* counts,
+    const int64_t* offsets,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+
+template void unpackDecimalSumState<int32_t>(
+    const int32_t* offsets,
+    const uint8_t* chars,
+    __int128_t* sums,
+    int64_t* counts,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+template void unpackDecimalSumState<int64_t>(
+    const int64_t* offsets,
+    const uint8_t* chars,
+    __int128_t* sums,
+    int64_t* counts,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+
+template void averageRoundDecimalSum<int64_t>(
+    const int64_t* sums,
+    const int64_t* counts,
+    int64_t* out,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+template void averageRoundDecimalSum<__int128_t>(
+    const __int128_t* sums,
+    const int64_t* counts,
+    __int128_t* out,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+
 std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMask(
     const cudf::column_view& sumCol,
     const cudf::column_view& countCol,
diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.h b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
index 25411ca1a2b..2a7b657f0a1 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.h
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
@@ -35,14 +35,13 @@ constexpr size_t kDecimalSumStateSize = 32;
 /**
  * Writes strings-style prefix offsets: offset[i] == i * kDecimalSumStateSize.
  *
- * @param use64BitOffsets whether offsets are INT64 (else INT32).
  * @param offsetsMutable output buffer of numRows + 1 offset elements.
  * @param numRows number of payload rows.
  * @param stream CUDA stream for the launch.
  */
+template <typename OffsetT>
 void fillOffsetsForDecimalSumState(
-    bool use64BitOffsets,
-    void* offsetsMutable,
+    OffsetT* offsetsMutable,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
 
@@ -50,21 +49,18 @@ void fillOffsetsForDecimalSumState(
  * Encodes each row's partial sum and count into the fixed-width device layout
  * used for VARBINARY interchange.
  *
- * @param sumType element type of sumPtr (DECIMAL64 or DECIMAL128).
- * @param use64BitOffsets whether offsetsPtr is INT64 (else INT32).
- * @param sumPtr per-row sums.
- * @param countPtr per-row int64 counts.
- * @param offsetsPtr per-row byte offsets into chars.
+ * @param sums per-row sums.
+ * @param counts per-row int64 counts.
+ * @param offsets per-row byte offsets into chars.
  * @param chars output payload buffer.
  * @param numRows number of rows.
  * @param stream CUDA stream for the launch.
  */
+template <typename SumT, typename OffsetT>
 void packDecimalSumState(
-    cudf::type_id sumType,
-    bool use64BitOffsets,
-    const void* sumPtr,
-    const int64_t* countPtr,
-    const void* offsetsPtr,
+    const SumT* sums,
+    const int64_t* counts,
+    const OffsetT* offsets,
     uint8_t* chars,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
@@ -72,17 +68,16 @@ void packDecimalSumState(
 /**
  * Inverse of packDecimalSumState.
  *
- * @param offsets64 whether offsetsPtr is INT64 (else INT32).
- * @param offsetsPtr per-row byte offsets into chars.
+ * @param offsets per-row byte offsets into chars.
  * @param chars packed payload buffer.
  * @param sums output per-row DECIMAL128 sums.
  * @param counts output per-row counts.
  * @param numRows number of rows.
  * @param stream CUDA stream for the launch.
  */
+template <typename OffsetT>
 void unpackDecimalSumState(
-    bool offsets64,
-    const void* offsetsPtr,
+    const OffsetT* offsets,
     const uint8_t* chars,
     __int128_t* sums,
     int64_t* counts,
@@ -93,18 +88,17 @@ void unpackDecimalSumState(
  * Per-row half-up integer divide of sum by count; count == 0 writes zero
  * (validity is applied separately).
  *
- * @param sumType element type of sums/out (DECIMAL64 or DECIMAL128).
  * @param sums per-row sums.
  * @param counts per-row counts.
  * @param out output per-row averages.
  * @param numRows number of rows.
  * @param stream CUDA stream for the launch.
  */
+template <typename SumT>
 void averageRoundDecimalSum(
-    cudf::type_id sumType,
-    const void* sums,
+    const SumT* sums,
     const int64_t* counts,
-    void* out,
+    SumT* out,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
 
diff --git a/velox/experimental/cudf/exec/DecimalAggregationState.cpp b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
index a0756d82a7c..446483f56fd 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationState.cpp
+++ b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-#include "velox/experimental/cudf/CudfNoDefaults.h"
 #include "velox/experimental/cudf/exec/DecimalAggregationDevice.h"
 #include "velox/experimental/cudf/exec/DecimalAggregationState.h"
 #include "velox/experimental/cudf/exec/GpuResources.h"
@@ -25,10 +24,86 @@
 #include <cudf/null_mask.hpp>
 #include <cudf/strings/strings_column_view.hpp>
 #include <cudf/strings/utilities.hpp>
+#include <cudf/utilities/type_dispatcher.hpp>
 
 #include <limits>
 
+#include "velox/experimental/cudf/CudfNoDefaults.h"
+
 namespace facebook::velox::cudf_velox {
+namespace {
+
+struct UnpackDecimalSumStateDispatcher {
+  const cudf::column_view& offsetsView;
+  const uint8_t* charsPtr;
+  __int128_t* sums;
+  int64_t* counts;
+  cudf::size_type numRows;
+  rmm::cuda_stream_view stream;
+
+  template <typename OffsetT>
+  void operator()() {
+    detail::unpackDecimalSumState(
+        offsetsView.data<OffsetT>(),
+        charsPtr,
+        sums,
+        counts,
+        numRows,
+        stream);
+  }
+};
+
+struct FillOffsetsForDecimalSumStateDispatcher {
+  cudf::mutable_column_view offsetsView;
+  cudf::size_type rowCount;
+  rmm::cuda_stream_view stream;
+
+  template <typename OffsetT>
+  void operator()() {
+    detail::fillOffsetsForDecimalSumState(
+        offsetsView.data<OffsetT>(), rowCount, stream);
+  }
+};
+
+struct PackDecimalSumStateDispatcher {
+  const cudf::column_view& sumCol;
+  const int64_t* countPtr;
+  cudf::column_view offsetsView;
+  uint8_t* charsPtr;
+  cudf::size_type rowCount;
+  rmm::cuda_stream_view stream;
+
+  template <typename SumT, typename OffsetT>
+  void operator()() {
+    detail::packDecimalSumState(
+        sumCol.data<SumT>(),
+        countPtr,
+        offsetsView.data<OffsetT>(),
+        charsPtr,
+        rowCount,
+        stream);
+  }
+};
+
+struct AverageRoundDecimalSumDispatcher {
+  const cudf::column_view& sumCol;
+  const int64_t* countPtr;
+  cudf::mutable_column_view outView;
+  cudf::size_type rowCount;
+  rmm::cuda_stream_view stream;
+
+  template <typename SumT>
+  void operator()() {
+    detail::averageRoundDecimalSum(
+        sumCol.data<SumT>(),
+        countPtr,
+        outView.data<SumT>(),
+        rowCount,
+        stream);
+  }
+};
+
+} // namespace
 
 DecimalSumStateColumns deserializeDecimalSumState(
     const cudf::column_view& stateCol,
@@ -81,7 +156,6 @@ DecimalSumStateColumns deserializeDecimalSumState(
   cudf::strings_column_view strings(stateCol);
 
   auto offsetsView = strings.offsets();
-  auto offsetsType = offsetsView.type().id();
   auto charsPtr = reinterpret_cast<const uint8_t*>(strings.chars_begin(stream));
 
   auto sumCol = cudf::make_fixed_width_column(
@@ -101,20 +175,21 @@ DecimalSumStateColumns deserializeDecimalSumState(
   auto countView = countCol->mutable_view();
 
   // numRows is guaranteed positive here
-  const bool offsets64 = (offsetsType == cudf::type_id::INT64);
+  auto const offsetsType = offsetsView.type().id();
   VELOX_CHECK(
-      offsets64 || offsetsType == cudf::type_id::INT32,
+      offsetsType == cudf::type_id::INT32 ||
+          offsetsType == cudf::type_id::INT64,
       "Decimal sum state requires INT32 or INT64 offsets (offset type is {})",
-      static_cast<int>(offsetsType));
-  detail::unpackDecimalSumState(
-      offsets64,
-      offsets64 ? static_cast<const void*>(offsetsView.data<int64_t>())
-                : static_cast<const void*>(offsetsView.data<int32_t>()),
-      charsPtr,
-      sumView.data<__int128_t>(),
-      countView.data<int64_t>(),
-      numRows,
-      stream);
+      cudf::type_to_name(offsetsView.type()));
+  cudf::type_dispatcher(
+      offsetsView.type(),
+      UnpackDecimalSumStateDispatcher{
+          offsetsView,
+          charsPtr,
+          sumView.data<__int128_t>(),
+          countView.data<int64_t>(),
+          numRows,
+          stream});
 
   if (stateCol.nullable()) {
     auto nullMask = cudf::copy_bitmask(stateCol, stream, mr);
@@ -183,35 +258,28 @@ std::unique_ptr<cudf::column> serializeDecimalSumState(
   rmm::device_buffer charsBuf(
       static_cast<size_t>(numRows) * detail::kDecimalSumStateSize, stream, mr);
 
-  detail::fillOffsetsForDecimalSumState(
-      useLargeOffsets,
-      useLargeOffsets ? static_cast<void*>(offsetsView.data<int64_t>())
-                      : static_cast<void*>(offsetsView.data<int32_t>()),
-      rowCount,
-      stream);
-
   auto charsPtr = reinterpret_cast<uint8_t*>(charsBuf.data());
-  const void* offsetsPtr = useLargeOffsets
-      ? static_cast<const void*>(offsetsView.data<int64_t>())
-      : static_cast<const void*>(offsetsView.data<int32_t>());
-  const auto sumType = sumCol.type().id();
+  cudf::type_dispatcher(
+      offsetsView.type(),
+      FillOffsetsForDecimalSumStateDispatcher{
+          offsetsView, rowCount, stream});
+
+  auto const sumType = sumCol.type().id();
   VELOX_CHECK(
       sumType == cudf::type_id::DECIMAL64 ||
           sumType == cudf::type_id::DECIMAL128,
       "Unsupported decimal sum column type (type is {})",
       cudf::type_to_name(sumCol.type()));
-  const void* sumPtr = sumType == cudf::type_id::DECIMAL64
-      ? static_cast<const void*>(sumCol.data<int64_t>())
-      : static_cast<const void*>(sumCol.data<__int128_t>());
-  detail::packDecimalSumState(
-      sumType,
-      useLargeOffsets,
-      sumPtr,
-      countCol.data<int64_t>(),
-      offsetsPtr,
-      charsPtr,
-      rowCount,
-      stream);
+  cudf::double_type_dispatcher<cudf::dispatch_storage_type>(
+      sumCol.type(),
+      offsetsView.type(),
+      PackDecimalSumStateDispatcher{
+          sumCol,
+          countCol.data<int64_t>(),
+          offsetsView,
+          charsPtr,
+          rowCount,
+          stream});
 
   auto [nullMask, nullCount] =
       detail::buildStateValidityMask(sumCol, countCol, stream, mr);
@@ -250,15 +318,14 @@ std::unique_ptr<cudf::column> computeDecimalAverage(
 
   if (numRows > 0) {
     auto const rowCount = static_cast<int32_t>(numRows);
-    const auto sumType = sumCol.type().id();
-    const void* sumsPtr = sumType == cudf::type_id::DECIMAL64
-        ? static_cast<const void*>(sumCol.data<int64_t>())
-        : static_cast<const void*>(sumCol.data<__int128_t>());
-    void* outPtr = sumType == cudf::type_id::DECIMAL64
-        ? static_cast<void*>(out->mutable_view().data<int64_t>())
-        : static_cast<void*>(out->mutable_view().data<__int128_t>());
-    detail::averageRoundDecimalSum(
-        sumType, sumsPtr, countCol.data<int64_t>(), outPtr, rowCount, stream);
+    cudf::type_dispatcher<cudf::dispatch_storage_type>(
+        sumCol.type(),
+        AverageRoundDecimalSumDispatcher{
+            sumCol,
+            countCol.data<int64_t>(),
+            out->mutable_view(),
+            rowCount,
+            stream});
   }
 
   auto [nullMask, nullCount] =

From 006e074f4fef6090ad8eef61ba33a485f14c3433 Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Wed, 10 Jun 2026 14:10:43 -0700
Subject: [PATCH 2/8] Use (apparently) standard if-constexpr to limit the
 subset of dispatched types

---
 .../cudf/exec/DecimalAggregationState.cpp     | 59 ++++++++++++-------
 1 file changed, 37 insertions(+), 22 deletions(-)

diff --git a/velox/experimental/cudf/exec/DecimalAggregationState.cpp b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
index 446483f56fd..2be0c1af696 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationState.cpp
+++ b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
@@ -27,6 +27,7 @@
 #include <cudf/utilities/type_dispatcher.hpp>
 
 #include <limits>
+#include <type_traits>
 
 #include "velox/experimental/cudf/CudfNoDefaults.h"
 
@@ -43,13 +44,16 @@ struct UnpackDecimalSumStateDispatcher {
 
   template <typename OffsetT>
   void operator()() {
-    detail::unpackDecimalSumState(
-        offsetsView.data<OffsetT>(),
-        charsPtr,
-        sums,
-        counts,
-        numRows,
-        stream);
+    if constexpr (std::is_same_v<OffsetT, int32_t> ||
+                  std::is_same_v<OffsetT, int64_t>) {
+      detail::unpackDecimalSumState(
+          offsetsView.data<OffsetT>(),
+          charsPtr,
+          sums,
+          counts,
+          numRows,
+          stream);
+    }
   }
 };
 
@@ -60,8 +64,11 @@ struct FillOffsetsForDecimalSumStateDispatcher {
 
   template <typename OffsetT>
   void operator()() {
-    detail::fillOffsetsForDecimalSumState(
-        offsetsView.data<OffsetT>(), rowCount, stream);
+    if constexpr (std::is_same_v<OffsetT, int32_t> ||
+                  std::is_same_v<OffsetT, int64_t>) {
+      detail::fillOffsetsForDecimalSumState(
+          offsetsView.data<OffsetT>(), rowCount, stream);
+    }
   }
 };
 
@@ -75,13 +82,18 @@ struct PackDecimalSumStateDispatcher {
 
   template <typename SumT, typename OffsetT>
   void operator()() {
-    detail::packDecimalSumState(
-        sumCol.data<SumT>(),
-        countPtr,
-        offsetsView.data<OffsetT>(),
-        charsPtr,
-        rowCount,
-        stream);
+    if constexpr ((std::is_same_v<SumT, int64_t> ||
+                   std::is_same_v<SumT, __int128_t>) &&
+                  (std::is_same_v<OffsetT, int32_t> ||
+                   std::is_same_v<OffsetT, int64_t>)) {
+      detail::packDecimalSumState(
+          sumCol.data<SumT>(),
+          countPtr,
+          offsetsView.data<OffsetT>(),
+          charsPtr,
+          rowCount,
+          stream);
+    }
   }
 };
 
@@ -94,12 +106,15 @@ struct AverageRoundDecimalSumDispatcher {
 
   template <typename SumT>
   void operator()() {
-    detail::averageRoundDecimalSum(
-        sumCol.data<SumT>(),
-        countPtr,
-        outView.data<SumT>(),
-        rowCount,
-        stream);
+    if constexpr (std::is_same_v<SumT, int64_t> ||
+                  std::is_same_v<SumT, __int128_t>) {
+      detail::averageRoundDecimalSum(
+          sumCol.data<SumT>(),
+          countPtr,
+          outView.data<SumT>(),
+          rowCount,
+          stream);
+    }
   }
 };
 

From e2025f713865ad87f6591bd8b86bac2c3f951c65 Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Thu, 11 Jun 2026 16:34:39 -0700
Subject: [PATCH 3/8] Second attempt

---
 .../cudf/exec/DecimalAggregationDevice.cu     | 107 ++++----
 .../cudf/exec/DecimalAggregationDevice.h      | 229 +++++++++++++++---
 .../cudf/exec/DecimalAggregationState.cpp     | 142 +++--------
 3 files changed, 273 insertions(+), 205 deletions(-)

diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
index e1a68622b09..fec07223dad 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
@@ -235,51 +235,50 @@ std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMaskImpl(
 namespace detail {
 
 template <typename OffsetT>
-void fillOffsetsForDecimalSumState(
+void launchFillOffsetsForDecimalSumState(
     OffsetT* offsetsMutable,
-    cudf::size_type numRows,
+    size_t offsetCount,
     rmm::cuda_stream_view stream) {
-  // The offsets buffer holds numRows + 1 entries.
-  auto const n = static_cast<size_t>(numRows) + 1;
-  launchFillOffsets(cuda::std::span<OffsetT>{offsetsMutable, n}, stream);
+  launchFillOffsets(
+      cuda::std::span<OffsetT>{offsetsMutable, offsetCount}, stream);
 }
 
-template <typename SumT, typename OffsetT>
-void packDecimalSumState(
-    const SumT* sums,
-    const int64_t* counts,
+template <typename OffsetT>
+void launchUnpackDecimalSumState(
     const OffsetT* offsets,
-    uint8_t* chars,
+    const uint8_t* chars,
+    __int128_t* sums,
+    int64_t* counts,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) {
   auto const n = static_cast<size_t>(numRows);
-  launchPackState(
-      cuda::std::span<const SumT>{sums, n},
-      cuda::std::span<const int64_t>{counts, n},
+  launchUnpackState(
       cuda::std::span<const OffsetT>{offsets, n},
       chars,
+      cuda::std::span<__int128_t>{sums, n},
+      cuda::std::span<int64_t>{counts, n},
       stream);
 }
 
-template <typename OffsetT>
-void unpackDecimalSumState(
+template <typename SumT, typename OffsetT>
+void launchPackDecimalSumState(
+    const SumT* sums,
+    const int64_t* counts,
     const OffsetT* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
+    uint8_t* chars,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) {
   auto const n = static_cast<size_t>(numRows);
-  launchUnpackState(
+  launchPackState(
+      cuda::std::span<const SumT>{sums, n},
+      cuda::std::span<const int64_t>{counts, n},
       cuda::std::span<const OffsetT>{offsets, n},
       chars,
-      cuda::std::span<__int128_t>{sums, n},
-      cuda::std::span<int64_t>{counts, n},
       stream);
 }
 
 template <typename SumT>
-void averageRoundDecimalSum(
+void launchAverageRoundDecimalSum(
     const SumT* sums,
     const int64_t* counts,
     SumT* out,
@@ -293,37 +292,60 @@ void averageRoundDecimalSum(
       stream);
 }
 
-template void fillOffsetsForDecimalSumState<int32_t>(
+std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMask(
+    const cudf::column_view& sumCol,
+    const cudf::column_view& countCol,
+    rmm::cuda_stream_view stream,
+    rmm::device_async_resource_ref mr) {
+  return buildStateValidityMaskImpl(sumCol, countCol, stream, mr);
+}
+
+template void launchFillOffsetsForDecimalSumState<int32_t>(
     int32_t* offsetsMutable,
-    cudf::size_type numRows,
+    size_t offsetCount,
     rmm::cuda_stream_view stream);
-template void fillOffsetsForDecimalSumState<int64_t>(
+template void launchFillOffsetsForDecimalSumState<int64_t>(
     int64_t* offsetsMutable,
+    size_t offsetCount,
+    rmm::cuda_stream_view stream);
+
+template void launchUnpackDecimalSumState<int32_t>(
+    const int32_t* offsets,
+    const uint8_t* chars,
+    __int128_t* sums,
+    int64_t* counts,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+template void launchUnpackDecimalSumState<int64_t>(
+    const int64_t* offsets,
+    const uint8_t* chars,
+    __int128_t* sums,
+    int64_t* counts,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
 
-template void packDecimalSumState<int64_t, int32_t>(
+template void launchPackDecimalSumState<int64_t, int32_t>(
     const int64_t* sums,
     const int64_t* counts,
     const int32_t* offsets,
     uint8_t* chars,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
-template void packDecimalSumState<int64_t, int64_t>(
+template void launchPackDecimalSumState<int64_t, int64_t>(
     const int64_t* sums,
     const int64_t* counts,
     const int64_t* offsets,
     uint8_t* chars,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
-template void packDecimalSumState<__int128_t, int32_t>(
+template void launchPackDecimalSumState<__int128_t, int32_t>(
     const __int128_t* sums,
     const int64_t* counts,
     const int32_t* offsets,
     uint8_t* chars,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
-template void packDecimalSumState<__int128_t, int64_t>(
+template void launchPackDecimalSumState<__int128_t, int64_t>(
     const __int128_t* sums,
     const int64_t* counts,
     const int64_t* offsets,
@@ -331,41 +353,18 @@ template void packDecimalSumState<__int128_t, int64_t>(
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
 
-template void unpackDecimalSumState<int32_t>(
-    const int32_t* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-template void unpackDecimalSumState<int64_t>(
-    const int64_t* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-
-template void averageRoundDecimalSum<int64_t>(
+template void launchAverageRoundDecimalSum<int64_t>(
     const int64_t* sums,
     const int64_t* counts,
     int64_t* out,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
-template void averageRoundDecimalSum<__int128_t>(
+template void launchAverageRoundDecimalSum<__int128_t>(
     const __int128_t* sums,
     const int64_t* counts,
     __int128_t* out,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream);
 
-std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMask(
-    const cudf::column_view& sumCol,
-    const cudf::column_view& countCol,
-    rmm::cuda_stream_view stream,
-    rmm::device_async_resource_ref mr) {
-  return buildStateValidityMaskImpl(sumCol, countCol, stream, mr);
-}
-
 } // namespace detail
 } // namespace facebook::velox::cudf_velox
diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.h b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
index 2a7b657f0a1..b0e791da2f1 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.h
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
@@ -17,90 +17,247 @@
 
 #include <cudf/column/column_view.hpp>
 #include <cudf/types.hpp>
+#include <cudf/utilities/type_dispatcher.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
 #include <rmm/device_buffer.hpp>
 #include <rmm/resource_ref.hpp>
 
+#include <cstddef>
 #include <cstdint>
+#include <type_traits>
 #include <utility>
 
 namespace facebook::velox::cudf_velox::detail {
 
+template <typename T>
+inline constexpr bool isDecimalSumStorageType =
+    std::is_same_v<T, int64_t> || std::is_same_v<T, __int128_t>;
+
+template <typename T>
+inline constexpr bool isOffsetStorageType =
+    std::is_same_v<T, int32_t> || std::is_same_v<T, int64_t>;
+
 // Size in bytes of each row's packed decimal SUM intermediate state in the
 // strings payload (count, overflow placeholder, and 128-bit sum split into
 // words).
 constexpr size_t kDecimalSumStateSize = 32;
 
+template <typename OffsetT>
+void launchFillOffsetsForDecimalSumState(
+    OffsetT* offsetsMutable,
+    size_t offsetCount,
+    rmm::cuda_stream_view stream);
+
+template <typename OffsetT>
+void launchUnpackDecimalSumState(
+    const OffsetT* offsets,
+    const uint8_t* chars,
+    __int128_t* sums,
+    int64_t* counts,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+
+template <typename SumT, typename OffsetT>
+void launchPackDecimalSumState(
+    const SumT* sums,
+    const int64_t* counts,
+    const OffsetT* offsets,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+
+template <typename SumT>
+void launchAverageRoundDecimalSum(
+    const SumT* sums,
+    const int64_t* counts,
+    SumT* out,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream);
+
 /**
  * Writes strings-style prefix offsets: offset[i] == i * kDecimalSumStateSize.
  *
- * @param offsetsMutable output buffer of numRows + 1 offset elements.
+ * @param offsetsView output offsets column of numRows + 1 elements.
  * @param numRows number of payload rows.
  * @param stream CUDA stream for the launch.
  */
-template <typename OffsetT>
-void fillOffsetsForDecimalSumState(
-    OffsetT* offsetsMutable,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
+struct fillOffsetsForDecimalSumState {
+  template <
+      typename OffsetT,
+      std::enable_if_t<isOffsetStorageType<OffsetT>, int> = 0>
+  void operator()(
+      cudf::mutable_column_view offsetsView,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {
+    launchFillOffsetsForDecimalSumState<OffsetT>(
+        offsetsView.data<OffsetT>(),
+        static_cast<size_t>(numRows) + 1,
+        stream);
+  }
+
+  template <
+      typename OffsetT,
+      std::enable_if_t<!isOffsetStorageType<OffsetT>, int> = 0>
+  void operator()(
+      cudf::mutable_column_view offsetsView,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {}
+};
+
+template <typename SumT>
+struct packDecimalSumStateWithOffset {
+  const SumT* sums;
+  const int64_t* counts;
+  cudf::column_view offsetsView;
+  uint8_t* chars;
+  cudf::size_type numRows;
+  rmm::cuda_stream_view stream;
+
+  template <
+      typename OffsetT,
+      std::enable_if_t<isOffsetStorageType<OffsetT>, int> = 0>
+  void operator()() const {
+    launchPackDecimalSumState<SumT, OffsetT>(
+        sums,
+        counts,
+        offsetsView.data<OffsetT>(),
+        chars,
+        numRows,
+        stream);
+  }
+
+  template <
+      typename OffsetT,
+      std::enable_if_t<!isOffsetStorageType<OffsetT>, int> = 0>
+  void operator()() const {}
+};
 
 /**
  * Encodes each row's partial sum and count into the fixed-width device layout
  * used for VARBINARY interchange.
  *
- * @param sums per-row sums.
+ * @param sumCol per-row sums.
  * @param counts per-row int64 counts.
- * @param offsets per-row byte offsets into chars.
+ * @param offsetsView per-row byte offsets into chars.
  * @param chars output payload buffer.
  * @param numRows number of rows.
  * @param stream CUDA stream for the launch.
  */
-template <typename SumT, typename OffsetT>
-void packDecimalSumState(
-    const SumT* sums,
-    const int64_t* counts,
-    const OffsetT* offsets,
-    uint8_t* chars,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
+struct packDecimalSumState {
+  template <
+      typename SumT,
+      std::enable_if_t<isDecimalSumStorageType<SumT>, int> = 0>
+  void operator()(
+      cudf::column_view sumCol,
+      const int64_t* counts,
+      cudf::column_view offsetsView,
+      uint8_t* chars,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {
+    cudf::type_dispatcher(
+        offsetsView.type(),
+        packDecimalSumStateWithOffset<SumT>{
+            sumCol.data<SumT>(),
+            counts,
+            offsetsView,
+            chars,
+            numRows,
+            stream});
+  }
+
+  template <
+      typename SumT,
+      std::enable_if_t<!isDecimalSumStorageType<SumT>, int> = 0>
+  void operator()(
+      cudf::column_view sumCol,
+      const int64_t* counts,
+      cudf::column_view offsetsView,
+      uint8_t* chars,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {}
+};
 
 /**
  * Inverse of packDecimalSumState.
  *
- * @param offsets per-row byte offsets into chars.
+ * @param offsetsView per-row byte offsets into chars.
  * @param chars packed payload buffer.
- * @param sums output per-row DECIMAL128 sums.
- * @param counts output per-row counts.
+ * @param sumView output per-row DECIMAL128 sums.
+ * @param countView output per-row counts.
  * @param numRows number of rows.
  * @param stream CUDA stream for the launch.
  */
-template <typename OffsetT>
-void unpackDecimalSumState(
-    const OffsetT* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
+struct unpackDecimalSumState {
+  template <
+      typename OffsetT,
+      std::enable_if_t<isOffsetStorageType<OffsetT>, int> = 0>
+  void operator()(
+      cudf::column_view offsetsView,
+      const uint8_t* chars,
+      cudf::mutable_column_view sumView,
+      cudf::mutable_column_view countView,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {
+    launchUnpackDecimalSumState<OffsetT>(
+        offsetsView.data<OffsetT>(),
+        chars,
+        sumView.data<__int128_t>(),
+        countView.data<int64_t>(),
+        numRows,
+        stream);
+  }
+
+  template <
+      typename OffsetT,
+      std::enable_if_t<!isOffsetStorageType<OffsetT>, int> = 0>
+  void operator()(
+      cudf::column_view offsetsView,
+      const uint8_t* chars,
+      cudf::mutable_column_view sumView,
+      cudf::mutable_column_view countView,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {}
+};
 
 /**
  * Per-row half-up integer divide of sum by count; count == 0 writes zero
  * (validity is applied separately).
  *
- * @param sums per-row sums.
+ * @param sumCol per-row sums.
  * @param counts per-row counts.
- * @param out output per-row averages.
+ * @param outView output per-row averages.
  * @param numRows number of rows.
  * @param stream CUDA stream for the launch.
  */
-template <typename SumT>
-void averageRoundDecimalSum(
-    const SumT* sums,
-    const int64_t* counts,
-    SumT* out,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
+struct averageRoundDecimalSum {
+  template <
+      typename SumT,
+      std::enable_if_t<isDecimalSumStorageType<SumT>, int> = 0>
+  void operator()(
+      cudf::column_view sumCol,
+      const int64_t* counts,
+      cudf::mutable_column_view outView,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {
+    launchAverageRoundDecimalSum<SumT>(
+        sumCol.data<SumT>(),
+        counts,
+        outView.data<SumT>(),
+        numRows,
+        stream);
+  }
+
+  template <
+      typename SumT,
+      std::enable_if_t<!isDecimalSumStorageType<SumT>, int> = 0>
+  void operator()(
+      cudf::column_view sumCol,
+      const int64_t* counts,
+      cudf::mutable_column_view outView,
+      cudf::size_type numRows,
+      rmm::cuda_stream_view stream) const {}
+};
 
 /**
  * Builds a null mask for rows where sum and count are both valid and count is
diff --git a/velox/experimental/cudf/exec/DecimalAggregationState.cpp b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
index 2be0c1af696..dcfe8cee39e 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationState.cpp
+++ b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
@@ -27,98 +27,10 @@
 #include <cudf/utilities/type_dispatcher.hpp>
 
 #include <limits>
-#include <type_traits>
 
 #include "velox/experimental/cudf/CudfNoDefaults.h"
 
 namespace facebook::velox::cudf_velox {
-namespace {
-
-struct UnpackDecimalSumStateDispatcher {
-  const cudf::column_view& offsetsView;
-  const uint8_t* charsPtr;
-  __int128_t* sums;
-  int64_t* counts;
-  cudf::size_type numRows;
-  rmm::cuda_stream_view stream;
-
-  template <typename OffsetT>
-  void operator()() {
-    if constexpr (std::is_same_v<OffsetT, int32_t> ||
-                  std::is_same_v<OffsetT, int64_t>) {
-      detail::unpackDecimalSumState(
-          offsetsView.data<OffsetT>(),
-          charsPtr,
-          sums,
-          counts,
-          numRows,
-          stream);
-    }
-  }
-};
-
-struct FillOffsetsForDecimalSumStateDispatcher {
-  cudf::mutable_column_view offsetsView;
-  cudf::size_type rowCount;
-  rmm::cuda_stream_view stream;
-
-  template <typename OffsetT>
-  void operator()() {
-    if constexpr (std::is_same_v<OffsetT, int32_t> ||
-                  std::is_same_v<OffsetT, int64_t>) {
-      detail::fillOffsetsForDecimalSumState(
-          offsetsView.data<OffsetT>(), rowCount, stream);
-    }
-  }
-};
-
-struct PackDecimalSumStateDispatcher {
-  const cudf::column_view& sumCol;
-  const int64_t* countPtr;
-  cudf::column_view offsetsView;
-  uint8_t* charsPtr;
-  cudf::size_type rowCount;
-  rmm::cuda_stream_view stream;
-
-  template <typename SumT, typename OffsetT>
-  void operator()() {
-    if constexpr ((std::is_same_v<SumT, int64_t> ||
-                   std::is_same_v<SumT, __int128_t>) &&
-                  (std::is_same_v<OffsetT, int32_t> ||
-                   std::is_same_v<OffsetT, int64_t>)) {
-      detail::packDecimalSumState(
-          sumCol.data<SumT>(),
-          countPtr,
-          offsetsView.data<OffsetT>(),
-          charsPtr,
-          rowCount,
-          stream);
-    }
-  }
-};
-
-struct AverageRoundDecimalSumDispatcher {
-  const cudf::column_view& sumCol;
-  const int64_t* countPtr;
-  cudf::mutable_column_view outView;
-  cudf::size_type rowCount;
-  rmm::cuda_stream_view stream;
-
-  template <typename SumT>
-  void operator()() {
-    if constexpr (std::is_same_v<SumT, int64_t> ||
-                  std::is_same_v<SumT, __int128_t>) {
-      detail::averageRoundDecimalSum(
-          sumCol.data<SumT>(),
-          countPtr,
-          outView.data<SumT>(),
-          rowCount,
-          stream);
-    }
-  }
-};
-
-} // namespace
 
 DecimalSumStateColumns deserializeDecimalSumState(
     const cudf::column_view& stateCol,
@@ -198,13 +110,13 @@ DecimalSumStateColumns deserializeDecimalSumState(
       cudf::type_to_name(offsetsView.type()));
   cudf::type_dispatcher(
       offsetsView.type(),
-      UnpackDecimalSumStateDispatcher{
-          offsetsView,
-          charsPtr,
-          sumView.data<__int128_t>(),
-          countView.data<int64_t>(),
-          numRows,
-          stream});
+      detail::unpackDecimalSumState{},
+      offsetsView,
+      charsPtr,
+      sumView,
+      countView,
+      numRows,
+      stream);
 
   if (stateCol.nullable()) {
     auto nullMask = cudf::copy_bitmask(stateCol, stream, mr);
@@ -276,25 +188,25 @@ std::unique_ptr<cudf::column> serializeDecimalSumState(
   auto charsPtr = reinterpret_cast<uint8_t*>(charsBuf.data());
   cudf::type_dispatcher(
       offsetsView.type(),
-      FillOffsetsForDecimalSumStateDispatcher{
-          offsetsView, rowCount, stream});
+      detail::fillOffsetsForDecimalSumState{},
+      offsetsView,
+      rowCount,
+      stream);
 
-  auto const sumType = sumCol.type().id();
   VELOX_CHECK(
-      sumType == cudf::type_id::DECIMAL64 ||
-          sumType == cudf::type_id::DECIMAL128,
+      sumCol.type().id() == cudf::type_id::DECIMAL64 ||
+          sumCol.type().id() == cudf::type_id::DECIMAL128,
       "Unsupported decimal sum column type (type is {})",
       cudf::type_to_name(sumCol.type()));
-  cudf::double_type_dispatcher<cudf::dispatch_storage_type>(
+  cudf::type_dispatcher<cudf::dispatch_storage_type>(
       sumCol.type(),
-      offsetsView.type(),
-      PackDecimalSumStateDispatcher{
-          sumCol,
-          countCol.data<int64_t>(),
-          offsetsView,
-          charsPtr,
-          rowCount,
-          stream});
+      detail::packDecimalSumState{},
+      sumCol,
+      countCol.data<int64_t>(),
+      offsetsView,
+      charsPtr,
+      rowCount,
+      stream);
 
   auto [nullMask, nullCount] =
       detail::buildStateValidityMask(sumCol, countCol, stream, mr);
@@ -335,12 +247,12 @@ std::unique_ptr<cudf::column> computeDecimalAverage(
     auto const rowCount = static_cast<int32_t>(numRows);
     cudf::type_dispatcher<cudf::dispatch_storage_type>(
         sumCol.type(),
-        AverageRoundDecimalSumDispatcher{
-            sumCol,
-            countCol.data<int64_t>(),
-            out->mutable_view(),
-            rowCount,
-            stream});
+        detail::averageRoundDecimalSum{},
+        sumCol,
+        countCol.data<int64_t>(),
+        out->mutable_view(),
+        rowCount,
+        stream);
   }
 
   auto [nullMask, nullCount] =

From 5c1162ef118104597aaefac1e9d4683e8fdaf485 Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Thu, 11 Jun 2026 16:41:43 -0700
Subject: [PATCH 4/8] Third attempt

---
 .../cudf/exec/DecimalAggregationDevice.cu     | 212 ++++++++++--------
 .../cudf/exec/DecimalAggregationDevice.h      |  99 +-------
 2 files changed, 119 insertions(+), 192 deletions(-)

diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
index fec07223dad..04deec2bb98 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
@@ -234,61 +234,145 @@ std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMaskImpl(
 
 namespace detail {
 
-template <typename OffsetT>
-void launchFillOffsetsForDecimalSumState(
-    OffsetT* offsetsMutable,
-    size_t offsetCount,
-    rmm::cuda_stream_view stream) {
+template <>
+void fillOffsetsForDecimalSumState::operator()<int32_t, 0>(
+    cudf::mutable_column_view offsetsView,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const {
   launchFillOffsets(
-      cuda::std::span<OffsetT>{offsetsMutable, offsetCount}, stream);
+      cuda::std::span<int32_t>{
+          offsetsView.data<int32_t>(),
+          static_cast<size_t>(numRows) + 1},
+      stream);
 }
 
-template <typename OffsetT>
-void launchUnpackDecimalSumState(
-    const OffsetT* offsets,
+template <>
+void fillOffsetsForDecimalSumState::operator()<int64_t, 0>(
+    cudf::mutable_column_view offsetsView,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const {
+  launchFillOffsets(
+      cuda::std::span<int64_t>{
+          offsetsView.data<int64_t>(),
+          static_cast<size_t>(numRows) + 1},
+      stream);
+}
+
+template <>
+void unpackDecimalSumState::operator()<int32_t, 0>(
+    cudf::column_view offsetsView,
     const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
+    cudf::mutable_column_view sumView,
+    cudf::mutable_column_view countView,
     cudf::size_type numRows,
-    rmm::cuda_stream_view stream) {
+    rmm::cuda_stream_view stream) const {
   auto const n = static_cast<size_t>(numRows);
   launchUnpackState(
-      cuda::std::span<const OffsetT>{offsets, n},
+      cuda::std::span<const int32_t>{offsetsView.data<int32_t>(), n},
       chars,
-      cuda::std::span<__int128_t>{sums, n},
-      cuda::std::span<int64_t>{counts, n},
+      cuda::std::span<__int128_t>{sumView.data<__int128_t>(), n},
+      cuda::std::span<int64_t>{countView.data<int64_t>(), n},
       stream);
 }
 
-template <typename SumT, typename OffsetT>
-void launchPackDecimalSumState(
-    const SumT* sums,
+template <>
+void unpackDecimalSumState::operator()<int64_t, 0>(
+    cudf::column_view offsetsView,
+    const uint8_t* chars,
+    cudf::mutable_column_view sumView,
+    cudf::mutable_column_view countView,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const {
+  auto const n = static_cast<size_t>(numRows);
+  launchUnpackState(
+      cuda::std::span<const int64_t>{offsetsView.data<int64_t>(), n},
+      chars,
+      cuda::std::span<__int128_t>{sumView.data<__int128_t>(), n},
+      cuda::std::span<int64_t>{countView.data<int64_t>(), n},
+      stream);
+}
+
+template <>
+void packDecimalSumState::operator()<int64_t, 0>(
+    cudf::column_view sumCol,
     const int64_t* counts,
-    const OffsetT* offsets,
+    cudf::column_view offsetsView,
     uint8_t* chars,
     cudf::size_type numRows,
-    rmm::cuda_stream_view stream) {
+    rmm::cuda_stream_view stream) const {
   auto const n = static_cast<size_t>(numRows);
-  launchPackState(
-      cuda::std::span<const SumT>{sums, n},
+  auto const sums = sumCol.data<int64_t>();
+  if (offsetsView.type().id() == cudf::type_id::INT32) {
+    launchPackState(
+        cuda::std::span<const int64_t>{sums, n},
+        cuda::std::span<const int64_t>{counts, n},
+        cuda::std::span<const int32_t>{offsetsView.data<int32_t>(), n},
+        chars,
+        stream);
+  } else {
+    launchPackState(
+        cuda::std::span<const int64_t>{sums, n},
+        cuda::std::span<const int64_t>{counts, n},
+        cuda::std::span<const int64_t>{offsetsView.data<int64_t>(), n},
+        chars,
+        stream);
+  }
+}
+
+template <>
+void packDecimalSumState::operator()<__int128_t, 0>(
+    cudf::column_view sumCol,
+    const int64_t* counts,
+    cudf::column_view offsetsView,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const {
+  auto const n = static_cast<size_t>(numRows);
+  auto const sums = sumCol.data<__int128_t>();
+  if (offsetsView.type().id() == cudf::type_id::INT32) {
+    launchPackState(
+        cuda::std::span<const __int128_t>{sums, n},
+        cuda::std::span<const int64_t>{counts, n},
+        cuda::std::span<const int32_t>{offsetsView.data<int32_t>(), n},
+        chars,
+        stream);
+  } else {
+    launchPackState(
+        cuda::std::span<const __int128_t>{sums, n},
+        cuda::std::span<const int64_t>{counts, n},
+        cuda::std::span<const int64_t>{offsetsView.data<int64_t>(), n},
+        chars,
+        stream);
+  }
+}
+
+template <>
+void averageRoundDecimalSum::operator()<int64_t, 0>(
+    cudf::column_view sumCol,
+    const int64_t* counts,
+    cudf::mutable_column_view outView,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const {
+  auto const n = static_cast<size_t>(numRows);
+  launchAvgRound(
+      cuda::std::span<const int64_t>{sumCol.data<int64_t>(), n},
       cuda::std::span<const int64_t>{counts, n},
-      cuda::std::span<const OffsetT>{offsets, n},
-      chars,
+      cuda::std::span<int64_t>{outView.data<int64_t>(), n},
       stream);
 }
 
-template <typename SumT>
-void launchAverageRoundDecimalSum(
-    const SumT* sums,
+template <>
+void averageRoundDecimalSum::operator()<__int128_t, 0>(
+    cudf::column_view sumCol,
     const int64_t* counts,
-    SumT* out,
+    cudf::mutable_column_view outView,
     cudf::size_type numRows,
-    rmm::cuda_stream_view stream) {
+    rmm::cuda_stream_view stream) const {
   auto const n = static_cast<size_t>(numRows);
   launchAvgRound(
-      cuda::std::span<const SumT>{sums, n},
+      cuda::std::span<const __int128_t>{sumCol.data<__int128_t>(), n},
       cuda::std::span<const int64_t>{counts, n},
-      cuda::std::span<SumT>{out, n},
+      cuda::std::span<__int128_t>{outView.data<__int128_t>(), n},
       stream);
 }
 
@@ -300,71 +384,5 @@ std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMask(
   return buildStateValidityMaskImpl(sumCol, countCol, stream, mr);
 }
 
-template void launchFillOffsetsForDecimalSumState<int32_t>(
-    int32_t* offsetsMutable,
-    size_t offsetCount,
-    rmm::cuda_stream_view stream);
-template void launchFillOffsetsForDecimalSumState<int64_t>(
-    int64_t* offsetsMutable,
-    size_t offsetCount,
-    rmm::cuda_stream_view stream);
-
-template void launchUnpackDecimalSumState<int32_t>(
-    const int32_t* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-template void launchUnpackDecimalSumState<int64_t>(
-    const int64_t* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-
-template void launchPackDecimalSumState<int64_t, int32_t>(
-    const int64_t* sums,
-    const int64_t* counts,
-    const int32_t* offsets,
-    uint8_t* chars,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-template void launchPackDecimalSumState<int64_t, int64_t>(
-    const int64_t* sums,
-    const int64_t* counts,
-    const int64_t* offsets,
-    uint8_t* chars,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-template void launchPackDecimalSumState<__int128_t, int32_t>(
-    const __int128_t* sums,
-    const int64_t* counts,
-    const int32_t* offsets,
-    uint8_t* chars,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-template void launchPackDecimalSumState<__int128_t, int64_t>(
-    const __int128_t* sums,
-    const int64_t* counts,
-    const int64_t* offsets,
-    uint8_t* chars,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-
-template void launchAverageRoundDecimalSum<int64_t>(
-    const int64_t* sums,
-    const int64_t* counts,
-    int64_t* out,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-template void launchAverageRoundDecimalSum<__int128_t>(
-    const __int128_t* sums,
-    const int64_t* counts,
-    __int128_t* out,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-
 } // namespace detail
 } // namespace facebook::velox::cudf_velox
diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.h b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
index b0e791da2f1..e69c0cff3e0 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.h
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
@@ -17,7 +17,6 @@
 
 #include <cudf/column/column_view.hpp>
 #include <cudf/types.hpp>
-#include <cudf/utilities/type_dispatcher.hpp>
 
 #include <rmm/cuda_stream_view.hpp>
 #include <rmm/device_buffer.hpp>
@@ -43,38 +42,6 @@ inline constexpr bool isOffsetStorageType =
 // words).
 constexpr size_t kDecimalSumStateSize = 32;
 
-template <typename OffsetT>
-void launchFillOffsetsForDecimalSumState(
-    OffsetT* offsetsMutable,
-    size_t offsetCount,
-    rmm::cuda_stream_view stream);
-
-template <typename OffsetT>
-void launchUnpackDecimalSumState(
-    const OffsetT* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-
-template <typename SumT, typename OffsetT>
-void launchPackDecimalSumState(
-    const SumT* sums,
-    const int64_t* counts,
-    const OffsetT* offsets,
-    uint8_t* chars,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-
-template <typename SumT>
-void launchAverageRoundDecimalSum(
-    const SumT* sums,
-    const int64_t* counts,
-    SumT* out,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-
 /**
  * Writes strings-style prefix offsets: offset[i] == i * kDecimalSumStateSize.
  *
@@ -89,12 +56,7 @@ struct fillOffsetsForDecimalSumState {
   void operator()(
       cudf::mutable_column_view offsetsView,
       cudf::size_type numRows,
-      rmm::cuda_stream_view stream) const {
-    launchFillOffsetsForDecimalSumState<OffsetT>(
-        offsetsView.data<OffsetT>(),
-        static_cast<size_t>(numRows) + 1,
-        stream);
-  }
+      rmm::cuda_stream_view stream) const;
 
   template <
       typename OffsetT,
@@ -105,34 +67,6 @@ struct fillOffsetsForDecimalSumState {
       rmm::cuda_stream_view stream) const {}
 };
 
-template <typename SumT>
-struct packDecimalSumStateWithOffset {
-  const SumT* sums;
-  const int64_t* counts;
-  cudf::column_view offsetsView;
-  uint8_t* chars;
-  cudf::size_type numRows;
-  rmm::cuda_stream_view stream;
-
-  template <
-      typename OffsetT,
-      std::enable_if_t<isOffsetStorageType<OffsetT>, int> = 0>
-  void operator()() const {
-    launchPackDecimalSumState<SumT, OffsetT>(
-        sums,
-        counts,
-        offsetsView.data<OffsetT>(),
-        chars,
-        numRows,
-        stream);
-  }
-
-  template <
-      typename OffsetT,
-      std::enable_if_t<!isOffsetStorageType<OffsetT>, int> = 0>
-  void operator()() const {}
-};
-
 /**
  * Encodes each row's partial sum and count into the fixed-width device layout
  * used for VARBINARY interchange.
@@ -154,17 +88,7 @@ struct packDecimalSumState {
       cudf::column_view offsetsView,
       uint8_t* chars,
       cudf::size_type numRows,
-      rmm::cuda_stream_view stream) const {
-    cudf::type_dispatcher(
-        offsetsView.type(),
-        packDecimalSumStateWithOffset<SumT>{
-            sumCol.data<SumT>(),
-            counts,
-            offsetsView,
-            chars,
-            numRows,
-            stream});
-  }
+      rmm::cuda_stream_view stream) const;
 
   template <
       typename SumT,
@@ -198,15 +122,7 @@ struct unpackDecimalSumState {
       cudf::mutable_column_view sumView,
       cudf::mutable_column_view countView,
       cudf::size_type numRows,
-      rmm::cuda_stream_view stream) const {
-    launchUnpackDecimalSumState<OffsetT>(
-        offsetsView.data<OffsetT>(),
-        chars,
-        sumView.data<__int128_t>(),
-        countView.data<int64_t>(),
-        numRows,
-        stream);
-  }
+      rmm::cuda_stream_view stream) const;
 
   template <
       typename OffsetT,
@@ -239,14 +155,7 @@ struct averageRoundDecimalSum {
       const int64_t* counts,
       cudf::mutable_column_view outView,
       cudf::size_type numRows,
-      rmm::cuda_stream_view stream) const {
-    launchAverageRoundDecimalSum<SumT>(
-        sumCol.data<SumT>(),
-        counts,
-        outView.data<SumT>(),
-        numRows,
-        stream);
-  }
+      rmm::cuda_stream_view stream) const;
 
   template <
       typename SumT,

From 033142fcbfc45dda1a3af3cc9eefc57f6e735807 Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Thu, 11 Jun 2026 16:43:34 -0700
Subject: [PATCH 5/8] Remove Cursor plan file

---
 ...tcher_decimal_aggregation_refactor.plan.md | 392 ------------------
 1 file changed, 392 deletions(-)
 delete mode 100644 .cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md

diff --git a/.cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md b/.cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md
deleted file mode 100644
index 1ee0e840345..00000000000
--- a/.cursor/plans/type_dispatcher_decimal_aggregation_refactor.plan.md
+++ /dev/null
@@ -1,392 +0,0 @@
----
-name: type_dispatcher decimal aggregation refactor
-overview: Replace all manual type branches and void* casts in decimal sum state pack/unpack/average paths with cudf::type_dispatcher (and double_type_dispatcher where two dimensions apply), using typed device helpers and dispatch_storage_type for decimal columns.
-todos:
-  - id: retype-unpack-api
-    content: Change unpackDecimalSumState to template<OffsetT>(const OffsetT* offsets, ...) in DecimalAggregationDevice.h/.cu with explicit int32_t/int64_t instantiations
-    status: pending
-  - id: retype-pack-fill-api
-    content: Change fillOffsetsForDecimalSumState to template<OffsetT> and packDecimalSumState to template<SumT, OffsetT> in DecimalAggregationDevice.h/.cu with explicit instantiations for all 4 combinations
-    status: pending
-  - id: retype-average-api
-    content: Change averageRoundDecimalSum to template<SumT>(const SumT* sums, SumT* out, ...) in DecimalAggregationDevice.h/.cu with explicit int64_t/__int128_t instantiations
-    status: pending
-  - id: dispatch-deserialize-call-site
-    content: Add type_dispatcher.hpp include and replace bool/void* branch at line 109 in DecimalAggregationState.cpp with UnpackDecimalSumStateDispatcher functor
-    status: pending
-  - id: dispatch-serialize-call-sites
-    content: Replace bool/void* branches at lines 187 and 207 with FillOffsetsDispatcher and nested/double_type_dispatcher for pack (sum + offset types)
-    status: pending
-  - id: dispatch-average-call-site
-    content: Replace sumType/void* branch at lines 257-265 in computeDecimalAverage with AverageRoundDecimalSumDispatcher using dispatch_storage_type
-    status: pending
-  - id: verify-tests
-    content: Run DecimalAggregationTest sum-state and average tests covering DECIMAL64/DECIMAL128 and INT32/INT64 offsets
-    status: pending
-isProject: true
----
-
-# Refactor decimal aggregation device calls to use type_dispatcher
-
-## Current state
-
-### Deserialize path (line 109)
-
-In [`DecimalAggregationState.cpp`](velox/experimental/cudf/exec/DecimalAggregationState.cpp) (lines 83–117), deserialize manually inspects the strings offsets type, sets `offsets64`, and casts `offsetsView.data<int64_t>()` / `data<int32_t>()` to `void*` before calling the device helper.
-
-The device implementation in [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu) already has fully typed kernels (`launchUnpackState<OffsetT>`); the `bool` + `void*` wrapper (lines 309–337) is redundant type erasure.
-
-### Serialize path (lines 187 and 207)
-
-The serialize path uses offset type-erasure for both `fillOffsetsForDecimalSumState` and `packDecimalSumState`, and additionally erases sum types via `sumType` + `sumPtr`:
-
-```187:215:velox/experimental/cudf/exec/DecimalAggregationState.cpp
-  detail::fillOffsetsForDecimalSumState(
-      useLargeOffsets,
-      useLargeOffsets ? static_cast<void*>(offsetsView.data<int64_t>())
-                      : static_cast<void*>(offsetsView.data<int32_t>()),
-      rowCount,
-      stream);
-  ...
-  const void* sumPtr = sumType == cudf::type_id::DECIMAL64
-      ? static_cast<const void*>(sumCol.data<int64_t>())
-      : static_cast<const void*>(sumCol.data<__int128_t>());
-  detail::packDecimalSumState(
-      sumType,
-      useLargeOffsets,
-      sumPtr,
-      countCol.data<int64_t>(),
-      offsetsPtr,
-      charsPtr,
-      rowCount,
-      stream);
-```
-
-`useLargeOffsets` is still needed to **choose** `offsetsType` when creating the offsets column (lines 172–179), but it should no longer be threaded into device calls or used for pointer casts.
-
-The device wrappers `fillOffsetsForDecimalSumState` (lines 237–253) and `packDecimalSumState` (lines 255–307) mirror the unpack problem: typed kernels (`launchFillOffsets<OffsetT>`, `launchPackState<SumT, OffsetT>`) exist, but the public API erases types via `bool`/`type_id` + `void*`.
-
-### Average path (lines 257–265)
-
-`computeDecimalAverage` applies the same sum-type erasure pattern:
-
-```257:265:velox/experimental/cudf/exec/DecimalAggregationState.cpp
-    const auto sumType = sumCol.type().id();
-    const void* sumsPtr = sumType == cudf::type_id::DECIMAL64
-        ? static_cast<const void*>(sumCol.data<int64_t>())
-        : static_cast<const void*>(sumCol.data<__int128_t>());
-    void* outPtr = sumType == cudf::type_id::DECIMAL64
-        ? static_cast<void*>(out->mutable_view().data<int64_t>())
-        : static_cast<void*>(out->mutable_view().data<__int128_t>());
-    detail::averageRoundDecimalSum(
-        sumType, sumsPtr, countCol.data<int64_t>(), outPtr, rowCount, stream);
-```
-
-The device wrapper `averageRoundDecimalSum` (lines 339–362) branches on `sumType` and casts to typed spans for `launchAvgRound<SumT>`.
-
-There is **no existing `type_dispatcher` usage** in `velox/experimental/cudf/` today — this will be the first adoption in this module.
-
-## Target design
-
-```mermaid
-flowchart TB
-  subgraph deserialize [deserializeDecimalSumState]
-    dOffsets["strings.offsets()"]
-    dDispatch["type_dispatcher(offsetsView.type())"]
-    dUnpack["unpackDecimalSumState&lt;OffsetT&gt;"]
-    dKernel["launchUnpackState&lt;OffsetT&gt;"]
-    dOffsets --> dDispatch --> dUnpack --> dKernel
-  end
-
-  subgraph serialize [serializeDecimalSumState]
-    sOffsets["offsetsCol mutable_view"]
-    sFillDispatch["type_dispatcher(offsetsView.type())"]
-    sPackDispatch["double dispatch sum + offset types"]
-    sFill["fillOffsetsForDecimalSumState&lt;OffsetT&gt;"]
-    sPack["packDecimalSumState&lt;SumT,OffsetT&gt;"]
-    sFillKernel["launchFillOffsets&lt;OffsetT&gt;"]
-    sPackKernel["launchPackState&lt;SumT,OffsetT&gt;"]
-    sOffsets --> sFillDispatch --> sFill --> sFillKernel
-    sOffsets --> sPackDispatch --> sPack --> sPackKernel
-  end
-
-  subgraph average [computeDecimalAverage]
-    aSum["sumCol.type()"]
-    aDispatch["type_dispatcher&lt;dispatch_storage_type&gt;"]
-    aAvg["averageRoundDecimalSum&lt;SumT&gt;"]
-    aKernel["launchAvgRound&lt;SumT&gt;"]
-    aSum --> aDispatch --> aAvg --> aKernel
-  end
-```
-
-### Dispatcher conventions
-
-- **Offset columns (INT32 / INT64):** use default `cudf::type_dispatcher` — `id_to_type` maps to `int32_t` / `int64_t`, matching `column_view::data<T>()`.
-- **Decimal sum columns (DECIMAL64 / DECIMAL128):** use `cudf::type_dispatcher<cudf::dispatch_storage_type>` so the dispatched `T` is the **device storage type** (`int64_t` / `__int128_t`), not the logical `numeric::decimal64` / `numeric::decimal128` type. This matches how `.data<int64_t>()` / `.data<__int128_t>()` are used today and is the libcudf-recommended pattern when operating on raw column buffers.
-- **Pack (two dimensions):** use **nested dispatch** — outer `type_dispatcher<dispatch_storage_type>` on `sumCol.type()`, inner `type_dispatcher` on `offsetsView.type()` — because `double_type_dispatcher` accepts only one `IdTypeMap` and we need `dispatch_storage_type` for sums but default `id_to_type` semantics also work for offsets via `dispatch_storage_type` (non-decimal types pass through unchanged). Alternatively, a single `double_type_dispatcher<cudf::dispatch_storage_type>` on `(sumCol.type(), offsetsView.type())` is equivalent and slightly cleaner if preferred during implementation.
-
-Add `#include <cudf/utilities/type_dispatcher.hpp>` **before** [`CudfNoDefaults.h`](velox/experimental/cudf/CudfNoDefaults.h) in [`DecimalAggregationState.cpp`](velox/experimental/cudf/exec/DecimalAggregationState.cpp).
-
-All call-site functors live in an anonymous namespace in that file.
-
----
-
-## Part A: Unpack refactor
-
-### A1. Retype `detail::unpackDecimalSumState` (header + .cu)
-
-In [`DecimalAggregationDevice.h`](velox/experimental/cudf/exec/DecimalAggregationDevice.h):
-
-```cpp
-template <typename OffsetT>
-void unpackDecimalSumState(
-    const OffsetT* offsets,
-    const uint8_t* chars,
-    __int128_t* sums,
-    int64_t* counts,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-```
-
-In [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu):
-
-- Collapse the `if (offsets64)` branch into the template body calling `launchUnpackState` with `cuda::std::span<const OffsetT>{offsets, n}`.
-- Explicit instantiations: `int32_t`, `int64_t`.
-
-### A2. Dispatch at deserialize call site (line 109)
-
-```cpp
-struct UnpackDecimalSumStateDispatcher {
-  const cudf::column_view& offsetsView;
-  const uint8_t* charsPtr;
-  __int128_t* sums;
-  int64_t* counts;
-  cudf::size_type numRows;
-  rmm::cuda_stream_view stream;
-
-  template <typename OffsetT>
-  void operator()() {
-    detail::unpackDecimalSumState(
-        offsetsView.data<OffsetT>(),
-        charsPtr,
-        sums,
-        counts,
-        numRows,
-        stream);
-  }
-};
-
-auto const offsetsType = offsetsView.type().id();
-VELOX_CHECK(
-    offsetsType == cudf::type_id::INT32 || offsetsType == cudf::type_id::INT64,
-    "Decimal sum state requires INT32 or INT64 offsets (offset type is {})",
-    cudf::type_to_name(offsetsView.type()));
-
-cudf::type_dispatcher(
-    offsetsView.type(),
-    UnpackDecimalSumStateDispatcher{...});
-```
-
----
-
-## Part B: Pack + fill refactor
-
-### B1. Retype device helpers (header + .cu)
-
-**`fillOffsetsForDecimalSumState`:**
-
-```cpp
-template <typename OffsetT>
-void fillOffsetsForDecimalSumState(
-    OffsetT* offsetsMutable,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-```
-
-**`packDecimalSumState`** — fully typed on both sum and offset dimensions; remove `cudf::type_id sumType`, `const void* sumPtr`, `bool use64BitOffsets`, and `const void* offsetsPtr`:
-
-```cpp
-template <typename SumT, typename OffsetT>
-void packDecimalSumState(
-    const SumT* sums,
-    const int64_t* counts,
-    const OffsetT* offsets,
-    uint8_t* chars,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-```
-
-In [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu):
-
-- `fillOffsetsForDecimalSumState`: call `launchFillOffsets` with `cuda::std::span<OffsetT>{offsetsMutable, n}`.
-- `packDecimalSumState`: call `launchPackState` directly with typed spans — no inner `sumType` or `use64BitOffsets` branches.
-- Explicit instantiations for all four `(SumT, OffsetT)` pairs: `(int64_t, int32_t)`, `(int64_t, int64_t)`, `(__int128_t, int32_t)`, `(__int128_t, int64_t)`.
-
-### B2. Dispatch at serialize call sites (lines 187 and 207)
-
-**Fill** — single dispatch on offset type:
-
-```cpp
-struct FillOffsetsForDecimalSumStateDispatcher {
-  cudf::column_view offsetsView;
-  cudf::size_type rowCount;
-  rmm::cuda_stream_view stream;
-
-  template <typename OffsetT>
-  void operator()() {
-    detail::fillOffsetsForDecimalSumState(
-        offsetsView.data<OffsetT>(), rowCount, stream);
-  }
-};
-
-cudf::type_dispatcher(
-    offsetsView.type(),
-    FillOffsetsForDecimalSumStateDispatcher{offsetsView, rowCount, stream});
-```
-
-**Pack** — nested dispatch on sum then offset types (eliminates `sumType`, `sumPtr`, `offsetsPtr`):
-
-```cpp
-struct PackDecimalSumStateDispatcher {
-  const cudf::column_view& sumCol;
-  const int64_t* countPtr;
-  cudf::column_view offsetsView;
-  uint8_t* charsPtr;
-  cudf::size_type rowCount;
-  rmm::cuda_stream_view stream;
-
-  template <typename SumT>
-  void operator()() {
-    struct PackWithOffset {
-      const cudf::column_view& sumCol;
-      const int64_t* countPtr;
-      cudf::column_view offsetsView;
-      uint8_t* charsPtr;
-      cudf::size_type rowCount;
-      rmm::cuda_stream_view stream;
-
-      template <typename OffsetT>
-      void operator()() {
-        detail::packDecimalSumState(
-            sumCol.data<SumT>(),
-            countPtr,
-            offsetsView.data<OffsetT>(),
-            charsPtr,
-            rowCount,
-            stream);
-      }
-    };
-    cudf::type_dispatcher(
-        offsetsView.type(),
-        PackWithOffset{sumCol, countPtr, offsetsView, charsPtr, rowCount, stream});
-  }
-};
-
-auto const sumType = sumCol.type().id();
-VELOX_CHECK(
-    sumType == cudf::type_id::DECIMAL64 ||
-        sumType == cudf::type_id::DECIMAL128,
-    "Unsupported decimal sum column type (type is {})",
-    cudf::type_to_name(sumCol.type()));
-
-cudf::type_dispatcher<cudf::dispatch_storage_type>(
-    sumCol.type(),
-    PackDecimalSumStateDispatcher{
-        sumCol,
-        countCol.data<int64_t>(),
-        offsetsView,
-        charsPtr,
-        rowCount,
-        stream});
-```
-
-**Alternative (equivalent):** replace the nested struct with a single call:
-
-```cpp
-cudf::double_type_dispatcher<cudf::dispatch_storage_type>(
-    sumCol.type(),
-    offsetsView.type(),
-    PackDecimalSumStateDoubleDispatcher{...});  // operator()<SumT, OffsetT>()
-```
-
-Pick whichever reads cleaner during implementation; both eliminate all `void*` casts in the serialize path.
-
----
-
-## Part C: Average refactor
-
-### C1. Retype `detail::averageRoundDecimalSum` (header + .cu)
-
-Replace `cudf::type_id sumType, const void* sums, void* out` with:
-
-```cpp
-template <typename SumT>
-void averageRoundDecimalSum(
-    const SumT* sums,
-    const int64_t* counts,
-    SumT* out,
-    cudf::size_type numRows,
-    rmm::cuda_stream_view stream);
-```
-
-In [`DecimalAggregationDevice.cu`](velox/experimental/cudf/exec/DecimalAggregationDevice.cu):
-
-- Template body calls `launchAvgRound` with `cuda::std::span<const SumT>{sums, n}` and `cuda::std::span<SumT>{out, n}`.
-- Explicit instantiations: `int64_t`, `__int128_t`.
-
-### C2. Dispatch at average call site (lines 257–265)
-
-Replace the `sumType` / `sumsPtr` / `outPtr` setup with:
-
-```cpp
-struct AverageRoundDecimalSumDispatcher {
-  const cudf::column_view& sumCol;
-  const int64_t* countPtr;
-  cudf::mutable_column_view outView;
-  cudf::size_type rowCount;
-  rmm::cuda_stream_view stream;
-
-  template <typename SumT>
-  void operator()() {
-    detail::averageRoundDecimalSum(
-        sumCol.data<SumT>(),
-        countPtr,
-        outView.data<SumT>(),
-        rowCount,
-        stream);
-  }
-};
-
-cudf::type_dispatcher<cudf::dispatch_storage_type>(
-    sumCol.type(),
-    AverageRoundDecimalSumDispatcher{
-        sumCol,
-        countCol.data<int64_t>(),
-        out->mutable_view(),
-        rowCount,
-        stream});
-```
-
-The existing `VELOX_CHECK` for DECIMAL64/DECIMAL128 at the top of `computeDecimalAverage` (lines 235–239) remains; it guards dispatch the same way as today.
-
----
-
-## Verification
-
-Run existing coverage in [`DecimalAggregationTest.cpp`](velox/experimental/cudf/tests/DecimalAggregationTest.cpp):
-
-**Sum state (Parts A + B):**
-- `decimalDeserializeSumState*` (INT32 offsets)
-- `decimalSerializeSumState*` / `decimalSumStateRoundTrip*`
-- `decimalSumStateRoundTripUsesInt64Offsets` / `decimalSerializeSumStateUsesInt64OffsetsWhenEnabled` (INT64 offsets)
-
-**Average (Part C):**
-- All `computeDecimalAverage` tests (lines ~1245–1448): DECIMAL64 and DECIMAL128 sums, null/zero-count edge cases, rounding behavior
-
-No new tests are strictly required unless you want explicit invalid-type regression tests.
-
-Target test command (adjust to your build setup):
-
-```bash
-./velox/experimental/cudf/tests/cudf_test --gtest_filter='*decimal*SumState*:*decimal*Average*:*Decimal*'
-```

From 1273f63479ad385644e6914fd3013bf3b0533356 Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Thu, 11 Jun 2026 16:58:30 -0700
Subject: [PATCH 6/8] Format

---
 velox/experimental/cudf/exec/DecimalAggregationDevice.cu | 6 ++----
 velox/experimental/cudf/exec/DecimalAggregationState.cpp | 3 +--
 2 files changed, 3 insertions(+), 6 deletions(-)

diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
index 04deec2bb98..e62765a3cf1 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
@@ -241,8 +241,7 @@ void fillOffsetsForDecimalSumState::operator()<int32_t, 0>(
     rmm::cuda_stream_view stream) const {
   launchFillOffsets(
       cuda::std::span<int32_t>{
-          offsetsView.data<int32_t>(),
-          static_cast<size_t>(numRows) + 1},
+          offsetsView.data<int32_t>(), static_cast<size_t>(numRows) + 1},
       stream);
 }
 
@@ -253,8 +252,7 @@ void fillOffsetsForDecimalSumState::operator()<int64_t, 0>(
     rmm::cuda_stream_view stream) const {
   launchFillOffsets(
       cuda::std::span<int64_t>{
-          offsetsView.data<int64_t>(),
-          static_cast<size_t>(numRows) + 1},
+          offsetsView.data<int64_t>(), static_cast<size_t>(numRows) + 1},
       stream);
 }
 
diff --git a/velox/experimental/cudf/exec/DecimalAggregationState.cpp b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
index dcfe8cee39e..6238bb3a885 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationState.cpp
+++ b/velox/experimental/cudf/exec/DecimalAggregationState.cpp
@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+#include "velox/experimental/cudf/CudfNoDefaults.h"
 #include "velox/experimental/cudf/exec/DecimalAggregationDevice.h"
 #include "velox/experimental/cudf/exec/DecimalAggregationState.h"
 #include "velox/experimental/cudf/exec/GpuResources.h"
@@ -28,8 +29,6 @@
 
 #include <limits>
 
-#include "velox/experimental/cudf/CudfNoDefaults.h"
-
 namespace facebook::velox::cudf_velox {
 
 DecimalSumStateColumns deserializeDecimalSumState(

From 13a3afc9df30440eb0cb5fc2b382a2ad033d707e Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Fri, 12 Jun 2026 10:04:51 -0700
Subject: [PATCH 7/8] Combine pairs of functions, per @bdice

---
 .../cudf/exec/DecimalAggregationDevice.cu     | 114 ++++++++----------
 1 file changed, 48 insertions(+), 66 deletions(-)

diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
index e62765a3cf1..37daa5ec2e0 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
@@ -234,30 +234,28 @@ std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMaskImpl(
 
 namespace detail {
 
-template <>
-void fillOffsetsForDecimalSumState::operator()<int32_t, 0>(
+template <typename OffsetT, std::enable_if_t<isOffsetStorageType<OffsetT>, int>>
+void fillOffsetsForDecimalSumState::operator()(
     cudf::mutable_column_view offsetsView,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const {
   launchFillOffsets(
-      cuda::std::span<int32_t>{
-          offsetsView.data<int32_t>(), static_cast<size_t>(numRows) + 1},
+      cuda::std::span<OffsetT>{
+          offsetsView.data<OffsetT>(), static_cast<size_t>(numRows) + 1},
       stream);
 }
 
-template <>
-void fillOffsetsForDecimalSumState::operator()<int64_t, 0>(
+template void fillOffsetsForDecimalSumState::operator()<int32_t, 0>(
     cudf::mutable_column_view offsetsView,
     cudf::size_type numRows,
-    rmm::cuda_stream_view stream) const {
-  launchFillOffsets(
-      cuda::std::span<int64_t>{
-          offsetsView.data<int64_t>(), static_cast<size_t>(numRows) + 1},
-      stream);
-}
+    rmm::cuda_stream_view stream) const;
+template void fillOffsetsForDecimalSumState::operator()<int64_t, 0>(
+    cudf::mutable_column_view offsetsView,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const;
 
-template <>
-void unpackDecimalSumState::operator()<int32_t, 0>(
+template <typename OffsetT, std::enable_if_t<isOffsetStorageType<OffsetT>, int>>
+void unpackDecimalSumState::operator()(
     cudf::column_view offsetsView,
     const uint8_t* chars,
     cudf::mutable_column_view sumView,
@@ -266,32 +264,30 @@ void unpackDecimalSumState::operator()<int32_t, 0>(
     rmm::cuda_stream_view stream) const {
   auto const n = static_cast<size_t>(numRows);
   launchUnpackState(
-      cuda::std::span<const int32_t>{offsetsView.data<int32_t>(), n},
+      cuda::std::span<const OffsetT>{offsetsView.data<OffsetT>(), n},
       chars,
       cuda::std::span<__int128_t>{sumView.data<__int128_t>(), n},
       cuda::std::span<int64_t>{countView.data<int64_t>(), n},
       stream);
 }
 
-template <>
-void unpackDecimalSumState::operator()<int64_t, 0>(
+template void unpackDecimalSumState::operator()<int32_t, 0>(
     cudf::column_view offsetsView,
     const uint8_t* chars,
     cudf::mutable_column_view sumView,
     cudf::mutable_column_view countView,
     cudf::size_type numRows,
-    rmm::cuda_stream_view stream) const {
-  auto const n = static_cast<size_t>(numRows);
-  launchUnpackState(
-      cuda::std::span<const int64_t>{offsetsView.data<int64_t>(), n},
-      chars,
-      cuda::std::span<__int128_t>{sumView.data<__int128_t>(), n},
-      cuda::std::span<int64_t>{countView.data<int64_t>(), n},
-      stream);
-}
+    rmm::cuda_stream_view stream) const;
+template void unpackDecimalSumState::operator()<int64_t, 0>(
+    cudf::column_view offsetsView,
+    const uint8_t* chars,
+    cudf::mutable_column_view sumView,
+    cudf::mutable_column_view countView,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const;
 
-template <>
-void packDecimalSumState::operator()<int64_t, 0>(
+template <typename SumT, std::enable_if_t<isDecimalSumStorageType<SumT>, int>>
+void packDecimalSumState::operator()(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::column_view offsetsView,
@@ -299,17 +295,17 @@ void packDecimalSumState::operator()<int64_t, 0>(
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const {
   auto const n = static_cast<size_t>(numRows);
-  auto const sums = sumCol.data<int64_t>();
+  auto const sums = sumCol.data<SumT>();
   if (offsetsView.type().id() == cudf::type_id::INT32) {
     launchPackState(
-        cuda::std::span<const int64_t>{sums, n},
+        cuda::std::span<const SumT>{sums, n},
         cuda::std::span<const int64_t>{counts, n},
         cuda::std::span<const int32_t>{offsetsView.data<int32_t>(), n},
         chars,
         stream);
   } else {
     launchPackState(
-        cuda::std::span<const int64_t>{sums, n},
+        cuda::std::span<const SumT>{sums, n},
         cuda::std::span<const int64_t>{counts, n},
         cuda::std::span<const int64_t>{offsetsView.data<int64_t>(), n},
         chars,
@@ -317,35 +313,23 @@ void packDecimalSumState::operator()<int64_t, 0>(
   }
 }
 
-template <>
-void packDecimalSumState::operator()<__int128_t, 0>(
+template void packDecimalSumState::operator()<int64_t, 0>(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::column_view offsetsView,
     uint8_t* chars,
     cudf::size_type numRows,
-    rmm::cuda_stream_view stream) const {
-  auto const n = static_cast<size_t>(numRows);
-  auto const sums = sumCol.data<__int128_t>();
-  if (offsetsView.type().id() == cudf::type_id::INT32) {
-    launchPackState(
-        cuda::std::span<const __int128_t>{sums, n},
-        cuda::std::span<const int64_t>{counts, n},
-        cuda::std::span<const int32_t>{offsetsView.data<int32_t>(), n},
-        chars,
-        stream);
-  } else {
-    launchPackState(
-        cuda::std::span<const __int128_t>{sums, n},
-        cuda::std::span<const int64_t>{counts, n},
-        cuda::std::span<const int64_t>{offsetsView.data<int64_t>(), n},
-        chars,
-        stream);
-  }
-}
+    rmm::cuda_stream_view stream) const;
+template void packDecimalSumState::operator()<__int128_t, 0>(
+    cudf::column_view sumCol,
+    const int64_t* counts,
+    cudf::column_view offsetsView,
+    uint8_t* chars,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const;
 
-template <>
-void averageRoundDecimalSum::operator()<int64_t, 0>(
+template <typename SumT, std::enable_if_t<isDecimalSumStorageType<SumT>, int>>
+void averageRoundDecimalSum::operator()(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::mutable_column_view outView,
@@ -353,26 +337,24 @@ void averageRoundDecimalSum::operator()<int64_t, 0>(
     rmm::cuda_stream_view stream) const {
   auto const n = static_cast<size_t>(numRows);
   launchAvgRound(
-      cuda::std::span<const int64_t>{sumCol.data<int64_t>(), n},
+      cuda::std::span<const SumT>{sumCol.data<SumT>(), n},
       cuda::std::span<const int64_t>{counts, n},
-      cuda::std::span<int64_t>{outView.data<int64_t>(), n},
+      cuda::std::span<SumT>{outView.data<SumT>(), n},
       stream);
 }
 
-template <>
-void averageRoundDecimalSum::operator()<__int128_t, 0>(
+template void averageRoundDecimalSum::operator()<int64_t, 0>(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::mutable_column_view outView,
     cudf::size_type numRows,
-    rmm::cuda_stream_view stream) const {
-  auto const n = static_cast<size_t>(numRows);
-  launchAvgRound(
-      cuda::std::span<const __int128_t>{sumCol.data<__int128_t>(), n},
-      cuda::std::span<const int64_t>{counts, n},
-      cuda::std::span<__int128_t>{outView.data<__int128_t>(), n},
-      stream);
-}
+    rmm::cuda_stream_view stream) const;
+template void averageRoundDecimalSum::operator()<__int128_t, 0>(
+    cudf::column_view sumCol,
+    const int64_t* counts,
+    cudf::mutable_column_view outView,
+    cudf::size_type numRows,
+    rmm::cuda_stream_view stream) const;
 
 std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMask(
     const cudf::column_view& sumCol,

From 7051e20c6dde4b412d474151915ff21b7310ee16 Mon Sep 17 00:00:00 2001
From: Simon Eves <seves@nvidia.com>
Date: Fri, 12 Jun 2026 10:12:26 -0700
Subject: [PATCH 8/8] Use C++20 requires/concepts instead of enable_if/traits,
 per @bdice

---
 .../cudf/exec/DecimalAggregationDevice.cu     | 28 ++++++-----
 .../cudf/exec/DecimalAggregationDevice.h      | 50 ++++++++-----------
 2 files changed, 37 insertions(+), 41 deletions(-)

diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
index 37daa5ec2e0..ea56801d86e 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.cu
@@ -234,7 +234,8 @@ std::pair<rmm::device_buffer, cudf::size_type> buildStateValidityMaskImpl(
 
 namespace detail {
 
-template <typename OffsetT, std::enable_if_t<isOffsetStorageType<OffsetT>, int>>
+template <typename OffsetT>
+  requires OffsetStorageType<OffsetT>
 void fillOffsetsForDecimalSumState::operator()(
     cudf::mutable_column_view offsetsView,
     cudf::size_type numRows,
@@ -245,16 +246,17 @@ void fillOffsetsForDecimalSumState::operator()(
       stream);
 }
 
-template void fillOffsetsForDecimalSumState::operator()<int32_t, 0>(
+template void fillOffsetsForDecimalSumState::operator()<int32_t>(
     cudf::mutable_column_view offsetsView,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const;
-template void fillOffsetsForDecimalSumState::operator()<int64_t, 0>(
+template void fillOffsetsForDecimalSumState::operator()<int64_t>(
     cudf::mutable_column_view offsetsView,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const;
 
-template <typename OffsetT, std::enable_if_t<isOffsetStorageType<OffsetT>, int>>
+template <typename OffsetT>
+  requires OffsetStorageType<OffsetT>
 void unpackDecimalSumState::operator()(
     cudf::column_view offsetsView,
     const uint8_t* chars,
@@ -271,14 +273,14 @@ void unpackDecimalSumState::operator()(
       stream);
 }
 
-template void unpackDecimalSumState::operator()<int32_t, 0>(
+template void unpackDecimalSumState::operator()<int32_t>(
     cudf::column_view offsetsView,
     const uint8_t* chars,
     cudf::mutable_column_view sumView,
     cudf::mutable_column_view countView,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const;
-template void unpackDecimalSumState::operator()<int64_t, 0>(
+template void unpackDecimalSumState::operator()<int64_t>(
     cudf::column_view offsetsView,
     const uint8_t* chars,
     cudf::mutable_column_view sumView,
@@ -286,7 +288,8 @@ template void unpackDecimalSumState::operator()<int64_t, 0>(
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const;
 
-template <typename SumT, std::enable_if_t<isDecimalSumStorageType<SumT>, int>>
+template <typename SumT>
+  requires DecimalSumStorageType<SumT>
 void packDecimalSumState::operator()(
     cudf::column_view sumCol,
     const int64_t* counts,
@@ -313,14 +316,14 @@ void packDecimalSumState::operator()(
   }
 }
 
-template void packDecimalSumState::operator()<int64_t, 0>(
+template void packDecimalSumState::operator()<int64_t>(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::column_view offsetsView,
     uint8_t* chars,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const;
-template void packDecimalSumState::operator()<__int128_t, 0>(
+template void packDecimalSumState::operator()<__int128_t>(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::column_view offsetsView,
@@ -328,7 +331,8 @@ template void packDecimalSumState::operator()<__int128_t, 0>(
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const;
 
-template <typename SumT, std::enable_if_t<isDecimalSumStorageType<SumT>, int>>
+template <typename SumT>
+  requires DecimalSumStorageType<SumT>
 void averageRoundDecimalSum::operator()(
     cudf::column_view sumCol,
     const int64_t* counts,
@@ -343,13 +347,13 @@ void averageRoundDecimalSum::operator()(
       stream);
 }
 
-template void averageRoundDecimalSum::operator()<int64_t, 0>(
+template void averageRoundDecimalSum::operator()<int64_t>(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::mutable_column_view outView,
     cudf::size_type numRows,
     rmm::cuda_stream_view stream) const;
-template void averageRoundDecimalSum::operator()<__int128_t, 0>(
+template void averageRoundDecimalSum::operator()<__int128_t>(
     cudf::column_view sumCol,
     const int64_t* counts,
     cudf::mutable_column_view outView,
diff --git a/velox/experimental/cudf/exec/DecimalAggregationDevice.h b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
index e69c0cff3e0..152b3652501 100644
--- a/velox/experimental/cudf/exec/DecimalAggregationDevice.h
+++ b/velox/experimental/cudf/exec/DecimalAggregationDevice.h
@@ -22,20 +22,20 @@
 #include <rmm/device_buffer.hpp>
 #include <rmm/resource_ref.hpp>
 
+#include <concepts>
 #include <cstddef>
 #include <cstdint>
-#include <type_traits>
 #include <utility>
 
 namespace facebook::velox::cudf_velox::detail {
 
 template <typename T>
-inline constexpr bool isDecimalSumStorageType =
-    std::is_same_v<T, int64_t> || std::is_same_v<T, __int128_t>;
+concept OffsetStorageType =
+    std::same_as<T, int32_t> || std::same_as<T, int64_t>;
 
 template <typename T>
-inline constexpr bool isOffsetStorageType =
-    std::is_same_v<T, int32_t> || std::is_same_v<T, int64_t>;
+concept DecimalSumStorageType =
+    std::same_as<T, int64_t> || std::same_as<T, __int128_t>;
 
 // Size in bytes of each row's packed decimal SUM intermediate state in the
 // strings payload (count, overflow placeholder, and 128-bit sum split into
@@ -50,17 +50,15 @@ constexpr size_t kDecimalSumStateSize = 32;
  * @param stream CUDA stream for the launch.
  */
 struct fillOffsetsForDecimalSumState {
-  template <
-      typename OffsetT,
-      std::enable_if_t<isOffsetStorageType<OffsetT>, int> = 0>
+  template <typename OffsetT>
+    requires OffsetStorageType<OffsetT>
   void operator()(
       cudf::mutable_column_view offsetsView,
       cudf::size_type numRows,
       rmm::cuda_stream_view stream) const;
 
-  template <
-      typename OffsetT,
-      std::enable_if_t<!isOffsetStorageType<OffsetT>, int> = 0>
+  template <typename OffsetT>
+    requires(!OffsetStorageType<OffsetT>)
   void operator()(
       cudf::mutable_column_view offsetsView,
       cudf::size_type numRows,
@@ -79,9 +77,8 @@ struct fillOffsetsForDecimalSumState {
  * @param stream CUDA stream for the launch.
  */
 struct packDecimalSumState {
-  template <
-      typename SumT,
-      std::enable_if_t<isDecimalSumStorageType<SumT>, int> = 0>
+  template <typename SumT>
+    requires DecimalSumStorageType<SumT>
   void operator()(
       cudf::column_view sumCol,
       const int64_t* counts,
@@ -90,9 +87,8 @@ struct packDecimalSumState {
       cudf::size_type numRows,
       rmm::cuda_stream_view stream) const;
 
-  template <
-      typename SumT,
-      std::enable_if_t<!isDecimalSumStorageType<SumT>, int> = 0>
+  template <typename SumT>
+    requires(!DecimalSumStorageType<SumT>)
   void operator()(
       cudf::column_view sumCol,
       const int64_t* counts,
@@ -113,9 +109,8 @@ struct packDecimalSumState {
  * @param stream CUDA stream for the launch.
  */
 struct unpackDecimalSumState {
-  template <
-      typename OffsetT,
-      std::enable_if_t<isOffsetStorageType<OffsetT>, int> = 0>
+  template <typename OffsetT>
+    requires OffsetStorageType<OffsetT>
   void operator()(
       cudf::column_view offsetsView,
       const uint8_t* chars,
@@ -124,9 +119,8 @@ struct unpackDecimalSumState {
       cudf::size_type numRows,
       rmm::cuda_stream_view stream) const;
 
-  template <
-      typename OffsetT,
-      std::enable_if_t<!isOffsetStorageType<OffsetT>, int> = 0>
+  template <typename OffsetT>
+    requires(!OffsetStorageType<OffsetT>)
   void operator()(
       cudf::column_view offsetsView,
       const uint8_t* chars,
@@ -147,9 +141,8 @@ struct unpackDecimalSumState {
  * @param stream CUDA stream for the launch.
  */
 struct averageRoundDecimalSum {
-  template <
-      typename SumT,
-      std::enable_if_t<isDecimalSumStorageType<SumT>, int> = 0>
+  template <typename SumT>
+    requires DecimalSumStorageType<SumT>
   void operator()(
       cudf::column_view sumCol,
       const int64_t* counts,
@@ -157,9 +150,8 @@ struct averageRoundDecimalSum {
       cudf::size_type numRows,
       rmm::cuda_stream_view stream) const;
 
-  template <
-      typename SumT,
-      std::enable_if_t<!isDecimalSumStorageType<SumT>, int> = 0>
+  template <typename SumT>
+    requires(!DecimalSumStorageType<SumT>)
   void operator()(
       cudf::column_view sumCol,
       const int64_t* counts,