Pipe: optimize InsertRows schema aggregation

[Caideyipi]

Description

Problem

An async-profiler CPU profile of an insertRecords workload with Pipe enabled showed that synchronous Pipe event construction dominated the sampled write path:

• PipeInsertionDataNodeListener.listenToInsertNode accounted for 1725 of 3614 total samples (47.73%).
• This was about 95.2% of the sampled insertRecords request branch.
• Most self time was under Stream.distinct(), especially HashMap.resize and HashMap.putVal.

For every additional row of the same device, the previous implementation rebuilt a Stream, a distinct set, and a result array from the accumulated and new measurements. This causes heavy temporary allocation for identical schemas and repeated scans when the measurement union grows.

Changes

• Keep the original measurement array for devices that occur only once.
• Create one LinkedHashSet only when a device repeats, preserving encounter order.
• Reuse that set for subsequent heterogeneous schemas and materialize the result array once.
• Skip set probes when a later schema equals the first schema. For a shorter schema, skip the longest leading portion already present as an ordered subsequence of the first schema and merge only the unmatched suffix.
• Preserve the previous null-array failure behavior and support null measurement elements used by partial inserts.
• Add unit tests covering identical and heterogeneous schemas, shorter ordered subsequences, equal-but-distinct String instances, encounter order, multiple devices, duplicate measurements, null elements, and null measurement arrays.
• Add a manually enabled performance UT comparing the legacy and optimized aggregation implementations.

All aggregation containers remain method-local; shared-state and concurrency behavior are unchanged.

Performance UT

TsFileEpochManagerPerformanceTest compares the legacy aggregation implementation from base commit 7ffb1364b84 with the optimized production implementation. It is skipped by default and contains no timing assertion.

Default behavior, disabled/skipped:

mvn -pl iotdb-core/datanode -am -Dtest=TsFileEpochManagerPerformanceTest -Dsurefire.failIfNoSpecifiedTests=false test

Manual performance run:

mvn -pl iotdb-core/datanode -am -Dtest=TsFileEpochManagerPerformanceTest -Dsurefire.failIfNoSpecifiedTests=false -Diotdb.pipe.epoch.insert.rows.aggregation.perf.enabled=true -Diotdb.pipe.epoch.insert.rows.aggregation.perf.rows=10000 -Diotdb.pipe.epoch.insert.rows.aggregation.perf.measurements=100 -Diotdb.pipe.epoch.insert.rows.aggregation.perf.warmup.iterations=5 -Diotdb.pipe.epoch.insert.rows.aggregation.perf.iterations=10 test

At head commit d3cf02ee1b4, the test was rerun through the isolated Java 17 JUnitCore harness in three independent JVMs with a 1 GiB maximum heap. Each result is the median of 10 measured iterations after 5 warmup iterations:

run 1: legacy median=28.763 ms/op, optimized median=7.671 ms/op, speedup=3.75x
run 2: legacy median=28.067 ms/op, optimized median=7.234 ms/op, speedup=3.88x
run 3: legacy median=30.654 ms/op, optimized median=8.173 ms/op, speedup=3.75x

Workload: 10,000 rows for one device. Every row has the same content-equivalent schema, with 100 independently decoded measurement names per row. This benchmark exercises the identical-schema path; it does not directly measure the shorter-subsequence fast path.

Environment: Java 17.0.15 Oracle HotSpot, Windows NT 10.0.26200.0 x64, 13th Gen Intel Core i9-13900H, benchmark JVM -Xmx1024m.

This is a method-level steady-state microbenchmark of schema aggregation, not an end-to-end write-throughput benchmark.

Validation

• mvn spotless:apply -pl iotdb-core/datanode
• DataNode Checkstyle: 0 violations
• git diff --check
• Java 17 isolated compilation of the changed production and test classes
• TsFileEpochManagerTest: 3 tests passed
• TsFileEpochManagerTest plus the default-skipped TsFileEpochManagerPerformanceTest: OK (4 tests)
• Three manually enabled performance runs completed successfully at d3cf02ee1b4; results are shown above.

The full Maven test path could not reach these tests locally because the workspace has stale cross-module/generated artifacts, and the reactor fallback is blocked while compiling generated Thrift code by a missing javax.annotation class. The isolated Java 17 runs above verify the changed classes; CI provides clean-reactor verification.

---

This PR has:

• been self-reviewed.
• added comments explaining the performance-sensitive design.
• added unit tests for the new aggregation path.

---

Key changed/added classes

• TsFileEpochManager
• TsFileEpochManagerTest
• TsFileEpochManagerPerformanceTest

[jt2594838]

measurements may be a subsequence of firstMeasurements.
To avoid adding measurements and double-iteration in this case,
may find the first position where firstMeasurements[i] == measurements[j].
Then, from j to measurements.length, if firstMeasurements[i + n] != measurements[j +n], add measurements[j +n].

Still, this cannot avoid all unnecessary additions. Maybe you can find some data structure that is more efficient for this case.

[Caideyipi]

Thanks. Addressed in d3cf02e. For shorter schemas, the aggregation now uses Objects.equals to match the longest prefix of measurements that is an ordered subsequence of firstMeasurements, and only sends the unmatched suffix through the LinkedHashSet. This avoids redundant set probes for true subsequences while retaining LinkedHashSet as the general fallback for reordered, disjoint, or newly added measurements and preserving the legacy encounter order, duplicates, and null handling. I also added regression coverage for a non-contiguous subsequence with equal-but-distinct String instances, an interleaved new measurement, and the null-measurement-array behavior.