Skip to content

Use column_device_view instead of raw pointers in Decimal Division Expression Functors #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
simoneves wants to merge 1 commit into
base: simoneves/decimal_pr4
Choose a base branch
from
Open

Use column_device_view instead of raw pointers in Decimal Division Expression Functors #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 18 additions & 65 deletions velox/experimental/cudf/expression/DecimalExpressionKernels.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,10 @@
#include "velox/type/DecimalUtil.h"
#include "velox/type/Type.h"

#include <cudf/binaryop.hpp>
#include <cudf/column/column_factories.hpp>
#include <cudf/copying.hpp>
#include <cudf/fixed_point/fixed_point.hpp>
#include <cudf/null_mask.hpp>
#include <cudf/scalar/scalar_factories.hpp>
#include <cudf/table/table_view.hpp>

namespace facebook::velox::cudf_velox {
namespace {
Expand Down Expand Up @@ -77,52 +74,19 @@ void checkDecimalDivideTypes(
}
}

} // namespace

// Scatters null values to positions where the divisor is zero.
// Returns a new column with nulls at zero-divisor positions.
std::unique_ptr<cudf::column> scatterNullsAtZeroDivisor(
std::unique_ptr<cudf::column> result,
const cudf::column_view& divisor,
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr) {
// Create zero scalar for comparison and null scalar for scattering.
std::unique_ptr<cudf::scalar> zeroScalar;
std::unique_ptr<cudf::scalar> nullScalar;
auto divisorScale = numeric::scale_type{divisor.type().scale()};
auto outputScale = numeric::scale_type{result->type().scale()};

if (divisor.type().id() == cudf::type_id::DECIMAL64) {
zeroScalar = cudf::make_fixed_point_scalar<numeric::decimal64>(
0, divisorScale, stream, mr);
nullScalar = cudf::make_fixed_point_scalar<numeric::decimal64>(
0, outputScale, stream, mr);
} else if (divisor.type().id() == cudf::type_id::DECIMAL128) {
zeroScalar = cudf::make_fixed_point_scalar<numeric::decimal128>(
0, divisorScale, stream, mr);
nullScalar = cudf::make_fixed_point_scalar<numeric::decimal128>(
0, outputScale, stream, mr);
} else {
VELOX_FAIL(
"Unsupported decimal type {} for division",
cudf::ast::type_id_to_string(divisor.type().id()));
void finalizeDivideOutputNullCount(
cudf::column& out,
rmm::cuda_stream_view stream) {
if (out.size() == 0) {
return;
}
nullScalar->set_valid_async(false, stream);

// Create boolean column: TRUE where divisor == 0, FALSE otherwise.
auto divisorIsZero = cudf::binary_operation(
divisor,
*zeroScalar,
cudf::binary_operator::EQUAL,
cudf::data_type{cudf::type_id::BOOL8},
stream,
mr);

// Scatter nulls where divisor is zero.
return cudf::copy_if_else(
*nullScalar, *result, divisorIsZero->view(), stream, mr);
auto const nullCount =
cudf::null_count(out.view().null_mask(), 0, out.size(), stream);
out.set_null_count(nullCount);
}

} // namespace

std::unique_ptr<cudf::column> decimalDivide(
const cudf::column_view& lhs,
const cudf::column_view& rhs,
Expand All @@ -147,13 +111,8 @@ std::unique_ptr<cudf::column> decimalDivide(
const auto outType = outputType.id();
checkDecimalDivideTypes(inType, outType);

// Combine input null masks (lhs and rhs nulls).
auto [nullMask, nullCount] =
cudf::bitmask_and(cudf::table_view({lhs, rhs}), stream, mr);

// Create output column with input null mask and perform division.
auto out = cudf::make_fixed_width_column(
outputType, lhs.size(), std::move(nullMask), nullCount, stream, mr);
outputType, lhs.size(), cudf::mask_state::ALL_VALID, stream, mr);

const __int128_t rescaleFactor = DecimalUtil::kPowersOfTen[aRescale];
VELOX_USER_CHECK(
Expand All @@ -167,8 +126,8 @@ std::unique_ptr<cudf::column> decimalDivide(
stream),
"Decimal overflow");

// Scatter nulls where divisor is zero.
return scatterNullsAtZeroDivisor(std::move(out), rhs, stream, mr);
finalizeDivideOutputNullCount(*out, stream);
return out;
}

std::unique_ptr<cudf::column> decimalDivide(
Expand All @@ -189,10 +148,8 @@ std::unique_ptr<cudf::column> decimalDivide(
return makeAllNullDecimalColumn(outputType, lhs.size(), stream, mr);
}

auto nullMask = cudf::copy_bitmask(lhs, stream, mr);
auto nullCount = lhs.null_count();
auto out = cudf::make_fixed_width_column(
outputType, lhs.size(), std::move(nullMask), nullCount, stream, mr);
outputType, lhs.size(), cudf::mask_state::ALL_VALID, stream, mr);

auto rhsValue = getDecimalScalarValue(rhs, stream);

Expand All @@ -211,6 +168,7 @@ std::unique_ptr<cudf::column> decimalDivide(
stream),
"Decimal overflow");

