fix(flink): preserve Avro fixed decimal widths in Parquet writes - #19522
Conversation
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR makes the Flink RowData Parquet writer honor the declared Avro fixed size for fixed-backed decimals (instead of deriving width from precision alone) and replaces the shift-based compact-decimal encoding with a sign-extending big-endian write that is safe when the fixed width exceeds eight bytes. I traced the encoding paths for both writers (LongUnscaledBytesWriter for precision ≤ 18, UnscaledBytesWriter for 19–38) across boundary widths and positive/negative/nullable/nested cases, and confirmed the Parquet schema length and the emitted binary length stay in sync (both now use decimalFixedLen(fieldSchema, precision) on the non-null schema), so writes remain internally consistent and read-compatible. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One small naming nit; otherwise the change is clean and readable.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR makes the Flink RowData Parquet writer honor the declared Avro fixed size when converting fixed-backed decimals, and adds big-endian sign-extension so compact decimals encode correctly when the declared fixed width exceeds eight bytes. I traced both the schema conversion and value-encoding paths: they now resolve the byte width through the shared resolveDecimalByteLength helper after stripping nullability, so the declared Parquet length and the encoded value width stay consistent; the sign-extension logic matches the test vectors, and Avro validation guarantees the fixed size is never smaller than the precision minimum, so widening only pads and never truncates. No issues flagged from this automated pass, a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
Honor declared Avro fixed sizes in both the Flink Parquet schema converter and RowData value writer. Safely sign-extend compact decimals when the declared fixed width exceeds eight bytes.
Rename the helper to describe both fixed-schema resolution and precision-based fallback behavior. Addresses review comment 3718983684.
c06e676 to
1046c0a
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR makes the Flink RowData Parquet writer honor the declared Avro fixed size for fixed-backed decimals so the emitted FIXED_LEN_BYTE_ARRAY schema width and the encoded value width stay consistent, and adds sign-extension for widths over eight bytes. I traced the compact (LongUnscaledBytesWriter) and non-compact (UnscaledBytesWriter) encoding paths across boundary, negative, and over-allocated-fixed cases, and confirmed the schema converter and value writer share the same width resolver. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One minor naming nit below, otherwise clean.
cc @yihua
| return numBytes; | ||
| } | ||
|
|
||
| static int resolveDecimalByteLength(HoodieSchema fieldSchema, int precision) { |
There was a problem hiding this comment.
🤖 nit: the sibling method uses compute as its verb (computeMinBytesForDecimalPrecision); could you align the name here — e.g. computeDecimalByteLength — so the two helpers read as a family?
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19522 +/- ##
============================================
- Coverage 76.40% 73.76% -2.64%
+ Complexity 32398 31014 -1384
============================================
Files 2520 2520
Lines 138985 138987 +2
Branches 16695 16697 +2
============================================
- Hits 106189 102523 -3666
- Misses 25166 28680 +3514
- Partials 7630 7784 +154
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
) * fix(flink): preserve Avro fixed decimal widths in Parquet writes Honor declared Avro fixed sizes in both the Flink Parquet schema converter and RowData value writer. Safely sign-extend compact decimals when the declared fixed width exceeds eight bytes. * style(flink): clarify decimal byte length resolver name Rename the helper to describe both fixed-schema resolution and precision-based fallback behavior. Addresses review comment 3718983684. (cherry picked from commit 0425c5c) Adaptations for release-1.2.1: The fix needs each field's HoodieSchema to read the Avro fixed size, and master had already plumbed that through. Here the schema reached RowDataParquetWriteSupport and was converted straight to a RowType and dropped, so the threading is backported alongside the fix, without master's VECTOR work: - HoodieSchemaUtils.getFieldSchema, matching master. - A convertToParquetMessageType(String, HoodieSchema) overload, and a fieldSchema argument threaded through convertToParquetType into ROW, ARRAY and MAP. - ParquetRowDataWriter takes master's (RecordConsumer, boolean, HoodieSchema) constructor. The RowType is derivable from the schema and the GroupType was unused. This also lets the upstream test file apply unchanged. Unlike master, convertToParquetMessageType(String, RowType) does not delegate to the HoodieSchema overload. Converting a RowType to a HoodieSchema is lossy for two types Parquet supports and Avro does not, and both are covered by existing tests here: a map with a non-string key, and a timestamp of precision 9. The field schema is therefore @nullable, and resolveDecimalByteLength already falls back to computeMinBytesForDecimalPrecision when it is absent.
) * fix(flink): preserve Avro fixed decimal widths in Parquet writes Honor declared Avro fixed sizes in both the Flink Parquet schema converter and RowData value writer. Safely sign-extend compact decimals when the declared fixed width exceeds eight bytes. * style(flink): clarify decimal byte length resolver name Rename the helper to describe both fixed-schema resolution and precision-based fallback behavior. Addresses review comment 3718983684. (cherry picked from commit 0425c5c) Adaptations for release-1.2.1: The fix needs each field's HoodieSchema to read the Avro fixed size, and master had already plumbed that through. Here the schema reached RowDataParquetWriteSupport and was converted straight to a RowType and dropped, so the threading is backported alongside the fix, without master's VECTOR work: - HoodieSchemaUtils.getFieldSchema, matching master. - A convertToParquetMessageType(String, HoodieSchema) overload, and a fieldSchema argument threaded through convertToParquetType into ROW, ARRAY and MAP. - ParquetRowDataWriter takes master's (RecordConsumer, boolean, HoodieSchema) constructor. The RowType is derivable from the schema and the GroupType was unused. This also lets the upstream test file apply unchanged. Unlike master, convertToParquetMessageType(String, RowType) does not delegate to the HoodieSchema overload. Converting a RowType to a HoodieSchema is lossy for two types Parquet supports and Avro does not, and both are covered by existing tests here: a map with a non-string key, and a timestamp of precision 9. The field schema is therefore @nullable, and resolveDecimalByteLength already falls back to computeMinBytesForDecimalPrecision when it is absent.
5 participants
Describe the issue this Pull Request addresses
Closes #19521.
The Flink
RowDataParquet writer previously derived the physical width of every decimal from precision alone. This produced a schema and encoded values that were narrower than the authoritative Hudi/Avro schema when a decimal used an over-allocated fixed backing type. The compact-decimal encoding path also relied on shifts that are unsafe when the declared fixed width exceeds eight bytes.Summary and Changelog
FIXED_LEN_BYTE_ARRAY, while retaining the precision-based minimum for non-fixed decimal schemas.RowDatavalue writer and sign-extend compact positive and negative decimal values when the declared fixed width is greater than eight bytes.Impact
Flink Parquet writes now preserve the declared physical width of fixed-backed decimal schemas passed to the writer. There are no public API or configuration changes. Decimal schemas without a fixed backing type retain the existing minimum-width behavior.
Risk Level
Low. The change is limited to Flink Parquet decimal schema conversion and value encoding. Targeted tests cover schema widths and encoded values for both fixed-backed and fallback decimal representations.
Documentation Update
None.
Contributor's checklist