This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12056-2e2e9b804004f7a57da4a84219ed2438232b26fc in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit f6d230ecd06e87b7206b3fa252b9a2eb89b5f689 Author: Guobin Sun <[email protected]> AuthorDate: Thu Sep 3 12:17:34 2026 +0000 [Feature][Format] Add Debezium JSON SerDe microbenchmarks (#12056) Co-authored-by: Cursor <[email protected]> --- .github/workflows/benchmarks.yml | 1 + seatunnel-benchmarks/pom.xml | 5 + .../benchmark/DebeziumJsonFormatBenchmark.java | 304 +++++++++++++++++++++ .../benchmark/DebeziumJsonFormatBenchmarkTest.java | 74 +++++ tools/benchmarks/suites/benchmarks_core.txt | 1 + 5 files changed, 385 insertions(+) diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index 9a0b4e5997..dcbd8fa654 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -36,6 +36,7 @@ on: - 'benchmarks_core' - 'SeaTunnelRowBenchmark' - 'IntermediateQueueBenchmark' + - 'DebeziumJsonFormatBenchmark' - 'SeaTunnelPipelineBenchmark' - 'CheckpointingTimeBenchmark' - 'CheckpointStorageBenchmark' diff --git a/seatunnel-benchmarks/pom.xml b/seatunnel-benchmarks/pom.xml index 975e7b3c61..8a3da5d15e 100644 --- a/seatunnel-benchmarks/pom.xml +++ b/seatunnel-benchmarks/pom.xml @@ -38,6 +38,11 @@ <artifactId>seatunnel-api</artifactId> <version>${project.version}</version> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-format-json</artifactId> + <version>${project.version}</version> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>seatunnel-engine-client</artifactId> diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/DebeziumJsonFormatBenchmark.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/DebeziumJsonFormatBenchmark.java new file mode 100644 index 0000000000..daa0700b07 --- /dev/null +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/DebeziumJsonFormatBenchmark.java @@ -0,0 +1,304 @@ +/* + * 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.seatunnel.benchmark; + +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.LocalTimeType; +import org.apache.seatunnel.api.table.type.MetadataUtil; +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.format.json.debezium.DebeziumJsonDeserializationSchema; +import org.apache.seatunnel.format.json.debezium.DebeziumJsonSerializationSchema; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; +import org.openjdk.jmh.runner.options.VerboseMode; + +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.Objects; + +/** + * Benchmarks Debezium JSON serialization and deserialization hot paths used by CDC pipelines. + * + * <p>Throughput scores count CDC events, not emitted {@link SeaTunnelRow} records. {@code + * deserializeUpdateEvent} consumes one UPDATE envelope that emits two rows, and {@code + * serializeMergedUpdateEvent} includes both {@code UPDATE_BEFORE} and {@code UPDATE_AFTER} calls. + * Do not compare INSERT and UPDATE scores as if they measured the same unit of work. + */ +public class DebeziumJsonFormatBenchmark extends BenchmarkBase { + + private static final String CATALOG_NAME = "default"; + private static final String DATABASE_NAME = "inventory"; + private static final String SCHEMA_NAME = "public"; + private static final String TABLE_NAME = "orders"; + + private static final long INSERT_EVENT_TIME = 1_589_355_606_100L; + private static final long UPDATE_EVENT_TIME = 1_589_361_987_936L; + + private static final LocalDateTime INSERT_TIMESTAMP = + LocalDateTime.ofInstant(Instant.ofEpochMilli(INSERT_EVENT_TIME), ZoneOffset.UTC); + private static final LocalDateTime UPDATE_TIMESTAMP = + LocalDateTime.ofInstant(Instant.ofEpochMilli(UPDATE_EVENT_TIME), ZoneOffset.UTC); + + private static final BigDecimal INSERT_AMOUNT = new BigDecimal("199.9900"); + private static final BigDecimal UPDATE_AMOUNT = new BigDecimal("249.5000"); + + private static final SeaTunnelRowType ROW_TYPE = + new SeaTunnelRowType( + new String[] {"id", "name", "enabled", "score", "amount", "updated_at"}, + new SeaTunnelDataType<?>[] { + BasicType.LONG_TYPE, + BasicType.STRING_TYPE, + BasicType.BOOLEAN_TYPE, + BasicType.DOUBLE_TYPE, + new DecimalType(20, 4), + LocalTimeType.LOCAL_DATE_TIME_TYPE + }); + + private static final String INSERT_EVENT_JSON = + "{\"before\":null," + + "\"after\":{\"id\":1001,\"name\":\"seatunnel-order\",\"enabled\":true," + + "\"score\":12.5,\"amount\":199.99,\"updated_at\":1589355606100}," + + "\"source\":{\"version\":\"1.9.7.Final\",\"connector\":\"mysql\"," + + "\"name\":\"dbserver1\",\"ts_ms\":1589355606100,\"snapshot\":\"false\"," + + "\"db\":\"inventory\",\"table\":\"orders\",\"server_id\":223344," + + "\"gtid\":null,\"file\":\"mysql-bin.000003\",\"pos\":154,\"row\":0," + + "\"thread\":7,\"query\":null}," + + "\"op\":\"c\",\"ts_ms\":1589355606100,\"transaction\":null}"; + + private static final String UPDATE_EVENT_JSON = + "{\"before\":{\"id\":1001,\"name\":\"seatunnel-order\",\"enabled\":true," + + "\"score\":12.5,\"amount\":199.99,\"updated_at\":1589355606100}," + + "\"after\":{\"id\":1001,\"name\":\"seatunnel-order\",\"enabled\":true," + + "\"score\":13.75,\"amount\":249.5,\"updated_at\":1589361987936}," + + "\"source\":{\"version\":\"1.9.7.Final\",\"connector\":\"mysql\"," + + "\"name\":\"dbserver1\",\"ts_ms\":1589361987936,\"snapshot\":\"false\"," + + "\"db\":\"inventory\",\"table\":\"orders\",\"server_id\":223344," + + "\"gtid\":null,\"file\":\"mysql-bin.000003\",\"pos\":4096,\"row\":0," + + "\"thread\":7,\"query\":null}," + + "\"op\":\"u\",\"ts_ms\":1589361987936,\"transaction\":null}"; + + private DebeziumJsonDeserializationSchema deserializer; + private DebeziumJsonSerializationSchema insertSerializer; + private DebeziumJsonSerializationSchema mergeUpdateSerializer; + private ReusableRowCollector collector; + private byte[] insertEventBytes; + private byte[] updateEventBytes; + private SeaTunnelRow insertRow; + private SeaTunnelRow updateBeforeRow; + private SeaTunnelRow updateAfterRow; + + public static void main(String[] args) throws RunnerException { + Options options = + new OptionsBuilder() + .verbosity(VerboseMode.NORMAL) + .include(".*" + DebeziumJsonFormatBenchmark.class.getCanonicalName() + ".*") + .build(); + new Runner(options).run(); + } + + @Setup + public void setUp() { + CatalogTable catalogTable = + CatalogTableUtil.getCatalogTable( + CATALOG_NAME, DATABASE_NAME, SCHEMA_NAME, TABLE_NAME, ROW_TYPE); + String tableId = catalogTable.getTablePath().toString(); + + deserializer = new DebeziumJsonDeserializationSchema(catalogTable, false, false); + insertSerializer = new DebeziumJsonSerializationSchema(ROW_TYPE); + mergeUpdateSerializer = + new DebeziumJsonSerializationSchema(ROW_TYPE, StandardCharsets.UTF_8, true); + + insertEventBytes = INSERT_EVENT_JSON.getBytes(StandardCharsets.UTF_8); + updateEventBytes = UPDATE_EVENT_JSON.getBytes(StandardCharsets.UTF_8); + + insertRow = newRow(1001L, "seatunnel-order", true, 12.5D, INSERT_AMOUNT, INSERT_TIMESTAMP); + insertRow.setTableId(tableId); + insertRow.setRowKind(RowKind.INSERT); + MetadataUtil.setEventTime(insertRow, INSERT_EVENT_TIME); + + updateBeforeRow = + newRow(1001L, "seatunnel-order", true, 12.5D, INSERT_AMOUNT, INSERT_TIMESTAMP); + updateBeforeRow.setTableId(tableId); + updateBeforeRow.setRowKind(RowKind.UPDATE_BEFORE); + MetadataUtil.setEventTime(updateBeforeRow, UPDATE_EVENT_TIME); + + updateAfterRow = + newRow(1001L, "seatunnel-order", true, 13.75D, UPDATE_AMOUNT, UPDATE_TIMESTAMP); + updateAfterRow.setTableId(tableId); + updateAfterRow.setRowKind(RowKind.UPDATE_AFTER); + MetadataUtil.setEventTime(updateAfterRow, UPDATE_EVENT_TIME); + + collector = new ReusableRowCollector(); + validateFixtures(); + collector.reset(); + } + + /** + * Deserialize one schema-less Debezium CREATE/INSERT event and consume the single emitted row. + */ + @Benchmark + public SeaTunnelRow deserializeInsertEvent() { + collector.reset(); + deserializer.deserialize(insertEventBytes, collector); + return collector.row0; + } + + /** + * Deserialize one schema-less Debezium UPDATE event. One invocation is one CDC event even + * though the deserializer emits {@code UPDATE_BEFORE} followed by {@code UPDATE_AFTER}. + */ + @Benchmark + public SeaTunnelRow deserializeUpdateEvent() { + collector.reset(); + deserializer.deserialize(updateEventBytes, collector); + SeaTunnelRow before = collector.row0; + SeaTunnelRow after = collector.row1; + return before.getArity() == after.getArity() ? after : before; + } + + /** Serialize one {@link RowKind#INSERT} row into a Debezium JSON envelope. */ + @Benchmark + public byte[] serializeInsertEvent() { + return insertSerializer.serialize(insertRow); + } + + /** + * Serialize one logical UPDATE by calling the merge-enabled serializer with {@code + * UPDATE_BEFORE} then {@code UPDATE_AFTER}. One invocation is one CDC event, not two. + */ + @Benchmark + public byte[] serializeMergedUpdateEvent() { + byte[] cachedBefore = mergeUpdateSerializer.serialize(updateBeforeRow); + byte[] updateEvent = mergeUpdateSerializer.serialize(updateAfterRow); + return cachedBefore == null ? updateEvent : cachedBefore; + } + + private void validateFixtures() { + collector.reset(); + deserializer.deserialize(insertEventBytes, collector); + if (collector.count != 1) { + throw new IllegalStateException( + "INSERT deserialization must emit exactly one row, but emitted " + + collector.count); + } + if (collector.row0.getRowKind() != RowKind.INSERT) { + throw new IllegalStateException( + "INSERT deserialization must emit RowKind.INSERT, but emitted " + + collector.row0.getRowKind()); + } + + collector.reset(); + deserializer.deserialize(updateEventBytes, collector); + if (collector.count != 2) { + throw new IllegalStateException( + "UPDATE deserialization must emit two rows, but emitted " + collector.count); + } + if (collector.row0.getRowKind() != RowKind.UPDATE_BEFORE + || collector.row1.getRowKind() != RowKind.UPDATE_AFTER) { + throw new IllegalStateException( + "UPDATE deserialization must emit UPDATE_BEFORE then UPDATE_AFTER, but emitted " + + collector.row0.getRowKind() + + " then " + + collector.row1.getRowKind()); + } + if (Objects.equals(collector.row0.getField(3), collector.row1.getField(3))) { + throw new IllegalStateException( + "UPDATE before and after payloads must differ on at least one field"); + } + + byte[] insertJson = insertSerializer.serialize(insertRow); + String insertEnvelope = new String(insertJson, StandardCharsets.UTF_8); + if (!insertEnvelope.contains("\"op\":\"c\"") || !insertEnvelope.contains("\"after\":{")) { + throw new IllegalStateException( + "INSERT serialization must produce a Debezium CREATE envelope: " + + insertEnvelope); + } + + byte[] cachedBefore = mergeUpdateSerializer.serialize(updateBeforeRow); + if (cachedBefore != null) { + throw new IllegalStateException( + "Merged UPDATE serialization must cache UPDATE_BEFORE and return null"); + } + byte[] updateJson = mergeUpdateSerializer.serialize(updateAfterRow); + if (updateJson == null) { + throw new IllegalStateException( + "Merged UPDATE serialization must emit an envelope for UPDATE_AFTER"); + } + String updateEnvelope = new String(updateJson, StandardCharsets.UTF_8); + if (!updateEnvelope.contains("\"op\":\"u\"") + || !updateEnvelope.contains("\"before\":{") + || !updateEnvelope.contains("\"after\":{")) { + throw new IllegalStateException( + "Merged UPDATE serialization must contain before and after payloads: " + + updateEnvelope); + } + } + + private static SeaTunnelRow newRow( + long id, + String name, + boolean enabled, + double score, + BigDecimal amount, + LocalDateTime updatedAt) { + return new SeaTunnelRow(new Object[] {id, name, enabled, score, amount, updatedAt}); + } + + private static final class ReusableRowCollector implements Collector<SeaTunnelRow> { + + private SeaTunnelRow row0; + private SeaTunnelRow row1; + private int count; + + @Override + public void collect(SeaTunnelRow record) { + if (count == 0) { + row0 = record; + } else if (count == 1) { + row1 = record; + } + count++; + } + + @Override + public Object getCheckpointLock() { + return null; + } + + private void reset() { + count = 0; + row0 = null; + row1 = null; + } + } +} diff --git a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/DebeziumJsonFormatBenchmarkTest.java b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/DebeziumJsonFormatBenchmarkTest.java new file mode 100644 index 0000000000..62f7f46416 --- /dev/null +++ b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/DebeziumJsonFormatBenchmarkTest.java @@ -0,0 +1,74 @@ +/* + * 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.seatunnel.benchmark; + +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; + +import org.junit.jupiter.api.Test; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.State; + +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class DebeziumJsonFormatBenchmarkTest { + + @Test + void shouldKeepThreadScopedMutableSerializerState() { + assertEquals( + Scope.Thread, DebeziumJsonFormatBenchmark.class.getAnnotation(State.class).value()); + } + + @Test + void shouldRunDebeziumJsonFormatBenchmarkMethods() { + DebeziumJsonFormatBenchmark benchmark = new DebeziumJsonFormatBenchmark(); + benchmark.setUp(); + + SeaTunnelRow insertRow = benchmark.deserializeInsertEvent(); + SeaTunnelRow updateAfterRow = benchmark.deserializeUpdateEvent(); + byte[] insertJson = benchmark.serializeInsertEvent(); + byte[] updateJson = benchmark.serializeMergedUpdateEvent(); + + assertEquals(RowKind.INSERT, insertRow.getRowKind()); + assertEquals(6, insertRow.getArity()); + assertEquals(1001L, insertRow.getField(0)); + assertEquals("seatunnel-order", insertRow.getField(1)); + assertEquals(RowKind.UPDATE_AFTER, updateAfterRow.getRowKind()); + assertEquals(6, updateAfterRow.getArity()); + assertEquals(13.75D, updateAfterRow.getField(3)); + assertNotNull(insertJson); + assertNotNull(updateJson); + + String insertEnvelope = new String(insertJson, StandardCharsets.UTF_8); + String updateEnvelope = new String(updateJson, StandardCharsets.UTF_8); + assertTrue(insertEnvelope.contains("\"op\":\"c\"")); + assertTrue(insertEnvelope.contains("\"after\":{")); + assertTrue(insertEnvelope.contains("\"before\":null")); + assertTrue(updateEnvelope.contains("\"op\":\"u\"")); + assertTrue(updateEnvelope.contains("\"before\":{")); + assertTrue(updateEnvelope.contains("\"after\":{")); + assertTrue(updateEnvelope.contains("12.5")); + assertTrue(updateEnvelope.contains("13.75")); + assertTrue(updateEnvelope.contains("199.99")); + assertTrue(updateEnvelope.contains("249.5")); + } +} diff --git a/tools/benchmarks/suites/benchmarks_core.txt b/tools/benchmarks/suites/benchmarks_core.txt index 5d847e7138..cbdba400e0 100644 --- a/tools/benchmarks/suites/benchmarks_core.txt +++ b/tools/benchmarks/suites/benchmarks_core.txt @@ -16,6 +16,7 @@ # Fundamental data-path operations. SeaTunnelRowBenchmark IntermediateQueueBenchmark +DebeziumJsonFormatBenchmark SeaTunnelPipelineBenchmark.sourceSink$ SeaTunnelPipelineBenchmark.sourceTransformSink$
