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-12177-c2bb21809472e0874c8594e3f31b3c441fc6dc19 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 10ed8d433f07b01591d1c8cc699a8d7b7fe0e2a9 Author: zhiweiniu <[email protected]> AuthorDate: Tue Sep 8 12:15:49 2026 +0000 [Test][Zeta] Add ProtoStuff serializer microbenchmarks (#12177) --- .github/workflows/benchmarks.yml | 1 + docs/en/engines/zeta/benchmark.md | 11 +++ docs/zh/engines/zeta/benchmark.md | 10 +++ seatunnel-benchmarks/README.md | 6 ++ .../benchmark/ProtoStuffSerializerBenchmark.java | 97 ++++++++++++++++++++ .../ProtoStuffSerializerBenchmarkTest.java | 100 +++++++++++++++++++++ 6 files changed, 225 insertions(+) diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index dcbd8fa654..62721a1476 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -37,6 +37,7 @@ on: - 'SeaTunnelRowBenchmark' - 'IntermediateQueueBenchmark' - 'DebeziumJsonFormatBenchmark' + - 'ProtoStuffSerializerBenchmark' - 'SeaTunnelPipelineBenchmark' - 'CheckpointingTimeBenchmark' - 'CheckpointStorageBenchmark' diff --git a/docs/en/engines/zeta/benchmark.md b/docs/en/engines/zeta/benchmark.md index ed760947cf..3400737f67 100644 --- a/docs/en/engines/zeta/benchmark.md +++ b/docs/en/engines/zeta/benchmark.md @@ -178,6 +178,17 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark \ For a quick functional validation, add `-f 1 -wi 0 -i 1 -r 1s` to shorten the run. A single un-warmed sample is not valid performance evidence. +### Run the ProtoStuff Serializer Microbenchmarks + +```bash +java -jar seatunnel-benchmarks/target/benchmarks.jar ProtoStuffSerializerBenchmark \ + -rf json -rff seatunnel-benchmarks/target/protostuff.json +``` + +Measures in-memory `IMapFileData` serialization and deserialization throughput (`ops/ms`) with four +threads by default. Both methods exercise shared schema-cache lookup, excluding file I/O and +Hazelcast scheduling. + ### Run the Checkpoint Benchmark ```bash diff --git a/docs/zh/engines/zeta/benchmark.md b/docs/zh/engines/zeta/benchmark.md index 754e20f96e..d30019eae6 100644 --- a/docs/zh/engines/zeta/benchmark.md +++ b/docs/zh/engines/zeta/benchmark.md @@ -169,6 +169,16 @@ java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark \ 快速功能验证时可以增加 `-f 1 -wi 0 -i 1 -r 1s` 缩短运行时间。没有预热且只有一个样本的 结果不能用于性能结论。 +### 运行 ProtoStuff 序列化微基准 + +```bash +java -jar seatunnel-benchmarks/target/benchmarks.jar ProtoStuffSerializerBenchmark \ + -rf json -rff seatunnel-benchmarks/target/protostuff.json +``` + +默认使用 4 个线程，测量 `IMapFileData` 的内存序列化和反序列化吞吐（`ops/ms`）。 +两个方法均覆盖共享 Schema 缓存的查询路径，不包含文件 I/O 和 Hazelcast 调度。 + ### 运行 Checkpoint 基准测试 ```bash diff --git a/seatunnel-benchmarks/README.md b/seatunnel-benchmarks/README.md index 10b95b7e90..b673cb9ec1 100644 --- a/seatunnel-benchmarks/README.md +++ b/seatunnel-benchmarks/README.md @@ -41,6 +41,12 @@ Run only the `SeaTunnelRow` benchmarks: java -jar seatunnel-benchmarks/target/benchmarks.jar SeaTunnelRowBenchmark ``` +Run only the `ProtoStuffSerializer` benchmarks: + +```bash +java -jar seatunnel-benchmarks/target/benchmarks.jar ProtoStuffSerializerBenchmark +``` + Write JSON results: ```bash diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/ProtoStuffSerializerBenchmark.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/ProtoStuffSerializerBenchmark.java new file mode 100644 index 0000000000..8cbfd99b5c --- /dev/null +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/ProtoStuffSerializerBenchmark.java @@ -0,0 +1,97 @@ +/* + * 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.engine.imap.storage.file.bean.IMapFileData; +import org.apache.seatunnel.engine.serializer.protobuf.ProtoStuffSerializer; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.Threads; + +import java.util.concurrent.TimeUnit; + +/** + * Pure in-memory microbenchmarks for the steady-state ProtoStuff WAL record codec. + * + * <p>One operation encodes or decodes one IMapFileData envelope, not its nested key/value objects. + * No filesystem, Hazelcast, or Zeta runtime is started. Thread-local fixtures share the production + * serializer's static schema cache without sharing mutable input objects. Four threads are used by + * default to exercise concurrent lookup on the standard four-vCPU runner. IMapFileData is not a + * serializer wrapper type: both measured methods call getSchema on every invocation, even after + * setup initializes the schema. + */ +@Threads(4) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Fork( + value = 3, + jvmArgsAppend = { + "-Xms4g", + "-Xmx4g", + "-XX:+UseG1GC", + "-XX:+AlwaysPreTouch", + "-XX:+DisableExplicitGC", + "-XX:ActiveProcessorCount=4", + "-Djava.net.preferIPv4Stack=true" + }) +public class ProtoStuffSerializerBenchmark extends BenchmarkBase { + + private ProtoStuffSerializer serializer; + private IMapFileData record; + private byte[] serializedRecord; + + /** Prepare a fixed WAL record and initialize its schema outside the measured methods. */ + @Setup + public void setUp() { + serializer = new ProtoStuffSerializer(); + char[] payload = new char[1024]; + for (int index = 0; index < payload.length; index++) { + payload[index] = (char) ('a' + index % 26); + } + record = + IMapFileData.builder() + .deleted(false) + .key(serializer.serialize(1001L)) + .keyClassName(Long.class.getName()) + .value(serializer.serialize(new String(payload))) + .valueClassName(String.class.getName()) + .timestamp(1_700_000_000_000L) + .build(); + serializedRecord = serializer.serialize(record); + if (!record.equals(serializer.deserialize(serializedRecord, IMapFileData.class))) { + throw new IllegalStateException("ProtoStuff WAL record fixture failed to round-trip"); + } + } + + /** Encode one WAL envelope, including the serializer's normal buffer/output allocations. */ + @Benchmark + public byte[] serializeWalRecord() { + return serializer.serialize(record); + } + + /** Decode one prepared WAL envelope into a newly allocated object. */ + @Benchmark + public IMapFileData deserializeWalRecord() { + return serializer.deserialize(serializedRecord, IMapFileData.class); + } +} diff --git a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/ProtoStuffSerializerBenchmarkTest.java b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/ProtoStuffSerializerBenchmarkTest.java new file mode 100644 index 0000000000..a11d4d4d30 --- /dev/null +++ b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/ProtoStuffSerializerBenchmarkTest.java @@ -0,0 +1,100 @@ +/* + * 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.engine.imap.storage.file.bean.IMapFileData; +import org.apache.seatunnel.engine.serializer.protobuf.ProtoStuffSerializer; + +import org.junit.jupiter.api.Test; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Threads; + +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotSame; + +class ProtoStuffSerializerBenchmarkTest { + + @Test + void shouldExerciseConcurrentSchemaLookupByDefault() { + assertEquals(4, ProtoStuffSerializerBenchmark.class.getAnnotation(Threads.class).value()); + assertArrayEquals( + new Mode[] {Mode.Throughput}, + ProtoStuffSerializerBenchmark.class.getAnnotation(BenchmarkMode.class).value()); + assertEquals( + TimeUnit.MILLISECONDS, + ProtoStuffSerializerBenchmark.class.getAnnotation(OutputTimeUnit.class).value()); + Fork fork = ProtoStuffSerializerBenchmark.class.getAnnotation(Fork.class); + assertEquals(3, fork.value()); + assertArrayEquals( + SeaTunnelPipelineBenchmark.class.getAnnotation(Fork.class).jvmArgsAppend(), + fork.jvmArgsAppend()); + } + + @Test + void shouldRoundTripTheWalEnvelopeAndItsPreparedPayload() { + ProtoStuffSerializerBenchmark benchmark = new ProtoStuffSerializerBenchmark(); + benchmark.setUp(); + ProtoStuffSerializer serializer = new ProtoStuffSerializer(); + + IMapFileData record = benchmark.deserializeWalRecord(); + assertFalse(record.isDeleted()); + assertEquals(Long.class.getName(), record.getKeyClassName()); + assertEquals(String.class.getName(), record.getValueClassName()); + assertEquals(1_700_000_000_000L, record.getTimestamp()); + assertEquals(Long.valueOf(1001L), serializer.deserialize(record.getKey(), Long.class)); + String payload = serializer.deserialize(record.getValue(), String.class); + assertEquals(1024, payload.length()); + for (int index = 0; index < payload.length(); index++) { + assertEquals((char) ('a' + index % 26), payload.charAt(index)); + } + for (int iteration = 0; iteration < 10; iteration++) { + IMapFileData decoded = + serializer.deserialize(benchmark.serializeWalRecord(), IMapFileData.class); + assertEquals(record, decoded); + assertEquals(record, benchmark.deserializeWalRecord()); + } + } + + @Test + void shouldReturnFreshResultsWithoutMutatingTheReusableInputs() { + ProtoStuffSerializerBenchmark benchmark = new ProtoStuffSerializerBenchmark(); + benchmark.setUp(); + + byte[] expectedBytes = benchmark.serializeWalRecord(); + byte[] output = benchmark.serializeWalRecord(); + assertNotSame(expectedBytes, output); + output[0] ^= 1; + assertArrayEquals(expectedBytes, benchmark.serializeWalRecord()); + + IMapFileData expected = benchmark.deserializeWalRecord(); + IMapFileData decoded = benchmark.deserializeWalRecord(); + assertNotSame(expected, decoded); + decoded.getKey()[0] ^= 1; + decoded.getValue()[0] ^= 1; + decoded.setTimestamp(0); + assertEquals(expected, benchmark.deserializeWalRecord()); + assertArrayEquals(expectedBytes, benchmark.serializeWalRecord()); + } +}
