[FLINK-39418] Fix decimal truncation when precision / scale is over 19

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.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 @@ static StringData convertToStringData(Object obj) {
                 "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.");

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/converter/InternalObjectConverter.java

@@ -87,7 +87,7 @@ public Function<Object, byte[]> visit(VarBinaryType varBinaryType) {
 
         @Override
         public Function<Object, DecimalData> visit(DecimalType decimalType) {
-            return CommonConverter::convertToDecimalData;
+            return obj -> CommonConverter.convertToDecimalData(obj, decimalType);
         }
 
         @Override

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
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/pipeline/PipelineOptions.java

@@ -136,5 +136,24 @@ public class PipelineOptions {
                     .withDescription(
                             "The timeout time for SchemaOperator to wait downstream SchemaChangeEvent applying finished, the default value is 3 minutes.");
 
+    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() {}
 }

flink-cdc-common/src/test/java/org/apache/flink/cdc/common/converter/InternalObjectConverterTest.java

@@ -195,7 +195,7 @@ void testConvertToDecimal() {
                 .hasToString("4.2");
         assertThat(convertToInternal(new BigDecimal("-3.1415926"), DataTypes.DECIMAL(20, 10)))
                 .isInstanceOf(DecimalData.class)
-                .hasToString("-3.1415926");
+                .hasToString("-3.1415926000");
 
         assertThat(
                         convertToInternal(

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java

@@ -195,6 +195,9 @@ private void translate(StreamExecutionEnvironment env, PipelineDef pipelineDef)
                         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(),

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.cdc.common.event.Event;
+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;
@@ -94,6 +95,7 @@ public DataStream<Event> translatePostTransform(
             DataStream<Event> input,
             List<TransformDef> transforms,
             String timezone,
+            DecimalPrecisionMode decimalPrecisionMode,
             List<UdfDef> udfFunctions,
             List<ModelDef> models,
             SupportedMetadataColumn[] supportedMetadataColumns,
@@ -117,6 +119,7 @@ public DataStream<Event> translatePostTransform(
                     supportedMetadataColumns);
         }
         postTransformFunctionBuilder.addTimezone(timezone);
+        postTransformFunctionBuilder.addDecimalPrecisionMode(decimalPrecisionMode);
         postTransformFunctionBuilder.addUdfFunctions(
                 udfFunctions.stream().map(this::udfDefToUDFTuple).collect(Collectors.toList()));
         postTransformFunctionBuilder.addUdfFunctions(