More DECIMAL32 experiments (not worth an upstream PR)#12
Open
simoneves wants to merge 12 commits into
Open
Conversation
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`.
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.
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.
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.
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.
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.
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.
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.
GPU table scans with decimal subfield filters can still fail during Parquet bloom-filter evaluation after 16db1f5 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).
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.
Selective Build Plan
Selective build plan |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.