[FLINK-40360][runtime] Fix DECIMAL precision truncation when arithmetic result precision exceeds 19 digits - #4503
Open
haruki-830 wants to merge 1 commit into
Open
[FLINK-40360][runtime] Fix DECIMAL precision truncation when arithmetic result precision exceeds 19 digits#4503haruki-830 wants to merge 1 commit into
haruki-830 wants to merge 1 commit into
Conversation
github-actions
Bot
added
docs
haruki-830
marked this pull request as ready for review
August 11, 2026 03:04
Contributor
There was a problem hiding this comment.
Pull request overview
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Adds a configurable DECIMAL precision mode for YAML Transform evaluation to avoid precision truncation beyond 19 digits, while keeping existing behavior as default.
Changes:
- Introduced
DecimalPrecisionMode(UP_TO_19 default, UP_TO_38 opt-in) and a Calcite type system override to drive DECIMAL inference. - Propagated precision mode through composer → translator → runtime operator → parser/Janino paths.
- Expanded unit/integration/spec coverage and documented the new pipeline option.
Reviewed changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java | Adds unit test asserting DECIMAL inference differs between UP_TO_19 and UP_TO_38. |
| flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java | Threads DecimalPrecisionMode through projection/filter type inference and SQL-to-Rel conversion. |
| flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java | Extends Janino compilation context to carry decimal precision mode into type deduction. |
| flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/FlinkCdcTypeSystem.java | New Calcite type system wrapper to control max numeric precision/scale (19 vs 38). |
| flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java | Passes precision mode into projection parsing/processor construction. |
| flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java | Passes precision mode into filter expression translation/cache key generation. |
| flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java | Stores configured precision mode and forwards it into operator construction. |
| flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java | Propagates precision mode into schema projection and filter/projection processors. |
| flink-cdc-composer/src/test/resources/specs/decimal.yaml | Adds test specs for arithmetic under both precision modes. |
| flink-cdc-composer/src/test/resources/specs/casting.yaml | Adds test specs for casts to DECIMAL(38, *) under both modes. |
| flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java | Parses YAML decimal-precision-mode and sets pipeline option for specs runs. |
| flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java | Parameterizes numeric casting IT over both precision modes and adds DECIMAL(38, *) expectations. |
| flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java | Wires precision mode into post-transform operator builder. |
| flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java | Reads pipeline option and forwards precision mode into translation. |
| flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java | Updates DECIMAL conversion test expectations to match target scale/precision conversion. |
| flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java | Adds transform.decimal.precision.mode config option with documentation. |
| flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/DecimalPrecisionMode.java | New enum defining precision modes. |
| flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java | Ensures DECIMAL conversion uses target DecimalType rather than source value’s precision/scale. |
| flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java | Implements DECIMAL conversion respecting the target precision/scale (including re-scaling existing DecimalData). |
| docs/content/docs/core-concept/data-pipeline.md | Documents the new transform.decimal.precision.mode option (EN). |
| docs/content.zh/docs/core-concept/data-pipeline.md | Documents the new transform.decimal.precision.mode option (ZH). |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
236
to
+240
| columns, | ||
| udfDescriptors, | ||
| supportedMetadataColumns, | ||
| transformFilter.getColumnNameMap()); | ||
| transformFilter.getColumnNameMap(), | ||
| decimalPrecisionMode); |
Comment on lines
+36
to
+45
| public static FlinkCdcTypeSystem of(DecimalPrecisionMode mode) { | ||
| switch (mode) { | ||
| case UP_TO_38: | ||
| return UP_TO_38; | ||
| case UP_TO_19: | ||
| return UP_TO_19; | ||
| default: | ||
| throw new IllegalArgumentException("Unexpected decimal precision mode: " + mode); | ||
| } | ||
| } |
Comment on lines
+618
to
627
| public static String translateFilterExpressionToJaninoExpression( | ||
| String filterExpression, | ||
| List<Column> columns, | ||
| List<UserDefinedFunctionDescriptor> udfDescriptors, | ||
| SupportedMetadataColumn[] supportedMetadataColumns, | ||
| Map<String, String> columnNameMap, | ||
| DecimalPrecisionMode decimalPrecisionMode) { | ||
| if (isNullOrWhitespaceOnly(filterExpression)) { | ||
| return ""; | ||
| } |
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
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.
Summary
This commit fixes DECIMAL precision truncation in YAML Transform by adding an explicit decimal precision mode. The default
UP_TO_19mode preserves existing Calcite behavior, while the opt-inUP_TO_38mode supports the maximum DECIMAL precision provided by Flink CDC.Key Changes
transform.decimal.precision.modepipeline option.UP_TO_19as the default value to preserve backward compatibility.UP_TO_38as an opt-in mode to support DECIMAL precision and scale up to 38.NULLinUP_TO_38mode.JIRA Reference
https://issues.apache.org/jira/browse/FLINK-40360