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 a629dfa9f [FLINK-39832] Fix Oracle NUMBER(p,0) with p>=19 incorrectly
mapped to BIGINT (#4424)
a629dfa9f is described below
commit a629dfa9f43f9004beb6af43780a9e25b927adf7
Author: mike <[email protected]>
AuthorDate: Tue Jun 30 20:42:21 2026 +0800
[FLINK-39832] Fix Oracle NUMBER(p,0) with p>=19 incorrectly mapped to
BIGINT (#4424)
---
.../source/OracleSchemaDataTypeInference.java | 25 ++
.../connectors/oracle/utils/OracleTypeUtils.java | 56 +++-
.../oracle/source/OracleFullTypesITCase.java | 44 ++--
.../source/OracleMetadataAccessorITCase.java | 14 +-
.../oracle/source/OraclePipelineITCase.java | 34 ++-
.../oracle/utils/OracleTypeUtilsTest.java | 282 +++++++++++++++++++++
.../flink/cdc/pipeline/tests/OracleE2eITCase.java | 88 +++----
7 files changed, 453 insertions(+), 90 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSchemaDataTypeInference.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSchemaDataTypeInference.java
index e8bfeee66..df0ec5f0c 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSchemaDataTypeInference.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/source/OracleSchemaDataTypeInference.java
@@ -20,10 +20,13 @@ package org.apache.flink.cdc.connectors.oracle.source;
import org.apache.flink.cdc.common.annotation.Internal;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.DecimalType;
import org.apache.flink.cdc.debezium.event.DebeziumSchemaDataTypeInference;
+import io.debezium.data.VariableScaleDecimal;
import io.debezium.data.geometry.Geometry;
import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Struct;
/** {@link DataType} inference for oracle debezium {@link Schema}. */
@Internal
@@ -31,11 +34,33 @@ public class OracleSchemaDataTypeInference extends
DebeziumSchemaDataTypeInferen
private static final long serialVersionUID = 1L;
+ @Override
protected DataType inferStruct(Object value, Schema schema) {
// the Geometry datatype in oracle will be converted to
// a String with Json format
if (Geometry.LOGICAL_NAME.equals(schema.name())) {
return DataTypes.STRING();
+ } else if (VariableScaleDecimal.LOGICAL_NAME.equals(schema.name())) {
+ // For Oracle bare NUMBER and explicit-precision positive-scale
+ // NUMBER with p - s > 18, Debezium encodes the value as a
+ // VariableScaleDecimal struct. We return DECIMAL(38, 19) to
+ // match OracleTypeUtils' schema-level choice (16-byte
+ // BinaryRecordData layout is the same for any DECIMAL).
+ //
+ // For negative-scale NUMBER with (p - s) > 18 (e.g. NUMBER(36,
-2)),
+ // Debezium also encodes as VariableScaleDecimal but the integer
+ // range exceeds BIGINT. OracleTypeUtils returns STRING for these
+ // to avoid LONG overflow. We must mirror that STRING choice here
+ // at runtime, otherwise the schema/runtime type mismatch would
+ // corrupt the BinaryRecordData layout.
+ if (value instanceof Struct) {
+ Struct struct = (Struct) value;
+ Integer dbzScale =
struct.getInt32(VariableScaleDecimal.SCALE_FIELD);
+ if (dbzScale != null && dbzScale < 0) {
+ return DataTypes.STRING();
+ }
+ }
+ return DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 19);
} else {
return super.inferStruct(value, schema);
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/utils/OracleTypeUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/utils/OracleTypeUtils.java
index f54c14af5..6a334e207 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/utils/OracleTypeUtils.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/main/java/org/apache/flink/cdc/connectors/oracle/utils/OracleTypeUtils.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.connectors.oracle.utils;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.DecimalType;
import io.debezium.relational.Column;
import oracle.jdbc.OracleTypes;
@@ -72,11 +73,56 @@ public class OracleTypeUtils {
return DataTypes.DOUBLE();
case Types.NUMERIC:
case Types.DECIMAL:
- return column.length() == 0
- || !column.scale().isPresent()
- || column.scale().get() <= 0
- ? DataTypes.BIGINT()
- : DataTypes.DECIMAL(column.length(),
column.scale().orElse(0));
+ {
+ // Bare NUMBER (scale unspecified): floating-point
semantic,
+ // can store both integers and decimals up to 38
significant
+ // digits. Use DECIMAL(38, 19) as a balanced universal
+ // numeric type, matching Debezium's VariableScaleDecimal
+ // encoding and OracleSchemaDataTypeInference.inferStruct.
+ if (!column.scale().isPresent()) {
+ return DataTypes.DECIMAL(DecimalType.MAX_PRECISION,
19);
+ }
+ int precision = column.length();
+ // precision == 0 (e.g. NUMBER(*, s)) means unspecified.
+ int p = precision > 0 ? precision :
DecimalType.MAX_PRECISION;
+ int scale = column.scale().get();
+
+ // scale < 0: value = unscaled * 10^|scale|. Debezium
+ // encodes these as INT8 / INT16 / INT32 / INT64 based on
+ // (precision - scale), which is the number of integer
+ // digits. The default Debezium runtime
+ // (DebeziumSchemaDataTypeInference) maps those to
+ // TINYINT / SMALLINT / INT / BIGINT — all of which live
+ // in the same 8-byte compact slot of BinaryRecordData's
+ // fixed area. So we mirror that family here to keep the
+ // schema and runtime layers in sync. When (precision -
+ // scale) exceeds 18 the value can no longer fit in a
+ // BIGINT and Debezium falls back to VariableScaleDecimal;
+ // we map that case to STRING because the integer range
+ // is genuinely beyond any native Flink CDC type.
+ if (scale < 0) {
+ int intDigits = p - scale; // p + |scale|
+ if (intDigits <= 18) {
+ return DataTypes.BIGINT();
+ }
+ return DataTypes.STRING();
+ }
+ // scale > 36: not safely representable as DECIMAL.
+ if (scale > 36) {
+ return DataTypes.STRING();
+ }
+ if (scale == 0) {
+ // Explicit integer: BIGINT for p <= 18 (Debezium
encodes
+ // as INT8/16/32/64, parent class runtime returns
+ // TINYINT/SMALLINT/INT/BIGINT — same 8-byte-or-less
+ // family), DECIMAL(p, 0) for p > 18 (Debezium encodes
+ // as VariableScaleDecimal, runtime returns DECIMAL,
+ // BinaryRecordData layout matches 16-byte DECIMAL).
+ return p <= 18 ? DataTypes.BIGINT() :
DataTypes.DECIMAL(p, 0);
+ }
+ // 1 <= scale <= 36: standard decimal with fractional part.
+ return DataTypes.DECIMAL(p, scale);
+ }
case Types.DATE:
return DataTypes.DATE();
case Types.TIMESTAMP:
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleFullTypesITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleFullTypesITCase.java
index 4e3ddfbc6..e8b234c24 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleFullTypesITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleFullTypesITCase.java
@@ -113,7 +113,7 @@ public class OracleFullTypesITCase extends
OracleSourceTestBase {
Object[] expectedSnapshot =
new Object[] {
- DecimalData.fromBigDecimal(new BigDecimal("1"), 9, 0),
+ 1L,
BinaryStringData.fromString("vc2"),
BinaryStringData.fromString("vc2"),
BinaryStringData.fromString("nvc2"),
@@ -128,17 +128,17 @@ public class OracleFullTypesITCase extends
OracleSourceTestBase {
(float) 6.66,
DecimalData.fromBigDecimal(new BigDecimal("1234.567891"),
10, 6),
DecimalData.fromBigDecimal(new BigDecimal("1234.567891"),
10, 6),
- DecimalData.fromBigDecimal(new BigDecimal("77.323"), 10,
3),
- DecimalData.fromBigDecimal(new BigDecimal(1), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(22), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(333), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(4444), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(5555), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(1), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(99), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(9999), 10, 0),
- DecimalData.fromBigDecimal(new BigDecimal(999999999), 38,
0),
- DecimalData.fromBigDecimal(new
BigDecimal(999999999999999999L), 38, 0),
+ DecimalData.fromBigDecimal(new BigDecimal("77.323"), 38,
19),
+ DecimalData.fromBigDecimal(new BigDecimal(1), 38, 19),
+ DecimalData.fromBigDecimal(new BigDecimal(22), 38, 19),
+ DecimalData.fromBigDecimal(new BigDecimal(333), 38, 19),
+ DecimalData.fromBigDecimal(new BigDecimal(4444), 38, 0),
+ DecimalData.fromBigDecimal(new BigDecimal(5555), 38, 0),
+ 1L,
+ 99L,
+ 9999L,
+ 999999999L,
+ 999999999999999999L,
90L,
9900L,
999999990L,
@@ -222,7 +222,7 @@ public class OracleFullTypesITCase extends
OracleSourceTestBase {
private static final RowType COMMON_TYPES =
RowType.of(
- DataTypes.DECIMAL(9, 0).notNull(),
+ DataTypes.BIGINT().notNull(),
DataTypes.STRING(),
DataTypes.STRING(),
DataTypes.STRING(),
@@ -237,17 +237,17 @@ public class OracleFullTypesITCase extends
OracleSourceTestBase {
DataTypes.FLOAT(),
DataTypes.DECIMAL(10, 6),
DataTypes.DECIMAL(10, 6),
- DataTypes.DECIMAL(10, 3),
- DataTypes.DECIMAL(10, 0),
- DataTypes.DECIMAL(10, 0),
- DataTypes.DECIMAL(10, 0),
+ DataTypes.DECIMAL(38, 19),
+ DataTypes.DECIMAL(38, 19),
+ DataTypes.DECIMAL(38, 19),
+ DataTypes.DECIMAL(38, 19),
DataTypes.DECIMAL(38, 0),
DataTypes.DECIMAL(38, 0),
- DataTypes.DECIMAL(1, 0),
- DataTypes.DECIMAL(2, 0),
- DataTypes.DECIMAL(4, 0),
- DataTypes.DECIMAL(9, 0),
- DataTypes.DECIMAL(18, 0),
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
DataTypes.BIGINT(),
DataTypes.BIGINT(),
DataTypes.BIGINT(),
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleMetadataAccessorITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleMetadataAccessorITCase.java
index 2ade43084..8ccfea167 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleMetadataAccessorITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OracleMetadataAccessorITCase.java
@@ -126,6 +126,12 @@ public class OracleMetadataAccessorITCase extends
OracleSourceTestBase {
DataTypes.FLOAT(),
DataTypes.DECIMAL(10, 6),
DataTypes.DECIMAL(10, 6),
+ DataTypes.DECIMAL(38, 19),
+ DataTypes.DECIMAL(38, 19),
+ DataTypes.DECIMAL(38, 19),
+ DataTypes.DECIMAL(38, 19),
+ DataTypes.DECIMAL(38, 0),
+ DataTypes.DECIMAL(38, 0),
DataTypes.BIGINT(),
DataTypes.BIGINT(),
DataTypes.BIGINT(),
@@ -135,13 +141,7 @@ public class OracleMetadataAccessorITCase extends
OracleSourceTestBase {
DataTypes.BIGINT(),
DataTypes.BIGINT(),
DataTypes.BIGINT(),
- DataTypes.BIGINT(),
- DataTypes.BIGINT(),
- DataTypes.BIGINT(),
- DataTypes.BIGINT(),
- DataTypes.BIGINT(),
- DataTypes.BIGINT(),
- DataTypes.BIGINT(),
+ DataTypes.STRING(),
DataTypes.TIMESTAMP(6),
DataTypes.TIMESTAMP(6),
DataTypes.TIMESTAMP(2),
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java
index 497fa54ea..700e87395 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/source/OraclePipelineITCase.java
@@ -20,6 +20,7 @@ package org.apache.flink.cdc.connectors.oracle.source;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.DecimalData;
import org.apache.flink.cdc.common.data.RecordData;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
import org.apache.flink.cdc.common.event.AddColumnEvent;
@@ -38,6 +39,7 @@ import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.common.source.FlinkSourceProvider;
import org.apache.flink.cdc.common.types.DataType;
import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.DecimalType;
import org.apache.flink.cdc.common.types.RowType;
import org.apache.flink.cdc.common.utils.SchemaUtils;
import org.apache.flink.cdc.connectors.base.options.StartupOptions;
@@ -71,6 +73,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.testcontainers.lifecycle.Startables;
+import java.math.BigDecimal;
import java.nio.file.Files;
import java.nio.file.Path;
import java.sql.Connection;
@@ -400,7 +403,10 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
Collections.singletonList(
new AddColumnEvent.ColumnWithPosition(
- Column.physicalColumn("COLS27",
DataTypes.BIGINT())))));
+ Column.physicalColumn(
+ "COLS27",
+ DataTypes.DECIMAL(
+
DecimalType.MAX_PRECISION, 19))))));
statement.execute(
String.format("ALTER TABLE %s.PRODUCTS ADD COLS28
TIMESTAMP(2)", "DEBEZIUM"));
expected.add(
@@ -983,7 +989,9 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
new CreateTableEvent(
TableId.tableId("DEBEZIUM", "MYLAKE"),
Schema.newBuilder()
- .physicalColumn("FEATURE_ID",
DataTypes.BIGINT().notNull())
+ .physicalColumn(
+ "FEATURE_ID",
+
DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 19).notNull())
.physicalColumn("NAME", DataTypes.VARCHAR(32))
.physicalColumn("SHAPE", DataTypes.STRING())
.primaryKey(Arrays.asList("FEATURE_ID"))
@@ -992,7 +1000,9 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
RowType rowType =
RowType.of(
new DataType[] {
- DataTypes.BIGINT(), DataTypes.VARCHAR(32),
DataTypes.STRING()
+ DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 19),
+ DataTypes.VARCHAR(32),
+ DataTypes.STRING()
},
new String[] {"FEATURE_ID", "NAME", "SHAPE"});
BinaryRecordDataGenerator generator = new
BinaryRecordDataGenerator(rowType);
@@ -1004,7 +1014,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 1L,
+ DecimalData.fromBigDecimal(new
BigDecimal(1), 38, 19),
BinaryStringData.fromString("center"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[116.6,24.343]]\",\"type\":\"Point\",\"srid\":4326}")
@@ -1014,7 +1024,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 2L,
+ DecimalData.fromBigDecimal(new
BigDecimal(2), 38, 19),
BinaryStringData.fromString("two-dimensional point"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[10.0,5.0]]\",\"type\":\"Point\",\"srid\":4326}")
@@ -1024,7 +1034,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 3L,
+ DecimalData.fromBigDecimal(new
BigDecimal(3), 38, 19),
BinaryStringData.fromString("straight line
segment"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[10.0,10.0],[20.0,20.0],[30.0,30.0]]\",\"type\":\"LineString\",\"srid\":4326}")
@@ -1034,7 +1044,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 4L,
+ DecimalData.fromBigDecimal(new
BigDecimal(4), 38, 19),
BinaryStringData.fromString("polyline"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[5.0,5.0],[15.0,10.0],[25.0,5.0],[35.0,10.0]]\",\"type\":\"LineString\",\"srid\":4326}")
@@ -1044,7 +1054,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 5L,
+ DecimalData.fromBigDecimal(new
BigDecimal(5), 38, 19),
BinaryStringData.fromString("rectangle"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[0.0,0.0],[10.0,0.0],[10.0,10.0],[0.0,10.0],[0.0,0.0]]\",\"type\":\"Polygon\",\"srid\":4326}")
@@ -1054,7 +1064,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 6L,
+ DecimalData.fromBigDecimal(new
BigDecimal(6), 38, 19),
BinaryStringData.fromString("Multi-Point"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[10.0,10.0],[20.0,20.0],[30.0,30.0]]\",\"type\":\"MultiPoint\",\"srid\":4326}")
@@ -1064,7 +1074,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 7L,
+ DecimalData.fromBigDecimal(new
BigDecimal(7), 38, 19),
BinaryStringData.fromString("Multi line
collection"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[10.0,10.0],[20.0,10.0],[10.0,20.0],[20.0,20.0]]\",\"type\":\"MultiLineString\",\"srid\":4326}")
@@ -1075,7 +1085,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 8L,
+ DecimalData.fromBigDecimal(new
BigDecimal(8), 38, 19),
BinaryStringData.fromString("Multi-Polygon"),
BinaryStringData.fromString(
"{\"coordinates\":\"[[0.0,0.0],[10.0,0.0],[10.0,10.0],[0.0,10.0],[0.0,0.0],[15.0,15.0],[25.0,15.0],[25.0,25.0],[15.0,25.0],[15.0,15.0]]\",\"type\":\"MultiPolygon\",\"srid\":4326}")
@@ -1085,7 +1095,7 @@ public class OraclePipelineITCase extends
OracleSourceTestBase {
tableId,
generator.generate(
new Object[] {
- 9L,
+ DecimalData.fromBigDecimal(new
BigDecimal(9), 38, 19),
BinaryStringData.fromString("Geometry
Collection"),
BinaryStringData.fromString(
"{\"geometries\":\"GEOMETRYCOLLECTION (POINT (25 25), LINESTRING (30 30, 35
25), POLYGON ((0 0, 10 0, 10 10, 0 10, 0
0)))\",\"type\":\"GeometryCollection\",\"srid\":4326}")
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/utils/OracleTypeUtilsTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/utils/OracleTypeUtilsTest.java
new file mode 100644
index 000000000..459f423f8
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-oracle/src/test/java/org/apache/flink/cdc/connectors/oracle/utils/OracleTypeUtilsTest.java
@@ -0,0 +1,282 @@
+/*
+ * 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.connectors.oracle.utils;
+
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.common.types.DecimalType;
+
+import io.debezium.relational.Column;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Types;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link OracleTypeUtils}. */
+class OracleTypeUtilsTest {
+
+ // --- Bare NUMBER (scale unspecified) tests ---
+
+ @Test
+ void testBareNumberShouldBeDecimal3819() {
+ // NUMBER (scale unspecified) → DECIMAL(38, 19)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(0)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+
assertThat(result).isEqualTo(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 19));
+ }
+
+ @Test
+ void testBareNumberNotNullShouldBeDecimal3819NotNull() {
+ // NOT NULL NUMBER → DECIMAL(38, 19).notNull()
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(0)
+ .optional(false)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+
assertThat(result).isEqualTo(DataTypes.DECIMAL(DecimalType.MAX_PRECISION,
19).notNull());
+ }
+
+ @Test
+ void testNumberPrecision10ScaleAbsentShouldBeDecimal3819() {
+ // NUMBER(10) with no scale info → DECIMAL(38, 19) (treated as bare
NUMBER)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(10)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+
assertThat(result).isEqualTo(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 19));
+ }
+
+ // --- Explicit integer (scale = 0) tests ---
+
+ @Test
+ void testNumberPrecision1ShouldBeBigint() {
+ // NUMBER(1, 0) → BIGINT (precision <= 18)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(1)
+ .scale(0)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.BIGINT());
+ }
+
+ @Test
+ void testNumberPrecision18ShouldBeBigint() {
+ // NUMBER(18, 0) → BIGINT (boundary: precision = 18)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(18)
+ .scale(0)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.BIGINT());
+ }
+
+ @Test
+ void testNumberPrecision19ShouldBeDecimal() {
+ // NUMBER(19, 0) → DECIMAL(19, 0) (boundary: precision = 19)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(19)
+ .scale(0)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.DECIMAL(19, 0));
+ }
+
+ @Test
+ void testNumberPrecision38ShouldBeDecimal() {
+ // NUMBER(38, 0) → DECIMAL(38, 0)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(38)
+ .scale(0)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.DECIMAL(38, 0));
+ }
+
+ @Test
+ void testNumberPrecision19NotNullShouldBeDecimalNotNull() {
+ // NOT NULL NUMBER(19, 0) → DECIMAL(19, 0).notNull()
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(19)
+ .scale(0)
+ .optional(false)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.DECIMAL(19, 0).notNull());
+ }
+
+ // --- Decimal (scale > 0) tests ---
+
+ @Test
+ void testNumberWithScaleShouldBeDecimal() {
+ // NUMBER(10, 6) → DECIMAL(10, 6)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(10)
+ .scale(6)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.DECIMAL(10, 6));
+ }
+
+ @Test
+ void testNumberWithScaleBoundaryShouldBeDecimal() {
+ // NUMBER(38, 36) → DECIMAL(38, 36) (scale at boundary)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(38)
+ .scale(36)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.DECIMAL(38, 36));
+ }
+
+ // --- NUMBER(*, s) (precision == 0, scale > 0) tests ---
+
+ @Test
+ void testNumberStarPrecisionShouldBeDecimal38() {
+ // NUMBER(*, 2) → DECIMAL(38, 2) (precision == 0 defaults to
MAX_PRECISION)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(0)
+ .scale(2)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+
assertThat(result).isEqualTo(DataTypes.DECIMAL(DecimalType.MAX_PRECISION, 2));
+ }
+
+ @Test
+ void testNumberStarPrecisionNotNullShouldBeDecimal38NotNull() {
+ // NOT NULL NUMBER(*, 4) → DECIMAL(38, 4).notNull()
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(0)
+ .scale(4)
+ .optional(false)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+
assertThat(result).isEqualTo(DataTypes.DECIMAL(DecimalType.MAX_PRECISION,
4).notNull());
+ }
+
+ // --- Scale < 0 (negative scale) downgrade tests ---
+
+ @Test
+ void testNegativeScaleShouldBeBigint() {
+ // NUMBER(10, -2) → BIGINT (negative scale, p+|s|=12 fits in BIGINT)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(10)
+ .scale(-2)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.BIGINT());
+ }
+
+ @Test
+ void testNegativeScaleStarPrecisionShouldBeString() {
+ // NUMBER(*, -3) → STRING (negative scale downgraded)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(0)
+ .scale(-3)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.STRING());
+ }
+
+ // --- Scale > 36 downgrade tests ---
+
+ @Test
+ void testScaleGreaterThan36ShouldBeString() {
+ // NUMBER(10, 37) → STRING (scale > 36 downgraded)
+ Column column =
+ Column.editor()
+ .name("col")
+ .type("NUMBER")
+ .jdbcType(Types.NUMERIC)
+ .length(10)
+ .scale(37)
+ .optional(true)
+ .create();
+ DataType result = OracleTypeUtils.fromDbzColumn(column);
+ assertThat(result).isEqualTo(DataTypes.STRING());
+ }
+}
diff --git
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
index e6b5387a2..498012e54 100644
---
a/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
+++
b/flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/OracleE2eITCase.java
@@ -171,49 +171,49 @@ public class OracleE2eITCase extends
PipelineTestEnvironment {
"DataChangeEvent{tableId=DEBEZIUM.PRODUCTS, before=[],
after=[105, hammer, 14oz carpenters hammer, 0.875], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS,
schema=columns={`ID` BIGINT NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS`
VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}");
+ "CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS,
schema=columns={`ID` DECIMAL(38, 0) NOT NULL,`NAME` VARCHAR(255) NOT
NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID,
options=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_1, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[101, user_1, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_4, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[109, user_4, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_3, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[103, user_3, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_2, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[102, user_2, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_13, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1012, user_13, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_14, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1013, user_14, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_11, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1010, user_11, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_12, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1011, user_12, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_21, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[2000, user_21, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_10, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1009, user_10, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_6, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[111, user_6, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_5, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[110, user_5, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_9, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[123, user_9, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_19, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1018, user_19, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_20, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1019, user_20, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_8, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[121, user_8, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_17, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1016, user_17, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_18, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1017, user_18, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_15, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1014, user_15, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691841, user_7, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[118, user_7, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[171798691842, user_16, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS, before=[],
after=[1015, user_16, Shanghai, 123567891234], op=INSERT, meta=()}");
stat.execute(
"UPDATE DEBEZIUM.PRODUCTS SET DESCRIPTION='18oz carpenter
hammer' WHERE ID=106 ");
@@ -225,49 +225,49 @@ public class OracleE2eITCase extends
PipelineTestEnvironment {
waitUntilSpecificEvent(
"DataChangeEvent{tableId=DEBEZIUM.PRODUCTS, before=[107,
rocks, box of assorted rocks, 5.3], after=[107, rocks, box of assorted rocks,
5.1], op=UPDATE, meta=()}");
waitUntilSpecificEvent(
- "CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS_1,
schema=columns={`ID` BIGINT NOT NULL,`NAME` VARCHAR(255) NOT NULL,`ADDRESS`
VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID, options=()}");
+ "CreateTableEvent{tableId=DEBEZIUM.CUSTOMERS_1,
schema=columns={`ID` DECIMAL(38, 0) NOT NULL,`NAME` VARCHAR(255) NOT
NULL,`ADDRESS` VARCHAR(1024),`PHONE_NUMBER` VARCHAR(512)}, primaryKeys=ID,
options=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_10, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1009, user_10, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_21, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[2000, user_21, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_5, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[110, user_5, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_14, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1013, user_14, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_6, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[111, user_6, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_13, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1012, user_13, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_12, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1011, user_12, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_11, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1010, user_11, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_4, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[109, user_4, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_2, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[102, user_2, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_3, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[103, user_3, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_1, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[101, user_1, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_18, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1017, user_18, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_9, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[123, user_9, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_17, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1016, user_17, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_16, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1015, user_16, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_15, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1014, user_15, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_7, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[118, user_7, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_20, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1019, user_20, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691841, user_8, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[121, user_8, Shanghai, 123567891234], op=INSERT, meta=()}");
waitUntilSpecificEvent(
- "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[171798691842, user_19, Shanghai, 123567891234], op=INSERT, meta=()}");
+ "DataChangeEvent{tableId=DEBEZIUM.CUSTOMERS_1, before=[],
after=[1018, user_19, Shanghai, 123567891234], op=INSERT, meta=()}");
} catch (Exception e) {
LOG.error("Update table for CDC failed.", e);
throw new RuntimeException(e);
