diff --git a/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp b/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp index a419bf5981d..92612c4ac17 100644 --- a/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp +++ b/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp @@ -18,6 +18,8 @@ #include "velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h" #include "velox/common/base/Exceptions.h" +#include "velox/type/DecimalUtil.h" +#include "velox/type/Type.h" #include #include @@ -118,6 +120,10 @@ std::unique_ptr decimalDivide( "Decimal divide requires matching input types"); VELOX_CHECK_GE( aRescale, 0, "Decimal divide requires non-negative rescale factor"); + // Rescale indexes DecimalUtil::kPowersOfTen; same bound as Presto divide + // init. + VELOX_USER_CHECK_LE( + aRescale, LongDecimalType::kMaxPrecision, "Decimal overflow"); const auto inType = lhs.type().id(); const auto outType = outputType.id(); @@ -143,8 +149,17 @@ std::unique_ptr decimalDivide( auto out = cudf::make_fixed_width_column( outputType, lhs.size(), std::move(nullMask), nullCount, stream, mr); - detail::decimalDivideColumnColumn( - inType, outType, lhs, rhs, out->mutable_view(), aRescale, stream); + const __int128_t rescaleFactor = DecimalUtil::kPowersOfTen[aRescale]; + VELOX_USER_CHECK( + detail::decimalDivideColumnColumn( + inType, + outType, + lhs, + rhs, + out->mutable_view(), + rescaleFactor, + stream), + "Decimal overflow"); // Scatter nulls where divisor is zero. return scatterNullsAtZeroDivisor(std::move(out), rhs, stream, mr); @@ -159,6 +174,10 @@ std::unique_ptr decimalDivide( rmm::device_async_resource_ref mr) { VELOX_CHECK_GE( aRescale, 0, "Decimal divide requires non-negative rescale factor"); + // Rescale indexes DecimalUtil::kPowersOfTen; same bound as Presto divide + // init. + VELOX_USER_CHECK_LE( + aRescale, LongDecimalType::kMaxPrecision, "Decimal overflow"); if (!rhs.is_valid(stream)) { return makeAllNullDecimalColumn(outputType, lhs.size(), stream, mr); @@ -187,8 +206,16 @@ std::unique_ptr decimalDivide( "Unexpected output type for decimal divide"); } - detail::decimalDivideColumnScalar( - inType, outType, lhs, rhsValue, out->mutable_view(), aRescale, stream); + VELOX_USER_CHECK( + detail::decimalDivideColumnScalar( + inType, + outType, + lhs, + rhsValue, + out->mutable_view(), + DecimalUtil::kPowersOfTen[aRescale], + stream), + "Decimal overflow"); return out; } @@ -202,6 +229,10 @@ std::unique_ptr decimalDivide( rmm::device_async_resource_ref mr) { VELOX_CHECK_GE( aRescale, 0, "Decimal divide requires non-negative rescale factor"); + // Rescale indexes DecimalUtil::kPowersOfTen; same bound as Presto divide + // init. + VELOX_USER_CHECK_LE( + aRescale, LongDecimalType::kMaxPrecision, "Decimal overflow"); if (!lhs.is_valid(stream)) { return makeAllNullDecimalColumn(outputType, rhs.size(), stream, mr); @@ -233,8 +264,17 @@ std::unique_ptr decimalDivide( "Unexpected output type for decimal divide"); } - detail::decimalDivideScalarColumn( - inType, outType, lhsValue, rhs, out->mutable_view(), aRescale, stream); + const __int128_t rescaleFactor = DecimalUtil::kPowersOfTen[aRescale]; + VELOX_USER_CHECK( + detail::decimalDivideScalarColumn( + inType, + outType, + lhsValue, + rhs, + out->mutable_view(), + rescaleFactor, + stream), + "Decimal overflow"); // Scatter nulls where divisor is zero. return scatterNullsAtZeroDivisor(std::move(out), rhs, stream, mr); diff --git a/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.cu b/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.cu index 29b7c13f59d..1d51221f55f 100644 --- a/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.cu +++ b/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.cu @@ -17,49 +17,136 @@ #include "velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h" #include +#include + +#include #include +#include #include -#include +#include #include namespace facebook::velox::cudf_velox { namespace { +// Device-safe int128 bounds (std::numeric_limits is host-only in CUDA). +constexpr unsigned __int128 kUnsigned128Max = + static_cast(-1); +constexpr unsigned __int128 kInt128MinMagnitude = + static_cast(1) << 127; +constexpr unsigned __int128 kInt128MaxMagnitude = kInt128MinMagnitude - 1; +// Bit pattern 2^127 maps to INT128_MIN without negating INT128_MIN (UB). +constexpr __int128_t kInt128Min = static_cast<__int128_t>(kInt128MinMagnitude); + +// Match DecimalUtil::kLongDecimal{Min,Max} (10^38 bounds); duplicated here +// because Velox headers cannot be included in this translation unit (nvcc). +constexpr __int128_t kLongDecimalPowerOfTen38 = + 1'000'000'000'000'000'000LL * (__int128_t)1'000'000'000'000'000'000LL * 100; +constexpr __int128_t kLongDecimalMax = kLongDecimalPowerOfTen38 - 1; +constexpr __int128_t kLongDecimalMin = -kLongDecimalPowerOfTen38 + 1; + +// Device threads cannot throw; record overflow for launchDecimalDivide to +// report to the host caller, matching Velox CPU decimal divide errors. +__device__ inline void markDecimalOverflow(int32_t* overflowFlag) { + atomicOr(overflowFlag, 1); +} + +// Extract absolute value in unsigned space. Signed negation of INT128_MIN is +// undefined; negating the unsigned bit pattern is always defined. +__device__ inline unsigned __int128 absToUnsigned( + __int128_t value, + bool& negative) { + if (value < 0) { + negative = !negative; + return -static_cast(value); + } + return static_cast(value); +} + +// Reapply sign after unsigned divide/round. Caller must ensure magnitude fits. +__device__ inline __int128_t signedFromUnsigned( + unsigned __int128 magnitude, + bool negative) { + if (!negative) { + return static_cast<__int128_t>(magnitude); + } + if (magnitude >= kInt128MinMagnitude) { + return kInt128Min; + } + return -static_cast<__int128_t>(magnitude); +} + +// Quotient magnitude must fit in int128 before signedFromUnsigned; rounding can +// push a representable unsigned quotient past INT128_MAX / INT128_MIN. +__device__ inline bool fitsRepresentableInt128( + unsigned __int128 magnitude, + bool negative) { + if (!negative) { + return magnitude <= kInt128MaxMagnitude; + } + return magnitude <= kInt128MinMagnitude; +} + +// Decimal divide with rescale (numerator * rescaleFactor / denom). Rounding +// matches Velox CPU DecimalUtil::divideWithRoundUp (increment unsigned +// quotient, then apply sign), not Java/Hive HALF_UP toward +infinity on ties. +// Overflow on rescale multiply, round-up, or out-of-range results sets +// overflowFlag (see launchDecimalDivide); intermediate math uses unsigned +// magnitudes so multiply, divide, mod, and abs never hit signed overflow UB. template -__device__ OutT -decimalDivideImpl(__int128_t numerator, __int128_t denom, __int128_t scale) { +__device__ OutT decimalDivideImpl( + __int128_t numerator, + __int128_t denom, + __int128_t rescaleFactor, + int32_t* overflowFlag) { if (denom == 0) { return OutT{0}; } - int sign = 1; - if (numerator < 0) { - numerator = -numerator; - sign = -sign; - } - if (denom < 0) { - denom = -denom; - sign = -sign; + + bool negative = false; + unsigned __int128 const uNum = absToUnsigned(numerator, negative); + unsigned __int128 const uDenom = absToUnsigned(denom, negative); + // rescaleFactor is DecimalUtil::kPowersOfTen[aRescale] from the host caller. + unsigned __int128 const uRescaleFactor = + static_cast(rescaleFactor); + + unsigned __int128 scaled = uNum * uRescaleFactor; + // Match Velox CPU checkedMultiply on rescale. + if (uRescaleFactor != 0 && scaled / uRescaleFactor != uNum) { + markDecimalOverflow(overflowFlag); + return OutT{0}; } - __int128_t scaled = numerator * scale; - __int128_t quotient = scaled / denom; - __int128_t remainder = scaled % denom; - if (remainder * 2 >= denom) { + + unsigned __int128 quotient = scaled / uDenom; + unsigned __int128 const remainder = scaled % uDenom; + + // Round ties away from zero (e.g. -1.5 -> -2), same as Velox CPU divide. + // Equivalent to 2 * remainder >= denom but avoids overflow when remainder is + // large. + if (remainder > (uDenom - 1) / 2) { + // Round-up would wrap unsigned quotient; CPU path would overflow too. + if (quotient >= kUnsigned128Max) { + markDecimalOverflow(overflowFlag); + return OutT{0}; + } ++quotient; } - if (sign < 0) { - quotient = -quotient; + + if (!fitsRepresentableInt128(quotient, negative)) { + markDecimalOverflow(overflowFlag); + return OutT{0}; } - return static_cast(quotient); -} -inline __int128_t pow10Int128(int32_t exp) { - __int128_t value = 1; - for (int32_t i = 0; i < exp; ++i) { - value *= 10; + __int128_t const result = signedFromUnsigned(quotient, negative); + // Match Velox CPU DecimalUtil::valueInRange after divide. + if (result < kLongDecimalMin || result > kLongDecimalMax) { + markDecimalOverflow(overflowFlag); + return OutT{0}; } - return value; + + return static_cast(result); } template @@ -67,10 +154,12 @@ struct DivideFunctor { const InT* lhs; const InT* rhs; OutT* out; - __int128_t scale; + __int128_t rescaleFactor; + int32_t* overflowFlag; - __device__ void operator()(int32_t idx) const { - out[idx] = decimalDivideImpl(lhs[idx], rhs[idx], scale); + __device__ void operator()(cudf::size_type idx) const { + out[idx] = decimalDivideImpl( + lhs[idx], rhs[idx], rescaleFactor, overflowFlag); } }; @@ -79,10 +168,12 @@ struct DivideLhsScalarFunctor { __int128_t lhsValue; const InColT* rhs; OutT* out; - __int128_t scale; + __int128_t rescaleFactor; + int32_t* overflowFlag; - __device__ void operator()(int32_t idx) const { - out[idx] = decimalDivideImpl(lhsValue, rhs[idx], scale); + __device__ void operator()(cudf::size_type idx) const { + out[idx] = decimalDivideImpl( + lhsValue, rhs[idx], rescaleFactor, overflowFlag); } }; @@ -91,132 +182,172 @@ struct DivideRhsScalarFunctor { const InColT* lhs; __int128_t rhsValue; OutT* out; - __int128_t scale; + __int128_t rescaleFactor; + int32_t* overflowFlag; - __device__ void operator()(int32_t idx) const { - out[idx] = decimalDivideImpl(lhs[idx], rhsValue, scale); + __device__ void operator()(cudf::size_type idx) const { + out[idx] = decimalDivideImpl( + lhs[idx], rhsValue, rescaleFactor, overflowFlag); } }; -template -void launchDivideKernel( - const cudf::column_view& lhs, - const cudf::column_view& rhs, - cudf::mutable_column_view out, - int32_t aRescale, +// Returns false if any row set overflowFlag during the kernel. +template +bool launchDecimalDivide( + cudf::size_type size, + BuildOp buildOp, rmm::cuda_stream_view stream) { - if (lhs.size() == 0) { - return; - } - DivideFunctor op{ - lhs.data(), - rhs.data(), - out.data(), - pow10Int128(aRescale)}; + if (size == 0) { + return true; + } + rmm::device_scalar overflowFlag{0, stream}; + auto op = buildOp(overflowFlag.data()); cub::DeviceFor::ForEachN( - thrust::counting_iterator(0), lhs.size(), op, stream.value()); + cuda::counting_iterator{0}, size, op, stream.value()); CUDF_CUDA_TRY(cudaGetLastError()); + return overflowFlag.value(stream) == 0; } -template -void launchDivideKernelLhsScalar( - __int128_t lhsValue, - const cudf::column_view& rhs, - cudf::mutable_column_view out, - int32_t aRescale, - rmm::cuda_stream_view stream) { - if (rhs.size() == 0) { - return; +} // namespace + +namespace detail { + +template +concept ValidDecimalDivideStorageTypes = + (std::same_as && + (std::same_as || std::same_as)) || + (std::same_as && std::same_as); + +struct divideColumnColumnKernel { + const cudf::column_view& lhs; + const cudf::column_view& rhs; + cudf::mutable_column_view out; + __int128_t rescaleFactor; + rmm::cuda_stream_view stream; + + template + requires ValidDecimalDivideStorageTypes + bool operator()() const { + return launchDecimalDivide( + lhs.size(), + [&](int32_t* overflowFlag) { + return DivideFunctor{ + lhs.data(), + rhs.data(), + out.data(), + rescaleFactor, + overflowFlag}; + }, + stream); } - DivideLhsScalarFunctor op{ - lhsValue, rhs.data(), out.data(), pow10Int128(aRescale)}; - cub::DeviceFor::ForEachN( - thrust::counting_iterator(0), rhs.size(), op, stream.value()); - CUDF_CUDA_TRY(cudaGetLastError()); -} -template -void launchDivideKernelRhsScalar( - const cudf::column_view& lhs, - __int128_t rhsValue, - cudf::mutable_column_view out, - int32_t aRescale, - rmm::cuda_stream_view stream) { - if (lhs.size() == 0) { - return; + template + requires(!ValidDecimalDivideStorageTypes) + bool operator()() const { + return true; } - DivideRhsScalarFunctor op{ - lhs.data(), rhsValue, out.data(), pow10Int128(aRescale)}; - cub::DeviceFor::ForEachN( - thrust::counting_iterator(0), lhs.size(), op, stream.value()); - CUDF_CUDA_TRY(cudaGetLastError()); -} +}; -} // namespace +struct divideColumnScalarKernel { + const cudf::column_view& lhs; + __int128_t rhsValue; + cudf::mutable_column_view out; + __int128_t rescaleFactor; + rmm::cuda_stream_view stream; -namespace detail { + template + requires ValidDecimalDivideStorageTypes + bool operator()() const { + return launchDecimalDivide( + lhs.size(), + [&](int32_t* overflowFlag) { + return DivideRhsScalarFunctor{ + lhs.data(), + rhsValue, + out.data(), + rescaleFactor, + overflowFlag}; + }, + stream); + } + + template + requires(!ValidDecimalDivideStorageTypes) + bool operator()() const { + return true; + } +}; + +struct divideScalarColumnKernel { + __int128_t lhsValue; + const cudf::column_view& rhs; + cudf::mutable_column_view out; + __int128_t rescaleFactor; + rmm::cuda_stream_view stream; + + template + requires ValidDecimalDivideStorageTypes + bool operator()() const { + return launchDecimalDivide( + rhs.size(), + [&](int32_t* overflowFlag) { + return DivideLhsScalarFunctor{ + lhsValue, + rhs.data(), + out.data(), + rescaleFactor, + overflowFlag}; + }, + stream); + } -void decimalDivideColumnColumn( + template + requires(!ValidDecimalDivideStorageTypes) + bool operator()() const { + return true; + } +}; + +bool decimalDivideColumnColumn( cudf::type_id inType, cudf::type_id outType, const cudf::column_view& lhs, const cudf::column_view& rhs, cudf::mutable_column_view out, - int32_t aRescale, + __int128_t rescaleFactor, rmm::cuda_stream_view stream) { - if (inType == cudf::type_id::DECIMAL64) { - if (outType == cudf::type_id::DECIMAL64) { - launchDivideKernel(lhs, rhs, out, aRescale, stream); - } else { - launchDivideKernel(lhs, rhs, out, aRescale, stream); - } - } else { - launchDivideKernel<__int128_t, __int128_t>(lhs, rhs, out, aRescale, stream); - } + return cudf::double_type_dispatcher( + cudf::data_type{inType}, + cudf::data_type{outType}, + divideColumnColumnKernel{lhs, rhs, out, rescaleFactor, stream}); } -void decimalDivideColumnScalar( +bool decimalDivideColumnScalar( cudf::type_id inType, cudf::type_id outType, const cudf::column_view& lhs, __int128_t rhsValue, cudf::mutable_column_view out, - int32_t aRescale, + __int128_t rescaleFactor, rmm::cuda_stream_view stream) { - if (inType == cudf::type_id::DECIMAL64) { - if (outType == cudf::type_id::DECIMAL64) { - launchDivideKernelRhsScalar( - lhs, rhsValue, out, aRescale, stream); - } else { - launchDivideKernelRhsScalar( - lhs, rhsValue, out, aRescale, stream); - } - } else { - launchDivideKernelRhsScalar<__int128_t, __int128_t>( - lhs, rhsValue, out, aRescale, stream); - } + return cudf::double_type_dispatcher( + cudf::data_type{inType}, + cudf::data_type{outType}, + divideColumnScalarKernel{lhs, rhsValue, out, rescaleFactor, stream}); } -void decimalDivideScalarColumn( +bool decimalDivideScalarColumn( cudf::type_id inType, cudf::type_id outType, __int128_t lhsValue, const cudf::column_view& rhs, cudf::mutable_column_view out, - int32_t aRescale, + __int128_t rescaleFactor, rmm::cuda_stream_view stream) { - if (inType == cudf::type_id::DECIMAL64) { - if (outType == cudf::type_id::DECIMAL64) { - launchDivideKernelLhsScalar( - lhsValue, rhs, out, aRescale, stream); - } else { - launchDivideKernelLhsScalar( - lhsValue, rhs, out, aRescale, stream); - } - } else { - launchDivideKernelLhsScalar<__int128_t, __int128_t>( - lhsValue, rhs, out, aRescale, stream); - } + return cudf::double_type_dispatcher( + cudf::data_type{inType}, + cudf::data_type{outType}, + divideScalarColumnKernel{lhsValue, rhs, out, rescaleFactor, stream}); } } // namespace detail diff --git a/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h b/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h index 310be3ee834..0093af13916 100644 --- a/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h +++ b/velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h @@ -24,39 +24,42 @@ namespace facebook::velox::cudf_velox::detail { -// Dispatches a per-row device loop: fixed-point divide (lhs * 10^aRescale) / +// Dispatches a per-row device loop: fixed-point divide (lhs * rescaleFactor) / // rhs with half-away-from-zero rounding on the remainder, writing into out. -// Zero divisors produce a numeric zero in out (callers patch nulls). inType / -// outType select DECIMAL64 vs DECIMAL128 storage widths for inputs and result. -void decimalDivideColumnColumn( +// rescaleFactor is DecimalUtil::kPowersOfTen[aRescale] from the caller. +// Zero divisors produce a numeric zero in out (callers patch nulls). Returns +// false if any row overflowed (caller should VELOX_USER_FAIL). inType / outType +// select DECIMAL64 vs DECIMAL128 storage widths via +// cudf::double_type_dispatcher. +bool decimalDivideColumnColumn( cudf::type_id inType, cudf::type_id outType, const cudf::column_view& lhs, const cudf::column_view& rhs, cudf::mutable_column_view out, - int32_t aRescale, + __int128_t rescaleFactor, rmm::cuda_stream_view stream); // Same kernel math as decimalDivideColumnColumn, but rhs is a single // __int128_t decimal payload (already decoded from a cuDF scalar). -void decimalDivideColumnScalar( +bool decimalDivideColumnScalar( cudf::type_id inType, cudf::type_id outType, const cudf::column_view& lhs, __int128_t rhsValue, cudf::mutable_column_view out, - int32_t aRescale, + __int128_t rescaleFactor, rmm::cuda_stream_view stream); // Same kernel math as decimalDivideColumnColumn, but lhs is a single // __int128_t decimal payload and rhs is per-row. -void decimalDivideScalarColumn( +bool decimalDivideScalarColumn( cudf::type_id inType, cudf::type_id outType, __int128_t lhsValue, const cudf::column_view& rhs, cudf::mutable_column_view out, - int32_t aRescale, + __int128_t rescaleFactor, rmm::cuda_stream_view stream); } // namespace facebook::velox::cudf_velox::detail