DanielLeens commented on PR #11140: URL: https://github.com/apache/seatunnel/pull/11140#issuecomment-6369105357
I prepared two alternative contributor patches for current head `8c2f7bc59c0d1480536b3585a2bcaafd2455f7b2`. They are proposals requiring fresh functional CI, not a merge-ready conclusion. The [current-head connector job](https://github.com/fallintoplace/seatunnel/actions/runs/31941821907/job/95152349361) directly reports: ```text CouchbaseIT.java:[299,18] error: no suitable method found for withUlimits(Ulimit) method HostConfig.withUlimits(Ulimit[]) is not applicable method HostConfig.withUlimits(List<Ulimit>) is not applicable ``` This was introduced by the current head's nofile-limit commit. A one-element `Ulimit[]` preserves the intended name/soft/hard limits and matches the actual docker-java API. The added TIME Javadocs document the current unit-resolution/flat-schema fallback; they do not change its arithmetic or defaults. 1. **Minimal patch for the current head:** apply the first diff directly to the clean source head. It corrects the compiler call and the existing review's fallback-documentation gap. It does not resolve the branch's current conflicts. 2. **Combined sync/conflict proposal:** merge upstream `dev` commit `14a27ac26a0a71b56722f86cdf8ec2779094ceaf`, then apply the second diff to the paused merge. This already includes the minimal fixes; do not stack the first patch before this marker-resolution patch. The combined proposal preserves both sides of the semantic conflict with upstream #12172: - Keep streaming payload/schema-descriptor reads, skips for large unused objects, cached table identity/root field arrays, pre-sized arrays and direct row allocation. - Thread the PR's immediate before/after logical TIME names through those optimized paths, without creating a full schema/envelope JsonNode tree. Existing schema-less conversion and dispatcher overloads remain. - Keep the PR's container nofile setup and upstream's Couchbase timer-flush collection, exact visibility assertion while the job is active, and cancellation cleanup. - Strengthen the existing logical-TIME test for schema-first/payload-first envelopes, all three units, INSERT and UPDATE, independent before/after unit assignments, and a skipped large nested source object. All existing tests and assertion expressions remain; no test is disabled or relaxed. For option 2, save the second diff as `pr11140-current-dev-conflict-resolution.patch` and start from the exact clean contributor head above: ```bash git fetch https://github.com/apache/seatunnel.git dev git -c merge.conflictStyle=merge -c rerere.enabled=false merge --no-commit --no-ff 14a27ac26a0a71b56722f86cdf8ec2779094ceaf # Three expected conflicts; the diff also updates the auto-merged existing SerDe test. git apply --check pr11140-current-dev-conflict-resolution.patch git apply pr11140-current-dev-conflict-resolution.patch ``` Review/stage the four resolved files, then commit/push normally from your fork. If either pinned commit or conflict output changes, regenerate the resolution instead of forcing hunks. Checks completed: scoped JSON-format/Couchbase-E2E Spotless; whitespace; patch applicability against the exact original head or saved merge-marker/test baseline; byte-for-byte reproduction of all four resolved files; test-method/old-assertion preservation; and a focused independent static read of the call signature/parser/cache reconciliation. Compilation, unit/IT/E2E execution and throughput/allocation measurements were not run locally. Current old-head CI does not validate either unpushed patch; fresh JSON/dispatcher/Couchbase and relevant connector CI remains necessary. One original-PR limitation remains explicit: logical names are a flat map, and ROW recursion forwards it. Same-named nested TIME fields can inherit a root field's unit. These proposals do not claim to resolve nested logical-schema semantics or all existing CI failures. Minimal current-head patch: ```diff diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java index b96f414954d..7fd4f951268 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java @@ -297,10 +297,12 @@ public class CouchbaseIT extends TestSuiteBase implements TestResource { ? HostConfig.newHostConfig() : command.getHostConfig(); hostConfig.withUlimits( - new Ulimit( - "nofile", - COUCHBASE_CONTAINER_NOFILE_LIMIT, - COUCHBASE_CONTAINER_NOFILE_LIMIT)); + new Ulimit[] { + new Ulimit( + "nofile", + COUCHBASE_CONTAINER_NOFILE_LIMIT, + COUCHBASE_CONTAINER_NOFILE_LIMIT) + }); command.withHostConfig(hostConfig); } } 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 d17bac56a65..4807b43ee08 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 @@ -221,6 +221,12 @@ public class DebeziumJsonDeserializationSchema implements DeserializationSchema< } } + /** + * Reads immediate column logical types from the selected before/after envelope schema. + * + * <p>The converter uses these names before its numeric TIME fallback. This flat map does not + * traverse nested ROW schemas; missing logical names retain the compatibility fallback. + */ 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); 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 2e6e103aef5..7d6a89e6239 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 @@ -246,6 +246,13 @@ public class DebeziumRowConverter implements Serializable { } } + /** + * Resolves the unit from Debezium's logical type before applying the schema-less fallback. + * + * <p>Without a logical type, integer-sized values retain the millisecond contract; larger + * non-negative values below one day in microseconds are treated as microseconds. The remaining + * values are nanoseconds. Ambiguous schema-less values cannot be disambiguated by magnitude. + */ private static long convertDebeziumTimeToNanos( long time, JsonNode value, String debeziumSchemaName) { if (DEBEZIUM_TIME_SCHEMA.equals(debeziumSchemaName)) { ``` Alternative combined merge-resolution patch: ```diff --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java @@ -47,6 +47,7 @@ import java.io.IOException; import java.time.Duration; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -68,16 +69,14 @@ private static final String COUCHBASE_BUCKET = "test_bucket"; private static final String COUCHBASE_SCOPE = "_default"; private static final String COUCHBASE_COLLECTION = "test_collection"; -<<<<<<< HEAD + private static final String COUCHBASE_COLLECTION_TIMER_FLUSH = "test_collection_timer_flush"; + /** * Couchbase 7.1 requires the couchbase user inside the container to have at least 200000 open * files. Configure a slightly higher per-container limit so GitHub-hosted Docker starts the * server reliably. */ private static final int COUCHBASE_CONTAINER_NOFILE_LIMIT = 262144; -======= - private static final String COUCHBASE_COLLECTION_TIMER_FLUSH = "test_collection_timer_flush"; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf /** Matches row.num in fake_source_to_couchbase.conf. */ private static final int EXPECTED_ROW_COUNT = 100; @@ -352,26 +351,10 @@ }); } -<<<<<<< HEAD /** - * Applies the minimum open-file limit required by the Couchbase image inside Docker. - * - * <p>GitHub-hosted runners expose a lower default limit (65536) to containers, which causes the - * Couchbase bootstrap script to exit before Testcontainers can finish the cluster - * initialization sequence. + * Verifies buffered rows become visible while the streaming job is still running. The exact + * ten-row assertion must succeed before cancellation can trigger an end-of-input flush. */ - private void applyCouchbaseFileLimit(CreateContainerCmd command) { - HostConfig hostConfig = - command.getHostConfig() == null - ? HostConfig.newHostConfig() - : command.getHostConfig(); - hostConfig.withUlimits( - new Ulimit( - "nofile", - COUCHBASE_CONTAINER_NOFILE_LIMIT, - COUCHBASE_CONTAINER_NOFILE_LIMIT)); - command.withHostConfig(hostConfig); -======= @TestTemplate @DisabledOnContainer( value = {}, @@ -385,8 +368,8 @@ COUCHBASE_BUCKET, COUCHBASE_SCOPE, COUCHBASE_COLLECTION_TIMER_FLUSH)); String jobId = String.valueOf(System.currentTimeMillis()); - java.util.concurrent.CompletableFuture<Container.ExecResult> jobFuture = - java.util.concurrent.CompletableFuture.supplyAsync( + CompletableFuture<Container.ExecResult> jobFuture = + CompletableFuture.supplyAsync( () -> { try { return container.executeJob( @@ -423,6 +406,27 @@ Assertions.assertEquals(0, cancelResult.getExitCode(), cancelResult.getStderr()); } } ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf + } + + /** + * Applies the minimum open-file limit required by the Couchbase image inside Docker. + * + * <p>GitHub-hosted runners expose a lower default limit (65536) to containers, which causes the + * Couchbase bootstrap script to exit before Testcontainers can finish the cluster + * initialization sequence. + */ + private void applyCouchbaseFileLimit(CreateContainerCmd command) { + HostConfig hostConfig = + command.getHostConfig() == null + ? HostConfig.newHostConfig() + : command.getHostConfig(); + hostConfig.withUlimits( + new Ulimit[] { + new Ulimit( + "nofile", + COUCHBASE_CONTAINER_NOFILE_LIMIT, + COUCHBASE_CONTAINER_NOFILE_LIMIT) + }); + command.withHostConfig(hostConfig); } } --- 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 @@ -118,15 +118,17 @@ } try { -<<<<<<< HEAD - JsonNode root = jsonDeserializer.deserializeToJsonNode(message); - JsonNode payload = getPayload(root); - parsePayload(out, tablePath, root, payload); -======= - // Materialize only CDC fields; skip large unused objects such as source/schema. - JsonNode payload = readDebeziumPayload(message); - parsePayload(out, tablePath, payload); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf + // Only schema-enabled messages need per-field logical type metadata. Keep the + // schema-less hot path free of metadata maps and continue skipping unused objects. + Map<String, Map<String, String>> fieldSchemaNames = + debeziumEnabledSchema ? new HashMap<>() : Collections.emptyMap(); + JsonNode payload = readDebeziumPayload(message, fieldSchemaNames); + parsePayload( + out, + tablePath, + payload, + fieldSchemaNames.getOrDefault(DATA_BEFORE, Collections.emptyMap()), + fieldSchemaNames.getOrDefault(DATA_AFTER, Collections.emptyMap())); } catch (Exception e) { // a big try catch to protect the processing. if (!ignoreParseErrors) { @@ -139,7 +141,6 @@ parsePayload(out, tablePath, payload); } -<<<<<<< HEAD void parsePayload(Collector<SeaTunnelRow> out, JsonNode root, JsonNode payload) throws IOException { parsePayload(out, tablePath, root, payload); @@ -154,11 +155,11 @@ payload, extractFieldSchemaNames(root, DATA_BEFORE), extractFieldSchemaNames(root, DATA_AFTER)); -======= + } + @Override public SeaTunnelDataType<SeaTunnelRow> getProducedType() { return this.rowType; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf } private void parsePayload(Collector<SeaTunnelRow> out, TablePath tablePath, JsonNode payload) @@ -222,7 +223,12 @@ } } -<<<<<<< HEAD + /** + * Reads immediate column logical types from the selected before/after envelope schema. + * + * <p>The converter uses these names before its numeric TIME fallback. This flat map does not + * traverse nested ROW schemas; missing logical names retain the compatibility fallback. + */ 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); @@ -257,10 +263,6 @@ return value == null || value.isNull() ? null : value.asText(); } - @Override - public SeaTunnelDataType<SeaTunnelRow> getProducedType() { - return this.rowType; -======= /** * Resolves the table id string for the emitted row. Call sites normally pass {@link * #tablePath}; the identity check reuses the cached {@link #tableId}, while the fallback covers @@ -275,7 +277,6 @@ return tableId; } return tablePath.toString(); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf } /** Applies table id and optional event-time metadata to a newly parsed row. */ @@ -290,11 +291,12 @@ /** * Parses a Debezium JSON envelope while skipping unused objects such as {@code source}, {@code - * transaction}, and the optional top-level {@code schema}. Only {@code op}, {@code ts_ms}, - * {@code before}, and {@code after} (under {@code payload} when schema is included) are - * materialized into JsonNodes. - */ - private JsonNode readDebeziumPayload(byte[] message) throws IOException { + * transaction}. Only CDC payload fields are materialized as JsonNodes. With schema enabled, + * retain the immediate before/after field logical names without building the schema tree; + * schema and payload may occur in either order in the envelope. + */ + private JsonNode readDebeziumPayload( + byte[] message, Map<String, Map<String, String>> fieldSchemaNames) throws IOException { ObjectMapper objectMapper = jsonDeserializer.getObjectMapper(); try (JsonParser parser = objectMapper.getFactory().createParser(message)) { if (parser.nextToken() != JsonToken.START_OBJECT) { @@ -307,6 +309,9 @@ parser.nextToken(); if (DATA_PAYLOAD.equals(fieldName)) { payload = readPayloadFields(objectMapper, parser); + } else if (DATA_SCHEMA.equals(fieldName)) { + fieldSchemaNames.clear(); + readEnvelopeFieldSchemaNames(parser, fieldSchemaNames); } else { parser.skipChildren(); } @@ -318,6 +323,100 @@ } return readPayloadFields(objectMapper, parser); } + } + + /** + * Reads only the immediate before/after schema descriptors. Schema attributes and nested + * metadata unrelated to TIME conversion remain streaming skips rather than JsonNode trees. + */ + private static void readEnvelopeFieldSchemaNames( + JsonParser parser, Map<String, Map<String, String>> fieldSchemaNames) + throws IOException { + if (parser.currentToken() != JsonToken.START_OBJECT) { + parser.skipChildren(); + return; + } + while (parser.nextToken() != JsonToken.END_OBJECT) { + String name = parser.currentName(); + parser.nextToken(); + if (SCHEMA_FIELDS.equals(name)) { + fieldSchemaNames.clear(); + if (parser.currentToken() == JsonToken.START_ARRAY) { + while (parser.nextToken() != JsonToken.END_ARRAY) { + readEnvelopeFieldSchema(parser, fieldSchemaNames); + } + } else { + parser.skipChildren(); + } + } else { + parser.skipChildren(); + } + } + } + + /** + * Preserves field-order independence, including descriptors whose fields precede field. Schema + * JSON property order must not determine which logical type names reach the converter. + */ + private static void readEnvelopeFieldSchema( + JsonParser parser, Map<String, Map<String, String>> fieldSchemaNames) + throws IOException { + if (parser.currentToken() != JsonToken.START_OBJECT) { + parser.skipChildren(); + return; + } + String envelopeField = null; + Map<String, String> logicalNames = Collections.emptyMap(); + while (parser.nextToken() != JsonToken.END_OBJECT) { + String name = parser.currentName(); + parser.nextToken(); + if (SCHEMA_FIELD.equals(name)) { + envelopeField = parser.getValueAsString(); + parser.skipChildren(); + } else if (SCHEMA_FIELDS.equals(name)) { + logicalNames = readFieldSchemaNames(parser); + } else { + parser.skipChildren(); + } + } + if ((DATA_BEFORE.equals(envelopeField) || DATA_AFTER.equals(envelopeField)) + && !fieldSchemaNames.containsKey(envelopeField)) { + fieldSchemaNames.put(envelopeField, logicalNames); + } + } + + /** + * Extracts only immediate field/name string pairs, matching the dispatcher tree path. Nested + * column-schema objects are skipped because the converter retains the existing flat contract. + */ + private static Map<String, String> readFieldSchemaNames(JsonParser parser) throws IOException { + if (parser.currentToken() != JsonToken.START_ARRAY) { + parser.skipChildren(); + return Collections.emptyMap(); + } + Map<String, String> logicalNames = new HashMap<>(); + while (parser.nextToken() != JsonToken.END_ARRAY) { + if (parser.currentToken() != JsonToken.START_OBJECT) { + parser.skipChildren(); + continue; + } + String fieldName = null; + String schemaName = null; + while (parser.nextToken() != JsonToken.END_OBJECT) { + String name = parser.currentName(); + parser.nextToken(); + if (SCHEMA_FIELD.equals(name)) { + fieldName = parser.getValueAsString(); + } else if (SCHEMA_NAME.equals(name)) { + schemaName = parser.getValueAsString(); + } + parser.skipChildren(); + } + if (fieldName != null && schemaName != null) { + logicalNames.put(fieldName, schemaName); + } + } + return logicalNames; } /** --- 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 @@ -207,57 +207,22 @@ LocalDate localDate = parsedTimestamp.query(TemporalQueries.localDate()); return LocalDateTime.of(localDate, localTime); case ARRAY: -<<<<<<< HEAD - List<Object> arrayValue = new ArrayList<>(); - for (JsonNode o : value) { - arrayValue.add( - getValue( - fieldName, - ((ArrayType) dataType).getElementType(), - o, - fieldSchemaNames)); -======= { List<Object> arrayValue = new ArrayList<>(value.size()); SeaTunnelDataType<?> elementType = ((ArrayType) dataType).getElementType(); for (JsonNode o : value) { - arrayValue.add(getValue(fieldName, elementType, o)); + arrayValue.add(getValue(fieldName, elementType, o, fieldSchemaNames)); } return arrayValue; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf } case MAP: Map<Object, Object> mapValue = new LinkedHashMap<>(); SeaTunnelDataType<?> mapValueType = ((MapType) dataType).getValueType(); for (Iterator<Map.Entry<String, JsonNode>> it = value.fields(); it.hasNext(); ) { Map.Entry<String, JsonNode> entry = it.next(); -<<<<<<< HEAD mapValue.put( entry.getKey(), - getValue( - null, - ((MapType) dataType).getValueType(), - entry.getValue(), - Collections.emptyMap())); - } - return mapValue; - case ROW: - SeaTunnelRowType rowType = (SeaTunnelRowType) dataType; - SeaTunnelRow row = new SeaTunnelRow(rowType.getTotalFields()); - for (int i = 0; i < rowType.getTotalFields(); i++) { - row.setField( - i, - getValue( - rowType.getFieldName(i), - rowType.getFieldType(i), - value.has(rowType.getFieldName(i)) - ? value.get(rowType.getFieldName(i)) - : null, - fieldSchemaNames)); - } - return row; -======= - mapValue.put(entry.getKey(), getValue(null, mapValueType, entry.getValue())); + getValue(null, mapValueType, entry.getValue(), Collections.emptyMap())); } return mapValue; case ROW: @@ -275,16 +240,23 @@ Object[] fields = new Object[arity]; for (int i = 0; i < arity; i++) { String name = fieldNames[i]; - fields[i] = getValue(name, fieldTypes[i], value.get(name)); + fields[i] = + getValue(name, fieldTypes[i], value.get(name), fieldSchemaNames); } return new SeaTunnelRow(fields); } ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf default: throw new UnsupportedOperationException("Unsupported type: " + sqlType); } } + /** + * Resolves the unit from Debezium's logical type before applying the schema-less fallback. + * + * <p>Without a logical type, integer-sized values retain the millisecond contract; larger + * non-negative values below one day in microseconds are treated as microseconds. The remaining + * values are nanoseconds. Ambiguous schema-less values cannot be disambiguated by magnitude. + */ private static long convertDebeziumTimeToNanos( long time, JsonNode value, String debeziumSchemaName) { if (DEBEZIUM_TIME_SCHEMA.equals(debeziumSchemaName)) { --- 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 @@ -20,6 +20,7 @@ import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.seatunnel.api.source.Collector; import org.apache.seatunnel.api.table.catalog.CatalogTable; @@ -624,8 +625,6 @@ DebeziumJsonDeserializationSchema deserializationSchema = new DebeziumJsonDeserializationSchema( CatalogTableUtil.getCatalogTable("default", rowType), false, true); - SimpleCollector collector = new SimpleCollector(); - String message = "{\"schema\":{\"type\":\"struct\",\"fields\":[" + "{\"type\":\"struct\",\"fields\":[" @@ -646,12 +645,65 @@ + "\"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)); + ObjectMapper mapper = new ObjectMapper(); + ObjectNode root = (ObjectNode) mapper.readTree(message); + ObjectNode payload = mapper.createObjectNode(); + // Unused nested objects still have to be skipped without losing subsequent CDC fields. + payload.putObject("source") + .putObject("unused") + .put("large", String.join("", Collections.nCopies(4096, "x"))); + payload.setAll((ObjectNode) root.get("payload")); + root.set("payload", payload); + for (boolean payloadFirst : new boolean[] {false, true}) { + String envelope = + payloadFirst + ? "{\"payload\":" + payload + ",\"schema\":" + root.get("schema") + "}" + : root.toString(); + SimpleCollector collector = new SimpleCollector(); + deserializationSchema.deserialize(envelope.getBytes(StandardCharsets.UTF_8), collector); + + Assertions.assertEquals(1, collector.getList().size()); + SeaTunnelRow row = collector.getList().get(0); + Assertions.assertEquals(RowKind.INSERT, row.getRowKind()); + 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)); + } + + // UPDATE must use the logical names for each envelope side, in either top-level order. + payload.set("before", payload.get("after")); + ObjectNode after = ((ObjectNode) payload.get("after")).deepCopy(); + after.put("time_millis", 120000); + after.put("time_micros", 120000000L); + after.put("time_nanos", 120000000000L); + payload.set("after", after); + ObjectNode beforeValues = (ObjectNode) payload.get("before"); + beforeValues.put("time_millis", 60000000L); + beforeValues.put("time_micros", 60000000000L); + beforeValues.put("time_nanos", 60000); + JsonNode beforeFields = root.get("schema").get("fields").get(0).get("fields"); + ((ObjectNode) beforeFields.get(0)).put("name", "io.debezium.time.MicroTime"); + ((ObjectNode) beforeFields.get(1)).put("name", "io.debezium.time.NanoTime"); + ((ObjectNode) beforeFields.get(2)).put("name", "io.debezium.time.Time"); + payload.put("op", "u"); + for (boolean payloadFirst : new boolean[] {false, true}) { + String envelope = + payloadFirst + ? "{\"payload\":" + payload + ",\"schema\":" + root.get("schema") + "}" + : root.toString(); + SimpleCollector collector = new SimpleCollector(); + deserializationSchema.deserialize(envelope.getBytes(StandardCharsets.UTF_8), collector); + + Assertions.assertEquals(2, collector.getList().size()); + SeaTunnelRow before = collector.getList().get(0); + SeaTunnelRow updated = collector.getList().get(1); + Assertions.assertEquals(RowKind.UPDATE_BEFORE, before.getRowKind()); + Assertions.assertEquals(RowKind.UPDATE_AFTER, updated.getRowKind()); + for (int i = 0; i < 3; i++) { + Assertions.assertEquals(LocalTime.of(0, 1), before.getField(i)); + Assertions.assertEquals(LocalTime.of(0, 2), updated.getField(i)); + } + } } /** ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
