[IOTDB-5]Combine deletion with FileNodeProcessor and FileNodeManager

iotdb/src/main/java/org/apache/iotdb/db/conf/IoTDBConstant.java

@@ -46,7 +46,7 @@ public class IoTDBConstant {
   public static final String OVERFLOW_LOG_NODE_SUFFIX = "-overflow";
 
   public static final String PATH_ROOT = "root";
-  public static final char PATH_SEPARATER = '.';
+  public static final char PATH_SEPARATOR = '.';
   public static final String ADMIN_NAME = "root";
   public static final String ADMIN_PW = "root";
   public static final String PROFILE_SUFFIX = ".profile";

iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwrite/BufferWriteProcessor.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
+
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBConstant;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -40,6 +41,7 @@
 import org.apache.iotdb.db.engine.pool.FlushManager;
 import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
 import org.apache.iotdb.db.engine.utils.FlushStatus;
+import org.apache.iotdb.db.engine.version.VersionController;
 import org.apache.iotdb.db.exception.BufferWriteProcessorException;
 import org.apache.iotdb.db.utils.MemUtils;
 import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
@@ -79,6 +81,7 @@ public class BufferWriteProcessor extends Processor {
   private String bufferWriteRelativePath;
 
   private WriteLogNode logNode;
+  private VersionController versionController;
 
   /**
    * constructor of BufferWriteProcessor.
@@ -91,8 +94,8 @@ public class BufferWriteProcessor extends Processor {
    * @throws BufferWriteProcessorException BufferWriteProcessorException
    */
   public BufferWriteProcessor(String baseDir, String processorName, String fileName,
-      Map<String, Action> parameters,
-      FileSchema fileSchema) throws BufferWriteProcessorException {
+                              Map<String, Action> parameters, VersionController versionController,
+                              FileSchema fileSchema) throws BufferWriteProcessorException {
     super(processorName);
     this.fileSchema = fileSchema;
     this.baseDir = baseDir;
@@ -131,6 +134,7 @@ public BufferWriteProcessor(String baseDir, String processorName, String fileNam
         throw new BufferWriteProcessorException(e);
       }
     }
+    this.versionController = versionController;
   }
 
   /**
@@ -216,8 +220,8 @@ private void checkMemThreshold4Flush(long addedMemory) throws BufferWriteProcess
    * @return corresponding chunk data and chunk metadata in memory
    */
   public Pair<ReadOnlyMemChunk, List<ChunkMetaData>> queryBufferWriteData(String deviceId,
-      String measurementId,
-      TSDataType dataType) {
+                                                                          String measurementId,
+                                                                          TSDataType dataType) {
     flushQueryLock.lock();
     try {
       MemSeriesLazyMerger memSeriesLazyMerger = new MemSeriesLazyMerger();
@@ -258,14 +262,15 @@ private void switchFlushToWork() {
     }
   }
 
-  private void flushOperation(String flushFunction) {
+  private void flushOperation(String flushFunction, long version) {
     long flushStartTime = System.currentTimeMillis();
     LOGGER.info("The bufferwrite processor {} starts flushing {}.", getProcessorName(),
         flushFunction);
     try {
       if (flushMemTable != null && !flushMemTable.isEmpty()) {
         // flush data
-        MemTableFlushUtil.flushMemTable(fileSchema, writer, flushMemTable);
+        MemTableFlushUtil.flushMemTable(fileSchema, writer, flushMemTable,
+                version);
         // write restore information
         writer.flush();
       }
@@ -346,13 +351,14 @@ private Future<?> flush(boolean synchronization) throws IOException {
       valueCount = 0;
       flushStatus.setFlushing();
       switchWorkToFlush();
+      long version = versionController.nextVersion();
       BasicMemController.getInstance().reportFree(this, memSize.get());
       memSize.set(0);
       // switch
       if (synchronization) {
-        flushOperation("synchronously");
+        flushOperation("synchronously", version);
       } else {
-        FlushManager.getInstance().submit(() -> flushOperation("asynchronously"));
+        FlushManager.getInstance().submit(() -> flushOperation("asynchronously", version));
       }
     }
     // TODO return a meaningful Future
@@ -500,4 +506,20 @@ public void setNewProcessor(boolean isNewProcessor) {
   public WriteLogNode getLogNode() {
     return logNode;
   }
+
+  /**
+   * Delete data whose timestamp <= 'timestamp' and belonging to timeseries deviceId.measurementId.
+   * Delete data in both working MemTable and flushing MemTable.
+   * @param deviceId the deviceId of the timeseries to be deleted.
+   * @param measurementId the measurementId of the timeseries to be deleted.
+   * @param timestamp the upper-bound of deletion time.
+   */
+  public void delete(String deviceId, String measurementId, long timestamp) {
+    workMemTable.delete(deviceId, measurementId, timestamp);
+    if (isFlush) {
+      // flushing MemTable cannot be directly modified since another thread is reading it
+      flushMemTable = flushMemTable.copy();
+      flushMemTable.delete(deviceId, measurementId, timestamp);
+    }
+  }
 }

iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java

@@ -52,7 +52,6 @@
 import org.apache.iotdb.db.monitor.IStatistic;
 import org.apache.iotdb.db.monitor.MonitorConstants;
 import org.apache.iotdb.db.monitor.StatMonitor;
-import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
 import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
 import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
 import org.apache.iotdb.db.query.control.FileReaderManager;
@@ -496,66 +495,66 @@ public void update(String deviceId, String measurementId, long startTime, long e
   /**
    * delete data.
    */
-  public void delete(String deviceId, String measurementId, long timestamp, TSDataType type)
-      throws FileNodeManagerException {
+  public void delete(String deviceId, String measurementId, long timestamp)
+          throws FileNodeManagerException {
 
     FileNodeProcessor fileNodeProcessor = getProcessor(deviceId, true);
     try {
       long lastUpdateTime = fileNodeProcessor.getLastUpdateTime(deviceId);
       // no tsfile data, the delete operation is invalid
       if (lastUpdateTime == -1) {
-        LOGGER.warn(
-            "The last update time is -1, delete overflow is invalid, the filenode processor is {}",
-            fileNodeProcessor.getProcessorName());
+        LOGGER.warn("The last update time is -1, delete overflow is invalid, "
+                        + "the filenode processor is {}",
+                fileNodeProcessor.getProcessorName());
       } else {
-        if (timestamp > lastUpdateTime) {
-          timestamp = lastUpdateTime;
-        }
-        String filenodeName = fileNodeProcessor.getProcessorName();
-        // get overflow processor
-        OverflowProcessor overflowProcessor;
         try {
-          overflowProcessor = fileNodeProcessor.getOverflowProcessor(filenodeName);
+          fileNodeProcessor.delete(deviceId, measurementId, timestamp);
         } catch (IOException e) {
-          LOGGER.error("Get the overflow processor failed, the filenode is {}, delete time is {}.",
-              filenodeName, timestamp);
           throw new FileNodeManagerException(e);
         }
-        overflowProcessor.delete(deviceId, measurementId, timestamp, type);
         // change the type of tsfile to overflowed
         fileNodeProcessor.changeTypeToChangedForDelete(deviceId, timestamp);
         fileNodeProcessor.setOverflowed(true);
-        // if (shouldMerge) {
-        // LOGGER.info(
-        // "The overflow file or metadata reaches the threshold,
-        // merge the filenode processor {}",
-        // filenodeName);
-        // fileNodeProcessor.submitToMerge();
-        // }
-        fileNodeProcessor.changeTypeToChangedForDelete(deviceId, timestamp);
-        fileNodeProcessor.setOverflowed(true);
 
-        // write wal
+        // TODO: support atomic deletion
+        /*// write wal
+        // get processors for wal
+        String filenodeName = fileNodeProcessor.getProcessorName();
+        OverflowProcessor overflowProcessor;
+        BufferWriteProcessor bufferWriteProcessor;
+        try {
+          overflowProcessor = fileNodeProcessor.getOverflowProcessor(filenodeName);
+          bufferWriteProcessor = fileNodeProcessor.getBufferWriteProcessor();
+        } catch (IOException | FileNodeProcessorException e) {
+          LOGGER.error("Getting the processor failed, the filenode is {}, delete time is {}.",
+                  filenodeName, timestamp);
+          throw new FileNodeManagerException(e);
+        }
         try {
           if (IoTDBDescriptor.getInstance().getConfig().enableWal) {
             overflowProcessor.getLogNode()
-                .write(new DeletePlan(timestamp, new Path(deviceId + "." + measurementId)));
+                    .write(new DeletePlan(timestamp,
+                            new Path(deviceId + "." + measurementId)));
+            bufferWriteProcessor.getLogNode()
+                    .write(new DeletePlan(timestamp,
+                            new Path(deviceId + "." + measurementId)));
           }
         } catch (IOException e) {
           throw new FileNodeManagerException(e);
-        }
+        }*/
       }
     } finally {
       fileNodeProcessor.writeUnlock();
     }
+
   }
 
   /**
    * try to delete the filenode processor.
    */
   private void delete(String processorName,
-      Iterator<Map.Entry<String, FileNodeProcessor>> processorIterator)
-      throws FileNodeManagerException {
+                      Iterator<Map.Entry<String, FileNodeProcessor>> processorIterator)
+          throws FileNodeManagerException {
     if (processorMap.containsKey(processorName)) {
       LOGGER.info("Try to delete the filenode processor {}.", processorName);
       FileNodeProcessor processor = processorMap.get(processorName);