Skip to content

[Fix][Format] Correct Debezium numeric TIME parsing #11140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fallintoplace wants to merge 3 commits into
base: dev
Choose a base branch
from
Open

[Fix][Format] Correct Debezium numeric TIME parsing #11140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
02ccddd
[Fix][Format] Correct Debezium numeric TIME parsing
fallintoplace Jun 25, 2026
08a2ae5
[Chore] Trigger CI
fallintoplace Jun 25, 2026
b838ce8
Merge branch 'dev' into fix/debezium-time-units
davidzollo Jul 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 81 additions & 6 deletions ...ain/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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, "
Expand Down Expand Up @@ -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) {
Expand All @@ -116,15 +125,42 @@ public void parsePayload(Collector<SeaTunnelRow> out, JsonNode payload) throws I
parsePayload(out, tablePath, payload);
}

void parsePayload(Collector<SeaTunnelRow> out, JsonNode root, JsonNode payload)
throws IOException {
parsePayload(out, tablePath, root, payload);
}

void parsePayload(
Collector<SeaTunnelRow> 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<SeaTunnelRow> out, TablePath tablePath, JsonNode payload)
throws IOException {
parsePayload(out, tablePath, payload, Collections.emptyMap(), Collections.emptyMap());
}

private void parsePayload(
Collector<SeaTunnelRow> out,
TablePath tablePath,
JsonNode payload,
Map<String, String> beforeFieldSchemaNames,
Map<String, String> 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());
Expand All @@ -135,7 +171,9 @@ private void parsePayload(Collector<SeaTunnelRow> 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"));
Expand All @@ -148,7 +186,8 @@ private void parsePayload(Collector<SeaTunnelRow> 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) {
Expand All @@ -161,7 +200,9 @@ private void parsePayload(Collector<SeaTunnelRow> 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"));
Expand All @@ -180,6 +221,40 @@ private void parsePayload(Collector<SeaTunnelRow> out, TablePath tablePath, Json
}
}

static Map<String, String> 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<String, String> fieldSchemaNames = new HashMap<>();
Iterator<JsonNode> 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<SeaTunnelRow> getProducedType() {
return this.rowType;
Expand Down
7 changes: 4 additions & 3 deletions ...rg/apache/seatunnel/format/json/debezium/DebeziumJsonDeserializationSchemaDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,19 +77,20 @@ public void deserialize(byte[] message, Collector<SeaTunnelRow> 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;
}
}
Expand Down
80 changes: 68 additions & 12 deletions ...at-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumRowConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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<String, DateTimeFormatter> fieldFormatterMap = new HashMap<>();
private final SeaTunnelRowType rowType;
Expand All @@ -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<String, String> fieldSchemaNames)
throws IOException {
return (SeaTunnelRow) getValue(null, rowType, node, fieldSchemaNames);
}

private Object getValue(
String fieldName,
SeaTunnelDataType<?> dataType,
JsonNode value,
Map<String, String> fieldSchemaNames)
throws IOException {
SqlType sqlType = dataType.getSqlType();
if (value == null || value.isNull()) {
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -194,7 +205,12 @@ private Object getValue(String fieldName, SeaTunnelDataType<?> dataType, JsonNod
case ARRAY:
List<Object> 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:
Expand All @@ -203,7 +219,11 @@ private Object getValue(String fieldName, SeaTunnelDataType<?> dataType, JsonNod
Map.Entry<String, JsonNode> 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:
Expand All @@ -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;
}
}
61 changes: 61 additions & 0 deletions .../src/test/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonSerDeSchemaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
Loading