From 02ccdddb726bb70b152a6c75b03e252b844bbf6f Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 25 Jun 2026 17:18:34 +0200 Subject: [PATCH 1/2] [Fix][Format] Correct Debezium numeric TIME parsing --- .../DebeziumJsonDeserializationSchema.java | 87 +++++++++++++++++-- ...umJsonDeserializationSchemaDispatcher.java | 7 +- .../json/debezium/DebeziumRowConverter.java | 80 ++++++++++++++--- .../debezium/DebeziumJsonSerDeSchemaTest.java | 61 +++++++++++++ 4 files changed, 214 insertions(+), 21 deletions(-) diff --git a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchema.java b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchema.java index d5fd640ce3e1..d17bac56a659 100644 --- a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchema.java +++ b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchema.java @@ -32,6 +32,10 @@ import org.apache.seatunnel.format.json.JsonDeserializationSchema; import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; import java.util.Optional; import static java.lang.String.format; @@ -48,6 +52,10 @@ public class DebeziumJsonDeserializationSchema implements DeserializationSchema< private static final String DATA_BEFORE = "before"; private static final String DATA_AFTER = "after"; private static final String DATA_TS = "ts_ms"; + private static final String DATA_SCHEMA = "schema"; + private static final String SCHEMA_FIELD = "field"; + private static final String SCHEMA_FIELDS = "fields"; + private static final String SCHEMA_NAME = "name"; private static final String REPLICA_IDENTITY_EXCEPTION = "The \"before\" field of %s operation is null, " @@ -102,8 +110,9 @@ public void deserializeMessage( } try { - JsonNode payload = getPayload(jsonDeserializer.deserializeToJsonNode(message)); - parsePayload(out, tablePath, payload); + JsonNode root = jsonDeserializer.deserializeToJsonNode(message); + JsonNode payload = getPayload(root); + parsePayload(out, tablePath, root, payload); } catch (Exception e) { // a big try catch to protect the processing. if (!ignoreParseErrors) { @@ -116,15 +125,42 @@ public void parsePayload(Collector out, JsonNode payload) throws I parsePayload(out, tablePath, payload); } + void parsePayload(Collector out, JsonNode root, JsonNode payload) + throws IOException { + parsePayload(out, tablePath, root, payload); + } + + void parsePayload( + Collector out, TablePath tablePath, JsonNode root, JsonNode payload) + throws IOException { + parsePayload( + out, + tablePath, + payload, + extractFieldSchemaNames(root, DATA_BEFORE), + extractFieldSchemaNames(root, DATA_AFTER)); + } + private void parsePayload(Collector out, TablePath tablePath, JsonNode payload) throws IOException { + parsePayload(out, tablePath, payload, Collections.emptyMap(), Collections.emptyMap()); + } + + private void parsePayload( + Collector out, + TablePath tablePath, + JsonNode payload, + Map beforeFieldSchemaNames, + Map afterFieldSchemaNames) + throws IOException { String op = payload.get(OP_KEY).asText(); JsonNode tsNode = payload.get(DATA_TS); switch (op) { case OP_CREATE: case OP_READ: - SeaTunnelRow insert = debeziumRowConverter.parse(payload.get(DATA_AFTER)); + SeaTunnelRow insert = + debeziumRowConverter.parse(payload.get(DATA_AFTER), afterFieldSchemaNames); insert.setRowKind(RowKind.INSERT); if (tablePath != null) { insert.setTableId(tablePath.toString()); @@ -135,7 +171,9 @@ private void parsePayload(Collector out, TablePath tablePath, Json out.collect(insert); break; case OP_UPDATE: - SeaTunnelRow before = debeziumRowConverter.parse(payload.get(DATA_BEFORE)); + SeaTunnelRow before = + debeziumRowConverter.parse( + payload.get(DATA_BEFORE), beforeFieldSchemaNames); if (before == null) { throw new IllegalStateException( String.format(REPLICA_IDENTITY_EXCEPTION, "UPDATE")); @@ -148,7 +186,8 @@ private void parsePayload(Collector out, TablePath tablePath, Json MetadataUtil.setEventTime(before, tsNode.asLong()); } - SeaTunnelRow after = debeziumRowConverter.parse(payload.get(DATA_AFTER)); + SeaTunnelRow after = + debeziumRowConverter.parse(payload.get(DATA_AFTER), afterFieldSchemaNames); after.setRowKind(RowKind.UPDATE_AFTER); if (tablePath != null) { @@ -161,7 +200,9 @@ private void parsePayload(Collector out, TablePath tablePath, Json out.collect(after); break; case OP_DELETE: - SeaTunnelRow delete = debeziumRowConverter.parse(payload.get(DATA_BEFORE)); + SeaTunnelRow delete = + debeziumRowConverter.parse( + payload.get(DATA_BEFORE), beforeFieldSchemaNames); if (delete == null) { throw new IllegalStateException( String.format(REPLICA_IDENTITY_EXCEPTION, "DELETE")); @@ -180,6 +221,40 @@ private void parsePayload(Collector out, TablePath tablePath, Json } } + static Map extractFieldSchemaNames(JsonNode root, String envelopeField) { + JsonNode schema = root == null ? null : root.get(DATA_SCHEMA); + JsonNode fields = schema == null ? null : schema.get(SCHEMA_FIELDS); + if (fields == null || !fields.isArray()) { + return Collections.emptyMap(); + } + for (JsonNode field : fields) { + if (!envelopeField.equals(getText(field, SCHEMA_FIELD))) { + continue; + } + JsonNode envelopeFields = field.get(SCHEMA_FIELDS); + if (envelopeFields == null || !envelopeFields.isArray()) { + return Collections.emptyMap(); + } + Map fieldSchemaNames = new HashMap<>(); + Iterator iterator = envelopeFields.elements(); + while (iterator.hasNext()) { + JsonNode envelopeFieldSchema = iterator.next(); + String fieldName = getText(envelopeFieldSchema, SCHEMA_FIELD); + String schemaName = getText(envelopeFieldSchema, SCHEMA_NAME); + if (fieldName != null && schemaName != null) { + fieldSchemaNames.put(fieldName, schemaName); + } + } + return fieldSchemaNames; + } + return Collections.emptyMap(); + } + + private static String getText(JsonNode node, String fieldName) { + JsonNode value = node == null ? null : node.get(fieldName); + return value == null || value.isNull() ? null : value.asText(); + } + @Override public SeaTunnelDataType getProducedType() { return this.rowType; diff --git a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchemaDispatcher.java b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchemaDispatcher.java index 0bd835193c94..233be1d6122a 100644 --- a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchemaDispatcher.java +++ b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchemaDispatcher.java @@ -77,19 +77,20 @@ public void deserialize(byte[] message, Collector out) { } try { - JsonNode payload = getPayload(JsonUtils.readTree(message)); + JsonNode root = JsonUtils.readTree(message); + JsonNode payload = getPayload(root); JsonNode source = payload.get(SOURCE); String database = getNodeValue(source, DATABASE); String schema = getNodeValue(source, SCHEMA); String table = getNodeValue(source, TABLE); TablePath tablePath = TablePath.of(database, schema, table); if (tableDeserializationMap.containsKey(tablePath)) { - tableDeserializationMap.get(tablePath).parsePayload(out, payload); + tableDeserializationMap.get(tablePath).parsePayload(out, root, payload); } else { if (isConnectorCanWithOutDB(source.get(CONNECTOR))) { tablePath = TablePath.of(null, schema, table); if (tableDeserializationMap.containsKey(tablePath)) { - tableDeserializationMap.get(tablePath).parsePayload(out, payload); + tableDeserializationMap.get(tablePath).parsePayload(out, root, payload); return; } } diff --git a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumRowConverter.java b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumRowConverter.java index db7798332474..dabd9d2966cb 100644 --- a/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumRowConverter.java +++ b/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumRowConverter.java @@ -43,6 +43,7 @@ import java.time.temporal.TemporalAccessor; import java.time.temporal.TemporalQueries; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; @@ -53,6 +54,10 @@ public class DebeziumRowConverter implements Serializable { private static final String DECIMAL_SCALE_KEY = "scale"; private static final String DECIMAL_VALUE_KEY = "value"; + private static final String DEBEZIUM_TIME_SCHEMA = "io.debezium.time.Time"; + private static final String DEBEZIUM_MICRO_TIME_SCHEMA = "io.debezium.time.MicroTime"; + private static final String DEBEZIUM_NANO_TIME_SCHEMA = "io.debezium.time.NanoTime"; + private static final long MICROS_PER_DAY = TimeUnit.DAYS.toMicros(1); private final Map fieldFormatterMap = new HashMap<>(); private final SeaTunnelRowType rowType; @@ -62,10 +67,19 @@ public DebeziumRowConverter(SeaTunnelRowType rowType) { } public SeaTunnelRow parse(JsonNode node) throws IOException { - return (SeaTunnelRow) getValue(null, rowType, node); + return parse(node, Collections.emptyMap()); } - private Object getValue(String fieldName, SeaTunnelDataType dataType, JsonNode value) + public SeaTunnelRow parse(JsonNode node, Map fieldSchemaNames) + throws IOException { + return (SeaTunnelRow) getValue(null, rowType, node, fieldSchemaNames); + } + + private Object getValue( + String fieldName, + SeaTunnelDataType dataType, + JsonNode value, + Map fieldSchemaNames) throws IOException { SqlType sqlType = dataType.getSqlType(); if (value == null || value.isNull()) { @@ -133,14 +147,11 @@ private Object getValue(String fieldName, SeaTunnelDataType dataType, JsonNod return dateFormatter.parse(dateStr).query(TemporalQueries.localDate()); case TIME: String timeStr = value.asText(); - if (value.canConvertToLong()) { + if (isIntegralValue(value)) { long time = Long.parseLong(timeStr); - if (timeStr.length() == 8) { - time = TimeUnit.SECONDS.toMicros(time); - } else if (timeStr.length() == 11) { - time = TimeUnit.MILLISECONDS.toMicros(time); - } - return LocalTime.ofNanoOfDay(time); + return LocalTime.ofNanoOfDay( + convertDebeziumTimeToNanos( + time, value, fieldSchemaNames.get(fieldName))); } DateTimeFormatter timeFormatter = fieldFormatterMap.get(fieldName); @@ -194,7 +205,12 @@ private Object getValue(String fieldName, SeaTunnelDataType dataType, JsonNod case ARRAY: List arrayValue = new ArrayList<>(); for (JsonNode o : value) { - arrayValue.add(getValue(fieldName, ((ArrayType) dataType).getElementType(), o)); + arrayValue.add( + getValue( + fieldName, + ((ArrayType) dataType).getElementType(), + o, + fieldSchemaNames)); } return arrayValue; case MAP: @@ -203,7 +219,11 @@ private Object getValue(String fieldName, SeaTunnelDataType dataType, JsonNod Map.Entry entry = it.next(); mapValue.put( entry.getKey(), - getValue(null, ((MapType) dataType).getValueType(), entry.getValue())); + getValue( + null, + ((MapType) dataType).getValueType(), + entry.getValue(), + Collections.emptyMap())); } return mapValue; case ROW: @@ -217,11 +237,47 @@ private Object getValue(String fieldName, SeaTunnelDataType dataType, JsonNod rowType.getFieldType(i), value.has(rowType.getFieldName(i)) ? value.get(rowType.getFieldName(i)) - : null)); + : null, + fieldSchemaNames)); } return row; default: throw new UnsupportedOperationException("Unsupported type: " + sqlType); } } + + private static boolean isIntegralValue(JsonNode value) { + if (value.canConvertToLong()) { + return true; + } + if (!value.isTextual()) { + return false; + } + try { + Long.parseLong(value.asText()); + return true; + } catch (NumberFormatException e) { + return false; + } + } + + private static long convertDebeziumTimeToNanos( + long time, JsonNode value, String debeziumSchemaName) { + if (DEBEZIUM_TIME_SCHEMA.equals(debeziumSchemaName)) { + return TimeUnit.MILLISECONDS.toNanos(time); + } + if (DEBEZIUM_MICRO_TIME_SCHEMA.equals(debeziumSchemaName)) { + return TimeUnit.MICROSECONDS.toNanos(time); + } + if (DEBEZIUM_NANO_TIME_SCHEMA.equals(debeziumSchemaName)) { + return time; + } + if (value.canConvertToInt()) { + return TimeUnit.MILLISECONDS.toNanos(time); + } + if (time >= 0 && time < MICROS_PER_DAY) { + return TimeUnit.MICROSECONDS.toNanos(time); + } + return time; + } } diff --git a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonSerDeSchemaTest.java b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonSerDeSchemaTest.java index fd3cbdf56147..359bae7f23f4 100644 --- a/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonSerDeSchemaTest.java +++ b/seatunnel-formats/seatunnel-format-json/src/test/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonSerDeSchemaTest.java @@ -41,6 +41,7 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.time.LocalTime; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -488,6 +489,66 @@ public void testDeserializationForOracle() throws Exception { Assertions.assertEquals("2024-12-17T15:23:42.119", row.getField(13).toString()); } + @Test + public void testDeserializationForSchemaLessMillisecondTime() throws Exception { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"time_millis"}, new SeaTunnelDataType[] {LOCAL_TIME_TYPE}); + DebeziumJsonDeserializationSchema deserializationSchema = + new DebeziumJsonDeserializationSchema( + CatalogTableUtil.getCatalogTable("default", rowType), false, false); + SimpleCollector collector = new SimpleCollector(); + + deserializationSchema.deserialize( + "{\"before\":null,\"after\":{\"time_millis\":60000},\"op\":\"c\"}" + .getBytes(StandardCharsets.UTF_8), + collector); + + SeaTunnelRow row = collector.getList().get(0); + Assertions.assertEquals(LocalTime.of(0, 1), row.getField(0)); + } + + @Test + public void testDeserializationForDebeziumTimeUnitsWithSchema() throws Exception { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"time_millis", "time_micros", "time_nanos"}, + new SeaTunnelDataType[] { + LOCAL_TIME_TYPE, LOCAL_TIME_TYPE, LOCAL_TIME_TYPE + }); + DebeziumJsonDeserializationSchema deserializationSchema = + new DebeziumJsonDeserializationSchema( + CatalogTableUtil.getCatalogTable("default", rowType), false, true); + SimpleCollector collector = new SimpleCollector(); + + String message = + "{\"schema\":{\"type\":\"struct\",\"fields\":[" + + "{\"type\":\"struct\",\"fields\":[" + + "{\"type\":\"int32\",\"name\":\"io.debezium.time.Time\"," + + "\"field\":\"time_millis\"}," + + "{\"type\":\"int64\",\"name\":\"io.debezium.time.MicroTime\"," + + "\"field\":\"time_micros\"}," + + "{\"type\":\"int64\",\"name\":\"io.debezium.time.NanoTime\"," + + "\"field\":\"time_nanos\"}],\"field\":\"before\"}," + + "{\"type\":\"struct\",\"fields\":[" + + "{\"type\":\"int32\",\"name\":\"io.debezium.time.Time\"," + + "\"field\":\"time_millis\"}," + + "{\"type\":\"int64\",\"name\":\"io.debezium.time.MicroTime\"," + + "\"field\":\"time_micros\"}," + + "{\"type\":\"int64\",\"name\":\"io.debezium.time.NanoTime\"," + + "\"field\":\"time_nanos\"}],\"field\":\"after\"}]}," + + "\"payload\":{\"before\":null,\"after\":{\"time_millis\":60000," + + "\"time_micros\":60000000,\"time_nanos\":60000000000}," + + "\"op\":\"c\"}}"; + + deserializationSchema.deserialize(message.getBytes(StandardCharsets.UTF_8), collector); + + SeaTunnelRow row = collector.getList().get(0); + Assertions.assertEquals(LocalTime.of(0, 1), row.getField(0)); + Assertions.assertEquals(LocalTime.of(0, 1), row.getField(1)); + Assertions.assertEquals(LocalTime.of(0, 1), row.getField(2)); + } + /** * create table all_types_1( id int8 primary key, f1 bool, f2 bool[], f3 bytea, f5 smallint, f6 * SMALLSERIAL, f7 smallint[], f8 int, f9 integer, f10 SERIAL, f11 int[], f12 bigint, f13 From 08a2ae540606594ed602feae4c87fe0317e62a6d Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 25 Jun 2026 17:39:33 +0200 Subject: [PATCH 2/2] [Chore] Trigger CI