finalizeDivideOutputNullCount(*out, stream);
return out;
}

Expand All @@ -232,13 +190,8 @@ std::unique_ptr<cudf::column> decimalDivide(
return makeAllNullDecimalColumn(outputType, rhs.size(), stream, mr);
}

// Copy rhs null mask.
auto nullMask = cudf::copy_bitmask(rhs, stream, mr);
auto nullCount = rhs.null_count();

// Create output column and perform division.
auto out = cudf::make_fixed_width_column(
outputType, rhs.size(), std::move(nullMask), nullCount, stream, mr);
outputType, rhs.size(), cudf::mask_state::ALL_VALID, stream, mr);

auto lhsValue = getDecimalScalarValue(lhs, stream);

Expand All @@ -258,8 +211,8 @@ std::unique_ptr<cudf::column> decimalDivide(
stream),
"Decimal overflow");

// Scatter nulls where divisor is zero.
return scatterNullsAtZeroDivisor(std::move(out), rhs, stream, mr);
finalizeDivideOutputNullCount(*out, stream);
return out;
}

} // namespace facebook::velox::cudf_velox
34 changes: 6 additions & 28 deletions velox/experimental/cudf/expression/DecimalExpressionKernels.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,8 @@ namespace facebook::velox::cudf_velox {
/**
* @brief Element-wise decimal division of two columns.
*
* Builds the output null mask as the bitwise AND of lhs and rhs validity, runs
* the GPU divide into outputType, and applies scatterNullsAtZeroDivisor so
* rows with a zero divisor are null.
* Runs the GPU divide into outputType. The kernel propagates input nulls and
* nulls rows where the divisor is zero.
*
* @param lhs Left-hand decimal operand column (DECIMAL64 or DECIMAL128).
* @param rhs Right-hand decimal operand column (same type as lhs).
Expand All @@ -54,8 +53,8 @@ std::unique_ptr<cudf::column> decimalDivide(
* @brief Element-wise decimal division of a column by a scalar.
*
* If the scalar is invalid, returns an all-null column of outputType;
* otherwise copies lhs nulls and divides without zero-divisor scattering (rhs
* is not per-row).
* otherwise the kernel propagates lhs nulls and nulls rows when dividing by
* zero.
*
* @param lhs Left-hand decimal operand column.
* @param rhs Right-hand decimal operand scalar.
Expand All @@ -77,9 +76,8 @@ std::unique_ptr<cudf::column> decimalDivide(
/**
* @brief Element-wise decimal division of a scalar by a column.
*
* Invalid lhs yields all-null output; otherwise rhs nulls are propagated, then
* divide and scatterNullsAtZeroDivisor on rhs so division-by-zero rows are
* null.
* Invalid lhs yields all-null output; otherwise the kernel propagates rhs
* nulls and nulls rows where the divisor is zero.
*
* @param lhs Left-hand decimal operand scalar.
* @param rhs Right-hand decimal operand column.
Expand All @@ -98,24 +96,4 @@ std::unique_ptr<cudf::column> decimalDivide(
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr);

/**
* @brief Nulls output rows where the divisor column equals zero.
*
* After a decimal divide, forces output rows to null where the divisor column
* compares equal to zero (DECIMAL64 or DECIMAL128), using copy_if_else. Kept in
* the .cpp translation unit so it can use Velox checks alongside cuDF APIs
* without pulling those into CUDA compilation units.
*
* @param result Column produced by decimal division.
* @param divisor Per-row divisor column used to detect division by zero.
* @param stream CUDA stream for GPU execution.
* @param mr Memory resource for output allocation.
* @return Column with nulls scattered at zero-divisor rows.
*/
std::unique_ptr<cudf::column> scatterNullsAtZeroDivisor(
std::unique_ptr<cudf::column> result,
const cudf::column_view& divisor,
rmm::cuda_stream_view stream,
rmm::device_async_resource_ref mr);

} // namespace facebook::velox::cudf_velox
71 changes: 39 additions & 32 deletions velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.cu
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

#include "velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h"

#include <cudf/column/column_device_view.cuh>
#include <cudf/utilities/error.hpp>
#include <cudf/utilities/type_dispatcher.hpp>

