modify summary string of DataType and fix test failure

whhe · whhe · commit 156bf685ab47 · 2025-03-04T15:40:13.000+08:00
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/ArrayType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/ArrayType.java
@@ -63,7 +63,7 @@ public String asSummaryString() {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT, elementType.asSerializableString());
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/BigIntType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/BigIntType.java
@@ -47,7 +47,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/BinaryType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/BinaryType.java
@@ -94,7 +94,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         if (length == EMPTY_LITERAL_LENGTH) {
             throw new IllegalArgumentException(
                     "Zero-length binary strings have no serializable string representation.");
@@ -104,7 +104,8 @@ public String asSerializableString() {
 
     @Override
     public String asSummaryString() {
-        return withNullability(FORMAT, length);
+        return withNullability(FORMAT, length)
+                + (getRawDataType() == null ? "" : " " + getRawDataType());
     }
 
     @Override
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/BooleanType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/BooleanType.java
@@ -44,7 +44,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/CharType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/CharType.java
@@ -93,7 +93,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         if (length == EMPTY_LITERAL_LENGTH) {
             throw new IllegalArgumentException(
                     "Zero-length character strings have no serializable string representation.");
@@ -103,7 +103,8 @@ public String asSerializableString() {
 
     @Override
     public String asSummaryString() {
-        return withNullability(FORMAT, length);
+        return withNullability(FORMAT, length)
+                + (getRawDataType() == null ? "" : " " + getRawDataType());
     }
 
     @Override
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DataType.java
@@ -132,12 +132,21 @@ public final DataType copy(boolean isNullable, RawDataType rawDataType) {
     }
 
     /**
-     * Returns a string that fully serializes this instance. The serialized string can be used for
-     * transmitting or persisting a type.
+     * Returns a string that fully serializes this instance except for the raw type.
      *
      * @return detailed string for transmission or persistence
      */
-    public abstract String asSerializableString();
+    protected abstract String asSerializableString();
+
+    /**
+     * Returns a string that fully serializes this instance, including the raw type. The serialized
+     * string can be used for transmitting or persisting a type.
+     *
+     * @return detailed string for transmission or persistence
+     */
+    public String asSerializableStringWithRawDataType() {
+        return asSerializableString() + (rawDataType == null ? "" : " " + rawDataType);
+    }
 
     /**
      * Returns a string that summarizes this type for printing to a console. An implementation might
@@ -148,7 +157,7 @@ public final DataType copy(boolean isNullable, RawDataType rawDataType) {
      * @return summary string of this type for debugging purposes
      */
     public String asSummaryString() {
-        return asSerializableString();
+        return asSerializableStringWithRawDataType();
     }
 
     public abstract List<DataType> getChildren();
@@ -157,7 +166,7 @@ public String asSummaryString() {
 
     @Override
     public String toString() {
-        return asSummaryString();
+        return asSummaryString() + (rawDataType == null ? "" : " " + rawDataType);
     }
 
     @Override
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DateType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DateType.java
@@ -50,7 +50,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DecimalType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DecimalType.java
@@ -89,7 +89,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT, precision, scale);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DoubleType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/DoubleType.java
@@ -44,7 +44,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/FloatType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/FloatType.java
@@ -47,7 +47,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/IntType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/IntType.java
@@ -44,7 +44,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/JdbcRawDataType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/JdbcRawDataType.java
@@ -75,4 +75,19 @@ public boolean equals(Object obj) {
     public int hashCode() {
         return Objects.hash(jdbcType, typeName, length, scale);
     }
+
+    @Override
+    public String toString() {
+        return "JdbcRawDataType{"
+                + "jdbcType="
+                + jdbcType
+                + ", typeName='"
+                + typeName
+                + '\''
+                + ", length="
+                + length
+                + ", scale="
+                + scale
+                + '}';
+    }
 }
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/LocalZonedTimestampType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/LocalZonedTimestampType.java
@@ -87,13 +87,14 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT, precision);
     }
 
     @Override
     public String asSummaryString() {
-        return withNullability(SUMMARY_FORMAT, precision);
+        return withNullability(SUMMARY_FORMAT, precision)
+                + (getRawDataType() == null ? "" : " " + getRawDataType());
     }
 
     @Override
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/MapType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/MapType.java
@@ -71,7 +71,7 @@ public String asSummaryString() {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(
                 FORMAT, keyType.asSerializableString(), valueType.asSerializableString());
     }
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/RowType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/RowType.java
@@ -101,7 +101,7 @@ public String asSummaryString() {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(
                 FORMAT,
                 fields.stream()
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/SmallIntType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/SmallIntType.java
@@ -44,7 +44,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/TimeType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/TimeType.java
@@ -77,7 +77,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT, precision);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/TimestampType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/TimestampType.java
@@ -80,15 +80,10 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT, precision);
     }
 
-    @Override
-    public String asSummaryString() {
-        return asSerializableString();
-    }
-
     @Override
     public List<DataType> getChildren() {
         return Collections.emptyList();
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/TinyIntType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/TinyIntType.java
@@ -44,7 +44,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT);
     }
 
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/VarBinaryType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/VarBinaryType.java
@@ -70,7 +70,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         if (length == MAX_LENGTH) {
             return withNullability(MAX_FORMAT);
         }
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/VarCharType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/VarCharType.java
@@ -74,7 +74,7 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         if (length == MAX_LENGTH) {
             return withNullability(MAX_FORMAT);
         }
diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/ZonedTimestampType.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/types/ZonedTimestampType.java
@@ -91,15 +91,10 @@ protected DataType copy(boolean isNullable) {
     }
 
     @Override
-    public String asSerializableString() {
+    protected String asSerializableString() {
         return withNullability(FORMAT, precision);
     }
 
-    @Override
-    public String asSummaryString() {
-        return asSerializableString();
-    }
-
     @Override
     public List<DataType> getChildren() {
         return Collections.emptyList();
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlFullTypesITCase.java
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlMetadataAccessorITCase.java
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlOnLineSchemaMigrationITCase.java
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MysqlPipelineNewlyAddedTableITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MysqlPipelineNewlyAddedTableITCase.java
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/testutils/MySqlTypeUtils.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/testutils/MySqlTypeUtils.java

Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ public String asSummaryString() {`
`63`	`63`	`}`
`64`	`64`
`65`	`65`	`@Override`
`66`		`- public String asSerializableString() {`
	`66`	`+ protected String asSerializableString() {`
`67`	`67`	`return withNullability(FORMAT, elementType.asSerializableString());`
`68`	`68`	`}`
`69`	`69`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ protected DataType copy(boolean isNullable) {`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`@Override`
`50`		`- public String asSerializableString() {`
	`50`	`+ protected String asSerializableString() {`
`51`	`51`	`return withNullability(FORMAT);`
`52`	`52`	`}`
`53`	`53`
Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ protected DataType copy(boolean isNullable) {`
`44`	`44`	`}`
`45`	`45`
`46`	`46`	`@Override`
`47`		`- public String asSerializableString() {`
	`47`	`+ protected String asSerializableString() {`
`48`	`48`	`return withNullability(FORMAT);`
`49`	`49`	`}`
`50`	`50`
Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ protected DataType copy(boolean isNullable) {`
`50`	`50`	`}`
`51`	`51`
`52`	`52`	`@Override`
`53`		`- public String asSerializableString() {`
	`53`	`+ protected String asSerializableString() {`
`54`	`54`	`return withNullability(FORMAT);`
`55`	`55`	`}`
`56`	`56`
Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,7 @@ protected DataType copy(boolean isNullable) {`
`89`	`89`	`}`
`90`	`90`
`91`	`91`	`@Override`
`92`		`- public String asSerializableString() {`
	`92`	`+ protected String asSerializableString() {`
`93`	`93`	`return withNullability(FORMAT, precision, scale);`
`94`	`94`	`}`
`95`	`95`