goutamadwant opened a new pull request, #13026:
URL: https://github.com/apache/seatunnel/pull/13026
<!--
Thank you for contributing to SeaTunnel! Please make sure that your code
changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
## Contribution Checklist
- Make sure that the pull request corresponds to a [GITHUB
issue](https://github.com/apache/seatunnel/issues).
- Name the pull request in the form "[Feature] [component] Title of the
pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
- Minor fixes should be named following this pattern: `[hotfix] [docs] Fix
typo in README.md doc`.
-->
### Purpose of this pull request
<!-- Describe the purpose of this pull request. For example: This pull
request adds checkstyle plugin.-->
Improve SeaTunnel nested Avro serialization by reusing record schemas
instead of rebuilding them for every ROW, including ROW elements in ARRAYs and
MAPs.
Fixes #12940. The nullable-ROW prerequisite #12900 is merged.
- Cache schemas within each converter by configured ROW type identity and
effective record name. Row values and map entry keys cannot grow the cache.
- Preserve legacy `SeaTunnelRecord` names and the nullable layout's
enclosing-field names. Different ROW structures with the same leaf name remain
separate.
- Build fresh records and values on every serialization. No new dependencies
or global cache.
Reusing enclosing schema branches directly would change legacy nested
GenericRecord names. Caching the existing schema-builder results preserves that
behavior in both layouts; type identity also avoids recursive structural
hashing on each ROW.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as
the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes
- provide the console output, description and/or an example to show the
behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to
the released SeaTunnel versions or within the unreleased branches such as dev.
If no, write 'No'.
If you are adding/modifying connector documents, please follow our new
specifications: https://github.com/apache/seatunnel/issues/4544.
-->
No functional, configuration or serialized-data changes. Existing default
bytes, record names, nullable unions and option defaults are preserved. This is
an internal allocation optimization.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some
test cases that check the changes thoroughly including negative and positive
cases if possible.
If it was tested in a way different from regular unit tests, please clarify
how you tested step by step, ideally copy and paste-able, so that other
reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why
it was difficult to add.
If you are adding E2E test cases, maybe refer to
https://github.com/apache/seatunnel/blob/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql.conf,
here is a good example.
-->
- Avro module verification on Java 11 and Java 17: 24 tests per JDK, zero
failures/errors/skips.
- Extended regressions for repeated ROW/ARRAY/MAP conversion, same-named
rows with different structures, one type under multiple names, changed values
after cache warm-up, and present/null transitions.
- Refreshed current-source scoped reactor packaging and Spotless checks
passed.
- Official Zeta Java 11 broker E2E: Kafka 3/3 and Pulsar 3/3 passed, zero
failures/skips/aborts. Both cover nullable nested-row round trip plus the
existing default Avro write/read methods. Runtime connector/starter hashes
matched the refreshed packages.
- The private Pulsar test launcher explicitly exits after successful JUnit
teardown; natural host-client thread shutdown is not claimed. This harness
adaptation is outside the source diff.
- Scoped independent correctness and security reviews completed with no
actionable findings.
Local serializer measurements used five alternating baseline/candidate JVM
pairs, 20,000 warm-up and 30,000 measured operations per fixture. Fixtures
cover flat rows, four nested ROW levels, and ARRAY/MAP containers with 32 ROW
values, in both nullable layouts. Allocation uses current-thread allocated
bytes; CPU uses current-thread CPU time. Every payload length and SHA-256
matched.
On native ARM64 Java 11, median allocation fell 92.0–96.6% and CPU time
56.8–83.0% for the nested/ARRAY/MAP fixtures. A longer flat control (500,000
warm-up / 1,000,000 measured operations, five pairs) remained 272 bytes/op; CPU
median differences were -0.43% and +0.89% for the legacy and nullable layouts.
These are diagnostic serializer measurements, not JMH scores or end-to-end
throughput results. The measurements were collected against baseline
`76e11cb80e36b4844ba6820d43e0522dbd912d3c`; the refreshed baseline is
`403efeee0b3806c77e55c38368abdd41fbf828bd`. Both baseline converter sources are
identical, and refreshed candidate class bytes match the measured candidate.
Java 17 checks used an x86 runtime under Rosetta; broker E2E uses Zeta on
native Java 11 with the pinned amd64 broker images under emulation. Hosted CI
and other engine/platform coverage remain separate.
<details>
<summary>Reproduce the diagnostic serializer measurements</summary>
Use the same source below on baseline
`403efeee0b3806c77e55c38368abdd41fbf828bd` and candidate
`5da03b810075c84974cc28a50f7109820602beb3`. Build the format module and
dependencies for each revision using the normal developer environment. Save the
source as `NestedAvroBenchmark.java` outside the checkout. Fixture setup and
output validation run outside the timed loop.
```bash
./mvnw -pl seatunnel-formats/seatunnel-format-avro -am install -DskipTests
-DskipIT
./mvnw -pl seatunnel-formats/seatunnel-format-avro
dependency:build-classpath -Dmdep.outputFile=/tmp/avro-deps.txt
avro_cp="seatunnel-formats/seatunnel-format-avro/target/classes:$(cat
/tmp/avro-deps.txt)"
javac -cp "$avro_cp" NestedAvroBenchmark.java
java -Xms256m -Xmx256m -XX:ActiveProcessorCount=2 -cp ".:$avro_cp"
NestedAvroBenchmark baseline
# Use the candidate checkout and label for the comparison.
# Run five fresh pairs, alternating baseline/candidate order.
```
```java
import com.sun.management.ThreadMXBean;
import java.lang.management.ManagementFactory;
import java.security.MessageDigest;
import java.util.*;
import org.apache.seatunnel.api.table.type.*;
import org.apache.seatunnel.format.avro.*;
/** Diagnostic workload using the public Avro serializer. */
public final class NestedAvroBenchmark {
private static volatile long blackhole;
private static final int WARMUP = 20000;
private static final int MEASURE = 30000;
private static final class Fixture {
final SeaTunnelRowType type;
final SeaTunnelRow row;
Fixture(SeaTunnelRowType t, SeaTunnelRow r) {type=t;row=r;}
}
private static Fixture leaf() {
return new Fixture(new SeaTunnelRowType(new
String[]{"id","name","enabled","amount"},
new
SeaTunnelDataType<?>[]{BasicType.INT_TYPE,BasicType.STRING_TYPE,BasicType.BOOLEAN_TYPE,BasicType.LONG_TYPE}),
new SeaTunnelRow(new
Object[]{123,"sample-nested-data",true,123456789L}));
}
private static Fixture fixture(String kind) {
Fixture f=leaf();
if(kind.equals("flat")) return f;
if(kind.equals("nested")) {
for(int depth=0;depth<4;depth++) {
f=new Fixture(new SeaTunnelRowType(new
String[]{"detail","sequence"},new
SeaTunnelDataType<?>[]{f.type,BasicType.INT_TYPE}),new SeaTunnelRow(new
Object[]{f.row,depth}));
}
return f;
}
if(kind.equals("array32")) {
SeaTunnelRow[] items=new
SeaTunnelRow[32];Arrays.fill(items,f.row);
return new Fixture(new SeaTunnelRowType(new
String[]{"items"},new SeaTunnelDataType<?>[]{ArrayType.of(f.type)}),new
SeaTunnelRow(new Object[]{items}));
}
Map<String,SeaTunnelRow> items=new LinkedHashMap<>();
for(int i=0;i<32;i++)items.put("item"+i,f.row);
return new Fixture(new SeaTunnelRowType(new String[]{"items"},new
SeaTunnelDataType<?>[]{new MapType<>(BasicType.STRING_TYPE,f.type)}),new
SeaTunnelRow(new Object[]{items}));
}
private static String hex(byte[] data) {StringBuilder s=new
StringBuilder();for(byte b:data)s.append(String.format("%02x",b));return
s.toString();}
public static void main(String[] args) throws Exception {
ThreadMXBean bean=(ThreadMXBean)ManagementFactory.getThreadMXBean();
bean.setThreadAllocatedMemoryEnabled(true);bean.setThreadCpuTimeEnabled(true);
long tid=Thread.currentThread().getId();
System.out.println("label,java,kind,nullable,iterations,bytes_per_op,cpu_ns_per_op,wall_ns_per_op,payload_bytes,sha256");
for(String kind:new String[]{"flat","nested","array32","map32"})
for(boolean nullable:new boolean[]{false,true}) {
Fixture fixture=fixture(kind);AvroSerializationSchema
serializer=new AvroSerializationSchema(fixture.type,nullable);
byte[] reference=serializer.serialize(fixture.row);
for(int
i=0;i<WARMUP;i++)blackhole+=serializer.serialize(fixture.row).length;
long
allocated=bean.getThreadAllocatedBytes(tid),cpu=bean.getCurrentThreadCpuTime(),start=System.nanoTime();
for(int i=0;i<MEASURE;i++) {byte[]
out=serializer.serialize(fixture.row);blackhole+=out.length;}
long
elapsed=System.nanoTime()-start,usedCpu=bean.getCurrentThreadCpuTime()-cpu,usedBytes=bean.getThreadAllocatedBytes(tid)-allocated;
System.out.printf(Locale.ROOT,"%s,%s,%s,%s,%d,%.2f,%.2f,%.2f,%d,%s%n",args[0],System.getProperty("java.version"),kind,nullable,MEASURE,(double)usedBytes/MEASURE,(double)usedCpu/MEASURE,(double)elapsed/MEASURE,reference.length,hex(MessageDigest.getInstance("SHA-256").digest(reference)));
}
}
}
```
</details>
### Check list
* [ ] If any new Jar binary package adding in your PR, please add License
Notice according
[New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
* [ ] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs
* [ ] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR.
* [ ] If you are contributing the connector code, please check that the
following files are updated:
1. Update
[plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties)
and add new connector information in it
2. Update the pom file of
[seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml)
3. Add ci label in
[label-scope-conf](https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml)
4. Add e2e testcase in
[seatunnel-e2e](https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/)
5. Update connector
[plugin_config](https://github.com/apache/seatunnel/blob/dev/config/plugin_config)
No new binary dependencies, connector registration changes, public API
changes or incompatible data formats are introduced.
--
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]