Expand Down Expand Up @@ -101,10 +102,6 @@ __device__ OutT decimalDivideImpl(
__int128_t denom,
__int128_t rescaleFactor,
int32_t* overflowFlag) {
if (denom == 0) {
return OutT{0};
}

bool negative = false;
unsigned __int128 const uNum = absToUnsigned(numerator, negative);
unsigned __int128 const uDenom = absToUnsigned(denom, negative);
Expand Down Expand Up @@ -151,43 +148,58 @@ __device__ OutT decimalDivideImpl(

template <typename InT, typename OutT>
struct DivideFunctor {
const InT* lhs;
const InT* rhs;
OutT* out;
cudf::column_device_view lhs;
cudf::column_device_view rhs;
cudf::mutable_column_device_view out;
__int128_t rescaleFactor;
int32_t* overflowFlag;

__device__ void operator()(cudf::size_type idx) const {
out[idx] = decimalDivideImpl<OutT>(
lhs[idx], rhs[idx], rescaleFactor, overflowFlag);
if (lhs.is_null(idx) || rhs.is_null(idx) || rhs.element<InT>(idx) == 0) {
out.set_null(idx);
return;
}
out.element<OutT>(idx) = decimalDivideImpl<OutT>(
lhs.element<InT>(idx),
rhs.element<InT>(idx),
rescaleFactor,
overflowFlag);
}
};

template <typename InColT, typename OutT>
struct DivideLhsScalarFunctor {
__int128_t lhsValue;
const InColT* rhs;
OutT* out;
cudf::column_device_view rhs;
cudf::mutable_column_device_view out;
__int128_t rescaleFactor;
int32_t* overflowFlag;

__device__ void operator()(cudf::size_type idx) const {
out[idx] = decimalDivideImpl<OutT>(
lhsValue, rhs[idx], rescaleFactor, overflowFlag);
if (rhs.is_null(idx) || rhs.element<InColT>(idx) == 0) {
out.set_null(idx);
return;
}
out.element<OutT>(idx) = decimalDivideImpl<OutT>(
lhsValue, rhs.element<InColT>(idx), rescaleFactor, overflowFlag);
}
};

template <typename InColT, typename OutT>
struct DivideRhsScalarFunctor {
const InColT* lhs;
cudf::column_device_view lhs;
__int128_t rhsValue;
OutT* out;
cudf::mutable_column_device_view out;
__int128_t rescaleFactor;
int32_t* overflowFlag;

__device__ void operator()(cudf::size_type idx) const {
out[idx] = decimalDivideImpl<OutT>(
lhs[idx], rhsValue, rescaleFactor, overflowFlag);
if (lhs.is_null(idx) || rhsValue == 0) {
out.set_null(idx);
return;
}
out.element<OutT>(idx) = decimalDivideImpl<OutT>(
lhs.element<InColT>(idx), rhsValue, rescaleFactor, overflowFlag);
}
};

Expand Down Expand Up @@ -228,15 +240,14 @@ struct divideColumnColumnKernel {
template <typename InT, typename OutT>
requires ValidDecimalDivideStorageTypes<InT, OutT>
bool operator()() const {
auto lhsDev = cudf::column_device_view::create(lhs, stream);
auto rhsDev = cudf::column_device_view::create(rhs, stream);
auto outDev = cudf::mutable_column_device_view::create(out, stream);
return launchDecimalDivide(
lhs.size(),
[&](int32_t* overflowFlag) {
return DivideFunctor<InT, OutT>{
lhs.data<InT>(),
rhs.data<InT>(),
out.data<OutT>(),
rescaleFactor,
overflowFlag};
*lhsDev, *rhsDev, *outDev, rescaleFactor, overflowFlag};
},
stream);
}
Expand All @@ -259,15 +270,13 @@ struct divideColumnScalarKernel {
template <typename InT, typename OutT>
requires ValidDecimalDivideStorageTypes<InT, OutT>
bool operator()() const {
auto lhsDev = cudf::column_device_view::create(lhs, stream);
auto outDev = cudf::mutable_column_device_view::create(out, stream);
return launchDecimalDivide(
lhs.size(),
[&](int32_t* overflowFlag) {
return DivideRhsScalarFunctor<InT, OutT>{
lhs.data<InT>(),
rhsValue,
out.data<OutT>(),
rescaleFactor,
overflowFlag};
*lhsDev, rhsValue, *outDev, rescaleFactor, overflowFlag};
},
stream);
}
Expand All @@ -290,15 +299,13 @@ struct divideScalarColumnKernel {
template <typename InT, typename OutT>
requires ValidDecimalDivideStorageTypes<InT, OutT>
bool operator()() const {
auto rhsDev = cudf::column_device_view::create(rhs, stream);
auto outDev = cudf::mutable_column_device_view::create(out, stream);
return launchDecimalDivide(
rhs.size(),
[&](int32_t* overflowFlag) {
return DivideLhsScalarFunctor<InT, OutT>{
lhsValue,
rhs.data<InT>(),
out.data<OutT>(),
rescaleFactor,
overflowFlag};
lhsValue, *rhsDev, *outDev, rescaleFactor, overflowFlag};
},
stream);
}
Expand Down
4 changes: 2 additions & 2 deletions velox/experimental/cudf/expression/DecimalExpressionKernelsGpu.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@ namespace facebook::velox::cudf_velox::detail {
* @brief Dispatches a per-row device loop for fixed-point decimal division.
*
* Computes (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).
* remainder, writing into out. Input nulls and zero divisors are written as
* null in the output null mask.
*
* @param inType DECIMAL64 or DECIMAL128 input storage width selector.
* @param outType DECIMAL64 or DECIMAL128 output storage width selector.
Expand Down
Loading