You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/10/12 07:51:46 UTC
(iotdb) branch master updated: Pipe: register aligned timeseries for each device id with schemas exactly once (#13715)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new afd57ba26c1 Pipe: register aligned timeseries for each device id with schemas exactly once (#13715)
afd57ba26c1 is described below
commit afd57ba26c163996667638180b4fc4f7af3df065
Author: V_Galaxy <vg...@apache.org>
AuthorDate: Sat Oct 12 15:51:39 2024 +0800
Pipe: register aligned timeseries for each device id with schemas exactly once (#13715)
---
.../batch/PipeTabletEventTsFileBatch.java | 34 +++++++++++++++-------
1 file changed, 24 insertions(+), 10 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventTsFileBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
index f9ca81996cd..bf8bd522369 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/batch/PipeTabletEventTsFileBatch.java
@@ -53,6 +53,7 @@ import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -345,17 +346,30 @@ public class PipeTabletEventTsFileBatch extends PipeTabletEventBatch {
}
final boolean isAligned = device2Aligned.get(deviceId);
- for (final Tablet tablet : tabletsToWrite) {
- if (isAligned) {
- try {
- fileWriter.registerAlignedTimeseries(
- new Path(tablet.getDeviceId()), tablet.getSchemas());
- } catch (final WriteProcessException ignore) {
- // Do nothing if the timeSeries has been registered
- }
-
+ if (isAligned) {
+ final Map<String, List<IMeasurementSchema>> deviceId2MeasurementSchemas = new HashMap<>();
+ tabletsToWrite.forEach(
+ tablet ->
+ deviceId2MeasurementSchemas.compute(
+ tablet.getDeviceId(),
+ (k, v) -> {
+ if (Objects.isNull(v)) {
+ return new ArrayList<>(tablet.getSchemas());
+ }
+ v.addAll(tablet.getSchemas());
+ return v;
+ }));
+ for (final Entry<String, List<IMeasurementSchema>> deviceIdWithMeasurementSchemas :
+ deviceId2MeasurementSchemas.entrySet()) {
+ fileWriter.registerAlignedTimeseries(
+ new Path(deviceIdWithMeasurementSchemas.getKey()),
+ deviceIdWithMeasurementSchemas.getValue());
+ }
+ for (final Tablet tablet : tabletsToWrite) {
fileWriter.writeAligned(tablet);
- } else {
+ }
+ } else {
+ for (final Tablet tablet : tabletsToWrite) {
for (final IMeasurementSchema schema : tablet.getSchemas()) {
try {
fileWriter.registerTimeseries(new Path(tablet.getDeviceId()), schema);