[GitHub] [iotdb] Cpaulyz commented on a diff in pull request #6102: [IOTDB-3296] ext-pipe suport .mods file.

[GitBox <gi...@apache.org>]

Cpaulyz commented on code in PR #6102:
URL: https://github.com/apache/iotdb/pull/6102#discussion_r941166925


##########
server/src/main/java/org/apache/iotdb/db/sync/datasource/DeletionGroup.java:
##########
@@ -0,0 +1,274 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.iotdb.db.sync.datasource;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.TreeMap;
+
+/**
+ * This class provides below functions
+ *
+ * <p>1) Save many deletion time-intervals.
+ *
+ * <p>2) Merge overlap intervals to 1 interval.
+ *
+ * <p>3) Check whether 1 time-range's data has been deleted according to saved deletion
+ * time-intervals.
+ *
+ * <p>4) Check whether 1 time-point's data has been deleted according to saved deletion
+ * time-intervals.
+ *
+ * <p>5) For time-ascending batch data, provide better-performance method to check whether 1
+ * time-point's data has been deleted.
+ */
+public class DeletionGroup {
+  // TreeMap: StartTime => EndTime
+  private TreeMap<Long, Long> delIntervalMap;
+
+  public enum DeletedType {
+    NO_DELETED,
+    PARTIAL_DELETED,
+    FULL_DELETED
+  }
+
+  public static class TsInterval {
+    public long startTime;
+    public long endTime;
+
+    public TsInterval(long startTime, long endTime) {
+      this.startTime = startTime;
+      this.endTime = endTime;
+    }
+
+    public void set(long startTime, long endTime) {
+      this.startTime = startTime;
+      this.endTime = endTime;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if (!(obj instanceof TsInterval)) {
+        return false;
+      }
+
+      TsInterval tsInterval = (TsInterval) obj;
+      return (tsInterval.startTime == this.startTime) && (tsInterval.endTime == this.endTime);
+    }
+  }
+
+  public static class IntervalCursor {
+    Iterator<Map.Entry<Long, Long>> iter = null;
+    boolean subsequentNoDelete = false;
+    public long startTime;
+    public long endTime;
+
+    public void reset() {
+      iter = null;
+      subsequentNoDelete = false;
+    }
+  }
+
+  public DeletionGroup() {
+    delIntervalMap = new TreeMap<>();
+  }
+
+  /**
+   * Insert delete time interval data for very deletion.
+   *
+   * @param startTime
+   * @param endTime
+   */
+  public void addDelInterval(long startTime, long endTime) {
+    if (startTime > endTime) {
+      throw new IllegalArgumentException("addDelInterval(), error: startTime > endTime.");
+    }
+
+    // == pay attention, intervalMap's Entries are not overlap.
+    Map.Entry<Long, Long> startEntry = delIntervalMap.floorEntry(startTime);
+    Map.Entry<Long, Long> endEntry = delIntervalMap.floorEntry(endTime);
+
+    if ((startEntry != null) && (startTime <= startEntry.getValue())) {
+      startTime = startEntry.getKey();
+    }
+    if ((endEntry != null) && (endTime < endEntry.getValue())) {
+      endTime = endEntry.getValue();
+    }
+
+    // == find existing overlap entries and remove them
+    Map<Long, Long> overlapEntries = delIntervalMap.subMap(startTime, true, endTime, true);
+    Iterator<Map.Entry<Long, Long>> iter = overlapEntries.entrySet().iterator();
+    while (iter.hasNext()) {
+      iter.next();
+      iter.remove();
+    }
+    //    Long[] keyList = overlapEntries.keySet().toArray(new Long[0]);

Review Comment:
   delete it



##########
server/src/main/java/org/apache/iotdb/db/sync/datasource/DeletionGroup.java:
##########
@@ -0,0 +1,274 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.iotdb.db.sync.datasource;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.TreeMap;
+
+/**
+ * This class provides below functions
+ *
+ * <p>1) Save many deletion time-intervals.
+ *
+ * <p>2) Merge overlap intervals to 1 interval.
+ *
+ * <p>3) Check whether 1 time-range's data has been deleted according to saved deletion
+ * time-intervals.
+ *
+ * <p>4) Check whether 1 time-point's data has been deleted according to saved deletion
+ * time-intervals.
+ *
+ * <p>5) For time-ascending batch data, provide better-performance method to check whether 1
+ * time-point's data has been deleted.
+ */
+public class DeletionGroup {
+  // TreeMap: StartTime => EndTime
+  private TreeMap<Long, Long> delIntervalMap;
+
+  public enum DeletedType {
+    NO_DELETED,
+    PARTIAL_DELETED,
+    FULL_DELETED
+  }
+
+  public static class TsInterval {
+    public long startTime;
+    public long endTime;
+
+    public TsInterval(long startTime, long endTime) {
+      this.startTime = startTime;
+      this.endTime = endTime;
+    }
+
+    public void set(long startTime, long endTime) {
+      this.startTime = startTime;
+      this.endTime = endTime;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if (!(obj instanceof TsInterval)) {
+        return false;
+      }
+
+      TsInterval tsInterval = (TsInterval) obj;
+      return (tsInterval.startTime == this.startTime) && (tsInterval.endTime == this.endTime);
+    }
+  }
+
+  public static class IntervalCursor {
+    Iterator<Map.Entry<Long, Long>> iter = null;
+    boolean subsequentNoDelete = false;
+    public long startTime;
+    public long endTime;
+
+    public void reset() {
+      iter = null;
+      subsequentNoDelete = false;
+    }
+  }
+
+  public DeletionGroup() {
+    delIntervalMap = new TreeMap<>();
+  }
+
+  /**
+   * Insert delete time interval data for very deletion.
+   *
+   * @param startTime
+   * @param endTime
+   */
+  public void addDelInterval(long startTime, long endTime) {
+    if (startTime > endTime) {
+      throw new IllegalArgumentException("addDelInterval(), error: startTime > endTime.");
+    }
+
+    // == pay attention, intervalMap's Entries are not overlap.
+    Map.Entry<Long, Long> startEntry = delIntervalMap.floorEntry(startTime);
+    Map.Entry<Long, Long> endEntry = delIntervalMap.floorEntry(endTime);
+
+    if ((startEntry != null) && (startTime <= startEntry.getValue())) {
+      startTime = startEntry.getKey();
+    }
+    if ((endEntry != null) && (endTime < endEntry.getValue())) {
+      endTime = endEntry.getValue();
+    }
+
+    // == find existing overlap entries and remove them
+    Map<Long, Long> overlapEntries = delIntervalMap.subMap(startTime, true, endTime, true);
+    Iterator<Map.Entry<Long, Long>> iter = overlapEntries.entrySet().iterator();
+    while (iter.hasNext()) {
+      iter.next();
+      iter.remove();
+    }
+    //    Long[] keyList = overlapEntries.keySet().toArray(new Long[0]);
+    //    for (Long key : keyList) {
+    //      delIntervalMap.remove(key);
+    //    }
+
+    delIntervalMap.put(startTime, endTime); // add new deletion interval
+  }
+
+  /**
+   * If this object has no deletion data (i.e delIntervalMap is empty), return true
+   *
+   * @return
+   */
+  public boolean isEmpty() {
+    return delIntervalMap.isEmpty();
+  }
+
+  /**
+   * Check the deletion-state of the data-points of specific time range according to the info of
+   * .mods
+   *
+   * @param startTime - the start time of data set, inclusive
+   * @param endTime - the end time of data set, inclusive
+   * @return

Review Comment:
   Please add simple annotation for return value or `DeletedType`



##########
server/src/test/java/org/apache/iotdb/db/sync/datasource/PipeOpManagerTest.java:
##########
@@ -181,27 +241,160 @@ public void testPipSrcManager() throws IOException {
     pipeOpManager.appendDataSrc(sgName2, tsFileOpBlock2);
 
     long count1 = tsFileOpBlock1.getDataCount();
-    assertEquals(6, count1);
+    assertEquals(8, count1);
     for (int i = 0; i < count1; i++) {
       Operation operation = pipeOpManager.getOperation(sgName1, i, 8);
       System.out.println("=== data" + i + ": " + operation + ", "); //
       assertEquals("root1", operation.getStorageGroup());
     }
 
     Operation operation = pipeOpManager.getOperation(sgName1, 0, 18);
+    InsertOperation insertOperation = (InsertOperation) operation;
     System.out.println("+++ data10" + ": " + operation + ", ");
+    assertEquals(
+        "root.lemming.device1.sensor1", insertOperation.getDataList().get(0).left.toString());
 
     pipeOpManager.commitData(sgName1, count1 - 1);
-    operation = pipeOpManager.getOperation(sgName1, 7, 18);
+    operation = pipeOpManager.getOperation(sgName1, 9, 18);
     System.out.println("+++ data11" + ": " + operation + ", ");
     assertNull(operation);
 
     operation = pipeOpManager.getOperation(sgName2, 6, 18);
     System.out.println("+++ data12" + ": " + operation + ", ");
     assertEquals(4, operation.getDataCount());
 
-    InsertOperation insertOperation = (InsertOperation) operation;
+    insertOperation = (InsertOperation) operation;
+    assertEquals(
+        "root2.lemming.device3.sensor3", insertOperation.getDataList().get(0).left.toString());
     assertEquals(1617206403003L, insertOperation.getDataList().get(0).right.get(0).getTimestamp());
-    assertEquals("33", insertOperation.getDataList().get(0).right.get(0).getValue().toString());
+    assertEquals("333", insertOperation.getDataList().get(0).right.get(0).getValue().toString());
+  }
+
+  @Test // (timeout = 10_000L)
+  public void testOpManager_Mods() throws IOException {
+    PipeOpManager pipeOpManager = new PipeOpManager(null);
+
+    String sgName1 = "root1";
+    // String sgName2 = "root2";
+
+    TsFileOpBlock tsFileOpBlock1 = new TsFileOpBlock(sgName1, seqTsFileName1, seqModsFileName1, 1);
+    pipeOpManager.appendDataSrc(sgName1, tsFileOpBlock1);
+    TsFileOpBlock tsFileOpBlock2 =
+        new TsFileOpBlock(sgName1, unSeqTsFileName1, unSeqModsFileName1, 2);
+    pipeOpManager.appendDataSrc(sgName1, tsFileOpBlock2);
+
+    long count1 = tsFileOpBlock1.getDataCount();
+    assertEquals(8, count1);
+    for (int i = 0; i < 18; i++) {
+      Operation operation = pipeOpManager.getOperation(sgName1, i, 8);
+      assertEquals(sgName1, operation.getStorageGroup());
+    }
+
+    // == test batch data in TsFile1 + .mods
+    Operation operation = pipeOpManager.getOperation(sgName1, 0, 18);
+    assertEquals(8, operation.getDataCount());
+
+    InsertOperation insertOperation = (InsertOperation) operation;
+    int i = 0;
+    assertEquals(1617206403001L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("1.1", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 1;
+    assertEquals(1617206403001L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("12", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 2;
+    assertEquals(1617206403001L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("13", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 3;
+    assertEquals(1, insertOperation.getDataList().get(i).right.size());
+    assertNull(insertOperation.getDataList().get(i).right.get(0));
+
+    i = 4;
+    assertEquals(1, insertOperation.getDataList().get(i).right.size());
+    assertNull(insertOperation.getDataList().get(i).right.get(0));
+
+    i = 5;
+    assertEquals(1617206403003L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("32", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 6;
+    assertEquals(1, insertOperation.getDataList().get(i).right.size());
+    assertNull(insertOperation.getDataList().get(i).right.get(0));
+
+    i = 7;
+    assertEquals(1617206403004L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("42", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    // == test batch data in TsFile2 + mods
+    operation = pipeOpManager.getOperation(sgName1, 8, 18);
+    assertEquals(10, operation.getDataCount());
+
+    insertOperation = (InsertOperation) operation;
+    i = 0;
+    assertEquals(
+        "root2.lemming.device1.sensor1", insertOperation.getDataList().get(i).left.toString());
+    assertEquals(1, insertOperation.getDataList().get(i).right.size());
+    assertNull(insertOperation.getDataList().get(i).right.get(0));
+
+    i = 1;
+    assertEquals(
+        "root2.lemming.device1.sensor2", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403001L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("12", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 2;
+    assertEquals(
+        "root2.lemming.device1.sensor3", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403001L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("13", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 3;
+    assertEquals(
+        "root2.lemming.device2.sensor2", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1, insertOperation.getDataList().get(i).right.size());
+    assertNull(insertOperation.getDataList().get(i).right.get(0));
+
+    i = 4;
+    assertEquals(
+        "root2.lemming.device3.sensor1", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403003L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("33.1", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 5;
+    assertEquals(
+        "root2.lemming.device3.sensor2", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403003L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("332", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 6;
+    assertEquals(
+        "root2.lemming.device3.sensor3", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403003L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("333", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 7;
+    assertEquals(
+        "root2.lemming.device3.sensor1", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403004L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("44.1", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 8;
+    assertEquals(
+        "root2.lemming.device3.sensor2", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403004L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("442", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    i = 9;
+    assertEquals(
+        "root2.lemming.device3.sensor3", insertOperation.getDataList().get(i).left.getFullPath());
+    assertEquals(1617206403004L, insertOperation.getDataList().get(i).right.get(0).getTimestamp());
+    assertEquals("443", insertOperation.getDataList().get(i).right.get(0).getValue().toString());
+
+    // == test commit
+    //    pipeOpManager.commitData(sgName1, count1 - 5);

Review Comment:
   Uncomment these codes



##########
server/src/main/java/org/apache/iotdb/db/sync/datasource/DeletionGroup.java:
##########
@@ -0,0 +1,274 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.iotdb.db.sync.datasource;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.TreeMap;
+
+/**
+ * This class provides below functions
+ *
+ * <p>1) Save many deletion time-intervals.
+ *
+ * <p>2) Merge overlap intervals to 1 interval.
+ *
+ * <p>3) Check whether 1 time-range's data has been deleted according to saved deletion
+ * time-intervals.
+ *
+ * <p>4) Check whether 1 time-point's data has been deleted according to saved deletion
+ * time-intervals.
+ *
+ * <p>5) For time-ascending batch data, provide better-performance method to check whether 1
+ * time-point's data has been deleted.
+ */
+public class DeletionGroup {
+  // TreeMap: StartTime => EndTime
+  private TreeMap<Long, Long> delIntervalMap;
+
+  public enum DeletedType {
+    NO_DELETED,
+    PARTIAL_DELETED,
+    FULL_DELETED
+  }
+
+  public static class TsInterval {

Review Comment:
   It seems that this internal class is not used？



-- 
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: reviews-unsubscribe@iotdb.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org