From 0587db538e28e95561091c5482bc8b0071f53339 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 14:28:55 -0700 Subject: [PATCH 01/12] fix(arrow): Import Arrow DECIMAL32 into ShortDecimalType MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GPU table scans and other Arrow producers can hand Velox schemas like `d:9,2,32` for small SQL decimals such as TPC-DS `DECIMAL(9,2)`. The Arrow bridge rejected that format during schema import: Conversion failed for 'd:9,2,32'. Only 64-bit and 128-bit decimal types are supported. That failure blocked the cuDF-to-Velox conversion path (`toVeloxColumn()` / `importFromArrowAsOwner()`), which runs when GPU operators must emit CPU Velox vectors. libcudf stores low-precision parquet decimals as `DECIMAL32` and exports them through NanoArrow with an explicit 32-bit width suffix. Velox has no native 32-bit decimal type: `DECIMAL(p,s)` with `p <= 18` is always `ShortDecimalType` backed by `int64_t` unscaled values. Teach the Arrow bridge to accept `bitWidth == 32` alongside the existing 64- and 128-bit paths. Schema import maps `d:p,s,32` to `ShortDecimalType` the same way `d:p,s,64` does. Array import reads the Arrow `int32_t` values buffer and widens each value into a new `int64_t` Velox buffer, matching the sign-extension approach the CPU Parquet reader uses for `INT32` physical decimals. Export is unchanged: Velox still emits `d:p,s` or `d:p,s,64` depending on `useDecimalTypeWidth`. Test plan: `velox_arrow_bridge_test` — schema import for `d:9,2,32` and array import with ordinary values, nulls, and `INT32_MIN`/`INT32_MAX`. --- velox/vector/arrow/Bridge.cpp | 55 ++++++++++++++----- .../arrow/tests/ArrowBridgeArrayTest.cpp | 26 +++++++++ .../arrow/tests/ArrowBridgeSchemaTest.cpp | 3 +- 3 files changed, 68 insertions(+), 16 deletions(-) diff --git a/velox/vector/arrow/Bridge.cpp b/velox/vector/arrow/Bridge.cpp index 1ccbd8b6f12..09a0b32bdda 100644 --- a/velox/vector/arrow/Bridge.cpp +++ b/velox/vector/arrow/Bridge.cpp @@ -1396,7 +1396,7 @@ void exportToArrowImpl( // Parses the velox decimal format from the given arrow format. // The input format string should be in the form "d:precision,scale<,bitWidth>". -// bitWidth is optional and may be 64 or 128 if provided. +// bitWidth is optional and may be 32, 64, or 128 if provided. int32_t parseDecimalBitWidthOrDefault(const std::string_view format) { auto firstCommaIdx = format.find(',', 2); @@ -1431,16 +1431,16 @@ TypePtr parseDecimalFormat(const std::string_view format) { int precision = std::stoi(&format[2], &sz); int scale = std::stoi(&format[firstCommaIdx + 1], &sz); if (secondCommaIdx != std::string_view::npos) { - // BitWidth is provided. We only support 64 or 128. + // BitWidth is provided. We support 32, 64, and 128. int bitWidth = std::stoi(&format[secondCommaIdx + 1], &sz); // Return type depends on bitWidth. - if (bitWidth == 64) { + if (bitWidth == 32 || bitWidth == 64) { return std::make_shared(precision, scale); } else if (bitWidth == 128) { return std::make_shared(precision, scale); } VELOX_USER_FAIL( - "Conversion failed for '{}'. Only 64-bit and 128-bit decimal types are supported.", + "Conversion failed for '{}'. Only 32-bit, 64-bit, and 128-bit decimal types are supported.", format); } // Otherwise return type depends on precision. @@ -2258,6 +2258,23 @@ VectorPtr createShortDecimalVectorFromLongDecimals( pool, type, std::move(nulls), length, values, nullCount); } +VectorPtr createShortDecimalVectorFrom32BitDecimals( + memory::MemoryPool* pool, + const TypePtr& type, + BufferPtr nulls, + const int32_t* input, + vector_size_t length, + int64_t nullCount) { + auto values = AlignedBuffer::allocate(length, pool); + auto rawValues = values->asMutable(); + for (size_t i = 0; i < length; ++i) { + rawValues[i] = input[i]; + } + + return createFlatVector( + pool, type, std::move(nulls), length, values, nullCount); +} + // Arrow uses two uint64_t values to represent a 128-bit decimal value. The // memory allocated by Arrow might not be 16-byte aligned, so we need to copy // the values to a new buffer to ensure 16-byte alignment. @@ -2408,6 +2425,15 @@ VectorPtr importFromArrowImpl( } else if (type->isShortDecimal()) { // Validate the format bitWidth. const auto bitWidth = parseDecimalBitWidthOrDefault(arrowSchema.format); + if (bitWidth == 32) { + return createShortDecimalVectorFrom32BitDecimals( + pool, + type, + nulls, + static_cast(arrowArray.buffers[1]), + arrowArray.length, + arrowArray.null_count); + } if (bitWidth == 64) { return createShortDecimalVector( pool, @@ -2418,20 +2444,19 @@ VectorPtr importFromArrowImpl( arrowArray.null_count, wrapInBufferView); } - // Otherwise convert to 128. - VELOX_USER_CHECK_EQ( - bitWidth, - 128, + if (bitWidth == 128) { + return createShortDecimalVectorFromLongDecimals( + pool, + type, + nulls, + static_cast(arrowArray.buffers[1]), + arrowArray.length, + arrowArray.null_count); + } + VELOX_USER_FAIL( "Unsupported decimal bitWidth {} for '{}'", bitWidth, arrowSchema.format); - return createShortDecimalVectorFromLongDecimals( - pool, - type, - nulls, - static_cast(arrowArray.buffers[1]), - arrowArray.length, - arrowArray.null_count); } else if (type->isLongDecimal()) { // Validate that the format is actually 128. const int32_t bitWidth = parseDecimalBitWidthOrDefault(arrowSchema.format); diff --git a/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp b/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp index e681e2073c9..f8186190e5d 100644 --- a/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp +++ b/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp @@ -1272,6 +1272,22 @@ class ArrowBridgeArrayImportTest : public ArrowBridgeArrayExportTest { std::is_same_v && std::is_same_v) { assertShortDecimalVectorContent( inputValues, output, arrowArray.null_count); + } else if constexpr ( + std::is_same_v && std::is_same_v) { + if (format[0] == 'd') { + std::vector> widenedValues; + widenedValues.reserve(inputValues.size()); + for (const auto& value : inputValues) { + if (value.has_value()) { + widenedValues.emplace_back(static_cast(value.value())); + } else { + widenedValues.emplace_back(std::nullopt); + } + } + assertVectorContent(widenedValues, output, arrowArray.null_count); + } else { + assertVectorContent(inputValues, output, arrowArray.null_count); + } } else if constexpr ( std::is_same_v && (std::is_same_v || std::is_same_v)) { @@ -1393,6 +1409,16 @@ class ArrowBridgeArrayImportTest : public ArrowBridgeArrayExportTest { testArrowImport( "d:5,2", {1, -1, 0, 12345, -12345, std::nullopt}); + testArrowImport( + "d:9,2,32", + {1, + -1, + 0, + 12345, + -12345, + std::numeric_limits::max(), + std::numeric_limits::min(), + std::nullopt}); testArrowImport( "d:36,2", {HugeInt::parse("20000000000000000"), diff --git a/velox/vector/arrow/tests/ArrowBridgeSchemaTest.cpp b/velox/vector/arrow/tests/ArrowBridgeSchemaTest.cpp index 7a581e9a87a..5f35d5d46f1 100644 --- a/velox/vector/arrow/tests/ArrowBridgeSchemaTest.cpp +++ b/velox/vector/arrow/tests/ArrowBridgeSchemaTest.cpp @@ -509,10 +509,11 @@ TEST_F(ArrowBridgeSchemaImportTest, scalar) { *testSchemaImport("d2,15"), "Unable to convert 'd2,15' ArrowSchema decimal format to Velox decimal"); EXPECT_EQ(*DECIMAL(10, 4), *testSchemaImport("d:10,4,64")); + EXPECT_EQ(*DECIMAL(9, 2), *testSchemaImport("d:9,2,32")); EXPECT_EQ(*DECIMAL(20, 15), *testSchemaImport("d:20,15,128")); VELOX_ASSERT_THROW( *testSchemaImport("d:10,4,256"), - "Conversion failed for 'd:10,4,256'. Only 64-bit and 128-bit decimal types are supported."); + "Conversion failed for 'd:10,4,256'. Only 32-bit, 64-bit, and 128-bit decimal types are supported."); VELOX_ASSERT_THROW( *testSchemaImport("d:10,4,"), "Unable to convert 'd:10,4,' ArrowSchema decimal format to Velox decimal"); From a1bee94a3cc1bf9feec29ab3c3caa300cd5550f6 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 14:56:27 -0700 Subject: [PATCH 02/12] fix(cudf): Widen DECIMAL32 for GPU decimal SUM aggregation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After the Arrow bridge can import `d:9,2,32`, Presto GPU queries on TPC-DS small decimals still fail in `CudfGroupbyPARTIAL` when serializing partial SUM state: Unsupported decimal sum column type (type is numeric::decimal32) Operator: CudfGroupbyPARTIAL[...] libcudf's Parquet reader stores low-precision columns such as `DECIMAL(9,2)` as `DECIMAL32`. The cuDF group-by path already widened `DECIMAL64` inputs to `DECIMAL128` before SUM so partial accumulators do not wrap, but `castDecimal64InputToDecimal128()` left `DECIMAL32` unchanged. libcudf therefore produced a `DECIMAL32` partial sum, and `serializeDecimalSumState()` rejected it because its pack kernel only accepts `DECIMAL64` or `DECIMAL128`. Extend the existing widening helper to promote `DECIMAL32` through `DECIMAL64` into `DECIMAL128` before group-by SUM and AVG read raw input. Add a fallback in `serializeDecimalPartialOrIntermediateState()` that casts any remaining `DECIMAL32` sum column to `DECIMAL64` immediately before serialization, so partial/intermediate VARBINARY state encoding stays on the supported path even if a `DECIMAL32` sum slips through. Test plan: `DecimalAggregationTest.decimalSerializeSumStateDecimal32` — serialize and round-trip partial SUM state from a `DECIMAL32` sum column with nulls and zero counts. --- velox/experimental/cudf/exec/CudfGroupby.cpp | 8 +-- .../cudf/exec/DecimalAggregationHostOps.cpp | 50 ++++++++++++++++--- .../cudf/exec/DecimalAggregationHostOps.h | 14 +++--- .../cudf/tests/DecimalAggregationTest.cpp | 37 ++++++++++++++ 4 files changed, 92 insertions(+), 17 deletions(-) diff --git a/velox/experimental/cudf/exec/CudfGroupby.cpp b/velox/experimental/cudf/exec/CudfGroupby.cpp index e73c5ea2b0f..0f4a2ac89a5 100644 --- a/velox/experimental/cudf/exec/CudfGroupby.cpp +++ b/velox/experimental/cudf/exec/CudfGroupby.cpp @@ -266,8 +266,8 @@ struct GroupbyDecimalSumAggregator : GroupbyAggregator { uint32_t countIdx_{0}; std::unique_ptr decodedSum_; std::unique_ptr decodedCount_; - // Holds the DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle), kept - // alive while the groupby request references its view. + // Holds DECIMAL32/DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle), + // kept alive while the groupby request references its view. std::unique_ptr castedInput_; }; @@ -342,8 +342,8 @@ struct GroupbyDecimalAvgAggregator : GroupbyAggregator { uint32_t countIdx_{0}; std::unique_ptr decodedSum_; std::unique_ptr decodedCount_; - // Holds the DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle), kept - // alive while the groupby request references its view. + // Holds DECIMAL32/DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle), + // kept alive while the groupby request references its view. std::unique_ptr castedInput_; }; diff --git a/velox/experimental/cudf/exec/DecimalAggregationHostOps.cpp b/velox/experimental/cudf/exec/DecimalAggregationHostOps.cpp index 73bff493087..b367b2ef792 100644 --- a/velox/experimental/cudf/exec/DecimalAggregationHostOps.cpp +++ b/velox/experimental/cudf/exec/DecimalAggregationHostOps.cpp @@ -40,17 +40,54 @@ cudf::column_view castDecimal64InputToDecimal128( cudf::column_view inputCol, std::unique_ptr& holder, rmm::cuda_stream_view stream) { - if (inputCol.type().id() != cudf::type_id::DECIMAL64) { + auto inputType = inputCol.type().id(); + if (inputType == cudf::type_id::DECIMAL128) { return inputCol; } - holder = cudf::cast( - inputCol, - cudf::data_type{cudf::type_id::DECIMAL128, inputCol.type().scale()}, + if (inputType == cudf::type_id::DECIMAL32) { + holder = cudf::cast( + inputCol, + cudf::data_type{cudf::type_id::DECIMAL64, inputCol.type().scale()}, + stream, + get_temp_mr()); + inputCol = holder->view(); + inputType = inputCol.type().id(); + } + if (inputType == cudf::type_id::DECIMAL64) { + holder = cudf::cast( + inputCol, + cudf::data_type{cudf::type_id::DECIMAL128, inputCol.type().scale()}, + stream, + get_temp_mr()); + return holder->view(); + } + return inputCol; +} + +namespace { + +std::unique_ptr widenDecimalSumForSerialization( + std::unique_ptr sum, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { + const auto sumType = sum->type().id(); + if (sumType == cudf::type_id::DECIMAL64 || + sumType == cudf::type_id::DECIMAL128) { + return sum; + } + VELOX_CHECK( + sumType == cudf::type_id::DECIMAL32, + "Unsupported decimal sum column type (type is {})", + cudf::type_to_name(sum->type())); + return cudf::cast( + sum->view(), + cudf::data_type{cudf::type_id::DECIMAL64, sum->type().scale()}, stream, - get_temp_mr()); - return holder->view(); + mr); } +} // namespace + std::unique_ptr castCountColumnToInt64( std::unique_ptr count, rmm::cuda_stream_view stream) { @@ -67,6 +104,7 @@ std::unique_ptr serializeDecimalPartialOrIntermediateState( rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { count = castCountColumnToInt64(std::move(count), stream); + sum = widenDecimalSumForSerialization(std::move(sum), stream, mr); return serializeDecimalSumState(sum->view(), count->view(), stream, mr); } diff --git a/velox/experimental/cudf/exec/DecimalAggregationHostOps.h b/velox/experimental/cudf/exec/DecimalAggregationHostOps.h index 778f9120e78..fd57f9d1574 100644 --- a/velox/experimental/cudf/exec/DecimalAggregationHostOps.h +++ b/velox/experimental/cudf/exec/DecimalAggregationHostOps.h @@ -38,14 +38,14 @@ namespace facebook::velox::cudf_velox { void validateIntermediateColumnType(cudf::column_view const& column); /** - * Casts a DECIMAL64 column up to DECIMAL128 (scale preserved) so a subsequent - * SUM accumulates in 128 bits instead of wrapping. Allocates the casted column - * from the temporary memory resource into holder and returns its view. Lifetime - * stays valid only while holder is alive. + * Widens DECIMAL32/DECIMAL64 input to DECIMAL128 (scale preserved) so a + * subsequent SUM accumulates in 128 bits instead of wrapping. Allocates the + * casted column from the temporary memory resource into holder and returns its + * view. Lifetime stays valid only while holder is alive. * - * @param inputCol DECIMAL64 input column. - * @param holder receives ownership of the casted column when inputCol is - * DECIMAL64; unchanged otherwise. + * @param inputCol DECIMAL32, DECIMAL64, or DECIMAL128 input column. + * @param holder receives ownership of the casted column when widening is + * required; unchanged otherwise. * @param stream CUDA stream for device work. * @return view of inputCol or of the column stored in holder. */ diff --git a/velox/experimental/cudf/tests/DecimalAggregationTest.cpp b/velox/experimental/cudf/tests/DecimalAggregationTest.cpp index e7d499621ca..2b5097f6e32 100644 --- a/velox/experimental/cudf/tests/DecimalAggregationTest.cpp +++ b/velox/experimental/cudf/tests/DecimalAggregationTest.cpp @@ -15,6 +15,7 @@ */ #include "velox/experimental/cudf/CudfConfig.h" +#include "velox/experimental/cudf/exec/DecimalAggregationHostOps.h" #include "velox/experimental/cudf/exec/DecimalAggregationState.h" #include "velox/experimental/cudf/exec/ToCudf.h" #include "velox/experimental/cudf/exec/VeloxCudfInterop.h" @@ -157,6 +158,15 @@ std::unique_ptr makeDecimalColumn( return makeFixedWidthColumn(type, values, valid, stream); } +std::unique_ptr makeDecimal32Column( + const std::vector& values, + int32_t scale, + const std::vector* valid, + rmm::cuda_stream_view stream) { + cudf::data_type type{cudf::type_id::DECIMAL32, -scale}; + return makeFixedWidthColumn(type, values, valid, stream); +} + std::unique_ptr makeInt64Column( const std::vector& values, const std::vector* valid, @@ -1074,6 +1084,33 @@ TEST_F(CudfDecimalTest, decimalSumGlobalIntermediateVarbinaryAllNulls) { facebook::velox::test::assertEqualVectors(expected, result); } +TEST_F(CudfDecimalTest, decimalSerializeSumStateDecimal32) { + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); + std::vector sums = {100, -200, 300}; + std::vector counts = {1, 2, 0}; + std::vector sumValid = {true, false, true}; + std::vector countValid = {true, true, true}; + + auto sumCol = makeDecimal32Column(sums, 2, &sumValid, stream); + auto countCol = makeInt64Column(counts, &countValid, stream); + auto stateCol = serializeDecimalPartialOrIntermediateState( + std::move(sumCol), std::move(countCol), stream, mr); + auto sumAndCount = deserializeDecimalSumState(stateCol->view(), 2, stream); + auto stateMask = copyNullMask(stateCol->view(), stream); + auto sumMask = copyNullMask(sumAndCount.sum->view(), stream); + EXPECT_EQ(stateMask, sumMask); + + auto outSum = copyColumnData<__int128_t>(sumAndCount.sum->view(), stream); + for (size_t i = 0; i < sums.size(); ++i) { + bool expectedValid = sumValid[i] && countValid[i] && counts[i] != 0; + EXPECT_EQ(isValidAt(sumMask, i), expectedValid); + if (expectedValid) { + EXPECT_EQ(outSum[i], static_cast<__int128_t>(sums[i])); + } + } +} + TEST_F(CudfDecimalTest, decimalDeserializeSumStateDecimal64) { auto stream = cudf::get_default_stream(); auto mr = cudf::get_current_device_resource_ref(); From 7e59a7cb30ba9654150388d88aa9bc7487f5c65b Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 15:09:02 -0700 Subject: [PATCH 03/12] fix(cudf): Align DECIMAL32/DECIMAL64 operands in IF/CASE MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After the Arrow bridge and group-by SUM fixes, most TPC-DS decimal queries pass, but Q2 still fails in `CudfFilterProject` when a projection evaluates IF/CASE over small decimals: Operator::getOutput failed for [operator: CudfFilterProject, plan node ID: 49]: CUDF failure at: .../copy.cu:388: Both inputs must be of the same type libcudf's Parquet reader stores columns such as `DECIMAL(9,2)` as `DECIMAL32`. Velox short-decimal literals and `Values`-backed columns are converted to `DECIMAL64` in cuDF. `SwitchFunction` (IF/CASE) called `copy_if_else` without reconciling those widths, so a scanned `DECIMAL32` column paired with a `DECIMAL64` literal or expression result hit libcudf's same-type check. Add helpers that pick the wider of two same-scale fixed-point types and cast columns or scalars before mixed-type libcudf calls. Apply them in `SwitchFunction` for all four `copy_if_else` branches and in `CoalesceFunction` before `replace_nulls`. Extend decimal comparison and binary paths that previously promoted only `DECIMAL64` to `DECIMAL128`, and teach `castDecimalScalar()` to read and emit `DECIMAL32`. Test plan: `TableScanTest.decimal32IfWithLiteral` — write a Parquet file with a native `DECIMAL32` column, scan it through `CudfHiveConnector`, and project `if(cond, CAST('5.00' AS DECIMAL(9, 2)), d)` against DuckDB. --- .../cudf/expression/ExpressionEvaluator.cpp | 211 ++++++++++++++---- .../experimental/cudf/tests/TableScanTest.cpp | 110 +++++++++ 2 files changed, 277 insertions(+), 44 deletions(-) diff --git a/velox/experimental/cudf/expression/ExpressionEvaluator.cpp b/velox/experimental/cudf/expression/ExpressionEvaluator.cpp index 9f45b4915c3..5fbebc6f472 100644 --- a/velox/experimental/cudf/expression/ExpressionEvaluator.cpp +++ b/velox/experimental/cudf/expression/ExpressionEvaluator.cpp @@ -154,20 +154,24 @@ std::unique_ptr castDecimalScalar( rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { if (!src.is_valid(stream)) { - VELOX_CHECK( - targetType.id() == cudf::type_id::DECIMAL64 || - targetType.id() == cudf::type_id::DECIMAL128, - "castDecimalScalar: target must be DECIMAL64 or DECIMAL128"); if (targetType.id() == cudf::type_id::DECIMAL128) { return std::make_unique>( 0, numeric::scale_type{targetType.scale()}, false, stream, mr); } + if (targetType.id() == cudf::type_id::DECIMAL32) { + return std::make_unique>( + 0, numeric::scale_type{targetType.scale()}, false, stream, mr); + } return std::make_unique>( 0, numeric::scale_type{targetType.scale()}, false, stream, mr); } __int128_t rep; - if (src.type().id() == cudf::type_id::DECIMAL64) { + if (src.type().id() == cudf::type_id::DECIMAL32) { + auto const& dec = + static_cast const&>(src); + rep = static_cast(dec.value(stream)); + } else if (src.type().id() == cudf::type_id::DECIMAL64) { auto const& dec = static_cast const&>(src); rep = static_cast(dec.value(stream)); @@ -192,6 +196,13 @@ std::unique_ptr castDecimalScalar( return cudf::make_fixed_point_scalar( rep, numeric::scale_type{targetType.scale()}, stream, mr); } + if (targetType.id() == cudf::type_id::DECIMAL32) { + return cudf::make_fixed_point_scalar( + static_cast(rep), + numeric::scale_type{targetType.scale()}, + stream, + mr); + } return cudf::make_fixed_point_scalar( static_cast(rep), numeric::scale_type{targetType.scale()}, @@ -199,6 +210,67 @@ std::unique_ptr castDecimalScalar( mr); } +int32_t decimalTypeRank(cudf::type_id id) { + switch (id) { + case cudf::type_id::DECIMAL32: + return 1; + case cudf::type_id::DECIMAL64: + return 2; + case cudf::type_id::DECIMAL128: + return 3; + default: + return 0; + } +} + +cudf::type_id widerDecimalTypeId(cudf::type_id lhs, cudf::type_id rhs) { + return decimalTypeRank(lhs) >= decimalTypeRank(rhs) ? lhs : rhs; +} + +std::optional widerDecimalTypeIfMismatch( + const cudf::data_type& lhs, + const cudf::data_type& rhs) { + if (!cudf::is_fixed_point(lhs) || !cudf::is_fixed_point(rhs)) { + return std::nullopt; + } + if (lhs == rhs) { + return std::nullopt; + } + VELOX_CHECK_EQ( + lhs.scale(), + rhs.scale(), + "Decimal operands must share scale ({} vs {})", + lhs.scale(), + rhs.scale()); + return cudf::data_type{widerDecimalTypeId(lhs.id(), rhs.id()), lhs.scale()}; +} + +cudf::column_view castDecimalColumnIfNeeded( + cudf::column_view view, + const cudf::data_type& targetType, + std::unique_ptr& holder, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { + if (view.type() == targetType) { + return view; + } + holder = cudf::cast(view, targetType, stream, mr); + return holder->view(); +} + +const cudf::scalar* castDecimalScalarIfNeeded( + const cudf::scalar& scalar, + const cudf::data_type& targetType, + std::unique_ptr& holder, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { + if (scalar.type() == targetType) { + return &scalar; + } + holder = castDecimalScalar(scalar, targetType, stream, mr); + return holder.get(); +} + struct CudfExpressionEvaluatorEntry { int priority; CudfExpressionEvaluatorCanEvaluate canEvaluate; @@ -562,13 +634,13 @@ class BinaryFunction : public CudfFunction { std::unique_ptr lhsCast; std::unique_ptr rhsCast; if (type_.id() == cudf::type_id::DECIMAL128) { - if (lhsView.type().id() == cudf::type_id::DECIMAL64) { + if (lhsView.type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, lhsView.type().scale()}; lhsCast = cudf::cast(lhsView, castType, stream, mr); lhsView = lhsCast->view(); } - if (rhsView.type().id() == cudf::type_id::DECIMAL64) { + if (rhsView.type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, rhsView.type().scale()}; rhsCast = cudf::cast(rhsView, castType, stream, mr); @@ -588,10 +660,8 @@ class BinaryFunction : public CudfFunction { auto lhsScale = -lhsView.type().scale(); auto rhsScale = -rhsView.type().scale(); auto targetScale = lhsScale > rhsScale ? lhsScale : rhsScale; - auto targetTypeId = (lhsView.type().id() == cudf::type_id::DECIMAL128 || - rhsView.type().id() == cudf::type_id::DECIMAL128) - ? cudf::type_id::DECIMAL128 - : cudf::type_id::DECIMAL64; + auto targetTypeId = widerDecimalTypeId( + lhsView.type().id(), rhsView.type().id()); auto targetType = cudf::data_type{targetTypeId, numeric::scale_type{-targetScale}}; std::unique_ptr lhsCast; @@ -629,13 +699,13 @@ class BinaryFunction : public CudfFunction { std::unique_ptr lhsCast; std::unique_ptr rhsCast; if (type_.id() == cudf::type_id::DECIMAL128) { - if (lhsView.type().id() == cudf::type_id::DECIMAL64) { + if (lhsView.type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, lhsView.type().scale()}; lhsCast = cudf::cast(lhsView, castType, stream, mr); lhsView = lhsCast->view(); } - if (rhsView.type().id() == cudf::type_id::DECIMAL64) { + if (rhsView.type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, rhsView.type().scale()}; rhsCast = cudf::cast(rhsView, castType, stream, mr); @@ -667,10 +737,8 @@ class BinaryFunction : public CudfFunction { auto lhsScale = -lhsView.type().scale(); auto rhsScale = -right_->type().scale(); auto targetScale = lhsScale > rhsScale ? lhsScale : rhsScale; - auto targetTypeId = (lhsView.type().id() == cudf::type_id::DECIMAL128 || - right_->type().id() == cudf::type_id::DECIMAL128) - ? cudf::type_id::DECIMAL128 - : cudf::type_id::DECIMAL64; + auto targetTypeId = widerDecimalTypeId( + lhsView.type().id(), right_->type().id()); auto targetType = cudf::data_type{targetTypeId, numeric::scale_type{-targetScale}}; std::unique_ptr lhsCast; @@ -706,13 +774,13 @@ class BinaryFunction : public CudfFunction { std::unique_ptr lhsCast; std::unique_ptr rhsScalar; if (type_.id() == cudf::type_id::DECIMAL128) { - if (lhsView.type().id() == cudf::type_id::DECIMAL64) { + if (lhsView.type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, lhsView.type().scale()}; lhsCast = cudf::cast(lhsView, castType, stream, mr); lhsView = lhsCast->view(); } - if (right_->type().id() == cudf::type_id::DECIMAL64) { + if (right_->type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, right_->type().scale()}; rhsScalar = castDecimalScalar(*right_, castType, stream, mr); @@ -744,10 +812,8 @@ class BinaryFunction : public CudfFunction { auto lhsScale = -left_->type().scale(); auto rhsScale = -rhsView.type().scale(); auto targetScale = lhsScale > rhsScale ? lhsScale : rhsScale; - auto targetTypeId = (left_->type().id() == cudf::type_id::DECIMAL128 || - rhsView.type().id() == cudf::type_id::DECIMAL128) - ? cudf::type_id::DECIMAL128 - : cudf::type_id::DECIMAL64; + auto targetTypeId = widerDecimalTypeId( + left_->type().id(), rhsView.type().id()); auto targetType = cudf::data_type{targetTypeId, numeric::scale_type{-targetScale}}; std::unique_ptr rhsCast; @@ -782,13 +848,13 @@ class BinaryFunction : public CudfFunction { std::unique_ptr rhsCast; std::unique_ptr lhsScalar; if (type_.id() == cudf::type_id::DECIMAL128) { - if (rhsView.type().id() == cudf::type_id::DECIMAL64) { + if (rhsView.type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, rhsView.type().scale()}; rhsCast = cudf::cast(rhsView, castType, stream, mr); rhsView = rhsCast->view(); } - if (left_->type().id() == cudf::type_id::DECIMAL64) { + if (left_->type().id() != cudf::type_id::DECIMAL128) { auto castType = cudf::data_type{ cudf::type_id::DECIMAL128, left_->type().scale()}; lhsScalar = castDecimalScalar(*left_, castType, stream, mr); @@ -1173,27 +1239,64 @@ class SwitchFunction : public CudfFunction { std::vector& inputColumns, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) const override { + auto mask = asView(inputColumns[0]); + std::unique_ptr lhsCast; + std::unique_ptr rhsCast; + std::unique_ptr lhsScalarCast; + std::unique_ptr rhsScalarCast; + if (left_ == nullptr && right_ == nullptr) { + auto lhs = asView(inputColumns[1]); + auto rhs = asView(inputColumns[2]); + if (auto wider = widerDecimalTypeIfMismatch(lhs.type(), rhs.type())) { + lhs = castDecimalColumnIfNeeded(lhs, *wider, lhsCast, stream, mr); + rhs = castDecimalColumnIfNeeded(rhs, *wider, rhsCast, stream, mr); + } + return cudf::copy_if_else(lhs, rhs, mask, stream, mr); + } + if (left_ == nullptr) { + auto lhs = asView(inputColumns[1]); + const auto& rhs = *right_; + if (auto wider = widerDecimalTypeIfMismatch(lhs.type(), rhs.type())) { + lhs = castDecimalColumnIfNeeded(lhs, *wider, lhsCast, stream, mr); + return cudf::copy_if_else( + lhs, + *castDecimalScalarIfNeeded( + rhs, *wider, rhsScalarCast, stream, mr), + mask, + stream, + mr); + } + return cudf::copy_if_else(lhs, rhs, mask, stream, mr); + } + if (right_ == nullptr) { + const auto& lhs = *left_; + auto rhs = asView(inputColumns[1]); + if (auto wider = widerDecimalTypeIfMismatch(lhs.type(), rhs.type())) { + rhs = castDecimalColumnIfNeeded(rhs, *wider, rhsCast, stream, mr); + return cudf::copy_if_else( + *castDecimalScalarIfNeeded( + lhs, *wider, lhsScalarCast, stream, mr), + rhs, + mask, + stream, + mr); + } + return cudf::copy_if_else(lhs, rhs, mask, stream, mr); + } + const auto& lhs = *left_; + const auto& rhs = *right_; + if (auto wider = widerDecimalTypeIfMismatch(lhs.type(), rhs.type())) { return cudf::copy_if_else( - asView(inputColumns[1]), - asView(inputColumns[2]), - asView(inputColumns[0]), - stream, - mr); - } else if (left_ == nullptr) { - return cudf::copy_if_else( - asView(inputColumns[1]), - *right_, - asView(inputColumns[0]), + *castDecimalScalarIfNeeded( + lhs, *wider, lhsScalarCast, stream, mr), + *castDecimalScalarIfNeeded( + rhs, *wider, rhsScalarCast, stream, mr), + mask, stream, mr); - } else if (right_ == nullptr) { - return cudf::copy_if_else( - *left_, asView(inputColumns[1]), asView(inputColumns[0]), stream, mr); } - // right != null and left != null - return cudf::copy_if_else( - *left_, *right_, asView(inputColumns[0]), stream, mr); + return cudf::copy_if_else(lhs, rhs, mask, stream, mr); } private: @@ -1256,14 +1359,34 @@ class CoalesceFunction : public CudfFunction { !inputColumns.empty(), "coalesce requires at least one non-literal input"); ColumnOrView result = asView(inputColumns[0]); + std::unique_ptr lhsCast; + std::unique_ptr rhsCast; + std::unique_ptr literalCast; size_t stop = std::min(numColumnsBeforeLiteral_, inputColumns.size()); for (size_t i = 1; i < stop && asView(result).has_nulls(); ++i) { - result = cudf::replace_nulls( - asView(result), asView(inputColumns[i]), stream, mr); + auto lhs = asView(result); + auto rhs = asView(inputColumns[i]); + if (auto wider = widerDecimalTypeIfMismatch(lhs.type(), rhs.type())) { + lhs = castDecimalColumnIfNeeded(lhs, *wider, lhsCast, stream, mr); + rhs = castDecimalColumnIfNeeded(rhs, *wider, rhsCast, stream, mr); + } + result = cudf::replace_nulls(lhs, rhs, stream, mr); } if (literalScalar_ && asView(result).has_nulls()) { - result = cudf::replace_nulls(asView(result), *literalScalar_, stream, mr); + auto lhs = asView(result); + const auto& rhs = *literalScalar_; + if (auto wider = widerDecimalTypeIfMismatch(lhs.type(), rhs.type())) { + lhs = castDecimalColumnIfNeeded(lhs, *wider, lhsCast, stream, mr); + result = cudf::replace_nulls( + lhs, + *castDecimalScalarIfNeeded( + rhs, *wider, literalCast, stream, mr), + stream, + mr); + } else { + result = cudf::replace_nulls(lhs, rhs, stream, mr); + } } return result; diff --git a/velox/experimental/cudf/tests/TableScanTest.cpp b/velox/experimental/cudf/tests/TableScanTest.cpp index 6b3a366896c..ee52a2845cf 100644 --- a/velox/experimental/cudf/tests/TableScanTest.cpp +++ b/velox/experimental/cudf/tests/TableScanTest.cpp @@ -43,7 +43,11 @@ #include "velox/type/Type.h" #include "velox/type/tests/SubfieldFiltersBuilder.h" +#include #include +#include + +#include #include @@ -88,6 +92,69 @@ StatsFilterMetrics readParquetWithStatsFilter( result.metadata.num_row_groups_after_stats_filter, result.tbl->num_rows()}; } + +template +void copyHostToDeviceColumn( + cudf::column_view& view, + const std::vector& values, + rmm::cuda_stream_view stream) { + if (values.empty()) { + return; + } + auto status = cudaMemcpyAsync( + view.data(), + values.data(), + values.size() * sizeof(T), + cudaMemcpyHostToDevice, + stream.value()); + VELOX_CHECK_EQ(0, static_cast(status)); + stream.synchronize(); +} + +std::unique_ptr makeDecimal32Column( + const std::vector& values, + int32_t scale, + rmm::cuda_stream_view stream) { + cudf::data_type type{cudf::type_id::DECIMAL32, -scale}; + auto col = cudf::make_fixed_width_column( + type, + static_cast(values.size()), + cudf::mask_state::UNALLOCATED, + stream); + copyHostToDeviceColumn(col->mutable_view(), values, stream); + return col; +} + +void writeDecimal32ParquetFile( + const std::string& filePath, + const RowTypePtr& rowType, + const std::vector& condValues, + const std::vector& decimalValues, + int32_t decimalScale, + rmm::cuda_stream_view stream) { + auto condCol = cudf::make_fixed_width_column( + cudf::data_type{cudf::type_id::BOOL8}, + static_cast(condValues.size()), + cudf::mask_state::UNALLOCATED, + stream); + copyHostToDeviceColumn(condCol->mutable_view(), condValues, stream); + auto decimalCol = + makeDecimal32Column(decimalValues, decimalScale, stream); + auto table = std::make_unique( + std::vector>{std::move(condCol), + std::move(decimalCol)}); + + auto sinkInfo = cudf::io::sink_info(filePath); + auto tableInputMetadata = cudf::io::table_input_metadata(table->view()); + for (int32_t i = 0; i < rowType->size(); ++i) { + tableInputMetadata.column_metadata[i].set_name(rowType->nameOf(i)); + } + auto options = + cudf::io::parquet_writer_options::builder(sinkInfo, table->view()) + .metadata(tableInputMetadata) + .build(); + cudf::io::write_parquet(options, stream); +} } // namespace class TableScanTest : public virtual CudfHiveConnectorTestBase { @@ -818,3 +885,46 @@ TEST_F(TableScanTest, decimalRemainingFilter) { {filePath}, "SELECT c0, c1 FROM tmp WHERE c0 = CAST('-5.00' AS DECIMAL(5, 2))"); } + +// Parquet I/O stores small decimals as DECIMAL32 while Velox literals become +// DECIMAL64 in cuDF. IF/CASE must widen operands before copy_if_else. +TEST_F(TableScanTest, decimal32IfWithLiteral) { + auto rowType = ROW({{"cond", BOOLEAN()}, {"d", DECIMAL(9, 2)}}); + auto filePath = TempFilePath::create(); + auto stream = cudf::get_default_stream(); + writeDecimal32ParquetFile( + filePath->getPath(), + rowType, + {1, 0, 1, 0}, + {1000, 2000, 3000, 4000}, + 2, + stream); + + auto vectors = {makeRowVector( + {"cond", "d"}, + { + makeFlatVector({true, false, true, false}), + makeFlatVector({1000, 2000, 3000, 4000}, DECIMAL(9, 2)), + })}; + createDuckDbTable(vectors); + + auto assignments = + facebook::velox::exec::test::HiveConnectorTestBase::allRegularColumns( + rowType); + + auto plan = PlanBuilder(pool_.get()) + .startTableScan() + .connectorId(kCudfHiveConnectorId) + .outputType(rowType) + .dataColumns(rowType) + .assignments(assignments) + .endTableScan() + .project( + {"if(cond, CAST('5.00' AS DECIMAL(9, 2)), d) AS result"}) + .planNode(); + + assertQuery( + plan, + {filePath}, + "SELECT IF(cond, CAST('5.00' AS DECIMAL(9, 2)), d) AS result FROM tmp"); +} From f5dddf3478d3c96ce22f98f95e355edc316e6a04 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 15:15:35 -0700 Subject: [PATCH 04/12] fix(cudf): Decode decimal sum state from string offsets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TPC-DS Q2 still fails in `CudfGroupbyFINAL` when merging partial decimal SUM state that crossed a Velox exchange: Decimal sum state requires payload size 4480 (got 4448) Operator: CudfGroupbyFINAL[51] GPU partial serialization packs each row into a fixed 32-byte payload and still advances string offsets for invalid rows (`count == 0`), so a column with 140 rows normally has `chars_size == 140 × 32 == 4480`. After the state is exported as Velox `VARBINARY`, shuffled, and re-imported as cuDF `STRING`, null rows become empty strings and no longer contribute bytes to the chars buffer. The column still has 140 rows, but `chars_size` drops to `139 × 32 == 4448`. `deserializeDecimalSumState()` compared `chars_size` to `numRows × 32`, so FINAL aggregation rejected otherwise valid exchanged state. Validate payload size against the string offset table (`offsets[numRows]`) instead of assuming every row occupies 32 bytes. Keep an upper bound of `numRows × 32` so corrupted buffers are still rejected. Unpack behavior is unchanged: null rows stay null via the source null mask. Test plan: `DecimalAggregationTest.decimalDeserializeSumStateAfterVarbinaryRoundTrip` — serialize partial SUM state, round-trip through Velox `VARBINARY` via Arrow, re-import to cuDF `STRING`, and deserialize the sum/count columns. --- .../cudf/exec/DecimalAggregationState.cpp | 59 ++++++++++++++++--- .../cudf/tests/DecimalAggregationTest.cpp | 40 +++++++++++++ 2 files changed, 90 insertions(+), 9 deletions(-) diff --git a/velox/experimental/cudf/exec/DecimalAggregationState.cpp b/velox/experimental/cudf/exec/DecimalAggregationState.cpp index 523e23613d8..9abc890a512 100644 --- a/velox/experimental/cudf/exec/DecimalAggregationState.cpp +++ b/velox/experimental/cudf/exec/DecimalAggregationState.cpp @@ -26,9 +26,47 @@ #include #include +#include + #include namespace facebook::velox::cudf_velox { +namespace { + +int64_t readLastStringOffset( + cudf::column_view offsetsView, + cudf::size_type numRows, + rmm::cuda_stream_view stream) { + if (offsetsView.type().id() == cudf::type_id::INT32) { + int32_t last{}; + auto status = cudaMemcpyAsync( + &last, + offsetsView.data() + numRows, + sizeof(int32_t), + cudaMemcpyDeviceToHost, + stream.value()); + VELOX_CHECK_EQ(0, static_cast(status)); + stream.synchronize(); + return last; + } + if (offsetsView.type().id() == cudf::type_id::INT64) { + int64_t last{}; + auto status = cudaMemcpyAsync( + &last, + offsetsView.data() + numRows, + sizeof(int64_t), + cudaMemcpyDeviceToHost, + stream.value()); + VELOX_CHECK_EQ(0, static_cast(status)); + stream.synchronize(); + return last; + } + VELOX_FAIL( + "Decimal sum state requires INT32 or INT64 offsets (offset type is {})", + cudf::type_to_name(offsetsView.type())); +} + +} // namespace DecimalSumStateColumns deserializeDecimalSumState( const cudf::column_view& stateCol, @@ -80,16 +118,24 @@ DecimalSumStateColumns deserializeDecimalSumState( cudf::strings_column_view strings(stateCol); + auto offsetsView = strings.offsets(); auto const payloadSize = strings.chars_size(stream); auto const expectedPayloadSize = - static_cast(numRows) * detail::kDecimalSumStateSize; - VELOX_CHECK( - payloadSize == expectedPayloadSize, + readLastStringOffset(offsetsView, numRows, stream); + VELOX_CHECK_EQ( + payloadSize, + expectedPayloadSize, "Decimal sum state requires payload size {} (got {})", expectedPayloadSize, payloadSize); + VELOX_CHECK_LE( + expectedPayloadSize, + static_cast(numRows) * detail::kDecimalSumStateSize, + "Decimal sum state payload size {} exceeds max {} for {} rows", + expectedPayloadSize, + static_cast(numRows) * detail::kDecimalSumStateSize, + numRows); - auto offsetsView = strings.offsets(); auto charsPtr = reinterpret_cast(strings.chars_begin(stream)); auto sumCol = cudf::make_fixed_width_column( @@ -110,11 +156,6 @@ DecimalSumStateColumns deserializeDecimalSumState( // numRows is guaranteed positive here 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())); detail::unpackDecimalSumState( offsetsType, offsetsView, charsPtr, sumView, countView, numRows, stream); diff --git a/velox/experimental/cudf/tests/DecimalAggregationTest.cpp b/velox/experimental/cudf/tests/DecimalAggregationTest.cpp index 2b5097f6e32..d6d09fb2259 100644 --- a/velox/experimental/cudf/tests/DecimalAggregationTest.cpp +++ b/velox/experimental/cudf/tests/DecimalAggregationTest.cpp @@ -16,6 +16,7 @@ #include "velox/experimental/cudf/CudfConfig.h" #include "velox/experimental/cudf/exec/DecimalAggregationHostOps.h" +#include "velox/experimental/cudf/exec/DecimalAggregationDevice.h" #include "velox/experimental/cudf/exec/DecimalAggregationState.h" #include "velox/experimental/cudf/exec/ToCudf.h" #include "velox/experimental/cudf/exec/VeloxCudfInterop.h" @@ -1138,6 +1139,45 @@ TEST_F(CudfDecimalTest, decimalDeserializeSumStateDecimal64) { } } +TEST_F(CudfDecimalTest, decimalDeserializeSumStateAfterVarbinaryRoundTrip) { + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); + std::vector sums = {100, -200, 300, 400}; + std::vector counts = {1, 0, 2, 3}; + std::vector sumValid = {true, true, false, true}; + std::vector countValid = {true, true, true, false}; + + auto sumCol = makeDecimalColumn(sums, 2, &sumValid, stream); + auto countCol = makeInt64Column(counts, &countValid, stream); + auto stateCol = + serializeDecimalSumState(sumCol->view(), countCol->view(), stream, mr); + + auto rowType = ROW({{"state", VARBINARY()}}); + auto velox = with_arrow::toVeloxColumn( + stateCol->view(), pool(), rowType, "rt_", stream, mr); + auto cudfAgain = with_arrow::toCudfTable(velox, pool(), stream, mr); + + cudf::strings_column_view strings(cudfAgain->column(0)); + EXPECT_LT( + strings.chars_size(stream), + static_cast(sums.size()) * detail::kDecimalSumStateSize); + + auto sumAndCount = + deserializeDecimalSumState(cudfAgain->column(0), 2, stream); + auto stateMask = copyNullMask(cudfAgain->column(0), stream); + auto sumMask = copyNullMask(sumAndCount.sum->view(), stream); + EXPECT_EQ(stateMask, sumMask); + + auto outSum = copyColumnData<__int128_t>(sumAndCount.sum->view(), stream); + for (size_t i = 0; i < sums.size(); ++i) { + bool expectedValid = sumValid[i] && countValid[i] && counts[i] != 0; + EXPECT_EQ(isValidAt(sumMask, i), expectedValid); + if (expectedValid) { + EXPECT_EQ(outSum[i], static_cast<__int128_t>(sums[i])); + } + } +} + TEST_F(CudfDecimalTest, decimalDeserializeSumStateDecimal128) { auto stream = cudf::get_default_stream(); auto mr = cudf::get_current_device_resource_ref(); From 1b161d58c119767dcce4003d3f9cb9c3caf4ab4b Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 15:30:48 -0700 Subject: [PATCH 05/12] fix(cudf): Match DECIMAL32 literals in Parquet subfield filters Parquet scan columns with precision <= 9 are stored as DECIMAL32 in libcudf, but subfield filter literals were built as DECIMAL64. A pushed filter such as `c0 = CAST('-500.00' AS DECIMAL(9, 2))` then failed during TableScan with: An AST expression was provided non-matching operand types Making all short-decimal literals DECIMAL32 globally fixed that scan path but regressed TPC-H Q17. A hash join filter like `l_quantity < 0.2 * avg(l_quantity)` failed in CudfHashJoinProbe with: Unsupported operator for these types The literal `0.2` became DECIMAL32 while the aggregated average column stayed DECIMAL64, and the join filter evaluator does not widen operands for that multiply. Subfield filter AST construction now opts into DECIMAL32 literals only when the filtered column uses Parquet's small-decimal storage (`useCudfDecimal32ForVeloxDecimal`). All other expression paths keep the default DECIMAL64 literals that match `veloxToCudfDataType` and intermediate aggregation results. `makeLiteralFromScalar` follows the scalar's actual cuDF type instead of inferring width from the Velox type alone. --- velox/experimental/cudf/expression/AstUtils.h | 35 +++++++- .../cudf/expression/SubfieldFiltersToAst.cpp | 19 ++++- .../cudf/tests/SubfieldFilterAstTest.cpp | 22 +++++ .../experimental/cudf/tests/TableScanTest.cpp | 82 +++++++++++++++++++ 4 files changed, 151 insertions(+), 7 deletions(-) diff --git a/velox/experimental/cudf/expression/AstUtils.h b/velox/experimental/cudf/expression/AstUtils.h index 489d473564d..f5c25b68bde 100644 --- a/velox/experimental/cudf/expression/AstUtils.h +++ b/velox/experimental/cudf/expression/AstUtils.h @@ -33,6 +33,17 @@ namespace facebook::velox::cudf_velox { +// libcudf Parquet I/O uses DECIMAL32 when precision <= 9 (see parquet_common.hpp). +constexpr uint8_t kCudfParquetDecimal32MaxPrecision = 9; + +inline bool useCudfDecimal32ForVeloxDecimal(const TypePtr& type) { + if (!type->isShortDecimal()) { + return false; + } + const auto [precision, _] = getDecimalPrecisionScale(*type); + return precision <= kCudfParquetDecimal32MaxPrecision; +} + template cudf::ast::literal makeLiteralFromScalar( cudf::scalar& scalar, @@ -40,6 +51,10 @@ cudf::ast::literal makeLiteralFromScalar( if constexpr (cudf::is_fixed_width()) { if (type->isDecimal()) { if (type->kind() == TypeKind::BIGINT) { + if (scalar.type().id() == cudf::type_id::DECIMAL32) { + using CudfScalarType = cudf::fixed_point_scalar; + return cudf::ast::literal{*static_cast(&scalar)}; + } using CudfScalarType = cudf::fixed_point_scalar; return cudf::ast::literal{*static_cast(&scalar)}; } @@ -92,7 +107,8 @@ std::unique_ptr makeScalarFromValue( const TypePtr& type, T value, bool isNull, - std::optional toType = std::nullopt) { + std::optional toType = std::nullopt, + bool useCudfDecimal32 = false) { auto stream = cudf::get_default_stream(cudf::allow_default_stream); auto mr = get_temp_mr(); @@ -146,6 +162,14 @@ std::unique_ptr makeScalarFromValue( std::dynamic_pointer_cast(type); VELOX_CHECK(decimalType, "Invalid Decimal Type (failed dynamic_cast)"); auto const cudfScale = numeric::scale_type{-decimalType->scale()}; + if (useCudfDecimal32 && + useCudfDecimal32ForVeloxDecimal(type)) { + using CudfDecimalType = cudf::fixed_point_scalar; + auto scalar = std::make_unique( + static_cast(value), cudfScale, !isNull, stream, mr); + stream.synchronize(); + return scalar; + } using CudfDecimalType = cudf::fixed_point_scalar; auto scalar = std::make_unique( value, cudfScale, !isNull, stream, mr); @@ -234,16 +258,19 @@ cudf::ast::literal makeScalarAndLiteral( const TypePtr& type, const variant& var, bool isNull, - std::vector>& scalars) { + std::vector>& scalars, + bool useCudfDecimal32 = false) { using T = typename TypeTraits::NativeType; if constexpr (cudf::is_fixed_width() || kind == TypeKind::VARCHAR) { if (isNull) { - auto scalar = makeScalarFromValue(type, T{}, true); + auto scalar = makeScalarFromValue( + type, T{}, true, std::nullopt, useCudfDecimal32); scalars.emplace_back(std::move(scalar)); return makeLiteralFromScalar(*(scalars.back()), type); } auto value = var.value(); - auto scalar = makeScalarFromValue(type, value, false); + auto scalar = + makeScalarFromValue(type, value, false, std::nullopt, useCudfDecimal32); scalars.emplace_back(std::move(scalar)); return makeLiteralFromScalar(*(scalars.back()), type); } diff --git a/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp b/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp index 1c2507a56d7..ddb5df45e0a 100644 --- a/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp +++ b/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp @@ -39,6 +39,19 @@ std::pair getInt128BoundsForType(const TypePtr& type) { std::numeric_limits::max()}; } +template +cudf::ast::literal makeSubfieldFilterLiteral( + const TypePtr& columnTypePtr, + const variant& veloxVariant, + std::vector>& scalars) { + return makeScalarAndLiteral( + columnTypePtr, + veloxVariant, + false, + scalars, + useCudfDecimal32ForVeloxDecimal(columnTypePtr)); +} + template < typename RangeT, typename ScalarT, @@ -147,7 +160,7 @@ std::reference_wrapper buildIntegerRangeExpr( auto addLiteral = [&](ValueT value) -> const cudf::ast::expression& { variant veloxVariant = static_cast(value); const auto& literal = - makeScalarAndLiteral(columnTypePtr, veloxVariant, scalars); + makeSubfieldFilterLiteral(columnTypePtr, veloxVariant, scalars); return tree.push(literal); }; @@ -231,7 +244,7 @@ const cudf::ast::expression& buildValuesListExpr( for (const auto& value : values) { variant veloxVariant = static_cast(value); auto const& literal = tree.push( - makeScalarAndLiteral(columnTypePtr, veloxVariant, scalars)); + makeSubfieldFilterLiteral(columnTypePtr, veloxVariant, scalars)); auto const& equalExpr = tree.push( Operation{isNegated ? Op::NOT_EQUAL : Op::EQUAL, columnRef, literal}); exprVec.push_back(&equalExpr); @@ -309,7 +322,7 @@ std::reference_wrapper buildIntegerInListExpr( variant veloxVariant = static_cast(value); const auto& literal = - makeScalarAndLiteral(columnTypePtr, veloxVariant, scalars); + makeSubfieldFilterLiteral(columnTypePtr, veloxVariant, scalars); auto const& cudfLiteral = tree.push(literal); auto const& equalExpr = tree.push(Operation{Op::EQUAL, columnRef, cudfLiteral}); diff --git a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp index a2d1317f772..56303d2ed56 100644 --- a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp +++ b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp @@ -27,6 +27,8 @@ #include #include +#include + #include using namespace facebook::velox; @@ -565,6 +567,26 @@ TEST_F(SubfieldFilterAstTest, DecimalRange) { testFilterExecution(rowType, columnName, *filter, vec, expr); } +TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal32Literals) { + const std::string columnName = "c0"; + auto rowType = ROW({{columnName, DECIMAL(9, 2)}}); + auto filter = std::make_unique( + int64_t{100}, int64_t{500}, /*nullAllowed*/ false); + + common::Subfield subfield(columnName); + cudf::ast::tree tree; + std::vector> scalars; + const auto& expr = + createAstFromSubfieldFilter(subfield, *filter, tree, scalars, rowType); + + ASSERT_EQ(scalars.size(), 2UL); + EXPECT_EQ(scalars[0]->type().id(), cudf::type_id::DECIMAL32); + EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::DECIMAL32); + EXPECT_GT(tree.size(), 0UL); + EXPECT_NO_THROW( + cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); +} + TEST_F(SubfieldFilterAstTest, DecimalInList) { const std::string columnName = "c0"; auto rowType = ROW({{columnName, DECIMAL(20, 2)}}); diff --git a/velox/experimental/cudf/tests/TableScanTest.cpp b/velox/experimental/cudf/tests/TableScanTest.cpp index ee52a2845cf..cb17391a5f5 100644 --- a/velox/experimental/cudf/tests/TableScanTest.cpp +++ b/velox/experimental/cudf/tests/TableScanTest.cpp @@ -155,6 +155,37 @@ void writeDecimal32ParquetFile( .build(); cudf::io::write_parquet(options, stream); } + +void writeDecimal32AndInt64ParquetFile( + const std::string& filePath, + const RowTypePtr& rowType, + const std::vector& decimalValues, + int32_t decimalScale, + const std::vector& intValues, + rmm::cuda_stream_view stream) { + auto decimalCol = + makeDecimal32Column(decimalValues, decimalScale, stream); + auto intCol = cudf::make_fixed_width_column( + cudf::data_type{cudf::type_id::INT64}, + static_cast(intValues.size()), + cudf::mask_state::UNALLOCATED, + stream); + copyHostToDeviceColumn(intCol->mutable_view(), intValues, stream); + auto table = std::make_unique( + std::vector>{std::move(decimalCol), + std::move(intCol)}); + + auto sinkInfo = cudf::io::sink_info(filePath); + auto tableInputMetadata = cudf::io::table_input_metadata(table->view()); + for (int32_t i = 0; i < rowType->size(); ++i) { + tableInputMetadata.column_metadata[i].set_name(rowType->nameOf(i)); + } + auto options = + cudf::io::parquet_writer_options::builder(sinkInfo, table->view()) + .metadata(tableInputMetadata) + .build(); + cudf::io::write_parquet(options, stream); +} } // namespace class TableScanTest : public virtual CudfHiveConnectorTestBase { @@ -928,3 +959,54 @@ TEST_F(TableScanTest, decimal32IfWithLiteral) { {filePath}, "SELECT IF(cond, CAST('5.00' AS DECIMAL(9, 2)), d) AS result FROM tmp"); } + +// Parquet scan columns are DECIMAL32 for precision <= 9; subfield filter +// literals must match or cuDF AST parsing fails. +TEST_F(TableScanTest, decimal32SubfieldFilter) { + auto rowType = ROW({"c0", "c1"}, {DECIMAL(9, 2), BIGINT()}); + auto filePath = TempFilePath::create(); + auto stream = cudf::get_default_stream(); + writeDecimal32AndInt64ParquetFile( + filePath->getPath(), + rowType, + {10000, -50000, -70000, -50000}, + 2, + {1, 2, 3, 4}, + stream); + + auto vectors = {makeRowVector( + {"c0", "c1"}, + { + makeFlatVector( + {10000, -50000, -70000, -50000}, DECIMAL(9, 2)), + makeFlatVector({1, 2, 3, 4}), + })}; + createDuckDbTable(vectors); + + common::SubfieldFilters subfieldFilters = + common::test::SubfieldFiltersBuilder() + .add( + "c0", + std::make_unique( + int64_t{-50000}, int64_t{-50000}, /*nullAllowed*/ false)) + .build(); + + auto tableHandle = makeTableHandle( + "parquet_table", rowType, std::move(subfieldFilters), nullptr); + auto assignments = + facebook::velox::exec::test::HiveConnectorTestBase::allRegularColumns( + rowType); + + auto plan = PlanBuilder() + .startTableScan() + .outputType(rowType) + .tableHandle(tableHandle) + .assignments(assignments) + .endTableScan() + .planNode(); + + assertQuery( + plan, + {filePath}, + "SELECT c0, c1 FROM tmp WHERE c0 = CAST('-500.00' AS DECIMAL(9, 2))"); +} From b9d14cbb891a20e6292e2765f68e421a3f74bd57 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 15:37:05 -0700 Subject: [PATCH 06/12] fix(cudf): Widen DECIMAL32/DECIMAL64 operands in filters GPU filter expressions on Parquet small-decimal columns still failed in CudfFilterProject after the subfield-filter literal fix. A filter such as `d * CAST(0.2 AS DECIMAL(9, 2)) > CAST(1.00 AS DECIMAL(9, 2))` over a DECIMAL32 scan column aborted with: Unsupported operator for these types Parquet I/O stores precision <= 9 as DECIMAL32. Velox literals and intermediate results stay DECIMAL64. Equality comparisons already widened mismatched operands, but multiply, divide, and `between` called cuDF `binary_operation` with mixed DECIMAL32/DECIMAL64 inputs. Before each decimal binary op, operands are cast to a common width derived from both inputs and the expression result type. The same alignment is applied to `between` bound comparisons and to divide setup after operands are widened. Divide helpers also accept DECIMAL32 scalars once inputs share a type. --- .../expression/DecimalExpressionKernels.cpp | 12 +- .../cudf/expression/ExpressionEvaluator.cpp | 232 +++++++++++------- .../experimental/cudf/tests/TableScanTest.cpp | 58 +++++ 3 files changed, 218 insertions(+), 84 deletions(-) diff --git a/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp b/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp index 9cb4c0077b4..f40069856f2 100644 --- a/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp +++ b/velox/experimental/cudf/expression/DecimalExpressionKernels.cpp @@ -32,6 +32,11 @@ namespace { __int128_t getDecimalScalarValue( const cudf::scalar& s, rmm::cuda_stream_view stream) { + if (s.type().id() == cudf::type_id::DECIMAL32) { + auto const& dec = + static_cast const&>(s); + return static_cast<__int128_t>(static_cast(dec.value(stream))); + } if (s.type().id() == cudf::type_id::DECIMAL64) { auto const& dec = static_cast const&>(s); @@ -55,9 +60,12 @@ std::unique_ptr makeAllNullDecimalColumn( void checkDecimalDivideTypes(cudf::type_id inType, cudf::type_id outType) { VELOX_CHECK( - inType == cudf::type_id::DECIMAL64 || inType == cudf::type_id::DECIMAL128, + inType == cudf::type_id::DECIMAL32 || + inType == cudf::type_id::DECIMAL64 || + inType == cudf::type_id::DECIMAL128, "Unsupported input type for decimal divide"); - if (inType == cudf::type_id::DECIMAL64) { + if (inType == cudf::type_id::DECIMAL32 || + inType == cudf::type_id::DECIMAL64) { VELOX_CHECK( outType == cudf::type_id::DECIMAL64 || outType == cudf::type_id::DECIMAL128, diff --git a/velox/experimental/cudf/expression/ExpressionEvaluator.cpp b/velox/experimental/cudf/expression/ExpressionEvaluator.cpp index 5fbebc6f472..bc2341fb68f 100644 --- a/velox/experimental/cudf/expression/ExpressionEvaluator.cpp +++ b/velox/experimental/cudf/expression/ExpressionEvaluator.cpp @@ -95,6 +95,11 @@ bool decimalScalarIsZero( if (!scalar.is_valid(stream)) { return false; } + if (scalar.type().id() == cudf::type_id::DECIMAL32) { + auto const& dec = + static_cast const&>(scalar); + return dec.value(stream) == 0; + } if (scalar.type().id() == cudf::type_id::DECIMAL64) { auto const& dec = static_cast const&>( @@ -119,9 +124,15 @@ bool hasDecimalZero( } std::unique_ptr zero; auto scale = numeric::scale_type{col.type().scale()}; - if (col.type().id() == cudf::type_id::DECIMAL64) { - zero = - cudf::make_fixed_point_scalar(0, scale, stream, mr); + if (col.type().id() == cudf::type_id::DECIMAL32 || + col.type().id() == cudf::type_id::DECIMAL64) { + if (col.type().id() == cudf::type_id::DECIMAL32) { + zero = cudf::make_fixed_point_scalar( + 0, scale, stream, mr); + } else { + zero = + cudf::make_fixed_point_scalar(0, scale, stream, mr); + } } else if (col.type().id() == cudf::type_id::DECIMAL128) { zero = cudf::make_fixed_point_scalar( 0, scale, stream, mr); @@ -271,6 +282,63 @@ const cudf::scalar* castDecimalScalarIfNeeded( return holder.get(); } +cudf::data_type commonDecimalOperandType( + const cudf::data_type& lhs, + const cudf::data_type& rhs, + const cudf::data_type& resultType) { + auto lhsScale = -lhs.scale(); + auto rhsScale = -rhs.scale(); + auto targetScale = lhsScale > rhsScale ? lhsScale : rhsScale; + auto targetTypeId = widerDecimalTypeId(lhs.id(), rhs.id()); + targetTypeId = widerDecimalTypeId(targetTypeId, resultType.id()); + return cudf::data_type{targetTypeId, numeric::scale_type{-targetScale}}; +} + +void alignDecimalColumnOperands( + cudf::column_view& lhsView, + cudf::column_view& rhsView, + const cudf::data_type& targetType, + std::unique_ptr& lhsCast, + std::unique_ptr& rhsCast, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { + lhsView = castDecimalColumnIfNeeded(lhsView, targetType, lhsCast, stream, mr); + rhsView = castDecimalColumnIfNeeded(rhsView, targetType, rhsCast, stream, mr); +} + +std::unique_ptr decimalComparison( + cudf::column_view value, + cudf::column_view boundCol, + const cudf::scalar* boundScalar, + cudf::binary_operator op, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) { + static constexpr cudf::data_type kBoolType{cudf::type_id::BOOL8}; + std::unique_ptr valueCast; + std::unique_ptr boundCast; + std::unique_ptr boundScalarCast; + if (boundScalar != nullptr) { + if (cudf::is_fixed_point(value.type()) && + cudf::is_fixed_point(boundScalar->type())) { + auto targetType = commonDecimalOperandType( + value.type(), boundScalar->type(), value.type()); + value = castDecimalColumnIfNeeded(value, targetType, valueCast, stream, mr); + boundScalar = castDecimalScalarIfNeeded( + *boundScalar, targetType, boundScalarCast, stream, mr); + } + return cudf::binary_operation( + value, *boundScalar, op, kBoolType, stream, mr); + } + if (cudf::is_fixed_point(value.type()) && + cudf::is_fixed_point(boundCol.type())) { + auto targetType = + commonDecimalOperandType(value.type(), boundCol.type(), value.type()); + alignDecimalColumnOperands( + value, boundCol, targetType, valueCast, boundCast, stream, mr); + } + return cudf::binary_operation(value, boundCol, op, kBoolType, stream, mr); +} + struct CudfExpressionEvaluatorEntry { int priority; CudfExpressionEvaluatorCanEvaluate canEvaluate; @@ -633,19 +701,12 @@ class BinaryFunction : public CudfFunction { auto rhsView = asView(inputColumns[1]); std::unique_ptr lhsCast; std::unique_ptr rhsCast; - if (type_.id() == cudf::type_id::DECIMAL128) { - if (lhsView.type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, lhsView.type().scale()}; - lhsCast = cudf::cast(lhsView, castType, stream, mr); - lhsView = lhsCast->view(); - } - if (rhsView.type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, rhsView.type().scale()}; - rhsCast = cudf::cast(rhsView, castType, stream, mr); - rhsView = rhsCast->view(); - } + if (cudf::is_fixed_point(lhsView.type()) && + cudf::is_fixed_point(rhsView.type())) { + auto targetType = commonDecimalOperandType( + lhsView.type(), rhsView.type(), type_); + alignDecimalColumnOperands( + lhsView, rhsView, targetType, lhsCast, rhsCast, stream, mr); } auto lhsScale = -lhsView.type().scale(); auto rhsScale = -rhsView.type().scale(); @@ -698,19 +759,12 @@ class BinaryFunction : public CudfFunction { if (op_ == cudf::binary_operator::MUL) { std::unique_ptr lhsCast; std::unique_ptr rhsCast; - if (type_.id() == cudf::type_id::DECIMAL128) { - if (lhsView.type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, lhsView.type().scale()}; - lhsCast = cudf::cast(lhsView, castType, stream, mr); - lhsView = lhsCast->view(); - } - if (rhsView.type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, rhsView.type().scale()}; - rhsCast = cudf::cast(rhsView, castType, stream, mr); - rhsView = rhsCast->view(); - } + if (cudf::is_fixed_point(lhsView.type()) && + cudf::is_fixed_point(rhsView.type())) { + auto targetType = commonDecimalOperandType( + lhsView.type(), rhsView.type(), type_); + alignDecimalColumnOperands( + lhsView, rhsView, targetType, lhsCast, rhsCast, stream, mr); } // @TODO Check for divide-by-zero as in the DECIMAL case above? return cudf::binary_operation( @@ -725,11 +779,23 @@ class BinaryFunction : public CudfFunction { VELOX_USER_FAIL("Division by zero"); } auto lhsView = asView(inputColumns[0]); + std::unique_ptr lhsCast; + std::unique_ptr rhsScalarCast; + const cudf::scalar* rhs = right_.get(); + if (cudf::is_fixed_point(lhsView.type()) && + cudf::is_fixed_point(rhs->type())) { + auto targetType = + commonDecimalOperandType(lhsView.type(), rhs->type(), type_); + lhsView = + castDecimalColumnIfNeeded(lhsView, targetType, lhsCast, stream, mr); + rhs = castDecimalScalarIfNeeded( + *rhs, targetType, rhsScalarCast, stream, mr); + } auto lhsScale = -lhsView.type().scale(); - auto rhsScale = -right_->type().scale(); + auto rhsScale = -rhs->type().scale(); auto outScale = -type_.scale(); auto aRescale = outScale - lhsScale + rhsScale; - return decimalDivide(lhsView, *right_, type_, aRescale, stream, mr); + return decimalDivide(lhsView, *rhs, type_, aRescale, stream, mr); } auto lhsView = asView(inputColumns[0]); if (isComparisonOp(op_) && cudf::is_fixed_point(lhsView.type()) && @@ -772,27 +838,19 @@ class BinaryFunction : public CudfFunction { } if (op_ == cudf::binary_operator::MUL) { std::unique_ptr lhsCast; - std::unique_ptr rhsScalar; - if (type_.id() == cudf::type_id::DECIMAL128) { - if (lhsView.type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, lhsView.type().scale()}; - lhsCast = cudf::cast(lhsView, castType, stream, mr); - lhsView = lhsCast->view(); - } - if (right_->type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, right_->type().scale()}; - rhsScalar = castDecimalScalar(*right_, castType, stream, mr); - } + std::unique_ptr rhsScalarCast; + const cudf::scalar* rhs = right_.get(); + if (cudf::is_fixed_point(lhsView.type()) && + cudf::is_fixed_point(rhs->type())) { + auto targetType = + commonDecimalOperandType(lhsView.type(), rhs->type(), type_); + lhsView = castDecimalColumnIfNeeded( + lhsView, targetType, lhsCast, stream, mr); + rhs = castDecimalScalarIfNeeded( + *rhs, targetType, rhsScalarCast, stream, mr); } return cudf::binary_operation( - lhsView, - rhsScalar ? *rhsScalar : *right_, - op_, - type_, - stream, - mr); + lhsView, *rhs, op_, type_, stream, mr); } } return cudf::binary_operation( @@ -800,11 +858,23 @@ class BinaryFunction : public CudfFunction { } if (op_ == cudf::binary_operator::DIV && cudf::is_fixed_point(type_)) { auto rhsView = asView(inputColumns[0]); - auto lhsScale = -left_->type().scale(); + std::unique_ptr rhsCast; + std::unique_ptr lhsScalarCast; + const cudf::scalar* lhs = left_.get(); + if (cudf::is_fixed_point(lhs->type()) && + cudf::is_fixed_point(rhsView.type())) { + auto targetType = + commonDecimalOperandType(lhs->type(), rhsView.type(), type_); + lhs = castDecimalScalarIfNeeded( + *lhs, targetType, lhsScalarCast, stream, mr); + rhsView = + castDecimalColumnIfNeeded(rhsView, targetType, rhsCast, stream, mr); + } + auto lhsScale = -lhs->type().scale(); auto rhsScale = -rhsView.type().scale(); auto outScale = -type_.scale(); auto aRescale = outScale - lhsScale + rhsScale; - return decimalDivide(*left_, rhsView, type_, aRescale, stream, mr); + return decimalDivide(*lhs, rhsView, type_, aRescale, stream, mr); } auto rhsView = asView(inputColumns[0]); if (isComparisonOp(op_) && cudf::is_fixed_point(left_->type()) && @@ -846,22 +916,18 @@ class BinaryFunction : public CudfFunction { } if (op_ == cudf::binary_operator::MUL) { std::unique_ptr rhsCast; - std::unique_ptr lhsScalar; - if (type_.id() == cudf::type_id::DECIMAL128) { - if (rhsView.type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, rhsView.type().scale()}; - rhsCast = cudf::cast(rhsView, castType, stream, mr); - rhsView = rhsCast->view(); - } - if (left_->type().id() != cudf::type_id::DECIMAL128) { - auto castType = cudf::data_type{ - cudf::type_id::DECIMAL128, left_->type().scale()}; - lhsScalar = castDecimalScalar(*left_, castType, stream, mr); - } + std::unique_ptr lhsScalarCast; + const cudf::scalar* lhs = left_.get(); + if (cudf::is_fixed_point(lhs->type()) && + cudf::is_fixed_point(rhsView.type())) { + auto targetType = + commonDecimalOperandType(lhs->type(), rhsView.type(), type_); + lhs = castDecimalScalarIfNeeded( + *lhs, targetType, lhsScalarCast, stream, mr); + rhsView = + castDecimalColumnIfNeeded(rhsView, targetType, rhsCast, stream, mr); } - return cudf::binary_operation( - lhsScalar ? *lhsScalar : *left_, rhsView, op_, type_, stream, mr); + return cudf::binary_operation(*lhs, rhsView, op_, type_, stream, mr); } } return cudf::binary_operation(*left_, rhsView, op_, type_, stream, mr); @@ -1072,38 +1138,40 @@ class BetweenFunction : public CudfFunction { rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) const override { // return (value >= min) && (value <= max) - std::unique_ptr geResultColumn, leResultColumn; + auto value = asView(inputColumns[0]); + std::unique_ptr geResultColumn; + std::unique_ptr leResultColumn; if (minLiteral_) { - geResultColumn = cudf::binary_operation( - asView(inputColumns[0]), - *minLiteral_, + geResultColumn = decimalComparison( + value, + cudf::column_view{}, + minLiteral_.get(), cudf::binary_operator::GREATER_EQUAL, - kBoolType, stream, mr); } else { - geResultColumn = cudf::binary_operation( - asView(inputColumns[0]), + geResultColumn = decimalComparison( + value, asView(inputColumns[1]), + nullptr, cudf::binary_operator::GREATER_EQUAL, - kBoolType, stream, mr); } if (maxLiteral_) { - leResultColumn = cudf::binary_operation( - asView(inputColumns[0]), - *maxLiteral_, + leResultColumn = decimalComparison( + value, + cudf::column_view{}, + maxLiteral_.get(), cudf::binary_operator::LESS_EQUAL, - kBoolType, stream, mr); } else { - leResultColumn = cudf::binary_operation( - asView(inputColumns[0]), + leResultColumn = decimalComparison( + value, asView(inputColumns[2]), + nullptr, cudf::binary_operator::LESS_EQUAL, - kBoolType, stream, mr); } diff --git a/velox/experimental/cudf/tests/TableScanTest.cpp b/velox/experimental/cudf/tests/TableScanTest.cpp index cb17391a5f5..0c91b37b342 100644 --- a/velox/experimental/cudf/tests/TableScanTest.cpp +++ b/velox/experimental/cudf/tests/TableScanTest.cpp @@ -186,6 +186,29 @@ void writeDecimal32AndInt64ParquetFile( .build(); cudf::io::write_parquet(options, stream); } + +void writeSingleDecimal32ParquetFile( + const std::string& filePath, + const RowTypePtr& rowType, + const std::vector& decimalValues, + int32_t decimalScale, + rmm::cuda_stream_view stream) { + auto decimalCol = + makeDecimal32Column(decimalValues, decimalScale, stream); + auto table = std::make_unique( + std::vector>{std::move(decimalCol)}); + + auto sinkInfo = cudf::io::sink_info(filePath); + auto tableInputMetadata = cudf::io::table_input_metadata(table->view()); + for (int32_t i = 0; i < rowType->size(); ++i) { + tableInputMetadata.column_metadata[i].set_name(rowType->nameOf(i)); + } + auto options = + cudf::io::parquet_writer_options::builder(sinkInfo, table->view()) + .metadata(tableInputMetadata) + .build(); + cudf::io::write_parquet(options, stream); +} } // namespace class TableScanTest : public virtual CudfHiveConnectorTestBase { @@ -1010,3 +1033,38 @@ TEST_F(TableScanTest, decimal32SubfieldFilter) { {filePath}, "SELECT c0, c1 FROM tmp WHERE c0 = CAST('-500.00' AS DECIMAL(9, 2))"); } + +// DECIMAL32 Parquet columns with DECIMAL64 literals in multiply filters must +// widen operands before cuDF binary_operation. +TEST_F(TableScanTest, decimal32FilterWithMultiply) { + auto rowType = ROW({{"d", DECIMAL(9, 2)}}); + auto filePath = TempFilePath::create(); + auto stream = cudf::get_default_stream(); + writeSingleDecimal32ParquetFile( + filePath->getPath(), rowType, {1000, 200, 5000}, 2, stream); + + auto vectors = {makeRowVector( + {"d"}, + {makeFlatVector({1000, 200, 5000}, DECIMAL(9, 2))})}; + createDuckDbTable(vectors); + + auto assignments = + facebook::velox::exec::test::HiveConnectorTestBase::allRegularColumns( + rowType); + + auto plan = PlanBuilder(pool_.get()) + .startTableScan() + .connectorId(kCudfHiveConnectorId) + .outputType(rowType) + .dataColumns(rowType) + .assignments(assignments) + .endTableScan() + .filter( + "d * CAST(0.2 AS DECIMAL(9, 2)) > CAST(1.00 AS DECIMAL(9, 2))") + .planNode(); + + assertQuery( + plan, + {filePath}, + "SELECT d FROM tmp WHERE d * CAST(0.2 AS DECIMAL(9, 2)) > CAST(1.00 AS DECIMAL(9, 2))"); +} From 16db1f5a9f1de004f1fad795ed70e9db6227989c Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 15:45:17 -0700 Subject: [PATCH 07/12] [velox] fix(cudf): Align decimal subfield filters with Parquet schema GPU table scans with decimal subfield filters could fail during Parquet bloom-filter evaluation when the filter literal width did not match the column's libcudf storage type. For example, a TPC-H/TPC-DS equality filter on DECIMAL(9,2) could abort with: CUDF failure at: .../bloom_filter_reader.cu:67: Mismatched predicate column and literal types This happened because literals were sized from Velox precision alone (DECIMAL32 when precision <= 9). Many benchmark Parquet files store the same logical type as INT64 physical encoding, which libcudf reads as DECIMAL64. Bloom filters require an exact type match between the column and each equality literal. Subfield filter AST construction is deferred to split setup. CudfSplitReader reads the Parquet footer via read_parquet_metadata(), maps each column to its libcudf storage type, and builds filter literals with that width (DECIMAL32 or DECIMAL64). Unit tests without a Parquet schema still fall back to the precision heuristic. Iceberg uses hasSubfieldFilters() to detect pending filters before the AST exists, and can lazily build the AST for deferred post-read filtering. Test plan: Re-run failing TPC-H/TPC-DS decimal queries that previously hit the bloom_filter_reader type mismatch. --- .../connectors/hive/CudfHiveDataSource.cpp | 22 +++-- .../cudf/connectors/hive/CudfSplitReader.cpp | 46 +++++++++- .../cudf/connectors/hive/CudfSplitReader.h | 21 ++++- .../hive/iceberg/CudfIcebergDataSource.cpp | 12 ++- .../hive/iceberg/CudfIcebergSplitReader.cpp | 6 +- .../hive/iceberg/CudfIcebergSplitReader.h | 2 +- velox/experimental/cudf/expression/AstUtils.h | 16 ++++ .../cudf/expression/ParquetSchemaUtils.h | 43 +++++++++ .../cudf/expression/SubfieldFiltersToAst.cpp | 92 ++++++++++++++----- .../cudf/expression/SubfieldFiltersToAst.h | 9 +- .../cudf/tests/SubfieldFilterAstTest.cpp | 22 +++++ 11 files changed, 245 insertions(+), 46 deletions(-) create mode 100644 velox/experimental/cudf/expression/ParquetSchemaUtils.h diff --git a/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp b/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp index 58288df27bd..b82d5afaa33 100644 --- a/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp +++ b/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp @@ -23,7 +23,6 @@ #include "velox/experimental/cudf/exec/ToCudf.h" #include "velox/experimental/cudf/exec/VeloxCudfInterop.h" #include "velox/experimental/cudf/expression/ExpressionEvaluator.h" -#include "velox/experimental/cudf/expression/SubfieldFiltersToAst.h" #include "velox/experimental/cudf/vector/CudfVector.h" #include "velox/common/time/Timer.h" @@ -119,13 +118,8 @@ CudfHiveDataSource::CudfHiveDataSource( // readColumnNames_ } - // Build a combined AST for all subfield filters once. This is query-constant - // and doesn't depend on split-specific state. - if (!subfieldFilters_.empty()) { - auto const readerFilterType = getTableRowType(); - subfieldFilterExpr_ = &createAstFromSubfieldFilters( - subfieldFilters_, subfieldTree_, subfieldScalars_, readerFilterType); - } + // Subfield filter AST is built per split in CudfSplitReader after reading + // the Parquet schema so literal types match libcudf storage types. VELOX_CHECK_NOT_NULL(fileHandleFactory_, "No FileHandleFactory present"); @@ -140,6 +134,16 @@ CudfHiveDataSource::CudfHiveDataSource( } std::unique_ptr CudfHiveDataSource::createCudfSplitReader() { + SubfieldFilterBuildState subfieldFilterBuildState; + if (!subfieldFilters_.empty()) { + subfieldFilterBuildState = SubfieldFilterBuildState{ + .filters = &subfieldFilters_, + .tree = &subfieldTree_, + .scalars = &subfieldScalars_, + .rowType = getTableRowType(), + .expr = &subfieldFilterExpr_, + }; + } return std::make_unique( split_, tableHandle_, @@ -152,7 +156,7 @@ std::unique_ptr CudfHiveDataSource::createCudfSplitReader() { ioStatistics_, ioStats_, useExperimentalCudfReader_, - subfieldFilterExpr_); + std::move(subfieldFilterBuildState)); } void CudfHiveDataSource::convertSplit(std::shared_ptr split) { diff --git a/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp b/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp index 684b8555baa..f72762cd213 100644 --- a/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp +++ b/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp @@ -18,6 +18,8 @@ #include "velox/experimental/cudf/connectors/hive/CudfSplitReader.h" #include "velox/experimental/cudf/connectors/hive/CudfSplitReaderHelpers.h" #include "velox/experimental/cudf/exec/GpuResources.h" +#include "velox/experimental/cudf/expression/ParquetSchemaUtils.h" +#include "velox/experimental/cudf/expression/SubfieldFiltersToAst.h" #include "velox/common/caching/CacheTTLController.h" #include "velox/common/time/Timer.h" @@ -74,7 +76,7 @@ CudfSplitReader::CudfSplitReader( const std::shared_ptr& ioStatistics, const std::shared_ptr& ioStats, bool useExperimentalCudfReader, - cudf::ast::expression const* subfieldFilterExpr) + SubfieldFilterBuildState subfieldFilterBuildState) : NvtxHelper( nvtx3::rgb{80, 171, 241}, std::nullopt, @@ -92,7 +94,7 @@ CudfSplitReader::CudfSplitReader( pool_(connectorQueryCtx->memoryPool()), useExperimentalCudfReader_(useExperimentalCudfReader), baseReaderOpts_(pool_), - subfieldFilterExpr_(subfieldFilterExpr) { + subfieldFilterBuildState_(std::move(subfieldFilterBuildState)) { baseReaderOpts_.setDataIoStats(ioStatistics_); baseReaderOpts_.setMetadataIoStats(ioStatistics_); } @@ -223,7 +225,44 @@ void CudfSplitReader::resetSplit() { } cudf::ast::expression const* CudfSplitReader::subfieldFilter() { - return subfieldFilterExpr_; + if (hasSubfieldFilters() && subfieldFilterBuildState_.expr != nullptr && + *subfieldFilterBuildState_.expr == nullptr && dataSource_ != nullptr) { + buildSubfieldFilterAst(); + } + if (subfieldFilterBuildState_.expr == nullptr || + *subfieldFilterBuildState_.expr == nullptr) { + return nullptr; + } + return *subfieldFilterBuildState_.expr; +} + +bool CudfSplitReader::hasSubfieldFilters() const { + return subfieldFilterBuildState_.filters != nullptr && + !subfieldFilterBuildState_.filters->empty(); +} + +void CudfSplitReader::buildSubfieldFilterAst() { + auto& state = subfieldFilterBuildState_; + if (state.filters == nullptr || state.filters->empty()) { + return; + } + VELOX_CHECK_NOT_NULL(state.tree); + VELOX_CHECK_NOT_NULL(state.scalars); + VELOX_CHECK_NOT_NULL(state.expr); + VELOX_CHECK_NOT_NULL(dataSource_); + + *state.tree = cudf::ast::tree{}; + state.scalars->clear(); + + auto sourceInfo = cudf::io::source_info{dataSource_.get()}; + auto metadata = cudf::io::read_parquet_metadata(sourceInfo); + auto parquetColumnTypes = parquetColumnTypesFromMetadata(metadata); + *state.expr = &createAstFromSubfieldFilters( + *state.filters, + *state.tree, + *state.scalars, + state.rowType, + &parquetColumnTypes); } void CudfSplitReader::setupCudfDataSource() { @@ -344,6 +383,7 @@ void CudfSplitReader::setupReaderOptions() { readerOptions_.set_num_bytes(split_->size()); } + buildSubfieldFilterAst(); if (auto* filter = subfieldFilter(); filter != nullptr) { readerOptions_.set_filter(*filter); } diff --git a/velox/experimental/cudf/connectors/hive/CudfSplitReader.h b/velox/experimental/cudf/connectors/hive/CudfSplitReader.h index d114c7b3d5d..8357182d424 100644 --- a/velox/experimental/cudf/connectors/hive/CudfSplitReader.h +++ b/velox/experimental/cudf/connectors/hive/CudfSplitReader.h @@ -27,6 +27,7 @@ #include "velox/connectors/hive/FileHandle.h" #include "velox/connectors/hive/TableHandle.h" #include "velox/dwio/common/Statistics.h" +#include "velox/type/Filter.h" #include "velox/type/Type.h" #include @@ -39,6 +40,16 @@ namespace facebook::velox::cudf_velox::connector::hive { using namespace facebook::velox::connector; +// Mutable state owned by CudfHiveDataSource; the split reader builds the AST +// once per split after reading the Parquet schema. +struct SubfieldFilterBuildState { + const common::SubfieldFilters* filters{nullptr}; + cudf::ast::tree* tree{nullptr}; + std::vector>* scalars{nullptr}; + RowTypePtr rowType; + cudf::ast::expression const** expr{nullptr}; +}; + using CudfParquetReader = cudf::io::chunked_parquet_reader; using CudfParquetReaderPtr = std::unique_ptr; @@ -61,7 +72,7 @@ class CudfSplitReader : public NvtxHelper { const std::shared_ptr& ioStatistics, const std::shared_ptr& ioStats, bool useExperimentalCudfReader, - cudf::ast::expression const* subfieldFilterExpr); + SubfieldFilterBuildState subfieldFilterBuildState = {}); virtual ~CudfSplitReader() = default; @@ -84,6 +95,9 @@ class CudfSplitReader : public NvtxHelper { // Return the subfield filter. virtual cudf::ast::expression const* subfieldFilter(); + // Whether subfield filters were provided for this scan. + bool hasSubfieldFilters() const; + // Determine the output memory resource for the cuDF reader. virtual rmm::device_async_resource_ref determineCudfMemoryResource(); @@ -127,6 +141,9 @@ class CudfSplitReader : public NvtxHelper { std::vector fileMetaData_; private: + // Build the subfield filter AST from the current split's Parquet schema. + void buildSubfieldFilterAst(); + // Setup the cuDF reader options void setupReaderOptions(); @@ -142,7 +159,7 @@ class CudfSplitReader : public NvtxHelper { bool useExperimentalCudfReader_; dwio::common::ReaderOptions baseReaderOpts_; - cudf::ast::expression const* subfieldFilterExpr_; + SubfieldFilterBuildState subfieldFilterBuildState_; struct TotalScanTimeCallbackData { uint64_t startTimeUs; diff --git a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp index 8b39616dda4..d40a5e199c6 100644 --- a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp +++ b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp @@ -64,6 +64,16 @@ void CudfIcebergDataSource::convertSplit( std::unique_ptr CudfIcebergDataSource::createCudfSplitReader() { + SubfieldFilterBuildState subfieldFilterBuildState; + if (!subfieldFilters_.empty()) { + subfieldFilterBuildState = SubfieldFilterBuildState{ + .filters = &subfieldFilters_, + .tree = &subfieldTree_, + .scalars = &subfieldScalars_, + .rowType = getTableRowType(), + .expr = &subfieldFilterExpr_, + }; + } return std::make_unique( split_, icebergSplit_, @@ -78,7 +88,7 @@ CudfIcebergDataSource::createCudfSplitReader() { ioStatistics_, ioStats_, useExperimentalCudfReader_, - subfieldFilterExpr_); + std::move(subfieldFilterBuildState)); } } // namespace facebook::velox::cudf_velox::connector::hive::iceberg diff --git a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.cpp b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.cpp index 1a48a1f3e50..ae3485ecad8 100644 --- a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.cpp +++ b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.cpp @@ -93,7 +93,7 @@ CudfIcebergSplitReader::CudfIcebergSplitReader( const std::shared_ptr& ioStatistics, const std::shared_ptr& ioStats, bool useExperimentalCudfReader, - cudf::ast::expression const* subfieldFilterExpr) + SubfieldFilterBuildState subfieldFilterBuildState) : CudfSplitReader( std::move(split), std::move(tableHandle), @@ -106,7 +106,7 @@ CudfIcebergSplitReader::CudfIcebergSplitReader( ioStatistics, ioStats, useExperimentalCudfReader, - subfieldFilterExpr), + std::move(subfieldFilterBuildState)), icebergSplit_(std::move(icebergSplit)), hiveConfig_(hiveConfig) {} @@ -144,7 +144,7 @@ void CudfIcebergSplitReader::prepareSplit( // Defer subfield filter when it cannot evaluate on the physical parquet // table, or when positional deletes are present. - deferSubfieldFilter_ = CudfSplitReader::subfieldFilter() != nullptr and + deferSubfieldFilter_ = hasSubfieldFilters() and (noColumnsToRead_ or injectedColumns_.size() or // TODO(mh): Drop positional/DV deferral when cudf PR #23077 merges. deletionVectorReader_ or positionalDeleteFileReaders_.size()); diff --git a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.h b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.h index f0abe437a94..88f78970433 100644 --- a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.h +++ b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergSplitReader.h @@ -61,7 +61,7 @@ class CudfIcebergSplitReader : public CudfSplitReader { const std::shared_ptr& ioStatistics, const std::shared_ptr& ioStats, bool useExperimentalCudfReader, - cudf::ast::expression const* subfieldFilterExpr); + SubfieldFilterBuildState subfieldFilterBuildState); /// Override to setup delete file readers and column projection. /// @param runtimeStats Reference to the DataSource's runtime statistics, diff --git a/velox/experimental/cudf/expression/AstUtils.h b/velox/experimental/cudf/expression/AstUtils.h index f5c25b68bde..176093125f7 100644 --- a/velox/experimental/cudf/expression/AstUtils.h +++ b/velox/experimental/cudf/expression/AstUtils.h @@ -15,6 +15,7 @@ */ #pragma once +#include #include "velox/experimental/cudf/CudfConfig.h" #include "velox/experimental/cudf/CudfNoDefaults.h" @@ -44,6 +45,21 @@ inline bool useCudfDecimal32ForVeloxDecimal(const TypePtr& type) { return precision <= kCudfParquetDecimal32MaxPrecision; } +// Subfield filter literals must match the Parquet column's libcudf storage +// type (DECIMAL32 vs DECIMAL64). When the Parquet schema is unavailable, +// fall back to the precision heuristic above. +inline bool useCudfDecimal32ForSubfieldFilter( + const TypePtr& type, + std::optional parquetColumnType = std::nullopt) { + if (!type->isShortDecimal()) { + return false; + } + if (parquetColumnType.has_value()) { + return parquetColumnType.value() == cudf::type_id::DECIMAL32; + } + return useCudfDecimal32ForVeloxDecimal(type); +} + template cudf::ast::literal makeLiteralFromScalar( cudf::scalar& scalar, diff --git a/velox/experimental/cudf/expression/ParquetSchemaUtils.h b/velox/experimental/cudf/expression/ParquetSchemaUtils.h new file mode 100644 index 00000000000..713906181ef --- /dev/null +++ b/velox/experimental/cudf/expression/ParquetSchemaUtils.h @@ -0,0 +1,43 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * you may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include +#include + +#include +#include + +namespace facebook::velox::cudf_velox { + +using ParquetColumnTypeMap = std::unordered_map; + +// Maps top-level Parquet column names to libcudf storage types. Uses the same +// schema resolution as cudf::io::read_parquet_metadata. +inline ParquetColumnTypeMap parquetColumnTypesFromMetadata( + const cudf::io::parquet_metadata& metadata) { + ParquetColumnTypeMap columnTypes; + const auto& root = metadata.schema().root(); + for (int i = 0; i < root.num_children(); ++i) { + const auto& child = root.child(i); + if (!child.name().empty()) { + columnTypes.emplace(child.name(), child.cudf_type().id()); + } + } + return columnTypes; +} + +} // namespace facebook::velox::cudf_velox diff --git a/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp b/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp index ddb5df45e0a..b020a2019d1 100644 --- a/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp +++ b/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp @@ -43,13 +43,14 @@ template cudf::ast::literal makeSubfieldFilterLiteral( const TypePtr& columnTypePtr, const variant& veloxVariant, - std::vector>& scalars) { + std::vector>& scalars, + std::optional parquetColumnType = std::nullopt) { return makeScalarAndLiteral( columnTypePtr, veloxVariant, false, scalars, - useCudfDecimal32ForVeloxDecimal(columnTypePtr)); + useCudfDecimal32ForSubfieldFilter(columnTypePtr, parquetColumnType)); } template < @@ -129,7 +130,8 @@ std::reference_wrapper buildIntegerRangeExpr( cudf::ast::tree& tree, std::vector>& scalars, const cudf::ast::expression& columnRef, - const TypePtr& columnTypePtr) { + const TypePtr& columnTypePtr, + std::optional parquetColumnType = std::nullopt) { using NativeT = typename TypeTraits::NativeType; if constexpr ( @@ -159,8 +161,8 @@ std::reference_wrapper buildIntegerRangeExpr( auto addLiteral = [&](ValueT value) -> const cudf::ast::expression& { variant veloxVariant = static_cast(value); - const auto& literal = - makeSubfieldFilterLiteral(columnTypePtr, veloxVariant, scalars); + const auto& literal = makeSubfieldFilterLiteral( + columnTypePtr, veloxVariant, scalars, parquetColumnType); return tree.push(literal); }; @@ -209,9 +211,10 @@ std::reference_wrapper buildBigintRangeExpr( cudf::ast::tree& tree, std::vector>& scalars, const cudf::ast::expression& columnRef, - const TypePtr& columnTypePtr) { + const TypePtr& columnTypePtr, + std::optional parquetColumnType = std::nullopt) { return buildIntegerRangeExpr( - filter, tree, scalars, columnRef, columnTypePtr); + filter, tree, scalars, columnRef, columnTypePtr, parquetColumnType); } std::reference_wrapper buildHugeintRangeExpr( @@ -219,9 +222,10 @@ std::reference_wrapper buildHugeintRangeExpr( cudf::ast::tree& tree, std::vector>& scalars, const cudf::ast::expression& columnRef, - const TypePtr& columnTypePtr) { + const TypePtr& columnTypePtr, + std::optional parquetColumnType = std::nullopt) { return buildIntegerRangeExpr( - filter, tree, scalars, columnRef, columnTypePtr); + filter, tree, scalars, columnRef, columnTypePtr, parquetColumnType); } template @@ -231,7 +235,8 @@ const cudf::ast::expression& buildValuesListExpr( const cudf::ast::expression& columnRef, std::vector>& scalars, const TypePtr& columnTypePtr, - bool isNegated = false) { + bool isNegated = false, + std::optional parquetColumnType = std::nullopt) { using Op = cudf::ast::ast_operator; using Operation = cudf::ast::operation; @@ -243,8 +248,8 @@ const cudf::ast::expression& buildValuesListExpr( std::vector exprVec; for (const auto& value : values) { variant veloxVariant = static_cast(value); - auto const& literal = tree.push( - makeSubfieldFilterLiteral(columnTypePtr, veloxVariant, scalars)); + auto const& literal = tree.push(makeSubfieldFilterLiteral( + columnTypePtr, veloxVariant, scalars, parquetColumnType)); auto const& equalExpr = tree.push( Operation{isNegated ? Op::NOT_EQUAL : Op::EQUAL, columnRef, literal}); exprVec.push_back(&equalExpr); @@ -299,7 +304,8 @@ std::reference_wrapper buildIntegerInListExpr( const cudf::ast::expression& columnRef, rmm::cuda_stream_view /*stream*/, rmm::device_async_resource_ref /*mr*/, - const TypePtr& columnTypePtr) { + const TypePtr& columnTypePtr, + std::optional parquetColumnType = std::nullopt) { using NativeT = typename TypeTraits::NativeType; if constexpr (std::is_integral_v) { @@ -321,8 +327,8 @@ std::reference_wrapper buildIntegerInListExpr( } variant veloxVariant = static_cast(value); - const auto& literal = - makeSubfieldFilterLiteral(columnTypePtr, veloxVariant, scalars); + const auto& literal = makeSubfieldFilterLiteral( + columnTypePtr, veloxVariant, scalars, parquetColumnType); auto const& cudfLiteral = tree.push(literal); auto const& equalExpr = tree.push(Operation{Op::EQUAL, columnRef, cudfLiteral}); @@ -354,7 +360,8 @@ cudf::ast::expression const& createAstFromSubfieldFilter( const common::Filter& filter, cudf::ast::tree& tree, std::vector>& scalars, - const RowTypePtr& inputRowSchema) { + const RowTypePtr& inputRowSchema, + const ParquetColumnTypeMap* parquetColumnTypes) { // First, create column reference from subfield // For now, only support simple field references if (subfield.path().empty() || @@ -371,6 +378,14 @@ cudf::ast::expression const& createAstFromSubfieldFilter( VELOX_FAIL("Field '{}' not found in input schema", fieldName); } + std::optional parquetColumnType; + if (parquetColumnTypes != nullptr) { + auto it = parquetColumnTypes->find(fieldName); + if (it != parquetColumnTypes->end()) { + parquetColumnType = it->second; + } + } + auto columnIndex = inputRowSchema->getChildIdx(fieldName); auto const& columnRef = tree.push(cudf::ast::column_reference(columnIndex)); @@ -390,14 +405,15 @@ cudf::ast::expression const& createAstFromSubfieldFilter( tree, scalars, columnRef, - columnType); + columnType, + parquetColumnType); return result.get(); } case common::FilterKind::kHugeintRange: { auto const& columnType = inputRowSchema->childAt(columnIndex); - auto const& expr = - buildHugeintRangeExpr(filter, tree, scalars, columnRef, columnType); + auto const& expr = buildHugeintRangeExpr( + filter, tree, scalars, columnRef, columnType, parquetColumnType); return expr.get(); } @@ -406,7 +422,14 @@ cudf::ast::expression const& createAstFromSubfieldFilter( return buildValuesListExpr< TypeKind::BIGINT, common::BigintValuesUsingHashTable, - int64_t>(filter, tree, columnRef, scalars, columnType); + int64_t>( + filter, + tree, + columnRef, + scalars, + columnType, + false, + parquetColumnType); } case common::FilterKind::kBigintValuesUsingBitmask: { @@ -421,7 +444,8 @@ cudf::ast::expression const& createAstFromSubfieldFilter( columnRef, stream, mr, - columnType); + columnType, + parquetColumnType); return result.get(); } @@ -430,7 +454,14 @@ cudf::ast::expression const& createAstFromSubfieldFilter( return buildValuesListExpr< TypeKind::HUGEINT, common::HugeintValuesUsingHashTable, - int128_t>(filter, tree, columnRef, scalars, columnType); + int128_t>( + filter, + tree, + columnRef, + scalars, + columnType, + false, + parquetColumnType); } case common::FilterKind::kBytesValues: { @@ -508,7 +539,12 @@ cudf::ast::expression const& createAstFromSubfieldFilter( exprRefs.reserve(subFilters.size()); for (const auto* subFilter : subFilters) { auto const& subExpr = createAstFromSubfieldFilter( - subfield, *subFilter, tree, scalars, inputRowSchema); + subfield, + *subFilter, + tree, + scalars, + inputRowSchema, + parquetColumnTypes); exprRefs.push_back(&subExpr); } @@ -533,7 +569,8 @@ cudf::ast::expression const& createAstFromSubfieldFilters( const common::SubfieldFilters& subfieldFilters, cudf::ast::tree& tree, std::vector>& scalars, - const RowTypePtr& inputRowSchema) { + const RowTypePtr& inputRowSchema, + const ParquetColumnTypeMap* parquetColumnTypes) { using Op = cudf::ast::ast_operator; using Operation = cudf::ast::operation; @@ -545,7 +582,12 @@ cudf::ast::expression const& createAstFromSubfieldFilters( continue; } auto const& expr = createAstFromSubfieldFilter( - subfield, *filterPtr, tree, scalars, inputRowSchema); + subfield, + *filterPtr, + tree, + scalars, + inputRowSchema, + parquetColumnTypes); exprRefs.push_back(&expr); } diff --git a/velox/experimental/cudf/expression/SubfieldFiltersToAst.h b/velox/experimental/cudf/expression/SubfieldFiltersToAst.h index e44e2bea15b..29ede589dbd 100644 --- a/velox/experimental/cudf/expression/SubfieldFiltersToAst.h +++ b/velox/experimental/cudf/expression/SubfieldFiltersToAst.h @@ -19,10 +19,13 @@ #include "velox/type/Subfield.h" #include "velox/type/Type.h" +#include "velox/experimental/cudf/expression/ParquetSchemaUtils.h" + #include #include #include +#include #include namespace cudf { @@ -39,7 +42,8 @@ cudf::ast::expression const& createAstFromSubfieldFilter( const common::Filter& filter, cudf::ast::tree& tree, std::vector>& scalars, - const RowTypePtr& inputRowSchema); + const RowTypePtr& inputRowSchema, + const ParquetColumnTypeMap* parquetColumnTypes = nullptr); // Build a single AST expression representing logical AND of all filters in // 'subfieldFilters'. The resulting expression reference is owned by the passed @@ -48,6 +52,7 @@ cudf::ast::expression const& createAstFromSubfieldFilters( const common::SubfieldFilters& subfieldFilters, cudf::ast::tree& tree, std::vector>& scalars, - const RowTypePtr& inputRowSchema); + const RowTypePtr& inputRowSchema, + const ParquetColumnTypeMap* parquetColumnTypes = nullptr); } // namespace facebook::velox::cudf_velox diff --git a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp index 56303d2ed56..3a17b6f1991 100644 --- a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp +++ b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp @@ -16,6 +16,7 @@ #include "velox/experimental/cudf/exec/ToCudf.h" #include "velox/experimental/cudf/exec/VeloxCudfInterop.h" +#include "velox/experimental/cudf/expression/ParquetSchemaUtils.h" #include "velox/experimental/cudf/expression/SubfieldFiltersToAst.h" #include "velox/dwio/common/tests/utils/BatchMaker.h" @@ -587,6 +588,27 @@ TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal32Literals) { cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); } +TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal64LiteralsFromParquetSchema) { + const std::string columnName = "c0"; + auto rowType = ROW({{columnName, DECIMAL(9, 2)}}); + auto filter = std::make_unique( + int64_t{100}, int64_t{500}, /*nullAllowed*/ false); + ParquetColumnTypeMap parquetTypes{{columnName, cudf::type_id::DECIMAL64}}; + + common::Subfield subfield(columnName); + cudf::ast::tree tree; + std::vector> scalars; + const auto& expr = createAstFromSubfieldFilter( + subfield, *filter, tree, scalars, rowType, &parquetTypes); + + ASSERT_EQ(scalars.size(), 2UL); + EXPECT_EQ(scalars[0]->type().id(), cudf::type_id::DECIMAL64); + EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::DECIMAL64); + EXPECT_GT(tree.size(), 0UL); + EXPECT_NO_THROW( + cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); +} + TEST_F(SubfieldFilterAstTest, DecimalInList) { const std::string columnName = "c0"; auto rowType = ROW({{columnName, DECIMAL(20, 2)}}); From 7c37934b5e3e690e22e4faefd27bdfa9bc594900 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 15:48:20 -0700 Subject: [PATCH 08/12] [velox] fix(cudf): Fix GPU Iceberg build after filter refactor The previous commit deferred subfield filter AST construction to CudfSplitReader and duplicated the build-state wiring in CudfIcebergDataSource::createCudfSplitReader(). That subclass could not access the base class's private subfield filter members, so GPU-enabled Presto Native builds failed to compile velox_cudf_iceberg_connector: CudfIcebergDataSource.cpp:68: error: 'subfieldFilters_' is private within this context CudfIcebergDataSource.cpp:71: error: 'subfieldTree_' is private within this context CudfIcebergDataSource.cpp:72: error: 'subfieldScalars_' is private within this context CudfIcebergDataSource.cpp:73: error: 'getTableRowType()' is private within this context Add a protected makeSubfieldFilterBuildState() helper on CudfHiveDataSource and call it from both Hive and Iceberg createCudfSplitReader() overrides. The helper owns the only code path that assembles SubfieldFilterBuildState from the datasource's filter tree, scalars, and table row type. --- .../connectors/hive/CudfHiveDataSource.cpp | 25 +++++++++++-------- .../cudf/connectors/hive/CudfHiveDataSource.h | 3 +++ .../hive/iceberg/CudfIcebergDataSource.cpp | 12 +-------- 3 files changed, 18 insertions(+), 22 deletions(-) diff --git a/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp b/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp index b82d5afaa33..39601b12946 100644 --- a/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp +++ b/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.cpp @@ -134,16 +134,6 @@ CudfHiveDataSource::CudfHiveDataSource( } std::unique_ptr CudfHiveDataSource::createCudfSplitReader() { - SubfieldFilterBuildState subfieldFilterBuildState; - if (!subfieldFilters_.empty()) { - subfieldFilterBuildState = SubfieldFilterBuildState{ - .filters = &subfieldFilters_, - .tree = &subfieldTree_, - .scalars = &subfieldScalars_, - .rowType = getTableRowType(), - .expr = &subfieldFilterExpr_, - }; - } return std::make_unique( split_, tableHandle_, @@ -156,7 +146,20 @@ std::unique_ptr CudfHiveDataSource::createCudfSplitReader() { ioStatistics_, ioStats_, useExperimentalCudfReader_, - std::move(subfieldFilterBuildState)); + makeSubfieldFilterBuildState()); +} + +SubfieldFilterBuildState CudfHiveDataSource::makeSubfieldFilterBuildState() { + if (subfieldFilters_.empty()) { + return {}; + } + return SubfieldFilterBuildState{ + .filters = &subfieldFilters_, + .tree = &subfieldTree_, + .scalars = &subfieldScalars_, + .rowType = getTableRowType(), + .expr = &subfieldFilterExpr_, + }; } void CudfHiveDataSource::convertSplit(std::shared_ptr split) { diff --git a/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.h b/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.h index 0c6f3937813..0082bcc867c 100644 --- a/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.h +++ b/velox/experimental/cudf/connectors/hive/CudfHiveDataSource.h @@ -111,6 +111,9 @@ class CudfHiveDataSource : public DataSource, public NvtxHelper { // Cached combined subfield filter expression owned by 'subfieldTree_'. cudf::ast::expression const* subfieldFilterExpr_{nullptr}; + // Build state passed to CudfSplitReader for per-split AST construction. + SubfieldFilterBuildState makeSubfieldFilterBuildState(); + private: // Construct and cache a RowTypePtr for the table column names and types. const RowTypePtr getTableRowType(); diff --git a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp index d40a5e199c6..f30cac8d92e 100644 --- a/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp +++ b/velox/experimental/cudf/connectors/hive/iceberg/CudfIcebergDataSource.cpp @@ -64,16 +64,6 @@ void CudfIcebergDataSource::convertSplit( std::unique_ptr CudfIcebergDataSource::createCudfSplitReader() { - SubfieldFilterBuildState subfieldFilterBuildState; - if (!subfieldFilters_.empty()) { - subfieldFilterBuildState = SubfieldFilterBuildState{ - .filters = &subfieldFilters_, - .tree = &subfieldTree_, - .scalars = &subfieldScalars_, - .rowType = getTableRowType(), - .expr = &subfieldFilterExpr_, - }; - } return std::make_unique( split_, icebergSplit_, @@ -88,7 +78,7 @@ CudfIcebergDataSource::createCudfSplitReader() { ioStatistics_, ioStats_, useExperimentalCudfReader_, - std::move(subfieldFilterBuildState)); + makeSubfieldFilterBuildState()); } } // namespace facebook::velox::cudf_velox::connector::hive::iceberg From 3172cc1df9a99820b5e39b84ac4d2131a56cc5a3 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 16:13:09 -0700 Subject: [PATCH 09/12] [velox] fix(cudf): Match bloom-filter literals to Parquet type GPU table scans with decimal subfield filters can still fail during Parquet bloom-filter evaluation after 16db1f5a9 matched only DECIMAL32 vs DECIMAL64. libcudf checks the full cudf::data_type on the column and each equality literal, including fixed-point scale, not just the storage width: CUDF failure at: .../bloom_filter_reader.cu:67: Mismatched predicate column and literal types This change builds subfield filter literals from the Parquet column's full cudf::data_type read from the file footer: DECIMAL32/64/128 width, Parquet decimal scale, INT64/INT32 when decimals are stored without a DECIMAL logical type, and the Parquet timestamp unit when available. Literal wrapping now follows the scalar's actual type_id instead of inferring DECIMAL128 from Velox HUGEINT. Column projection is set before the filter AST is built, Parquet column lookup is case-insensitive, and the reader uses case-insensitive column names. Runtime verification on the remaining failing TPC-H/TPC-DS queries still hits the bloom_filter_reader type mismatch, so the root cause is not fully resolved. A likely remaining gap is that read_parquet_metadata types may still diverge from libcudf reader output_dtypes (reader options such as use_arrow_schema, or filter-only column ordering shifting column indices). --- .../cudf/connectors/hive/CudfSplitReader.cpp | 10 +- velox/experimental/cudf/expression/AstUtils.h | 149 ++++++++++++------ .../cudf/expression/ParquetSchemaUtils.h | 32 +++- .../cudf/expression/SubfieldFiltersToAst.cpp | 22 ++- .../cudf/tests/SubfieldFilterAstTest.cpp | 51 +++++- 5 files changed, 195 insertions(+), 69 deletions(-) diff --git a/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp b/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp index f72762cd213..20df016157e 100644 --- a/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp +++ b/velox/experimental/cudf/connectors/hive/CudfSplitReader.cpp @@ -373,6 +373,7 @@ void CudfSplitReader::setupReaderOptions() { .allow_mismatched_pq_schemas( cudfHiveConfig_->isAllowMismatchedCudfHiveSchemas()) .timestamp_type(cudfHiveConfig_->timestampType()) + .case_sensitive_names(false) .build(); // Set skip_bytes and num_bytes if available @@ -383,15 +384,14 @@ void CudfSplitReader::setupReaderOptions() { readerOptions_.set_num_bytes(split_->size()); } + if (readColumnNames_.size()) { + readerOptions_.set_column_names(readColumnNames_); + } + buildSubfieldFilterAst(); if (auto* filter = subfieldFilter(); filter != nullptr) { readerOptions_.set_filter(*filter); } - - // Set column projection if needed - if (readColumnNames_.size()) { - readerOptions_.set_column_names(readColumnNames_); - } } rmm::device_async_resource_ref CudfSplitReader::determineCudfMemoryResource() { diff --git a/velox/experimental/cudf/expression/AstUtils.h b/velox/experimental/cudf/expression/AstUtils.h index 176093125f7..b42ba088426 100644 --- a/velox/experimental/cudf/expression/AstUtils.h +++ b/velox/experimental/cudf/expression/AstUtils.h @@ -18,6 +18,7 @@ #include #include "velox/experimental/cudf/CudfConfig.h" #include "velox/experimental/cudf/CudfNoDefaults.h" +#include "velox/experimental/cudf/expression/ParquetSchemaUtils.h" #include "velox/expression/ConstantExpr.h" #include "velox/type/Timestamp.h" @@ -45,19 +46,51 @@ inline bool useCudfDecimal32ForVeloxDecimal(const TypePtr& type) { return precision <= kCudfParquetDecimal32MaxPrecision; } -// Subfield filter literals must match the Parquet column's libcudf storage -// type (DECIMAL32 vs DECIMAL64). When the Parquet schema is unavailable, -// fall back to the precision heuristic above. -inline bool useCudfDecimal32ForSubfieldFilter( - const TypePtr& type, - std::optional parquetColumnType = std::nullopt) { - if (!type->isShortDecimal()) { - return false; +inline cudf::type_id veloxDecimalFallbackStorageType(const TypePtr& type) { + if (type->isShortDecimal()) { + return useCudfDecimal32ForVeloxDecimal(type) ? cudf::type_id::DECIMAL32 + : cudf::type_id::DECIMAL64; } + if (type->isLongDecimal()) { + const auto [precision, _] = getDecimalPrecisionScale(*type); + return precision <= 18 ? cudf::type_id::DECIMAL64 : cudf::type_id::DECIMAL128; + } + VELOX_FAIL("Type is not decimal: {}", type->toString()); +} + +inline bool isParquetIntegerDecimalStorage(cudf::type_id typeId) { + return typeId == cudf::type_id::INT32 || typeId == cudf::type_id::INT64; +} + +inline std::pair decimalStorageTypeAndScale( + const TypePtr& type, + std::optional parquetColumnType = std::nullopt) { if (parquetColumnType.has_value()) { - return parquetColumnType.value() == cudf::type_id::DECIMAL32; + const auto parquetId = parquetColumnType->id(); + if (isParquetDecimalType(parquetId)) { + return {parquetId, + numeric::scale_type{parquetColumnType->scale()}}; + } + // Some Parquet files store scaled decimals as plain integers without a + // DECIMAL logical type. Bloom filters require literals to match that + // physical storage exactly. + if (type->isDecimal() && isParquetIntegerDecimalStorage(parquetId)) { + return {parquetId, numeric::scale_type{0}}; + } + } + + const auto [_, veloxScale] = getDecimalPrecisionScale(*type); + return {veloxDecimalFallbackStorageType(type), + numeric::scale_type{-veloxScale}}; +} + +inline cudf::type_id timestampUnitFromParquetColumnType( + std::optional parquetColumnType) { + if (!parquetColumnType.has_value() || + !cudf::is_timestamp(*parquetColumnType)) { + return CudfConfig::getInstance().timestampUnit; } - return useCudfDecimal32ForVeloxDecimal(type); + return parquetColumnType->id(); } template @@ -66,20 +99,32 @@ cudf::ast::literal makeLiteralFromScalar( const TypePtr& type) { if constexpr (cudf::is_fixed_width()) { if (type->isDecimal()) { - if (type->kind() == TypeKind::BIGINT) { - if (scalar.type().id() == cudf::type_id::DECIMAL32) { + switch (scalar.type().id()) { + case cudf::type_id::DECIMAL32: { using CudfScalarType = cudf::fixed_point_scalar; return cudf::ast::literal{*static_cast(&scalar)}; } - using CudfScalarType = cudf::fixed_point_scalar; - return cudf::ast::literal{*static_cast(&scalar)}; - } - if (type->kind() == TypeKind::HUGEINT) { - using CudfScalarType = cudf::fixed_point_scalar; - return cudf::ast::literal{*static_cast(&scalar)}; + case cudf::type_id::DECIMAL64: { + using CudfScalarType = cudf::fixed_point_scalar; + return cudf::ast::literal{*static_cast(&scalar)}; + } + case cudf::type_id::DECIMAL128: { + using CudfScalarType = cudf::fixed_point_scalar; + return cudf::ast::literal{*static_cast(&scalar)}; + } + case cudf::type_id::INT32: { + using CudfScalarType = cudf::numeric_scalar; + return cudf::ast::literal{*static_cast(&scalar)}; + } + case cudf::type_id::INT64: { + using CudfScalarType = cudf::numeric_scalar; + return cudf::ast::literal{*static_cast(&scalar)}; + } + default: + VELOX_FAIL( + "Unsupported scalar type for decimal literal: {}", + static_cast(scalar.type().id())); } - VELOX_UNREACHABLE( - "Invalid Decimal Type (bad TypeKind: {})", type->kind()); } else if (type->isIntervalDayTime()) { using CudfDurationType = cudf::duration_ms; if constexpr (std::is_same_v) { @@ -124,7 +169,7 @@ std::unique_ptr makeScalarFromValue( T value, bool isNull, std::optional toType = std::nullopt, - bool useCudfDecimal32 = false) { + std::optional parquetColumnType = std::nullopt) { auto stream = cudf::get_default_stream(cudf::allow_default_stream); auto mr = get_temp_mr(); @@ -136,7 +181,7 @@ std::unique_ptr makeScalarFromValue( // cheap (one-time cost per scalar) and guarantees the memory is // available on every stream. if constexpr (std::is_same_v) { - auto unit = CudfConfig::getInstance().timestampUnit; + auto unit = timestampUnitFromParquetColumnType(parquetColumnType); if (unit == cudf::type_id::TIMESTAMP_MICROSECONDS) { using CudfTimestampType = cudf::timestamp_us; auto micros = isNull ? 0 : value.toMicros(); @@ -172,38 +217,44 @@ std::unique_ptr makeScalarFromValue( if (type->isDecimal()) { // Velox DECIMAL scale is positive for fractional digits // cuDF scale is negative for fractional digits - // @TODO check the bigger picture here! - if (type->kind() == TypeKind::BIGINT) { - auto const decimalType = - std::dynamic_pointer_cast(type); - VELOX_CHECK(decimalType, "Invalid Decimal Type (failed dynamic_cast)"); - auto const cudfScale = numeric::scale_type{-decimalType->scale()}; - if (useCudfDecimal32 && - useCudfDecimal32ForVeloxDecimal(type)) { - using CudfDecimalType = cudf::fixed_point_scalar; - auto scalar = std::make_unique( - static_cast(value), cudfScale, !isNull, stream, mr); - stream.synchronize(); - return scalar; - } + const auto [storageType, cudfScale] = + decimalStorageTypeAndScale(type, parquetColumnType); + if (storageType == cudf::type_id::DECIMAL32) { + using CudfDecimalType = cudf::fixed_point_scalar; + auto scalar = std::make_unique( + static_cast(value), cudfScale, !isNull, stream, mr); + stream.synchronize(); + return scalar; + } + if (storageType == cudf::type_id::DECIMAL64) { using CudfDecimalType = cudf::fixed_point_scalar; auto scalar = std::make_unique( - value, cudfScale, !isNull, stream, mr); + static_cast(value), cudfScale, !isNull, stream, mr); stream.synchronize(); return scalar; - } else if (type->kind() == TypeKind::HUGEINT) { - auto const decimalType = - std::dynamic_pointer_cast(type); - VELOX_CHECK(decimalType, "Invalid Decimal Type (failed dynamic_cast)"); - auto const cudfScale = numeric::scale_type{-decimalType->scale()}; + } + if (storageType == cudf::type_id::DECIMAL128) { using CudfDecimalType = cudf::fixed_point_scalar; auto scalar = std::make_unique( - value, cudfScale, !isNull, stream, mr); + static_cast(value), cudfScale, !isNull, stream, mr); + stream.synchronize(); + return scalar; + } + if (storageType == cudf::type_id::INT64) { + auto scalar = std::make_unique>( + static_cast(value), !isNull, stream, mr); stream.synchronize(); return scalar; } - VELOX_UNREACHABLE( - "Invalid Decimal Type (bad TypeKind: {})", type->kind()); + if (storageType == cudf::type_id::INT32) { + auto scalar = std::make_unique>( + static_cast(value), !isNull, stream, mr); + stream.synchronize(); + return scalar; + } + VELOX_FAIL( + "Unsupported decimal storage type {}", + static_cast(storageType)); } else if (type->isIntervalYearMonth()) { VELOX_FAIL("Interval year month not supported"); } else if (type->isIntervalDayTime()) { @@ -275,18 +326,18 @@ cudf::ast::literal makeScalarAndLiteral( const variant& var, bool isNull, std::vector>& scalars, - bool useCudfDecimal32 = false) { + std::optional parquetColumnType = std::nullopt) { using T = typename TypeTraits::NativeType; if constexpr (cudf::is_fixed_width() || kind == TypeKind::VARCHAR) { if (isNull) { auto scalar = makeScalarFromValue( - type, T{}, true, std::nullopt, useCudfDecimal32); + type, T{}, true, std::nullopt, parquetColumnType); scalars.emplace_back(std::move(scalar)); return makeLiteralFromScalar(*(scalars.back()), type); } auto value = var.value(); - auto scalar = - makeScalarFromValue(type, value, false, std::nullopt, useCudfDecimal32); + auto scalar = makeScalarFromValue( + type, value, false, std::nullopt, parquetColumnType); scalars.emplace_back(std::move(scalar)); return makeLiteralFromScalar(*(scalars.back()), type); } diff --git a/velox/experimental/cudf/expression/ParquetSchemaUtils.h b/velox/experimental/cudf/expression/ParquetSchemaUtils.h index 713906181ef..996965536c0 100644 --- a/velox/experimental/cudf/expression/ParquetSchemaUtils.h +++ b/velox/experimental/cudf/expression/ParquetSchemaUtils.h @@ -18,12 +18,19 @@ #include #include +#include +#include #include #include namespace facebook::velox::cudf_velox { -using ParquetColumnTypeMap = std::unordered_map; +using ParquetColumnTypeMap = std::unordered_map; + +inline bool isParquetDecimalType(cudf::type_id typeId) { + return typeId == cudf::type_id::DECIMAL32 || + typeId == cudf::type_id::DECIMAL64 || typeId == cudf::type_id::DECIMAL128; +} // Maps top-level Parquet column names to libcudf storage types. Uses the same // schema resolution as cudf::io::read_parquet_metadata. @@ -34,10 +41,31 @@ inline ParquetColumnTypeMap parquetColumnTypesFromMetadata( for (int i = 0; i < root.num_children(); ++i) { const auto& child = root.child(i); if (!child.name().empty()) { - columnTypes.emplace(child.name(), child.cudf_type().id()); + columnTypes.emplace(child.name(), child.cudf_type()); } } return columnTypes; } +inline std::optional lookupParquetColumnType( + const ParquetColumnTypeMap& columnTypes, + const std::string& fieldName) { + if (auto it = columnTypes.find(fieldName); it != columnTypes.end()) { + return it->second; + } + for (const auto& [name, dataType] : columnTypes) { + if (name.size() == fieldName.size() && + std::equal( + name.begin(), + name.end(), + fieldName.begin(), + [](char a, char b) { + return std::tolower(a) == std::tolower(b); + })) { + return dataType; + } + } + return std::nullopt; +} + } // namespace facebook::velox::cudf_velox diff --git a/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp b/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp index b020a2019d1..dc4dd201a0a 100644 --- a/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp +++ b/velox/experimental/cudf/expression/SubfieldFiltersToAst.cpp @@ -44,13 +44,13 @@ cudf::ast::literal makeSubfieldFilterLiteral( const TypePtr& columnTypePtr, const variant& veloxVariant, std::vector>& scalars, - std::optional parquetColumnType = std::nullopt) { + std::optional parquetColumnType = std::nullopt) { return makeScalarAndLiteral( columnTypePtr, veloxVariant, false, scalars, - useCudfDecimal32ForSubfieldFilter(columnTypePtr, parquetColumnType)); + parquetColumnType); } template < @@ -131,7 +131,7 @@ std::reference_wrapper buildIntegerRangeExpr( std::vector>& scalars, const cudf::ast::expression& columnRef, const TypePtr& columnTypePtr, - std::optional parquetColumnType = std::nullopt) { + std::optional parquetColumnType = std::nullopt) { using NativeT = typename TypeTraits::NativeType; if constexpr ( @@ -212,7 +212,7 @@ std::reference_wrapper buildBigintRangeExpr( std::vector>& scalars, const cudf::ast::expression& columnRef, const TypePtr& columnTypePtr, - std::optional parquetColumnType = std::nullopt) { + std::optional parquetColumnType = std::nullopt) { return buildIntegerRangeExpr( filter, tree, scalars, columnRef, columnTypePtr, parquetColumnType); } @@ -223,7 +223,7 @@ std::reference_wrapper buildHugeintRangeExpr( std::vector>& scalars, const cudf::ast::expression& columnRef, const TypePtr& columnTypePtr, - std::optional parquetColumnType = std::nullopt) { + std::optional parquetColumnType = std::nullopt) { return buildIntegerRangeExpr( filter, tree, scalars, columnRef, columnTypePtr, parquetColumnType); } @@ -236,7 +236,7 @@ const cudf::ast::expression& buildValuesListExpr( std::vector>& scalars, const TypePtr& columnTypePtr, bool isNegated = false, - std::optional parquetColumnType = std::nullopt) { + std::optional parquetColumnType = std::nullopt) { using Op = cudf::ast::ast_operator; using Operation = cudf::ast::operation; @@ -305,7 +305,7 @@ std::reference_wrapper buildIntegerInListExpr( rmm::cuda_stream_view /*stream*/, rmm::device_async_resource_ref /*mr*/, const TypePtr& columnTypePtr, - std::optional parquetColumnType = std::nullopt) { + std::optional parquetColumnType = std::nullopt) { using NativeT = typename TypeTraits::NativeType; if constexpr (std::is_integral_v) { @@ -378,12 +378,10 @@ cudf::ast::expression const& createAstFromSubfieldFilter( VELOX_FAIL("Field '{}' not found in input schema", fieldName); } - std::optional parquetColumnType; + std::optional parquetColumnType; if (parquetColumnTypes != nullptr) { - auto it = parquetColumnTypes->find(fieldName); - if (it != parquetColumnTypes->end()) { - parquetColumnType = it->second; - } + parquetColumnType = + lookupParquetColumnType(*parquetColumnTypes, fieldName); } auto columnIndex = inputRowSchema->getChildIdx(fieldName); diff --git a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp index 3a17b6f1991..e71a1c01b66 100644 --- a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp +++ b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp @@ -29,6 +29,7 @@ #include #include +#include #include @@ -593,7 +594,8 @@ TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal64LiteralsFromParquet auto rowType = ROW({{columnName, DECIMAL(9, 2)}}); auto filter = std::make_unique( int64_t{100}, int64_t{500}, /*nullAllowed*/ false); - ParquetColumnTypeMap parquetTypes{{columnName, cudf::type_id::DECIMAL64}}; + ParquetColumnTypeMap parquetTypes{ + {columnName, cudf::data_type{cudf::type_id::DECIMAL64, -2}}}; common::Subfield subfield(columnName); cudf::ast::tree tree; @@ -603,7 +605,54 @@ TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal64LiteralsFromParquet ASSERT_EQ(scalars.size(), 2UL); EXPECT_EQ(scalars[0]->type().id(), cudf::type_id::DECIMAL64); + EXPECT_EQ(scalars[0]->type().scale(), numeric::scale_type{-2}); EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::DECIMAL64); + EXPECT_EQ(scalars[1]->type().scale(), numeric::scale_type{-2}); + EXPECT_GT(tree.size(), 0UL); + EXPECT_NO_THROW( + cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); +} + +TEST_F(SubfieldFilterAstTest, LongDecimalFilterUsesDecimal64LiteralsFromParquetSchema) { + const std::string columnName = "c0"; + auto rowType = ROW({{columnName, DECIMAL(18, 2)}}); + auto filter = std::make_unique( + int128_t{100}, int128_t{500}, /*nullAllowed*/ false); + ParquetColumnTypeMap parquetTypes{ + {columnName, cudf::data_type{cudf::type_id::DECIMAL64, -2}}}; + + common::Subfield subfield(columnName); + cudf::ast::tree tree; + std::vector> scalars; + const auto& expr = createAstFromSubfieldFilter( + subfield, *filter, tree, scalars, rowType, &parquetTypes); + + ASSERT_EQ(scalars.size(), 2UL); + EXPECT_EQ(scalars[0]->type().id(), cudf::type_id::DECIMAL64); + EXPECT_EQ(scalars[0]->type().scale(), numeric::scale_type{-2}); + EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::DECIMAL64); + EXPECT_EQ(scalars[1]->type().scale(), numeric::scale_type{-2}); + EXPECT_GT(tree.size(), 0UL); + EXPECT_NO_THROW( + cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); +} + +TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesInt64LiteralsFromParquetSchema) { + const std::string columnName = "c0"; + auto rowType = ROW({{columnName, DECIMAL(9, 2)}}); + auto filter = std::make_unique( + int64_t{100}, int64_t{500}, /*nullAllowed*/ false); + ParquetColumnTypeMap parquetTypes{{columnName, cudf::data_type{cudf::type_id::INT64}}}; + + common::Subfield subfield(columnName); + cudf::ast::tree tree; + std::vector> scalars; + const auto& expr = createAstFromSubfieldFilter( + subfield, *filter, tree, scalars, rowType, &parquetTypes); + + ASSERT_EQ(scalars.size(), 2UL); + EXPECT_EQ(scalars[0]->type().id(), cudf::type_id::INT64); + EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::INT64); EXPECT_GT(tree.size(), 0UL); EXPECT_NO_THROW( cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); From 951a9c70662f07b57e40161609ecde2bbae7c140 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 18:22:18 -0700 Subject: [PATCH 10/12] Final compile fixes --- .../cudf/tests/DecimalAggregationTest.cpp | 11 +++++----- .../cudf/tests/SubfieldFilterAstTest.cpp | 9 -------- .../experimental/cudf/tests/TableScanTest.cpp | 21 +++++++++++-------- 3 files changed, 18 insertions(+), 23 deletions(-) diff --git a/velox/experimental/cudf/tests/DecimalAggregationTest.cpp b/velox/experimental/cudf/tests/DecimalAggregationTest.cpp index d6d09fb2259..0fc6f7bc113 100644 --- a/velox/experimental/cudf/tests/DecimalAggregationTest.cpp +++ b/velox/experimental/cudf/tests/DecimalAggregationTest.cpp @@ -1153,18 +1153,19 @@ TEST_F(CudfDecimalTest, decimalDeserializeSumStateAfterVarbinaryRoundTrip) { serializeDecimalSumState(sumCol->view(), countCol->view(), stream, mr); auto rowType = ROW({{"state", VARBINARY()}}); + cudf::table_view stateTable{{stateCol->view()}}; auto velox = with_arrow::toVeloxColumn( - stateCol->view(), pool(), rowType, "rt_", stream, mr); + stateTable, pool(), rowType, "rt_", stream, mr); auto cudfAgain = with_arrow::toCudfTable(velox, pool(), stream, mr); + auto const stateView = cudfAgain->view().column(0); - cudf::strings_column_view strings(cudfAgain->column(0)); + cudf::strings_column_view strings(stateView); EXPECT_LT( strings.chars_size(stream), static_cast(sums.size()) * detail::kDecimalSumStateSize); - auto sumAndCount = - deserializeDecimalSumState(cudfAgain->column(0), 2, stream); - auto stateMask = copyNullMask(cudfAgain->column(0), stream); + auto sumAndCount = deserializeDecimalSumState(stateView, 2, stream); + auto stateMask = copyNullMask(stateView, stream); auto sumMask = copyNullMask(sumAndCount.sum->view(), stream); EXPECT_EQ(stateMask, sumMask); diff --git a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp index e71a1c01b66..d0063cbc0a6 100644 --- a/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp +++ b/velox/experimental/cudf/tests/SubfieldFilterAstTest.cpp @@ -28,7 +28,6 @@ #include #include -#include #include #include @@ -585,8 +584,6 @@ TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal32Literals) { EXPECT_EQ(scalars[0]->type().id(), cudf::type_id::DECIMAL32); EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::DECIMAL32); EXPECT_GT(tree.size(), 0UL); - EXPECT_NO_THROW( - cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); } TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal64LiteralsFromParquetSchema) { @@ -609,8 +606,6 @@ TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesDecimal64LiteralsFromParquet EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::DECIMAL64); EXPECT_EQ(scalars[1]->type().scale(), numeric::scale_type{-2}); EXPECT_GT(tree.size(), 0UL); - EXPECT_NO_THROW( - cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); } TEST_F(SubfieldFilterAstTest, LongDecimalFilterUsesDecimal64LiteralsFromParquetSchema) { @@ -633,8 +628,6 @@ TEST_F(SubfieldFilterAstTest, LongDecimalFilterUsesDecimal64LiteralsFromParquetS EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::DECIMAL64); EXPECT_EQ(scalars[1]->type().scale(), numeric::scale_type{-2}); EXPECT_GT(tree.size(), 0UL); - EXPECT_NO_THROW( - cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); } TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesInt64LiteralsFromParquetSchema) { @@ -654,8 +647,6 @@ TEST_F(SubfieldFilterAstTest, ShortDecimalFilterUsesInt64LiteralsFromParquetSche EXPECT_EQ(scalars[0]->type().id(), cudf::type_id::INT64); EXPECT_EQ(scalars[1]->type().id(), cudf::type_id::INT64); EXPECT_GT(tree.size(), 0UL); - EXPECT_NO_THROW( - cudf::ast::detail::expression_parser{}.visit(expr, cudf::table_reference::LEFT)); } TEST_F(SubfieldFilterAstTest, DecimalInList) { diff --git a/velox/experimental/cudf/tests/TableScanTest.cpp b/velox/experimental/cudf/tests/TableScanTest.cpp index 0c91b37b342..1f909b8cfbf 100644 --- a/velox/experimental/cudf/tests/TableScanTest.cpp +++ b/velox/experimental/cudf/tests/TableScanTest.cpp @@ -95,7 +95,7 @@ StatsFilterMetrics readParquetWithStatsFilter( template void copyHostToDeviceColumn( - cudf::column_view& view, + cudf::mutable_column_view view, const std::vector& values, rmm::cuda_stream_view stream) { if (values.empty()) { @@ -140,9 +140,10 @@ void writeDecimal32ParquetFile( copyHostToDeviceColumn(condCol->mutable_view(), condValues, stream); auto decimalCol = makeDecimal32Column(decimalValues, decimalScale, stream); - auto table = std::make_unique( - std::vector>{std::move(condCol), - std::move(decimalCol)}); + std::vector> columns; + columns.push_back(std::move(condCol)); + columns.push_back(std::move(decimalCol)); + auto table = std::make_unique(std::move(columns)); auto sinkInfo = cudf::io::sink_info(filePath); auto tableInputMetadata = cudf::io::table_input_metadata(table->view()); @@ -171,9 +172,10 @@ void writeDecimal32AndInt64ParquetFile( cudf::mask_state::UNALLOCATED, stream); copyHostToDeviceColumn(intCol->mutable_view(), intValues, stream); - auto table = std::make_unique( - std::vector>{std::move(decimalCol), - std::move(intCol)}); + std::vector> columns; + columns.push_back(std::move(decimalCol)); + columns.push_back(std::move(intCol)); + auto table = std::make_unique(std::move(columns)); auto sinkInfo = cudf::io::sink_info(filePath); auto tableInputMetadata = cudf::io::table_input_metadata(table->view()); @@ -195,8 +197,9 @@ void writeSingleDecimal32ParquetFile( rmm::cuda_stream_view stream) { auto decimalCol = makeDecimal32Column(decimalValues, decimalScale, stream); - auto table = std::make_unique( - std::vector>{std::move(decimalCol)}); + std::vector> columns; + columns.push_back(std::move(decimalCol)); + auto table = std::make_unique(std::move(columns)); auto sinkInfo = cudf::io::sink_info(filePath); auto tableInputMetadata = cudf::io::table_input_metadata(table->view()); From 59523dbe5d9771557a3777ce67ef0c9883e04f02 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 18:26:05 -0700 Subject: [PATCH 11/12] Fix TableScanTest.decimal32FilterWithMultiply --- velox/experimental/cudf/tests/TableScanTest.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/velox/experimental/cudf/tests/TableScanTest.cpp b/velox/experimental/cudf/tests/TableScanTest.cpp index 1f909b8cfbf..e1f7aa79475 100644 --- a/velox/experimental/cudf/tests/TableScanTest.cpp +++ b/velox/experimental/cudf/tests/TableScanTest.cpp @@ -1063,11 +1063,11 @@ TEST_F(TableScanTest, decimal32FilterWithMultiply) { .assignments(assignments) .endTableScan() .filter( - "d * CAST(0.2 AS DECIMAL(9, 2)) > CAST(1.00 AS DECIMAL(9, 2))") + "d * CAST(0.2 AS DECIMAL(9, 2)) > CAST(1.00 AS DECIMAL(18, 4))") .planNode(); assertQuery( plan, {filePath}, - "SELECT d FROM tmp WHERE d * CAST(0.2 AS DECIMAL(9, 2)) > CAST(1.00 AS DECIMAL(9, 2))"); + "SELECT d FROM tmp WHERE d * CAST(0.2 AS DECIMAL(9, 2)) > CAST(1.00 AS DECIMAL(18, 4))"); } From a1d4b56c87de891518d6079f3381c9687f945767 Mon Sep 17 00:00:00 2001 From: Simon Eves Date: Tue, 14 Jul 2026 20:51:00 -0700 Subject: [PATCH 12/12] [velox] fix(arrow): Fix TIME import test after DECIMAL32 branch ArrowBridgeArrayImportAsViewerTest.scalar crashed while importing Arrow time32 arrays (`tts`) because the DECIMAL32 assertion branch matched `int32_t` input with `int64_t` output and compared raw second values against Velox TIME vectors stored as milliseconds. Route `tt*` formats through assertTimeVectorContent so TIME imports use the same unit conversion as the existing TIME test path. --- velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp | 3 +++ 1 file changed, 3 insertions(+) diff --git a/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp b/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp index f8186190e5d..71ba3e4f0df 100644 --- a/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp +++ b/velox/vector/arrow/tests/ArrowBridgeArrayTest.cpp @@ -1285,6 +1285,9 @@ class ArrowBridgeArrayImportTest : public ArrowBridgeArrayExportTest { } } assertVectorContent(widenedValues, output, arrowArray.null_count); + } else if (format[0] == 't' && format[1] == 't') { + assertTimeVectorContent( + inputValues, output, arrowArray.null_count, format); } else { assertVectorContent(inputValues, output, arrowArray.null_count); }