From dd865bf440b13b34fde40d4644bad7e517fa70f7 Mon Sep 17 00:00:00 2001
From: Shawn Huang <hx0698@gmail.com>
Date: Thu, 9 Jan 2025 14:18:25 +0800
Subject: [PATCH] [FLINK-36970][cdc-common] Merge result of data type BIGINT
 and DOUBLE should be DOUBLE instead of STRING

This closes #3821
---
 .../flink/cdc/common/utils/SchemaMergingUtils.java     |  4 +++-
 .../flink/cdc/common/utils/SchemaMergingUtilsTest.java | 10 +++++-----
 2 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
index 173aeb42d..dd93d5ca3 100644
--- a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
+++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/SchemaMergingUtils.java
@@ -810,7 +810,9 @@ public class SchemaMergingUtils {
         mergingTree.put(DoubleType.class, ImmutableList.of(doubleType, stringType));
         mergingTree.put(FloatType.class, ImmutableList.of(floatType, doubleType, stringType));
         mergingTree.put(DecimalType.class, ImmutableList.of(stringType));
-        mergingTree.put(BigIntType.class, ImmutableList.of(bigIntType, decimalType, stringType));
+        mergingTree.put(
+                BigIntType.class,
+                ImmutableList.of(bigIntType, decimalType, doubleType, stringType));
         mergingTree.put(
                 IntType.class,
                 ImmutableList.of(intType, bigIntType, decimalType, doubleType, stringType));
diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaMergingUtilsTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaMergingUtilsTest.java
index 5710206f0..c5021d593 100644
--- a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaMergingUtilsTest.java
+++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/utils/SchemaMergingUtilsTest.java
@@ -975,7 +975,7 @@ class SchemaMergingUtilsTest {
                         DECIMAL, FLOAT, DOUBLE, STRING, STRING, STRING, STRING, STRING, STRING,
                         STRING));
 
-        // 16-bit TINYINT could fit into FLOAT (24 sig bits) or DOUBLE (53 sig bits)
+        // 16-bit SMALLINT could fit into FLOAT (24 sig bits) or DOUBLE (53 sig bits)
         assertTypeMergingVector(
                 SMALLINT,
                 Arrays.asList(
@@ -983,7 +983,7 @@ class SchemaMergingUtilsTest {
                         DECIMAL, FLOAT, DOUBLE, STRING, STRING, STRING, STRING, STRING, STRING,
                         STRING));
 
-        // 32-bit TINYINT could fit into DOUBLE (53 sig bits)
+        // 32-bit INT could fit into DOUBLE (53 sig bits)
         assertTypeMergingVector(
                 INT,
                 Arrays.asList(
@@ -994,7 +994,7 @@ class SchemaMergingUtilsTest {
                 BIGINT,
                 Arrays.asList(
                         STRING, STRING, STRING, STRING, STRING, BIGINT, BIGINT, BIGINT, BIGINT,
-                        DECIMAL, STRING, STRING, STRING, STRING, STRING, STRING, STRING, STRING,
+                        DECIMAL, DOUBLE, DOUBLE, STRING, STRING, STRING, STRING, STRING, STRING,
                         STRING));
 
         assertTypeMergingVector(
@@ -1007,14 +1007,14 @@ class SchemaMergingUtilsTest {
         assertTypeMergingVector(
                 FLOAT,
                 Arrays.asList(
-                        STRING, STRING, STRING, STRING, STRING, FLOAT, FLOAT, DOUBLE, STRING,
+                        STRING, STRING, STRING, STRING, STRING, FLOAT, FLOAT, DOUBLE, DOUBLE,
                         STRING, FLOAT, DOUBLE, STRING, STRING, STRING, STRING, STRING, STRING,
                         STRING));
 
         assertTypeMergingVector(
                 DOUBLE,
                 Arrays.asList(
-                        STRING, STRING, STRING, STRING, STRING, DOUBLE, DOUBLE, DOUBLE, STRING,
+                        STRING, STRING, STRING, STRING, STRING, DOUBLE, DOUBLE, DOUBLE, DOUBLE,
                         STRING, DOUBLE, DOUBLE, STRING, STRING, STRING, STRING, STRING, STRING,
                         STRING));