Caideyipi commented on code in PR #18195:
URL: https://github.com/apache/iotdb/pull/18195#discussion_r3575754750
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java:
##########
@@ -84,17 +86,34 @@ public PipeRealtimeEvent
bindPipeInsertNodeTabletInsertionEvent(
: Collections.singletonMap(node.getDeviceID(),
node.getMeasurements()));
}
- private Map<IDeviceID, String[]> getDevice2MeasurementsMapFromInsertRowsNode(
+ static Map<IDeviceID, String[]> getDevice2MeasurementsMapFromInsertRowsNode(
InsertRowsNode insertRowsNode) {
- return insertRowsNode.getInsertRowNodeList().stream()
- .collect(
- Collectors.toMap(
- InsertNode::getDeviceID,
- InsertNode::getMeasurements,
- (oldMeasurements, newMeasurements) ->
- Stream.of(Arrays.asList(oldMeasurements),
Arrays.asList(newMeasurements))
- .flatMap(Collection::stream)
- .distinct()
- .toArray(String[]::new)));
+ final Map<IDeviceID, String[]> device2Measurements = new HashMap<>();
+ final Map<IDeviceID, Set<String>> device2DistinctMeasurements = new
HashMap<>();
+
+ // This method runs synchronously in the write path. Rebuilding a stream,
a hash set, and an
+ // array for every row is expensive when an InsertRowsNode contains many
rows for one device.
+ // Keep one set per repeated device instead and materialize its array only
once.
+ for (final InsertNode insertRowNode :
insertRowsNode.getInsertRowNodeList()) {
+ final IDeviceID deviceID = insertRowNode.getDeviceID();
+ final String[] measurements =
Objects.requireNonNull(insertRowNode.getMeasurements());
+ final String[] firstMeasurements =
device2Measurements.putIfAbsent(deviceID, measurements);
+ if (firstMeasurements == null) {
+ continue;
+ }
+
+ final Set<String> distinctMeasurements =
+ device2DistinctMeasurements.computeIfAbsent(
+ deviceID, key -> new
LinkedHashSet<>(Arrays.asList(firstMeasurements)));
+ if (!Arrays.equals(firstMeasurements, measurements)) {
+ Collections.addAll(distinctMeasurements, measurements);
+ }
Review Comment:
Thanks. Addressed in d3cf02ee1b4. 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.
--
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]
