This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 4e3767a75 [FLINK-40360][runtime] Fix DECIMAL precision truncation when
arithmetic result precision exceeds 19 digits (#4503)
4e3767a75 is described below
commit 4e3767a75a7f0fd38b68941d061c7f12f1316ce1
Author: haruki <[email protected]>
AuthorDate: Tue Aug 18 21:03:14 2026 +0800
[FLINK-40360][runtime] Fix DECIMAL precision truncation when arithmetic
result precision exceeds 19 digits (#4503)
Co-authored-by: 春栖 <[email protected]>
---
docs/content.zh/docs/core-concept/data-pipeline.md | 1 +
docs/content/docs/core-concept/data-pipeline.md | 2 +-
.../cdc/common/converter/CommonConverter.java | 15 +-
.../common/converter/InternalObjectConverter.java | 2 +-
.../cdc/common/pipeline/DecimalPrecisionMode.java | 34 ++++
.../flink/cdc/common/pipeline/PipelineOptions.java | 19 ++
.../converter/InternalObjectConverterTest.java | 2 +-
.../cdc/composer/flink/FlinkPipelineComposer.java | 3 +
.../flink/translator/TransformTranslator.java | 3 +
.../flink/FlinkPipelineTransformITCase.java | 90 ++++++++--
.../cdc/composer/specs/TransformSpecsITCase.java | 11 +-
.../src/test/resources/specs/casting.yaml | 120 +++++++++++++
.../src/test/resources/specs/decimal.yaml | 196 ++++++++++++++++++---
.../operators/transform/PostTransformOperator.java | 14 +-
.../transform/PostTransformOperatorBuilder.java | 11 +-
.../transform/TransformFilterProcessor.java | 14 +-
.../transform/TransformProjectionProcessor.java | 7 +-
.../cdc/runtime/parser/FlinkCdcTypeSystem.java | 56 ++++++
.../flink/cdc/runtime/parser/JaninoCompiler.java | 47 ++++-
.../flink/cdc/runtime/parser/TransformParser.java | 97 ++++++++--
.../cdc/runtime/parser/TransformParserTest.java | 34 ++++
21 files changed, 706 insertions(+), 72 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/data-pipeline.md
b/docs/content.zh/docs/core-concept/data-pipeline.md
index 928f6880c..bf4f3212e 100644
--- a/docs/content.zh/docs/core-concept/data-pipeline.md
+++ b/docs/content.zh/docs/core-concept/data-pipeline.md
@@ -124,5 +124,6 @@ under the License.
| `schema-operator.rpc-timeout` | SchemaOperator 等待下游 SchemaChangeEvent
应用完成的超时时间，默认值是 3 分钟。
[...]
| `operator.uid.prefix` | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的
UID。 建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。
[...]
| `sink.partitioning.strategy` | Sink
写入数据时使用的分区策略。数据类型：String。默认值：`SINK_DEFINED`。备注：可配置的值如下：`SINK_DEFINED`：使用 Sink
定义的分区策略；`PRIMARY_KEY`：按表 ID 和主键分区；`TABLE_ID`：仅按表 ID 分区。
| optional |
+| `transform.decimal.precision.mode` | transform 表达式求值中 DECIMAL
类型的最大精度模式。可选值：`UP_TO_19`（默认，使用 Calcite 默认类型系统）或 `UP_TO_38`（允许 DECIMAL 精度最高为 38
位）。
| optional |
注意：虽然上述参数都是可选的，但至少需要指定其中一个。`pipeline` 部分是必需的，不能为空。
diff --git a/docs/content/docs/core-concept/data-pipeline.md
b/docs/content/docs/core-concept/data-pipeline.md
index 68f3ca212..4b35ad6ca 100644
--- a/docs/content/docs/core-concept/data-pipeline.md
+++ b/docs/content/docs/core-concept/data-pipeline.md
@@ -126,6 +126,6 @@ Note that whilst the parameters are each individually
optional, at least one of
| `schema-operator.rpc-timeout` | The timeout time for SchemaOperator to wait
downstream SchemaChangeEvent applying finished, the default value is 3 minutes.
[...]
| `operator.uid.prefix` | The prefix to use for all pipeline operator
UIDs. If not set, all pipeline operator UIDs will be generated by Flink. It is
recommended to set this parameter to ensure stable and recognizable operator
UIDs, which can help with stateful upgrades, troubleshooting, and Flink UI
diagnostics.
[...]
| `sink.partitioning.strategy` | The partitioning strategy used when writing
data to the sink. Data type: String. Default value: `SINK_DEFINED`. Available
values: `SINK_DEFINED`: uses the partitioning strategy defined by the sink;
`PRIMARY_KEY`: partitions by table ID and primary key; `TABLE_ID`: partitions
only by table ID.
[...]
+| `transform.decimal.precision.mode` | Maximum precision mode for DECIMAL type
in transform expression evaluation. One of: `UP_TO_19` (default, match
Calcite's default type system) or `UP_TO_38` (allow DECIMAL precision up to 38
digits).
[...]
NOTE: Whilst the above parameters are each individually optional, at least one
of them must be specified. The `pipeline` section is mandatory and cannot be
empty.
-
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java
index feb0e4459..8260c305d 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/CommonConverter.java
@@ -33,6 +33,7 @@ import org.apache.flink.cdc.common.data.ZonedTimestampData;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
import org.apache.flink.cdc.common.types.ArrayType;
import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DecimalType;
import org.apache.flink.cdc.common.types.MapType;
import org.apache.flink.cdc.common.types.RowType;
import org.apache.flink.cdc.common.types.VariantType;
@@ -153,13 +154,21 @@ public class CommonConverter {
"Cannot convert " + obj + " of type " + obj.getClass() + " to
STRING DATA.");
}
- static DecimalData convertToDecimalData(Object obj) {
+ static DecimalData convertToDecimalData(Object obj, DecimalType
decimalType) {
if (obj instanceof DecimalData) {
- return (DecimalData) obj;
+ DecimalData dd = (DecimalData) obj;
+ // Re-convert to target precision and scale if different
+ if (dd.precision() == decimalType.getPrecision()
+ && dd.scale() == decimalType.getScale()) {
+ return dd;
+ }
+ return DecimalData.fromBigDecimal(
+ dd.toBigDecimal(), decimalType.getPrecision(),
decimalType.getScale());
}
if (obj instanceof BigDecimal) {
BigDecimal bd = (BigDecimal) obj;
- return DecimalData.fromBigDecimal(bd, bd.precision(), bd.scale());
+ return DecimalData.fromBigDecimal(
+ bd, decimalType.getPrecision(), decimalType.getScale());
}
throw new RuntimeException(
"Cannot convert " + obj + " of type " + obj.getClass() + " to
DECIMAL DATA.");
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java
index 7f1cc97bd..0dc664e54 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java
@@ -87,7 +87,7 @@ public class InternalObjectConverter {
@Override
public Function<Object, DecimalData> visit(DecimalType decimalType) {
- return CommonConverter::convertToDecimalData;
+ return obj -> CommonConverter.convertToDecimalData(obj,
decimalType);
}
@Override
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/DecimalPrecisionMode.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/DecimalPrecisionMode.java
new file mode 100644
index 000000000..79a733fb1
--- /dev/null
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/DecimalPrecisionMode.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package org.apache.flink.cdc.common.pipeline;
+
+import org.apache.flink.cdc.common.annotation.PublicEvolving;
+
+/**
+ * Maximum precision mode for DECIMAL type in transform expressions. Controls
the upper bound of
+ * numeric precision used by the SQL type system during expression evaluation.
+ */
+@PublicEvolving
+public enum DecimalPrecisionMode {
+
+ /** Limits DECIMAL precision to 19 digits, matching Calcite's default type
system behavior. */
+ UP_TO_19,
+
+ /** Allows DECIMAL precision up to 38 digits, matching Flink CDC's
extended type system. */
+ UP_TO_38
+}
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
index 7c6e49b9e..d46c94c65 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java
@@ -156,5 +156,24 @@ public class PipelineOptions {
"TABLE_ID: Hash by
TableId only. All events from the same table will land on the same subtask,
ensuring per-table ordering semantics. This strategy is not supported for
paimon, fluss, or maxcompute sinks.")))
.build());
+ public static final ConfigOption<DecimalPrecisionMode>
+ PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE =
+ ConfigOptions.key("transform.decimal.precision.mode")
+ .enumType(DecimalPrecisionMode.class)
+ .defaultValue(DecimalPrecisionMode.UP_TO_19)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Maximum precision mode
for DECIMAL type in transform expression evaluation. ")
+ .linebreak()
+ .add(
+ ListElement.list(
+ text(
+ "UP_TO_19:
Limits DECIMAL precision to 19 digits, matching Calcite's default type system. "
+ +
"This is the default behavior for all versions."),
+ text(
+ "UP_TO_38:
Allows DECIMAL precision up to 38 digits, matching Flink CDC's extended type
system.")))
+ .build());
+
private PipelineOptions() {}
}
diff --git
a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java
index 2a4c3e1b3..ee4ff5859 100644
---
a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java
+++
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java
@@ -195,7 +195,7 @@ class InternalObjectConverterTest {
.hasToString("4.2");
assertThat(convertToInternal(new BigDecimal("-3.1415926"),
DataTypes.DECIMAL(20, 10)))
.isInstanceOf(DecimalData.class)
- .hasToString("-3.1415926");
+ .hasToString("-3.1415926000");
assertThat(
convertToInternal(
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
index 71fc865f9..c0d693ed2 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
@@ -197,6 +197,9 @@ public class FlinkPipelineComposer implements
PipelineComposer {
stream,
pipelineDef.getTransforms(),
pipelineDef.getConfig().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
+ pipelineDef
+ .getConfig()
+
.get(PipelineOptions.PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE),
pipelineDef.getUdfs(),
pipelineDef.getModels(),
dataSource.supportedMetadataColumns(),
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
index f24e1a99a..20c8bc54e 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
@@ -24,6 +24,7 @@ import org.apache.flink.cdc.common.factories.FactoryHelper;
import org.apache.flink.cdc.common.model.AiModelClient;
import org.apache.flink.cdc.common.model.AiModelClientFactory;
import org.apache.flink.cdc.common.model.ModelContext;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.composer.definition.ModelDef;
import org.apache.flink.cdc.composer.definition.TransformDef;
@@ -110,6 +111,7 @@ public class TransformTranslator {
DataStream<Event> input,
List<TransformDef> transforms,
String timezone,
+ DecimalPrecisionMode decimalPrecisionMode,
List<UdfDef> udfFunctions,
List<ModelDef> models,
SupportedMetadataColumn[] supportedMetadataColumns,
@@ -134,6 +136,7 @@ public class TransformTranslator {
supportedMetadataColumns);
}
postTransformFunctionBuilder.addTimezone(timezone);
+
postTransformFunctionBuilder.addDecimalPrecisionMode(decimalPrecisionMode);
postTransformFunctionBuilder.addUdfFunctions(
udfFunctions.stream().map(this::udfDefToUDFTuple).collect(Collectors.toList()));
postTransformFunctionBuilder.addUdfFunctions(
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
index 9a774642e..f1b89a94b 100644
---
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
@@ -33,6 +33,7 @@ import org.apache.flink.cdc.common.event.DropColumnEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.RenameColumnEvent;
import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.schema.Column;
@@ -2179,7 +2180,8 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=default_namespace.default_schema.mytable2, before=[4,
Derrida, 25, student, Derrida, 26, extras], after=[], op=DELETE, meta=()}");
}
- String[] runNumericCastingWith(String expression) throws Exception {
+ String[] runNumericCastingWith(DecimalPrecisionMode decimalPrecisionMode,
String expression)
+ throws Exception {
try {
FlinkPipelineComposer composer =
FlinkPipelineComposer.ofMiniCluster();
@@ -2207,6 +2209,9 @@ class FlinkPipelineTransformITCase {
pipelineConfig.set(PipelineOptions.PIPELINE_PARALLELISM, 1);
pipelineConfig.set(
PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR,
SchemaChangeBehavior.EVOLVE);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE,
+ decimalPrecisionMode);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
@@ -2253,9 +2258,10 @@ class FlinkPipelineTransformITCase {
.collect(Collectors.joining(", "));
}
- @Test
- void testNumericCastingsWithTruncation() throws Exception {
- assertThat(runNumericCastingWith("*"))
+ @ParameterizedTest(name = "Decimal mode: {0}")
+ @EnumSource
+ void testNumericCastingsWithTruncation(DecimalPrecisionMode mode) throws
Exception {
+ assertThat(runNumericCastingWith(mode, "*"))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` TINYINT,`small_c`
SMALLINT,`int_c` INT,`bigint_c` BIGINT,`float_c` FLOAT,`double_c`
DOUBLE,`decimal_c` DECIMAL(10, 2),`valid_char_c` VARCHAR(17),`invalid_char_c`
VARCHAR(17)}, primaryKeys=id, options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -6.7, -8.9, -10.11, -12.13, foo], op=INSERT,
meta=()}",
@@ -2263,7 +2269,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 6.7, 8.9, 10.11, 12.13, baz], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("BOOLEAN")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("BOOLEAN")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` BOOLEAN,`small_c` BOOLEAN,`int_c`
BOOLEAN,`bigint_c` BOOLEAN,`float_c` BOOLEAN,`double_c` BOOLEAN,`decimal_c`
BOOLEAN,`valid_char_c` BOOLEAN,`invalid_char_c` BOOLEAN}, primaryKeys=id,
options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, true, true, true, true, true, true, true, false, false], op=INSERT,
meta=()}",
@@ -2271,7 +2277,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, true, true, true, true, true, true, true, false, false], op=INSERT,
meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("TINYINT")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("TINYINT")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` TINYINT,`small_c` TINYINT,`int_c`
TINYINT,`bigint_c` TINYINT,`float_c` TINYINT,`double_c` TINYINT,`decimal_c`
TINYINT,`valid_char_c` TINYINT,`invalid_char_c` TINYINT}, primaryKeys=id,
options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -6, -8, -10, -12, null], op=INSERT, meta=()}",
@@ -2279,7 +2285,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 6, 8, 10, 12, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("SMALLINT")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("SMALLINT")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` SMALLINT,`small_c`
SMALLINT,`int_c` SMALLINT,`bigint_c` SMALLINT,`float_c` SMALLINT,`double_c`
SMALLINT,`decimal_c` SMALLINT,`valid_char_c` SMALLINT,`invalid_char_c`
SMALLINT}, primaryKeys=id, options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -6, -8, -10, -12, null], op=INSERT, meta=()}",
@@ -2287,7 +2293,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 6, 8, 10, 12, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("INT")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("INT")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` INT,`small_c` INT,`int_c`
INT,`bigint_c` INT,`float_c` INT,`double_c` INT,`decimal_c` INT,`valid_char_c`
INT,`invalid_char_c` INT}, primaryKeys=id, options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -6, -8, -10, -12, null], op=INSERT, meta=()}",
@@ -2295,7 +2301,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 6, 8, 10, 12, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("BIGINT")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("BIGINT")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` BIGINT,`small_c` BIGINT,`int_c`
BIGINT,`bigint_c` BIGINT,`float_c` BIGINT,`double_c` BIGINT,`decimal_c`
BIGINT,`valid_char_c` BIGINT,`invalid_char_c` BIGINT}, primaryKeys=id,
options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -6, -8, -10, -12, null], op=INSERT, meta=()}",
@@ -2303,7 +2309,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 6, 8, 10, 12, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("FLOAT")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("FLOAT")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` FLOAT,`small_c` FLOAT,`int_c`
FLOAT,`bigint_c` FLOAT,`float_c` FLOAT,`double_c` FLOAT,`decimal_c`
FLOAT,`valid_char_c` FLOAT,`invalid_char_c` FLOAT}, primaryKeys=id,
options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.0, -3.0, -4.0, -5.0, -6.7, -8.9, -10.11, -12.13, null],
op=INSERT, meta=()}",
@@ -2311,7 +2317,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.0, 3.0, 4.0, 5.0, 6.7, 8.9, 10.11, 12.13, null], op=INSERT,
meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("DOUBLE")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("DOUBLE")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DOUBLE,`small_c` DOUBLE,`int_c`
DOUBLE,`bigint_c` DOUBLE,`float_c` DOUBLE,`double_c` DOUBLE,`decimal_c`
DOUBLE,`valid_char_c` DOUBLE,`invalid_char_c` DOUBLE}, primaryKeys=id,
options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.0, -3.0, -4.0, -5.0, -6.699999809265137, -8.9, -10.11, -12.13,
null], op=INSERT, meta=()}",
@@ -2319,7 +2325,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.0, 3.0, 4.0, 5.0, 6.699999809265137, 8.9, 10.11, 12.13, null],
op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("DECIMAL(1, 0)")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(1,
0)")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(1, 0),`small_c`
DECIMAL(1, 0),`int_c` DECIMAL(1, 0),`bigint_c` DECIMAL(1, 0),`float_c`
DECIMAL(1, 0),`double_c` DECIMAL(1, 0),`decimal_c` DECIMAL(1, 0),`valid_char_c`
DECIMAL(1, 0),`invalid_char_c` DECIMAL(1, 0)}, primaryKeys=id, options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -7, -9, null, null, null], op=INSERT, meta=()}",
@@ -2327,7 +2333,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 7, 9, null, null, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("DECIMAL(2, 0)")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(2,
0)")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(2, 0),`small_c`
DECIMAL(2, 0),`int_c` DECIMAL(2, 0),`bigint_c` DECIMAL(2, 0),`float_c`
DECIMAL(2, 0),`double_c` DECIMAL(2, 0),`decimal_c` DECIMAL(2, 0),`valid_char_c`
DECIMAL(2, 0),`invalid_char_c` DECIMAL(2, 0)}, primaryKeys=id, options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -7, -9, -10, -12, null], op=INSERT, meta=()}",
@@ -2335,7 +2341,7 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 7, 9, 10, 12, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("DECIMAL(3, 1)")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(3,
1)")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(3, 1),`small_c`
DECIMAL(3, 1),`int_c` DECIMAL(3, 1),`bigint_c` DECIMAL(3, 1),`float_c`
DECIMAL(3, 1),`double_c` DECIMAL(3, 1),`decimal_c` DECIMAL(3, 1),`valid_char_c`
DECIMAL(3, 1),`invalid_char_c` DECIMAL(3, 1)}, primaryKeys=id, options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.0, -3.0, -4.0, -5.0, -6.7, -8.9, -10.1, -12.1, null], op=INSERT,
meta=()}",
@@ -2343,13 +2349,67 @@ class FlinkPipelineTransformITCase {
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.0, 3.0, 4.0, 5.0, 6.7, 8.9, 10.1, 12.1, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
- assertThat(runNumericCastingWith(generateCastTo("DECIMAL(19, 10)")))
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(19,
10)")))
.containsExactly(
"CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(19, 10),`small_c`
DECIMAL(19, 10),`int_c` DECIMAL(19, 10),`bigint_c` DECIMAL(19, 10),`float_c`
DECIMAL(19, 10),`double_c` DECIMAL(19, 10),`decimal_c` DECIMAL(19,
10),`valid_char_c` DECIMAL(19, 10),`invalid_char_c` DECIMAL(19, 10)},
primaryKeys=id, options=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.0000000000, -3.0000000000, -4.0000000000, -5.0000000000,
-6.7000000000, -8.9000000000, -10.1100000000, -12.1300000000, null], op=INSERT,
meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[0, 0.0000000000, 0.0000000000, 0.0000000000, 0.0000000000, 0.0000000000,
0.0000000000, 0.0000000000, 0.0000000000, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.0000000000, 3.0000000000, 4.0000000000, 5.0000000000, 6.7000000000,
8.9000000000, 10.1100000000, 12.1300000000, null], op=INSERT, meta=()}",
"DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
+
+ // Test DECIMAL with maximum precision (38)
+
+ if (mode.equals(DecimalPrecisionMode.UP_TO_38)) {
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(38,
0)")))
+ .containsExactly(
+ "CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(38, 0),`small_c`
DECIMAL(38, 0),`int_c` DECIMAL(38, 0),`bigint_c` DECIMAL(38, 0),`float_c`
DECIMAL(38, 0),`double_c` DECIMAL(38, 0),`decimal_c` DECIMAL(38,
0),`valid_char_c` DECIMAL(38, 0),`invalid_char_c` DECIMAL(38, 0)},
primaryKeys=id, options=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -7, -9, -10, -12, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[0, 0, 0, 0, 0, 0, 0, 0, 0, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 7, 9, 10, 12, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
+
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(38,
10)")))
+ .containsExactly(
+ "CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(38, 10),`small_c`
DECIMAL(38, 10),`int_c` DECIMAL(38, 10),`bigint_c` DECIMAL(38, 10),`float_c`
DECIMAL(38, 10),`double_c` DECIMAL(38, 10),`decimal_c` DECIMAL(38,
10),`valid_char_c` DECIMAL(38, 10),`invalid_char_c` DECIMAL(38, 10)},
primaryKeys=id, options=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.0000000000, -3.0000000000, -4.0000000000, -5.0000000000,
-6.7000000000, -8.9000000000, -10.1100000000, -12.1300000000, null], op=INSERT,
meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[0, 0.0000000000, 0.0000000000, 0.0000000000, 0.0000000000, 0.0000000000,
0.0000000000, 0.0000000000, 0.0000000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.0000000000, 3.0000000000, 4.0000000000, 5.0000000000, 6.7000000000,
8.9000000000, 10.1100000000, 12.1300000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
+
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(38,
18)")))
+ .containsExactly(
+ "CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(38, 18),`small_c`
DECIMAL(38, 18),`int_c` DECIMAL(38, 18),`bigint_c` DECIMAL(38, 18),`float_c`
DECIMAL(38, 18),`double_c` DECIMAL(38, 18),`decimal_c` DECIMAL(38,
18),`valid_char_c` DECIMAL(38, 18),`invalid_char_c` DECIMAL(38, 18)},
primaryKeys=id, options=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.000000000000000000, -3.000000000000000000, -4.000000000000000000,
-5.000000000000000000, -6.700000000000000000, -8.900000000000000000,
-10.110000000000000000, -12.130000000000000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[0, 0.000000000000000000, 0.000000000000000000, 0.000000000000000000,
0.000000000000000000, 0.000000000000000000, 0.000000000000000000,
0.000000000000000000, 0.000000000000000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.000000000000000000, 3.000000000000000000, 4.000000000000000000,
5.000000000000000000, 6.700000000000000000, 8.900000000000000000,
10.110000000000000000, 12.130000000000000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
+ } else if (mode.equals(DecimalPrecisionMode.UP_TO_19)) {
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(38,
0)")))
+ .containsExactly(
+ "CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(19, 0),`small_c`
DECIMAL(19, 0),`int_c` DECIMAL(19, 0),`bigint_c` DECIMAL(19, 0),`float_c`
DECIMAL(19, 0),`double_c` DECIMAL(19, 0),`decimal_c` DECIMAL(19,
0),`valid_char_c` DECIMAL(19, 0),`invalid_char_c` DECIMAL(19, 0)},
primaryKeys=id, options=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2, -3, -4, -5, -7, -9, -10, -12, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[0, 0, 0, 0, 0, 0, 0, 0, 0, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2, 3, 4, 5, 7, 9, 10, 12, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
+
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(38,
10)")))
+ .containsExactly(
+ "CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(19, 10),`small_c`
DECIMAL(19, 10),`int_c` DECIMAL(19, 10),`bigint_c` DECIMAL(19, 10),`float_c`
DECIMAL(19, 10),`double_c` DECIMAL(19, 10),`decimal_c` DECIMAL(19,
10),`valid_char_c` DECIMAL(19, 10),`invalid_char_c` DECIMAL(19, 10)},
primaryKeys=id, options=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.0000000000, -3.0000000000, -4.0000000000, -5.0000000000,
-6.7000000000, -8.9000000000, -10.1100000000, -12.1300000000, null], op=INSERT,
meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[0, 0.0000000000, 0.0000000000, 0.0000000000, 0.0000000000, 0.0000000000,
0.0000000000, 0.0000000000, 0.0000000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.0000000000, 3.0000000000, 4.0000000000, 5.0000000000, 6.7000000000,
8.9000000000, 10.1100000000, 12.1300000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
+
+ assertThat(runNumericCastingWith(mode, generateCastTo("DECIMAL(38,
18)")))
+ .containsExactly(
+ "CreateTableEvent{tableId=ns.scm.tbl,
schema=columns={`id` BIGINT NOT NULL,`tiny_c` DECIMAL(19, 18),`small_c`
DECIMAL(19, 18),`int_c` DECIMAL(19, 18),`bigint_c` DECIMAL(19, 18),`float_c`
DECIMAL(19, 18),`double_c` DECIMAL(19, 18),`decimal_c` DECIMAL(19,
18),`valid_char_c` DECIMAL(19, 18),`invalid_char_c` DECIMAL(19, 18)},
primaryKeys=id, options=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[-1, -2.000000000000000000, -3.000000000000000000, -4.000000000000000000,
-5.000000000000000000, -6.700000000000000000, -8.900000000000000000, null,
null, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[0, 0.000000000000000000, 0.000000000000000000, 0.000000000000000000,
0.000000000000000000, 0.000000000000000000, 0.000000000000000000,
0.000000000000000000, 0.000000000000000000, null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[1, 2.000000000000000000, 3.000000000000000000, 4.000000000000000000,
5.000000000000000000, 6.700000000000000000, 8.900000000000000000, null, null,
null], op=INSERT, meta=()}",
+ "DataChangeEvent{tableId=ns.scm.tbl, before=[],
after=[2, null, null, null, null, null, null, null, null, null], op=INSERT,
meta=()}");
+ } else {
+ Assertions.fail("Unexpected decimal precision mode: " + mode);
+ }
}
@Test
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java
index 0d8079e71..337df690b 100644
---
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/specs/TransformSpecsITCase.java
@@ -32,6 +32,7 @@ import org.apache.flink.cdc.common.event.CreateTableEvent;
import org.apache.flink.cdc.common.event.DataChangeEvent;
import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
import org.apache.flink.cdc.common.schema.Schema;
@@ -342,7 +343,6 @@ class TransformSpecsITCase {
private static final ObjectMapper mapper = new ObjectMapper(new
YAMLFactory());
- @SuppressWarnings("unchecked")
private static Stream<TestSpec> loadTestSpec(Path specPath) {
List<TestSpec> specs = new ArrayList<>();
try {
@@ -356,6 +356,11 @@ class TransformSpecsITCase {
if (specNode.has("time-zone")) {
spec.timeZone = asTextOrNull(specNode.get("time-zone"));
}
+ if (specNode.has("decimal-precision-mode")) {
+ spec.decimalPrecisionMode =
+ DecimalPrecisionMode.valueOf(
+
specNode.get("decimal-precision-mode").asText().toUpperCase());
+ }
if (specNode.has("projection")) {
spec.projectionRules =
List.of(
@@ -399,6 +404,7 @@ class TransformSpecsITCase {
public String name;
public String ignore;
public String timeZone = "UTC";
+ public DecimalPrecisionMode decimalPrecisionMode =
DecimalPrecisionMode.UP_TO_19;
public List<String> projectionRules = new ArrayList<>();
public @Nullable String filterRule;
public @Nullable String primaryKey;
@@ -474,6 +480,9 @@ class TransformSpecsITCase {
pipelineConfig.set(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE,
spec.timeZone);
pipelineConfig.set(
PipelineOptions.PIPELINE_SCHEMA_CHANGE_BEHAVIOR,
SchemaChangeBehavior.EVOLVE);
+ pipelineConfig.set(
+ PipelineOptions.PIPELINE_TRANSFORM_DECIMAL_PRECISION_MODE,
+ spec.decimalPrecisionMode);
PipelineDef pipelineDef =
new PipelineDef(
sourceDef,
diff --git a/flink-cdc-composer/src/test/resources/specs/casting.yaml
b/flink-cdc-composer/src/test/resources/specs/casting.yaml
index 288b8df16..817f52b8d 100644
--- a/flink-cdc-composer/src/test/resources/specs/casting.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/casting.yaml
@@ -220,6 +220,126 @@
DataChangeEvent{tableId=foo.bar.baz, before=[-1, null, 0.00000, 1.00000,
2.22000, 333.00000, 44.44000, 5555555.00000, null], after=[], op=DELETE,
meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, 0.00000,
1.00000, 2.22000, 333.00000, 44.44000, 5555555.00000, null], op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[0, null, 0.00000, 1.00000,
2.22000, 333.00000, 44.44000, 5555555.00000, null], after=[], op=DELETE,
meta=()}
+- do: Cast To Decimal(38, 0) (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
+ projection: |-
+ id_
+ CAST(null AS DECIMAL(38, 0)) AS comp_0
+ CAST(0 AS DECIMAL(38, 0)) AS comp_1
+ CAST(1 AS DECIMAL(38, 0)) AS comp_2
+ CAST('2.22' AS DECIMAL(38, 0)) AS comp_3
+ CAST('333' AS DECIMAL(38, 0)) AS comp_4
+ CAST('44.44' AS DECIMAL(38, 0)) AS comp_5
+ CAST('5555555' AS DECIMAL(38, 0)) AS comp_6
+ CAST('FOOBAR' AS DECIMAL(38, 0)) AS comp_7
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_0` DECIMAL(19, 0),`comp_1` DECIMAL(19, 0),`comp_2`
DECIMAL(19, 0),`comp_3` DECIMAL(19, 0),`comp_4` DECIMAL(19, 0),`comp_5`
DECIMAL(19, 0),`comp_6` DECIMAL(19, 0),`comp_7` DECIMAL(19, 0)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, null, 0, 1, 2,
333, 44, 5555555, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, null, 0, 1, 2, 333, 44,
5555555, null], after=[-1, null, 0, 1, 2, 333, 44, 5555555, null], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, null, 0, 1, 2, 333, 44,
5555555, null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, 0, 1, 2,
333, 44, 5555555, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, 0, 1, 2, 333, 44,
5555555, null], after=[], op=DELETE, meta=()}
+- do: Cast To Decimal(38, 0) (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ CAST(null AS DECIMAL(38, 0)) AS comp_0
+ CAST(0 AS DECIMAL(38, 0)) AS comp_1
+ CAST(1 AS DECIMAL(38, 0)) AS comp_2
+ CAST('2.22' AS DECIMAL(38, 0)) AS comp_3
+ CAST('333' AS DECIMAL(38, 0)) AS comp_4
+ CAST('44.44' AS DECIMAL(38, 0)) AS comp_5
+ CAST('5555555' AS DECIMAL(38, 0)) AS comp_6
+ CAST('FOOBAR' AS DECIMAL(38, 0)) AS comp_7
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_0` DECIMAL(38, 0),`comp_1` DECIMAL(38, 0),`comp_2`
DECIMAL(38, 0),`comp_3` DECIMAL(38, 0),`comp_4` DECIMAL(38, 0),`comp_5`
DECIMAL(38, 0),`comp_6` DECIMAL(38, 0),`comp_7` DECIMAL(38, 0)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, null, 0, 1, 2,
333, 44, 5555555, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, null, 0, 1, 2, 333, 44,
5555555, null], after=[-1, null, 0, 1, 2, 333, 44, 5555555, null], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, null, 0, 1, 2, 333, 44,
5555555, null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, 0, 1, 2,
333, 44, 5555555, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, 0, 1, 2, 333, 44,
5555555, null], after=[], op=DELETE, meta=()}
+- do: Cast To Decimal(38, 10) (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
+ projection: |-
+ id_
+ CAST(null AS DECIMAL(38, 10)) AS comp_0
+ CAST(0 AS DECIMAL(38, 10)) AS comp_1
+ CAST(1 AS DECIMAL(38, 10)) AS comp_2
+ CAST('2.22' AS DECIMAL(38, 10)) AS comp_3
+ CAST('333' AS DECIMAL(38, 10)) AS comp_4
+ CAST('44.44' AS DECIMAL(38, 10)) AS comp_5
+ CAST('5555555' AS DECIMAL(38, 10)) AS comp_6
+ CAST('FOOBAR' AS DECIMAL(38, 10)) AS comp_7
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_0` DECIMAL(19, 10),`comp_1` DECIMAL(19, 10),`comp_2`
DECIMAL(19, 10),`comp_3` DECIMAL(19, 10),`comp_4` DECIMAL(19, 10),`comp_5`
DECIMAL(19, 10),`comp_6` DECIMAL(19, 10),`comp_7` DECIMAL(19, 10)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, null,
0.0000000000, 1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000,
5555555.0000000000, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, null, 0.0000000000,
1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000, 5555555.0000000000,
null], after=[-1, null, 0.0000000000, 1.0000000000, 2.2200000000,
333.0000000000, 44.4400000000, 5555555.0000000000, null], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, null, 0.0000000000,
1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000, 5555555.0000000000,
null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null,
0.0000000000, 1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000,
5555555.0000000000, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, 0.0000000000,
1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000, 5555555.0000000000,
null], after=[], op=DELETE, meta=()}
+- do: Cast To Decimal(38, 10) (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ CAST(null AS DECIMAL(38, 10)) AS comp_0
+ CAST(0 AS DECIMAL(38, 10)) AS comp_1
+ CAST(1 AS DECIMAL(38, 10)) AS comp_2
+ CAST('2.22' AS DECIMAL(38, 10)) AS comp_3
+ CAST('333' AS DECIMAL(38, 10)) AS comp_4
+ CAST('44.44' AS DECIMAL(38, 10)) AS comp_5
+ CAST('5555555' AS DECIMAL(38, 10)) AS comp_6
+ CAST('FOOBAR' AS DECIMAL(38, 10)) AS comp_7
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_0` DECIMAL(38, 10),`comp_1` DECIMAL(38, 10),`comp_2`
DECIMAL(38, 10),`comp_3` DECIMAL(38, 10),`comp_4` DECIMAL(38, 10),`comp_5`
DECIMAL(38, 10),`comp_6` DECIMAL(38, 10),`comp_7` DECIMAL(38, 10)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, null,
0.0000000000, 1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000,
5555555.0000000000, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, null, 0.0000000000,
1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000, 5555555.0000000000,
null], after=[-1, null, 0.0000000000, 1.0000000000, 2.2200000000,
333.0000000000, 44.4400000000, 5555555.0000000000, null], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, null, 0.0000000000,
1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000, 5555555.0000000000,
null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null,
0.0000000000, 1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000,
5555555.0000000000, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, 0.0000000000,
1.0000000000, 2.2200000000, 333.0000000000, 44.4400000000, 5555555.0000000000,
null], after=[], op=DELETE, meta=()}
+- do: Cast To Decimal(38, 18) (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
+ projection: |-
+ id_
+ CAST(null AS DECIMAL(38, 18)) AS comp_0
+ CAST(0 AS DECIMAL(38, 18)) AS comp_1
+ CAST(1 AS DECIMAL(38, 18)) AS comp_2
+ CAST('2.22' AS DECIMAL(38, 18)) AS comp_3
+ CAST('333' AS DECIMAL(38, 18)) AS comp_4
+ CAST('44.44' AS DECIMAL(38, 18)) AS comp_5
+ CAST('5555555' AS DECIMAL(38, 18)) AS comp_6
+ CAST('FOOBAR' AS DECIMAL(38, 18)) AS comp_7
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_0` DECIMAL(19, 18),`comp_1` DECIMAL(19, 18),`comp_2`
DECIMAL(19, 18),`comp_3` DECIMAL(19, 18),`comp_4` DECIMAL(19, 18),`comp_5`
DECIMAL(19, 18),`comp_6` DECIMAL(19, 18),`comp_7` DECIMAL(19, 18)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000, null, null,
null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000, null, null,
null, null], after=[-1, null, 0.000000000000000000, 1.000000000000000000,
2.220000000000000000, null, null, null, null], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000, null, null,
null, null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000, null, null,
null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000, null, null,
null, null], after=[], op=DELETE, meta=()}
+- do: Cast To Decimal(38, 18) (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ CAST(null AS DECIMAL(38, 18)) AS comp_0
+ CAST(0 AS DECIMAL(38, 18)) AS comp_1
+ CAST(1 AS DECIMAL(38, 18)) AS comp_2
+ CAST('2.22' AS DECIMAL(38, 18)) AS comp_3
+ CAST('333' AS DECIMAL(38, 18)) AS comp_4
+ CAST('44.44' AS DECIMAL(38, 18)) AS comp_5
+ CAST('5555555' AS DECIMAL(38, 18)) AS comp_6
+ CAST('FOOBAR' AS DECIMAL(38, 18)) AS comp_7
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_0` DECIMAL(38, 18),`comp_1` DECIMAL(38, 18),`comp_2`
DECIMAL(38, 18),`comp_3` DECIMAL(38, 18),`comp_4` DECIMAL(38, 18),`comp_5`
DECIMAL(38, 18),`comp_6` DECIMAL(38, 18),`comp_7` DECIMAL(38, 18)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000,
333.000000000000000000, 44.440000000000000000, 5555555.000000000000000000,
null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000,
333.000000000000000000, 44.440000000000000000, 5555555.000000000000000000,
null], after=[-1, null, 0.000000000000000000, 1.000000000000000000,
2.220000000000000000, 333.000000000000000000, 44.440000000000000000,
5555555.000000000000000000, null], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000,
333.000000000000000000, 44.440000000000000000, 5555555.000000000000000000,
null], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000,
333.000000000000000000, 44.440000000000000000, 5555555.000000000000000000,
null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null,
0.000000000000000000, 1.000000000000000000, 2.220000000000000000,
333.000000000000000000, 44.440000000000000000, 5555555.000000000000000000,
null], after=[], op=DELETE, meta=()}
- do: Cast To Timestamp (UTC)
projection: |-
id_
diff --git a/flink-cdc-composer/src/test/resources/specs/decimal.yaml
b/flink-cdc-composer/src/test/resources/specs/decimal.yaml
index f1418dd32..7e60cba37 100644
--- a/flink-cdc-composer/src/test/resources/specs/decimal.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/decimal.yaml
@@ -13,7 +13,8 @@
# limitations under the License.
################################################################################
-- do: Add Op
+- do: Add Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
projection: |-
id_
decimal_10_0_ + CAST(1 AS DECIMAL(1, 0)) AS comp_1
@@ -25,7 +26,38 @@
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567891,
null], op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567891, null],
after=[-1, -9876543209, null], op=UPDATE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543209, null],
after=[], op=DELETE, meta=()}
-- do: Subtract Op
+
+- do: Add Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ decimal_10_0_ + CAST(1 AS DECIMAL(1, 0)) AS comp_1
+ decimal_20_2_ + CAST(1 AS DECIMAL(1, 0)) AS comp_2
+ non-null: 'true'
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(11, 0),`comp_2` DECIMAL(21, 2)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567891,
123456789012345679.90], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567891,
123456789012345679.90], after=[-1, -9876543209, -987654321098765431.10],
op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543209,
-987654321098765431.10], after=[], op=DELETE, meta=()}
+
+- do: Nested Expressions (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ IFNULL(decimal_20_2_ + CAST(1 AS DECIMAL(1, 0)), CAST(0 AS DECIMAL(1, 0)))
AS ifnull_decimal
+ NULLIF(decimal_20_2_ + CAST(1 AS DECIMAL(1, 0)), CAST(0 AS DECIMAL(1, 0)))
AS nullif_decimal
+ ARRAY[decimal_20_2_ + CAST(1 AS DECIMAL(1, 0)), CAST(0 AS DECIMAL(1, 0))]
AS decimal_array
+ non-null: 'true'
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`ifnull_decimal` DECIMAL(21, 2) NOT NULL,`nullif_decimal`
DECIMAL(21, 2),`decimal_array` ARRAY<DECIMAL(21, 2)>}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1,
123456789012345679.90, 123456789012345679.90, [123456789012345679.90, 0.00]],
op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 123456789012345679.90,
123456789012345679.90, [123456789012345679.90, 0.00]], after=[-1,
-987654321098765431.10, -987654321098765431.10, [-987654321098765431.10,
0.00]], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -987654321098765431.10,
-987654321098765431.10, [-987654321098765431.10, 0.00]], after=[], op=DELETE,
meta=()}
+
+- do: Subtract Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
projection: |-
id_
decimal_10_0_ - CAST(1 AS DECIMAL(1, 0)) AS comp_1
@@ -37,31 +69,83 @@
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567889,
null], op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567889, null],
after=[-1, -9876543211, null], op=UPDATE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543211, null],
after=[], op=DELETE, meta=()}
-- do: Multiply Op
+
+- do: Subtract Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ decimal_10_0_ - CAST(1 AS DECIMAL(1, 0)) AS comp_1
+ decimal_20_2_ - CAST(1 AS DECIMAL(1, 0)) AS comp_2
+ primary-key: id_
+ non-null: 'true'
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(11, 0),`comp_2` DECIMAL(21, 2)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567889,
123456789012345677.90], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567889,
123456789012345677.90], after=[-1, -9876543211, -987654321098765433.10],
op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543211,
-987654321098765433.10], after=[], op=DELETE, meta=()}
+
+- do: Multiply Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
projection: |-
id_
+ decimal_10_0_, decimal_20_2_
decimal_10_0_ * CAST(2 AS DECIMAL(1, 0)) AS comp_1
decimal_20_2_ * CAST(2 AS DECIMAL(1, 0)) AS comp_2
primary-key: id_
non-null: 'true'
expect: |-
- CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(11, 0),`comp_2` DECIMAL(19, 2)},
primaryKeys=id_, options=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 2469135780,
null], op=INSERT, meta=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[1, 2469135780, null],
after=[-1, -19753086420, null], op=UPDATE, meta=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[-1, -19753086420, null],
after=[], op=DELETE, meta=()}
-- do: Divide Op
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`decimal_10_0_` DECIMAL(10, 0),`decimal_20_2_` DECIMAL(20,
2),`comp_1` DECIMAL(11, 0),`comp_2` DECIMAL(19, 2)}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678.90, 2469135780, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678.90, 2469135780, null], after=[-1, -9876543210,
-987654321098765432.10, -19753086420, null], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432.10, -19753086420, null], after=[], op=DELETE, meta=()}
+
+- do: Multiply Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ decimal_10_0_, decimal_20_2_
+ decimal_10_0_ * CAST(2 AS DECIMAL(1, 0)) AS comp_1
+ decimal_20_2_ * CAST(2 AS DECIMAL(1, 0)) AS comp_2
+ primary-key: id_
+ non-null: 'true'
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`decimal_10_0_` DECIMAL(10, 0),`decimal_20_2_` DECIMAL(20,
2),`comp_1` DECIMAL(11, 0),`comp_2` DECIMAL(21, 2)}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678.90, 2469135780, 246913578024691357.80], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678.90, 2469135780, 246913578024691357.80], after=[-1,
-9876543210, -987654321098765432.10, -19753086420, -1975308642197530864.20],
op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432.10, -19753086420, -1975308642197530864.20], after=[],
op=DELETE, meta=()}
+
+- do: Divide Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
projection: |-
id_
+ decimal_10_0_, decimal_20_2_
decimal_10_0_ / CAST(2 AS DECIMAL(1, 0)) AS comp_1
decimal_20_2_ / CAST(2 AS DECIMAL(1, 0)) AS comp_2
primary-key: id_
non-null: 'true'
expect: |-
- CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(16, 6),`comp_2` DECIMAL(19, 2)},
primaryKeys=id_, options=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 617.283945,
61728394506172839.45], op=INSERT, meta=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[1, 617.283945,
61728394506172839.45], after=[-1, -4938.271605, null], op=UPDATE, meta=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[-1, -4938.271605, null],
after=[], op=DELETE, meta=()}
-- do: Abs Op
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`decimal_10_0_` DECIMAL(10, 0),`decimal_20_2_` DECIMAL(20,
2),`comp_1` DECIMAL(16, 6),`comp_2` DECIMAL(19, 2)}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678.90, 617283945.000000, 61728394506172839.45], op=INSERT,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678.90, 617283945.000000, 61728394506172839.45], after=[-1,
-9876543210, -987654321098765432.10, -4938271605.000000, null], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432.10, -4938271605.000000, null], after=[], op=DELETE, meta=()}
+
+- do: Divide Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ decimal_10_0_, decimal_20_2_
+ decimal_10_0_ / CAST(2 AS DECIMAL(1, 0)) AS comp_1
+ decimal_20_2_ / CAST(2 AS DECIMAL(1, 0)) AS comp_2
+ primary-key: id_
+ non-null: 'true'
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`decimal_10_0_` DECIMAL(10, 0),`decimal_20_2_` DECIMAL(20,
2),`comp_1` DECIMAL(16, 6),`comp_2` DECIMAL(24, 6)}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678.90, 617283945.000000, 61728394506172839.450000], op=INSERT,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678.90, 617283945.000000, 61728394506172839.450000], after=[-1,
-9876543210, -987654321098765432.10, -4938271605.000000,
-493827160549382716.050000], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432.10, -4938271605.000000, -493827160549382716.050000],
after=[], op=DELETE, meta=()}
+
+- do: Abs Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
projection: |-
id_
ABS(decimal_10_0_) AS comp_1
@@ -74,7 +158,24 @@
DataChangeEvent{tableId=foo.bar.baz, before=[-1, 9876543210, null],
after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null],
op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[],
op=DELETE, meta=()}
-- do: Ceil Op
+
+- do: Abs Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ ABS(decimal_10_0_) AS comp_1
+ ABS(decimal_20_2_) AS comp_2
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(10, 0),`comp_2` DECIMAL(20, 2)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678.90], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678.90], after=[-1, 9876543210, 987654321098765432.10],
op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, 9876543210,
987654321098765432.10], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null],
op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[],
op=DELETE, meta=()}
+
+- do: Ceil Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
projection: |-
id_
CEIL(decimal_10_0_) AS comp_1
@@ -87,7 +188,24 @@
DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null],
op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[],
op=DELETE, meta=()}
-- do: Floor Op
+
+- do: Ceil Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ CEIL(decimal_10_0_) AS comp_1
+ CEIL(decimal_20_2_) AS comp_2
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(10, 0),`comp_2` DECIMAL(20, 0)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345679], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345679], after=[-1, -9876543210, -987654321098765432], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null],
op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[],
op=DELETE, meta=()}
+
+- do: Floor Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
projection: |-
id_
FLOOR(decimal_10_0_) AS comp_1
@@ -100,16 +218,50 @@
DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765433], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null],
op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[],
op=DELETE, meta=()}
-- do: Round Op
+
+- do: Floor Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
projection: |-
id_
- ROUND(decimal_10_0_, 1) AS comp_1
- ROUND(decimal_20_2_, 1) AS comp_2
+ FLOOR(decimal_10_0_) AS comp_1
+ FLOOR(decimal_20_2_) AS comp_2
primary-key: id_
expect: |-
- CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(10, 0),`comp_2` DECIMAL(19, 1)},
primaryKeys=id_, options=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 12345678900,
123456789012345678.9], op=INSERT, meta=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[1, 12345678900,
123456789012345678.9], after=[-1, -98765432100, -987654321098765432.1],
op=UPDATE, meta=()}
- DataChangeEvent{tableId=foo.bar.baz, before=[-1, -98765432100,
-987654321098765432.1], after=[], op=DELETE, meta=()}
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`comp_1` DECIMAL(10, 0),`comp_2` DECIMAL(20, 0)},
primaryKeys=id_, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678], after=[-1, -9876543210, -987654321098765433], op=UPDATE,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765433], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null],
op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null], after=[],
op=DELETE, meta=()}
+
+- do: Round Op (Decimal-19 mode)
+ decimal-precision-mode: up_to_19
+ projection: |-
+ id_
+ decimal_10_0_, decimal_20_2_
+ ROUND(decimal_10_0_, 1) AS comp_1
+ ROUND(decimal_20_2_, 1) AS comp_2
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`decimal_10_0_` DECIMAL(10, 0),`decimal_20_2_` DECIMAL(20,
2),`comp_1` DECIMAL(10, 0),`comp_2` DECIMAL(19, 1)}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678.90, 1234567890, 123456789012345678.9], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678.90, 1234567890, 123456789012345678.9], after=[-1,
-9876543210, -987654321098765432.10, -9876543210, -987654321098765432.1],
op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432.10, -9876543210, -987654321098765432.1], after=[],
op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null,
null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null],
after=[], op=DELETE, meta=()}
+
+- do: Round Op (Decimal-38 mode)
+ decimal-precision-mode: up_to_38
+ projection: |-
+ id_
+ decimal_10_0_, decimal_20_2_
+ ROUND(decimal_10_0_, 1) AS comp_1
+ ROUND(decimal_20_2_, 1) AS comp_2
+ primary-key: id_
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`decimal_10_0_` DECIMAL(10, 0),`decimal_20_2_` DECIMAL(20,
2),`comp_1` DECIMAL(10, 0),`comp_2` DECIMAL(20, 1)}, primaryKeys=id_,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 1234567890,
123456789012345678.90, 1234567890, 123456789012345678.9], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 1234567890,
123456789012345678.90, 1234567890, 123456789012345678.9], after=[-1,
-9876543210, -987654321098765432.10, -9876543210, -987654321098765432.1],
op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, -9876543210,
-987654321098765432.10, -9876543210, -987654321098765432.1], after=[],
op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null,
null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null],
after=[], op=DELETE, meta=()}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
index 9c6552fbe..36deef753 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperator.java
@@ -30,6 +30,7 @@ import org.apache.flink.cdc.common.event.Event;
import org.apache.flink.cdc.common.event.SchemaChangeEvent;
import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.schema.Selectors;
import org.apache.flink.cdc.common.udf.UserDefinedFunctionContext;
@@ -77,6 +78,7 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
private static final Logger LOG =
LoggerFactory.getLogger(PostTransformOperator.class);
private final String timezone;
+ private final DecimalPrecisionMode decimalPrecisionMode;
private final List<TransformRule> transformRules;
private final Map<TableId, Boolean> hasAsteriskMap;
private final Map<TableId, List<String>> projectedColumnsMap;
@@ -107,9 +109,11 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
PostTransformOperator(
List<TransformRule> transformRules,
String timezone,
+ DecimalPrecisionMode decimalPrecisionMode,
List<Tuple3<String, String, Map<String, String>>> udfFunctions,
Map<String, AiModelClient> modelClients) {
this.timezone = timezone;
+ this.decimalPrecisionMode = decimalPrecisionMode;
this.transformRules = transformRules;
this.hasAsteriskMap = new HashMap<>();
this.projectedColumnsMap = new HashMap<>();
@@ -383,7 +387,8 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
.orElse(null),
preSchema.getColumns(),
udfDescriptors,
- transformer.getSupportedMetadataColumns());
+ transformer.getSupportedMetadataColumns(),
+ decimalPrecisionMode);
return preSchema.copy(
projectionColumns.stream()
.map(ProjectionColumn::getColumn)
@@ -460,6 +465,7 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
.map(TransformProjection::getProjection)
.orElse(null),
timezone,
+ decimalPrecisionMode,
udfDescriptors,
udfFunctionInstances,
postTransformer.getSupportedMetadataColumns(),
@@ -476,7 +482,10 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
TableId tableId, PostTransformer postTransformer) {
if (!filterProcessors.contains(tableId, postTransformer)) {
if (!postTransformer.getFilter().isPresent()) {
- filterProcessors.put(tableId, postTransformer,
TransformFilterProcessor.ofNoOp());
+ filterProcessors.put(
+ tableId,
+ postTransformer,
+ TransformFilterProcessor.ofNoOp(decimalPrecisionMode));
} else {
PostTransformChangeInfo changeInfo =
postTransformInfoMap.get(tableId);
filterProcessors.put(
@@ -486,6 +495,7 @@ public class PostTransformOperator extends
AbstractStreamOperatorAdapter<Event>
changeInfo,
postTransformer.getFilter().orElse(null),
timezone,
+ decimalPrecisionMode,
udfDescriptors,
udfFunctionInstances,
postTransformer.getSupportedMetadataColumns(),
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
index be2e8823b..a5de56c75 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorBuilder.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.pipeline.PipelineOptions;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
@@ -34,6 +35,7 @@ import java.util.Map;
public class PostTransformOperatorBuilder {
private final List<TransformRule> transformRules = new ArrayList<>();
private String timezone;
+ private DecimalPrecisionMode decimalPrecisionMode =
DecimalPrecisionMode.UP_TO_19;
private final List<Tuple3<String, String, Map<String, String>>>
udfFunctions =
new ArrayList<>();
private final Map<String, AiModelClient> modelClients = new
LinkedHashMap<>();
@@ -108,6 +110,12 @@ public class PostTransformOperatorBuilder {
return this;
}
+ public PostTransformOperatorBuilder addDecimalPrecisionMode(
+ DecimalPrecisionMode decimalPrecisionMode) {
+ this.decimalPrecisionMode = decimalPrecisionMode;
+ return this;
+ }
+
public PostTransformOperatorBuilder addUdfFunctions(
List<Tuple3<String, String, Map<String, String>>> udfFunctions) {
this.udfFunctions.addAll(udfFunctions);
@@ -120,6 +128,7 @@ public class PostTransformOperatorBuilder {
}
public PostTransformOperator build() {
- return new PostTransformOperator(transformRules, timezone,
udfFunctions, modelClients);
+ return new PostTransformOperator(
+ transformRules, timezone, decimalPrecisionMode, udfFunctions,
modelClients);
}
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
index 6781ea191..02b2c5171 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
@@ -20,6 +20,7 @@ package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cdc.common.converter.JavaClassConverter;
import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.runtime.parser.JaninoCompiler;
@@ -46,6 +47,7 @@ public class TransformFilterProcessor {
private final PostTransformChangeInfo tableInfo;
private final TransformFilter transformFilter;
private final String timezone;
+ private final DecimalPrecisionMode decimalPrecisionMode;
private final List<Object> udfFunctionInstances;
private final Map<String, SupportedMetadataColumn>
supportedMetadataColumns;
private final Map<String, AiModelClient> modelClients;
@@ -58,6 +60,7 @@ public class TransformFilterProcessor {
PostTransformChangeInfo tableInfo,
TransformFilter transformFilter,
String timezone,
+ DecimalPrecisionMode decimalPrecisionMode,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
Map<String, SupportedMetadataColumn> supportedMetadataColumns,
@@ -66,6 +69,7 @@ public class TransformFilterProcessor {
this.tableInfo = tableInfo;
this.transformFilter = transformFilter;
this.timezone = timezone;
+ this.decimalPrecisionMode = decimalPrecisionMode;
this.udfFunctionInstances = udfFunctionInstances;
this.supportedMetadataColumns = supportedMetadataColumns;
this.modelClients = modelClients;
@@ -87,14 +91,16 @@ public class TransformFilterProcessor {
}
}
- public static TransformFilterProcessor ofNoOp() {
- return new TransformFilterProcessor(true, null, null, null, null,
null, null, null);
+ public static TransformFilterProcessor ofNoOp(DecimalPrecisionMode
decimalPrecisionMode) {
+ return new TransformFilterProcessor(
+ true, null, null, null, decimalPrecisionMode, null, null,
null, null);
}
public static TransformFilterProcessor of(
PostTransformChangeInfo tableInfo,
TransformFilter transformFilter,
String timezone,
+ DecimalPrecisionMode decimalPrecisionMode,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
SupportedMetadataColumn[] supportedMetadataColumns,
@@ -109,6 +115,7 @@ public class TransformFilterProcessor {
tableInfo,
transformFilter,
timezone,
+ decimalPrecisionMode,
udfDescriptors,
udfFunctionInstances,
supportedMetadataColumnsMap,
@@ -238,7 +245,8 @@ public class TransformFilterProcessor {
columns,
udfDescriptors,
supportedMetadataColumns,
- transformFilter.getColumnNameMap());
+ transformFilter.getColumnNameMap(),
+ decimalPrecisionMode);
return TransformExpressionKey.of(
transformFilter.getExpression(),
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
index 09fe5fb19..84b4d3a9a 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformProjectionProcessor.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.utils.Preconditions;
import org.apache.flink.cdc.runtime.parser.TransformParser;
@@ -48,6 +49,7 @@ public class TransformProjectionProcessor {
private final PostTransformChangeInfo changeInfo;
private final String projectionExpression;
private final String timezone;
+ private final DecimalPrecisionMode decimalPrecisionMode;
private final List<UserDefinedFunctionDescriptor> udfDescriptors;
private final List<Object> udfFunctionInstances;
private final List<ProjectionColumnProcessor> columnProcessors;
@@ -59,6 +61,7 @@ public class TransformProjectionProcessor {
PostTransformChangeInfo changeInfo,
String projectionExpression,
String timezone,
+ DecimalPrecisionMode decimalPrecisionMode,
List<UserDefinedFunctionDescriptor> udfDescriptors,
List<Object> udfFunctionInstances,
SupportedMetadataColumn[] supportedMetadataColumns,
@@ -66,6 +69,7 @@ public class TransformProjectionProcessor {
this.changeInfo = changeInfo;
this.projectionExpression = projectionExpression;
this.timezone = timezone;
+ this.decimalPrecisionMode = decimalPrecisionMode;
this.udfDescriptors = udfDescriptors;
this.udfFunctionInstances = udfFunctionInstances;
this.supportedMetadataColumns = supportedMetadataColumns;
@@ -97,7 +101,8 @@ public class TransformProjectionProcessor {
projectionExpression,
changeInfo.getPreTransformedSchema().getColumns(),
udfDescriptors,
- supportedMetadataColumns);
+ supportedMetadataColumns,
+ decimalPrecisionMode);
List<ProjectionColumnProcessor> columnProcessors =
projectionColumns.stream()
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/FlinkCdcTypeSystem.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/FlinkCdcTypeSystem.java
new file mode 100644
index 000000000..8bc0f8aae
--- /dev/null
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/FlinkCdcTypeSystem.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package org.apache.flink.cdc.runtime.parser;
+
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
+
+import org.apache.calcite.rel.type.RelDataTypeSystemImpl;
+
+/** A customized version of {@link
org.apache.calcite.rel.type.RelDataTypeSystem}. */
+public class FlinkCdcTypeSystem extends RelDataTypeSystemImpl {
+
+ public static final FlinkCdcTypeSystem UP_TO_38 = new
FlinkCdcTypeSystem(38);
+ public static final FlinkCdcTypeSystem UP_TO_19 = new
FlinkCdcTypeSystem(19);
+
+ private final int maxPrecision;
+
+ private FlinkCdcTypeSystem(int maxPrecision) {
+ this.maxPrecision = maxPrecision;
+ }
+
+ 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);
+ }
+ }
+
+ @Override
+ public int getMaxNumericPrecision() {
+ return maxPrecision;
+ }
+
+ @Override
+ public int getMaxNumericScale() {
+ return maxPrecision;
+ }
+}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
index 4caef2768..d03664a35 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
@@ -21,6 +21,7 @@ import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.io.ParseException;
import org.apache.flink.cdc.common.annotation.VisibleForTesting;
import org.apache.flink.cdc.common.converter.JavaClassConverter;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.types.DataType;
@@ -502,7 +503,8 @@ public class JaninoCompiler {
context.columns,
sqlBasicCall,
context.udfDescriptors,
- context.supportedMetadataColumns);
+ context.supportedMetadataColumns,
+ context.decimalPrecisionMode);
return generateCollectionConstructorOperation(
context, sqlBasicCall, functionName, resultType);
}
@@ -550,7 +552,8 @@ public class JaninoCompiler {
context.columns,
operand,
context.udfDescriptors,
- context.supportedMetadataColumns);
+ context.supportedMetadataColumns,
+ context.decimalPrecisionMode);
atoms[i] = generateImplicitTypeConvertMethod(operandType,
targetType, atoms[i]);
}
}
@@ -771,7 +774,8 @@ public class JaninoCompiler {
context.columns,
sqlBasicCall,
context.udfDescriptors,
- context.supportedMetadataColumns);
+ context.supportedMetadataColumns,
+ context.decimalPrecisionMode);
if (resultType.is(DataTypeRoot.DECIMAL)) {
return new Java.MethodInvocation(Location.NOWHERE, null,
handler, atoms);
}
@@ -950,7 +954,8 @@ public class JaninoCompiler {
context.columns,
sqlBasicCall,
context.udfDescriptors,
- context.supportedMetadataColumns);
+ context.supportedMetadataColumns,
+ context.decimalPrecisionMode);
// Get the Java class for the result type and add a cast
// Use getCanonicalName() to correctly handle array types (e.g.,
byte[] instead of "[B")
@@ -1035,7 +1040,8 @@ public class JaninoCompiler {
context.columns,
sqlBasicCall,
context.udfDescriptors,
- context.supportedMetadataColumns);
+ context.supportedMetadataColumns,
+ context.decimalPrecisionMode);
Java.Rvalue[] coercedAtoms = new Java.Rvalue[atoms.length];
for (int index = 0; index < atoms.length; index++) {
coercedAtoms[index] = generateNumericTypeConvertMethod(resultType,
atoms[index]);
@@ -1058,7 +1064,8 @@ public class JaninoCompiler {
context.columns,
value,
context.udfDescriptors,
- context.supportedMetadataColumns);
+ context.supportedMetadataColumns,
+ context.decimalPrecisionMode);
return castToJavaType(resultType, operation);
}
@@ -1273,15 +1280,20 @@ public class JaninoCompiler {
// Readable metadata columns
public final SupportedMetadataColumn[] supportedMetadataColumns;
+ // Maximum precision mode for DECIMAL type evaluation
+ public final DecimalPrecisionMode decimalPrecisionMode;
+
private Context(
List<Column> columns,
Map<String, String> columnNameMap,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ DecimalPrecisionMode decimalPrecisionMode) {
this.columns = columns;
this.columnNameMap = columnNameMap;
this.udfDescriptors = udfDescriptors;
this.supportedMetadataColumns = supportedMetadataColumns;
+ this.decimalPrecisionMode = decimalPrecisionMode;
}
public static Context of(
@@ -1289,7 +1301,26 @@ public class JaninoCompiler {
Map<String, String> columnNameMap,
List<UserDefinedFunctionDescriptor> udfDescriptors,
SupportedMetadataColumn[] supportedMetadataColumns) {
- return new Context(columns, columnNameMap, udfDescriptors,
supportedMetadataColumns);
+ return of(
+ columns,
+ columnNameMap,
+ udfDescriptors,
+ supportedMetadataColumns,
+ DecimalPrecisionMode.UP_TO_19);
+ }
+
+ public static Context of(
+ List<Column> columns,
+ Map<String, String> columnNameMap,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ DecimalPrecisionMode decimalPrecisionMode) {
+ return new Context(
+ columns,
+ columnNameMap,
+ udfDescriptors,
+ supportedMetadataColumns,
+ decimalPrecisionMode);
}
}
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
index 5fcf1b608..1020a3970 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.runtime.parser;
import org.apache.flink.api.common.io.ParseException;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.common.types.DataType;
@@ -41,7 +42,6 @@ import org.apache.calcite.rel.RelRoot;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
-import org.apache.calcite.rel.type.RelDataTypeSystem;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.schema.ScalarFunction;
import org.apache.calcite.schema.SchemaPlus;
@@ -158,7 +158,8 @@ public class TransformParser {
List<Column> columns,
SqlNode sqlNode,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ DecimalPrecisionMode decimalPrecisionMode) {
List<Column> columnsWithMetadata =
copyFillMetadataColumn(columns, supportedMetadataColumns);
CalciteSchema rootSchema = CalciteSchema.createRootSchema(true);
@@ -202,7 +203,8 @@ public class TransformParser {
throw new RuntimeException("Failed to resolve UDF: " + udf, e);
}
}
- SqlTypeFactoryImpl factory = new
SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+ SqlTypeFactoryImpl factory =
+ new
SqlTypeFactoryImpl(FlinkCdcTypeSystem.of(decimalPrecisionMode));
CalciteCatalogReader calciteCatalogReader =
new CalciteCatalogReader(
rootSchema,
@@ -328,6 +330,20 @@ public class TransformParser {
List<Column> columns,
List<UserDefinedFunctionDescriptor> udfDescriptors,
SupportedMetadataColumn[] supportedMetadataColumns) {
+ return generateProjectionColumns(
+ projectionExpression,
+ columns,
+ udfDescriptors,
+ supportedMetadataColumns,
+ DecimalPrecisionMode.UP_TO_19);
+ }
+
+ public static List<ProjectionColumn> generateProjectionColumns(
+ String projectionExpression,
+ List<Column> columns,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ DecimalPrecisionMode decimalPrecisionMode) {
if (isNullOrWhitespaceOnly(projectionExpression)) {
return new ArrayList<>();
}
@@ -345,7 +361,8 @@ public class TransformParser {
originalColumnMap,
sqlSelect,
udfDescriptors,
- supportedMetadataColumns);
+ supportedMetadataColumns,
+ decimalPrecisionMode);
List<ProjectionColumn> projectionColumns = new ArrayList<>();
Map<String, Integer> addedProjectionColumnNames = new HashMap<>();
@@ -415,7 +432,8 @@ public class TransformParser {
columns,
columnNameMap,
udfDescriptors,
- supportedMetadataColumns),
+ supportedMetadataColumns,
+ decimalPrecisionMode),
exprNode),
originalColumnNames,
columnNameMap);
@@ -455,10 +473,16 @@ public class TransformParser {
Map<String, Column> originalColumnMap,
SqlSelect sqlSelect,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ DecimalPrecisionMode decimalPrecisionMode) {
try {
RelNode relNode =
- sqlToRel(columns, sqlSelect, udfDescriptors,
supportedMetadataColumns);
+ sqlToRel(
+ columns,
+ sqlSelect,
+ udfDescriptors,
+ supportedMetadataColumns,
+ decimalPrecisionMode);
return relNode.getRowType().getFieldList().stream()
.map(RelDataTypeField::getType)
.toArray(RelDataType[]::new);
@@ -467,7 +491,8 @@ public class TransformParser {
// Keep Calcite as the primary type inference path. This
fallback only covers
// transform predicates that Janino can evaluate but Calcite
may fail to convert
// while building projection columns.
- SqlTypeFactoryImpl typeFactory = new
SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+ SqlTypeFactoryImpl typeFactory =
+ new
SqlTypeFactoryImpl(FlinkCdcTypeSystem.of(decimalPrecisionMode));
List<RelDataType> relDataTypes = new ArrayList<>();
for (SqlNode sqlNode : sqlSelect.getSelectList()) {
relDataTypes.add(
@@ -477,7 +502,8 @@ public class TransformParser {
originalColumnMap,
unwrapAsExpression(sqlNode),
udfDescriptors,
- supportedMetadataColumns));
+ supportedMetadataColumns,
+ decimalPrecisionMode));
}
return relDataTypes.toArray(new RelDataType[0]);
} catch (RuntimeException fallbackException) {
@@ -504,7 +530,8 @@ public class TransformParser {
Map<String, Column> originalColumnMap,
SqlNode exprNode,
List<UserDefinedFunctionDescriptor> udfDescriptors,
- SupportedMetadataColumn[] supportedMetadataColumns) {
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ DecimalPrecisionMode decimalPrecisionMode) {
if (exprNode instanceof SqlIdentifier) {
String columnName =
((SqlIdentifier) exprNode)
@@ -522,7 +549,11 @@ public class TransformParser {
return toRelDataType(
typeFactory,
deduceSubExpressionType(
- columns, exprNode, udfDescriptors,
supportedMetadataColumns));
+ columns,
+ exprNode,
+ udfDescriptors,
+ supportedMetadataColumns,
+ decimalPrecisionMode));
}
private static DataType findIdentifierDataType(
@@ -635,6 +666,22 @@ public class TransformParser {
List<UserDefinedFunctionDescriptor> udfDescriptors,
SupportedMetadataColumn[] supportedMetadataColumns,
Map<String, String> columnNameMap) {
+ return translateFilterExpressionToJaninoExpression(
+ filterExpression,
+ columns,
+ udfDescriptors,
+ supportedMetadataColumns,
+ columnNameMap,
+ DecimalPrecisionMode.UP_TO_19);
+ }
+
+ public static String translateFilterExpressionToJaninoExpression(
+ String filterExpression,
+ List<Column> columns,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ Map<String, String> columnNameMap,
+ DecimalPrecisionMode decimalPrecisionMode) {
if (isNullOrWhitespaceOnly(filterExpression)) {
return "";
}
@@ -645,7 +692,11 @@ public class TransformParser {
SqlNode where = sqlSelect.getWhere();
return JaninoCompiler.translateSqlNodeToJaninoExpression(
JaninoCompiler.Context.of(
- columns, columnNameMap, udfDescriptors,
supportedMetadataColumns),
+ columns,
+ columnNameMap,
+ udfDescriptors,
+ supportedMetadataColumns,
+ decimalPrecisionMode),
where);
}
@@ -863,6 +914,20 @@ public class TransformParser {
SqlNode subExpression,
List<UserDefinedFunctionDescriptor> udfDescriptors,
SupportedMetadataColumn[] supportedMetadataColumns) {
+ return deduceSubExpressionType(
+ columns,
+ subExpression,
+ udfDescriptors,
+ supportedMetadataColumns,
+ DecimalPrecisionMode.UP_TO_19);
+ }
+
+ public static DataType deduceSubExpressionType(
+ List<Column> columns,
+ SqlNode subExpression,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ DecimalPrecisionMode decimalPrecisionMode) {
SqlSelect sqlSelect =
new SqlSelect(
SqlParserPos.QUOTED_ZERO,
@@ -877,7 +942,13 @@ public class TransformParser {
null,
null,
null);
- RelNode relNode = sqlToRel(columns, sqlSelect, udfDescriptors,
supportedMetadataColumns);
+ RelNode relNode =
+ sqlToRel(
+ columns,
+ sqlSelect,
+ udfDescriptors,
+ supportedMetadataColumns,
+ decimalPrecisionMode);
RelDataType[] relDataTypes =
relNode.getRowType().getFieldList().stream()
.map(RelDataTypeField::getType)
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
index 621188bbb..16136323f 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.runtime.parser;
import org.apache.flink.api.common.io.ParseException;
+import org.apache.flink.cdc.common.pipeline.DecimalPrecisionMode;
import org.apache.flink.cdc.common.schema.Column;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
@@ -1092,6 +1093,39 @@ class TransformParserTest {
Assertions.assertThat(result).hasToString("[" + String.join(", ",
expected) + "]");
}
+ @Test
+ void testGenerateProjectionColumnsWithDecimalPrecisionMode() {
+ List<Column> columns =
+ Arrays.asList(
+ Column.physicalColumn("id", DataTypes.INT()),
+ Column.physicalColumn("deposit", DataTypes.DECIMAL(20,
2)));
+ String projection =
+ "deposit + CAST(1 AS DECIMAL(1, 0)) AS amount, "
+ + "id IS DISTINCT FROM 1 AS changed";
+
+ List<ProjectionColumn> upTo19 =
+ TransformParser.generateProjectionColumns(
+ projection,
+ columns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0],
+ DecimalPrecisionMode.UP_TO_19);
+ List<ProjectionColumn> upTo38 =
+ TransformParser.generateProjectionColumns(
+ projection,
+ columns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0],
+ DecimalPrecisionMode.UP_TO_38);
+
+ Assertions.assertThat(upTo19)
+ .extracting(ProjectionColumn::getDataType)
+ .containsExactly(DataTypes.DECIMAL(19, 2),
DataTypes.BOOLEAN());
+ Assertions.assertThat(upTo38)
+ .extracting(ProjectionColumn::getDataType)
+ .containsExactly(DataTypes.DECIMAL(21, 2),
DataTypes.BOOLEAN());
+ }
+
@Test
void testGenerateReferencedColumns() {
List<Column> testColumns =
