jt2594838 commented on code in PR #18195:
URL: https://github.com/apache/iotdb/pull/18195#discussion_r3570063329
##########
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:
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.
--
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]
