[IOTDB-294]online upgrade from v0.8.0 to current version

server/src/assembly/resources/conf/iotdb-engine.properties

@@ -191,6 +191,15 @@ chunk_buffer_pool_enable=false
 # data.
 # default_ttl=36000000
 
+####################
+### Upgrade Configurations
+####################
+
+# When there exists old version(v0.8.x) data, how many thread will be set up to perform upgrade tasks, 1 by default.
+# Set to 1 when less than or equal to 0.
+upgrade_thread_num=1
+
+
 ####################
 ### Merge Configurations
 ####################

server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -336,6 +336,11 @@ public class IoTDBConfig {
    */
   private long mergeMemoryBudget = (long) (Runtime.getRuntime().maxMemory() * 0.2);
 
+  /**
+   * How many threads will be set up to perform upgrade tasks.
+   */
+  private int upgradeThreadNum = 1;
+
   /**
    * How many threads will be set up to perform main merge tasks.
    */
@@ -1153,6 +1158,14 @@ public void setHdfsPort(String hdfsPort) {
     this.hdfsPort = hdfsPort;
   }
 
+  public int getUpgradeThreadNum() {
+    return upgradeThreadNum;
+  }
+
+  public void setUpgradeThreadNum(int upgradeThreadNum) {
+    this.upgradeThreadNum = upgradeThreadNum;
+  }
+
   public String getDfsNameServices() {
     return dfsNameServices;
   }

server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -233,6 +233,8 @@ private void loadProps() {
       conf.setExternalSortThreshold(Integer.parseInt(properties
           .getProperty("external_sort_threshold",
               Integer.toString(conf.getExternalSortThreshold()))));
+      conf.setUpgradeThreadNum(Integer.parseInt(properties.getProperty("upgrade_thread_num",
+          Integer.toString(conf.getUpgradeThreadNum()))));
       conf.setMergeMemoryBudget(Long.parseLong(properties.getProperty("merge_memory_budget",
           Long.toString(conf.getMergeMemoryBudget()))));
       conf.setMergeThreadNum(Integer.parseInt(properties.getProperty("merge_thread_num",

server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java

@@ -52,6 +52,7 @@
 import org.apache.iotdb.db.service.IService;
 import org.apache.iotdb.db.service.ServiceType;
 import org.apache.iotdb.db.utils.FilePathUtils;
+import org.apache.iotdb.db.utils.UpgradeUtils;
 import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
@@ -94,6 +95,8 @@ private StorageEngine() {
       throw new StorageEngineFailureException("create system directory failed!");
     }
 
+    // recover upgrade process
+    UpgradeUtils.recoverUpgrade();
     /*
      * recover all storage group processors.
      */
@@ -209,6 +212,7 @@ public boolean insert(InsertPlan insertPlan) throws ProcessorException {
 
   /**
    * insert a BatchInsertPlan to a storage group
+   *
    * @return result of each row
    */
   public Integer[] insertBatch(BatchInsertPlan batchInsertPlan) throws StorageEngineException {
@@ -311,6 +315,33 @@ public List<String> getOverlapFiles(String storageGroupName, TsFileResource appe
     return Collections.emptyList();
   }
 
+  /**
+   * count all Tsfiles which need to be upgraded
+   * @return total num of the tsfiles which need to be upgraded
+   */
+  public int countUpgradeFiles() {
+    int totalUpgradeFileNum = 0;
+    for (StorageGroupProcessor storageGroupProcessor : processorMap.values()) {
+      totalUpgradeFileNum += storageGroupProcessor.countUpgradeFiles();
+    }
+    return totalUpgradeFileNum;
+  }
+
+  /**
+   * upgrade all storage groups.
+   *
+   * @throws StorageEngineException StorageEngineException
+   */
+  public void upgradeAll() throws StorageEngineException {
+    if (IoTDBDescriptor.getInstance().getConfig().isReadOnly()) {
+      throw new StorageEngineException(
+          "Current system mode is read only, does not support file upgrade");
+    }
+    for (StorageGroupProcessor storageGroupProcessor : processorMap.values()) {
+      storageGroupProcessor.upgrade();
+    }
+  }
+
   /**
    * merge all storage groups.
    *
@@ -371,7 +402,7 @@ public void setTTL(String storageGroup, long dataTTL) throws StorageEngineExcept
   public void deleteStorageGroup(String storageGroupName) {
     deleteAllDataFilesInOneStorageGroup(storageGroupName);
     StorageGroupProcessor processor = processorMap.remove(storageGroupName);
-    if(processor != null) {
+    if (processor != null) {
       processor.deleteFolder(systemDir);
     }
   }

server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java

@@ -36,6 +36,7 @@
 import org.apache.iotdb.db.query.reader.IPointReader;
 import org.apache.iotdb.db.query.reader.resourceRelated.CachedUnseqResourceMergeReader;
 import org.apache.iotdb.db.utils.MergeUtils;
+import org.apache.iotdb.db.utils.UpgradeUtils;
 import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
 import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
 import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -69,14 +70,15 @@ public class MergeResource {
   private boolean cacheDeviceMeta = false;
 
   public MergeResource(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles) {
-    this.seqFiles =
-        seqFiles.stream().filter(this::filterResource).collect(Collectors.toList());
-    this.unseqFiles =
-        unseqFiles.stream().filter(this::filterResource).collect(Collectors.toList());
+    this.seqFiles = seqFiles.stream().filter(p -> p.isClosed() && !UpgradeUtils.isNeedUpgrade(p))
+        .collect(Collectors.toList());
+    this.unseqFiles = unseqFiles.stream().filter(p -> p.isClosed() && !UpgradeUtils.isNeedUpgrade(p))
+        .collect(Collectors.toList());
   }
 
   private boolean filterResource(TsFileResource res) {
-    return res.isClosed() && !res.isDeleted() && res.stillLives(timeLowerBound);
+    return res.isClosed() && !res.isDeleted() && res.stillLives(timeLowerBound) && !UpgradeUtils
+        .isNeedUpgrade(res);
   }
 
   public MergeResource(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles,

server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java

@@ -116,7 +116,7 @@ private void moveMergedToOld(TsFileResource seqFile) throws IOException {
       return;
     }
 
-    seqFile.getMergeQueryLock().writeLock().lock();
+    seqFile.getWriteQueryLock().writeLock().lock();
     try {
       TsFileMetaDataCache.getInstance().remove(seqFile);
       DeviceMetaDataCache.getInstance().remove(seqFile);
@@ -163,7 +163,7 @@ private void moveMergedToOld(TsFileResource seqFile) throws IOException {
               new File(nextMergeVersionFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX));
       seqFile.setFile(nextMergeVersionFile);
     } finally {
-      seqFile.getMergeQueryLock().writeLock().unlock();
+      seqFile.getWriteQueryLock().writeLock().unlock();
     }
   }
 
@@ -217,7 +217,7 @@ private void moveUnmergedToNew(TsFileResource seqFile) throws IOException {
     mergeLogger.logFileMergeEnd();
     logger.debug("{} moved unmerged chunks of {} to the new file", taskName, seqFile);
 
-    seqFile.getMergeQueryLock().writeLock().lock();
+    seqFile.getWriteQueryLock().writeLock().lock();
     try {
       resource.removeFileReader(seqFile);
       TsFileMetaDataCache.getInstance().remove(seqFile);
@@ -232,7 +232,7 @@ private void moveUnmergedToNew(TsFileResource seqFile) throws IOException {
               new File(nextMergeVersionFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX));
       seqFile.setFile(nextMergeVersionFile);
     } finally {
-      seqFile.getMergeQueryLock().writeLock().unlock();
+      seqFile.getWriteQueryLock().writeLock().unlock();
     }
   }
 

server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java

@@ -72,6 +72,8 @@
 import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
 import org.apache.iotdb.db.query.context.QueryContext;
 import org.apache.iotdb.db.query.control.JobFileManager;
+import org.apache.iotdb.db.utils.UpgradeUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
 import org.apache.iotdb.db.utils.CopyOnReadLinkedList;
 import org.apache.iotdb.db.utils.TestOnly;
 import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer;
@@ -185,8 +187,8 @@ public class StorageGroupProcessor {
   private static final int MAX_CACHE_SENSORS = 5000;
 
   /**
-   * when the data in a storage group is older than dataTTL, it is considered invalid and will
-   * be eventually removed.
+   * when the data in a storage group is older than dataTTL, it is considered invalid and will be
+   * eventually removed.
    */
   private long dataTTL = Long.MAX_VALUE;
 
@@ -360,7 +362,7 @@ public void addMeasurement(String measurementId, TSDataType dataType, TSEncoding
   public boolean insert(InsertPlan insertPlan) throws QueryProcessorException {
     // reject insertions that are out of ttl
     if (!checkTTL(insertPlan.getTime())) {
-     throw new OutOfTTLException(insertPlan.getTime(), (System.currentTimeMillis() - dataTTL));
+      throw new OutOfTTLException(insertPlan.getTime(), (System.currentTimeMillis() - dataTTL));
     }
     writeLock();
     try {
@@ -417,8 +419,6 @@ public Integer[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProces
   }
 
   /**
-   *
-   * @param time
    * @return whether the given time falls in ttl
    */
   private boolean checkTTL(long time) {
@@ -668,7 +668,7 @@ private void checkFileTTL(TsFileResource resource, long timeLowerBound, boolean
         return;
       }
       // ensure that the file is not used by any queries
-      if (resource.getMergeQueryLock().writeLock().tryLock()) {
+      if (resource.getWriteQueryLock().writeLock().tryLock()) {
         try {
           // physical removal
           resource.remove();
@@ -682,7 +682,7 @@ private void checkFileTTL(TsFileResource resource, long timeLowerBound, boolean
             unSequenceFileList.remove(resource);
           }
         } finally {
-          resource.getMergeQueryLock().writeLock().unlock();
+          resource.getWriteQueryLock().writeLock().unlock();
         }
       }
     } finally {
@@ -830,9 +830,6 @@ private List<TsFileResource> getFileReSourceListForQuery(List<TsFileResource> ts
   }
 
   /**
-   *
-   * @param tsFileResource
-   * @param deviceId
    * @return true if the device is contained in the TsFile and it lives beyond TTL
    */
   private boolean testResourceDevice(TsFileResource tsFileResource, String deviceId) {
@@ -976,6 +973,35 @@ private void closeUnsealedTsFileProcessor(
     }
   }
 
+  /**
+   * count all Tsfiles in the storage group which need to be upgraded
+   *
+   * @return total num of the tsfiles which need to be upgraded in the storage group
+   */
+  public int countUpgradeFiles() {
+    int cntUpgradeFileNum = 0;
+    for (TsFileResource seqTsFileResource : sequenceFileList) {
+      if (UpgradeUtils.isNeedUpgrade(seqTsFileResource)) {
+        cntUpgradeFileNum += 1;
+      }
+    }
+    for (TsFileResource unseqTsFileResource : unSequenceFileList) {
+      if (UpgradeUtils.isNeedUpgrade(unseqTsFileResource)) {
+        cntUpgradeFileNum += 1;
+      }
+    }
+    return cntUpgradeFileNum;
+  }
+
+  public void upgrade() {
+    for (TsFileResource seqTsFileResource : sequenceFileList) {
+      seqTsFileResource.doUpgrade();
+    }
+    for (TsFileResource unseqTsFileResource : unSequenceFileList) {
+      unseqTsFileResource.doUpgrade();
+    }
+  }
+
   public void merge(boolean fullMerge) {
     writeLock();
     try {
@@ -993,7 +1019,8 @@ public void merge(boolean fullMerge) {
 
       long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget();
       long timeLowerBound = System.currentTimeMillis() - dataTTL;
-      MergeResource mergeResource = new MergeResource(sequenceFileList, unSequenceFileList, timeLowerBound);
+      MergeResource mergeResource = new MergeResource(sequenceFileList, unSequenceFileList,
+          timeLowerBound);
 
       IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource);
       try {
@@ -1059,17 +1086,17 @@ private void removeUnseqFiles(List<TsFileResource> unseqFiles) {
     }
 
     for (TsFileResource unseqFile : unseqFiles) {
-      unseqFile.getMergeQueryLock().writeLock().lock();
+      unseqFile.getWriteQueryLock().writeLock().lock();
       try {
         unseqFile.remove();
       } finally {
-        unseqFile.getMergeQueryLock().writeLock().unlock();
+        unseqFile.getWriteQueryLock().writeLock().unlock();
       }
     }
   }
 
   private void updateMergeModification(TsFileResource seqFile) {
-    seqFile.getMergeQueryLock().writeLock().lock();
+    seqFile.getWriteQueryLock().writeLock().lock();
     try {
       // remove old modifications and write modifications generated during merge
       seqFile.removeModFile();
@@ -1082,7 +1109,7 @@ private void updateMergeModification(TsFileResource seqFile) {
       logger.error("{} cannot clean the ModificationFile of {} after merge", storageGroupName,
           seqFile.getFile(), e);
     } finally {
-      seqFile.getMergeQueryLock().writeLock().unlock();
+      seqFile.getWriteQueryLock().writeLock().unlock();
     }
   }
 
@@ -1314,12 +1341,12 @@ public void deleteTsfile(File deletedTsfile) {
     if (deletedTsFileResource == null) {
       return;
     }
-    deletedTsFileResource.getMergeQueryLock().writeLock().lock();
+    deletedTsFileResource.getWriteQueryLock().writeLock().lock();
     try {
       logger.info("Delete tsfile {} in sync loading process.", deletedTsFileResource.getFile());
       deletedTsFileResource.remove();
     } finally {
-      deletedTsFileResource.getMergeQueryLock().writeLock().unlock();
+      deletedTsFileResource.getWriteQueryLock().writeLock().unlock();
     }
   }
 

server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java

@@ -18,7 +18,10 @@
  */
 package org.apache.iotdb.db.engine.storagegroup;
 
-import java.io.*;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -28,6 +31,9 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.iotdb.db.engine.modification.ModificationFile;
 import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.engine.upgrade.UpgradeTask;
+import org.apache.iotdb.db.service.UpgradeSevice;
+import org.apache.iotdb.db.utils.UpgradeUtils;
 import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
 import org.apache.iotdb.tsfile.fileSystem.fsFactory.FSFactory;
 import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
@@ -59,6 +65,7 @@ public class TsFileResource {
   private volatile boolean deleted = false;
   private volatile boolean isMerging = false;
 
+
   /**
    * Chunk metadata list of unsealed tsfile. Only be set in a temporal TsFileResource in a query
    * process.
@@ -70,7 +77,7 @@ public class TsFileResource {
    */
   private ReadOnlyMemChunk readOnlyMemChunk;
 
-  private ReentrantReadWriteLock mergeQueryLock = new ReentrantReadWriteLock();
+  private ReentrantReadWriteLock writeQueryLock = new ReentrantReadWriteLock();
 
   private FSFactory fsFactory = FSFactoryProducer.getFSFactory();
 
@@ -230,8 +237,14 @@ public TsFileProcessor getUnsealedFileProcessor() {
     return processor;
   }
 
-  public ReentrantReadWriteLock getMergeQueryLock() {
-    return mergeQueryLock;
+  public ReentrantReadWriteLock getWriteQueryLock() {
+    return writeQueryLock;
+  }
+
+  public void doUpgrade() {
+    if (UpgradeUtils.isNeedUpgrade(this)) {
+      UpgradeSevice.getINSTANCE().submitUpgradeTask(new UpgradeTask(this));
+    }
   }
 
   public void removeModFile() throws IOException {