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

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

@@ -192,10 +192,10 @@ chunk_buffer_pool_enable=false
 # default_ttl=36000000
 
 ####################
-### Update Configurations
+### Upgrade Configurations
 ####################
 
-# When there exists old version(v0.8.0) data, how many thread will be set up to perform upgrade tasks, 1 by default.
+# 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
 

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

@@ -315,6 +315,10 @@ public List<String> getOverlapFiles(String storageGroupName, TsFileResource appe
     return Collections.emptyList();
   }
 
+  /**
+   * count all Tsfiles which need to upgrade
+   * @return total num of the tsfiles which need to upgrade
+   */
   public int countUpgradeFiles() {
     int totalUpgradeFileNum = 0;
     for (StorageGroupProcessor storageGroupProcessor : processorMap.values()) {

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

@@ -187,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;
 
@@ -362,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 {
@@ -419,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) {
@@ -832,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) {
@@ -978,7 +973,11 @@ private void closeUnsealedTsFileProcessor(
     }
   }
 
-
+  /**
+   * count all Tsfiles in the storage group which need to upgrade
+   *
+   * @return total num of the tsfiles which need to upgrade in the storage group
+   */
   public int countUpgradeFiles() {
     int cntUpgradeFileNum = 0;
     for (TsFileResource seqTsFileResource : sequenceFileList) {
@@ -994,22 +993,15 @@ public int countUpgradeFiles() {
     return cntUpgradeFileNum;
   }
 
-
   public void upgrade() {
-    insertLock.readLock().lock();
-    try {
-      for (TsFileResource seqTsFileResource : sequenceFileList) {
-        seqTsFileResource.doUpgrade();
-      }
-      for (TsFileResource unseqTsFileResource : unSequenceFileList) {
-        unseqTsFileResource.doUpgrade();
-      }
-    } finally {
-      insertLock.readLock().unlock();
+    for (TsFileResource seqTsFileResource : sequenceFileList) {
+      seqTsFileResource.doUpgrade();
+    }
+    for (TsFileResource unseqTsFileResource : unSequenceFileList) {
+      unseqTsFileResource.doUpgrade();
     }
   }
 
-
   public void merge(boolean fullMerge) {
     writeLock();
     try {
@@ -1027,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 {

server/src/main/java/org/apache/iotdb/db/engine/upgrade/UpgradeCheckStatus.java

@@ -18,9 +18,22 @@
  */
 package org.apache.iotdb.db.engine.upgrade;
 
-public class UpgradeCheckStatus {
+public enum UpgradeCheckStatus {
 
-  public static final long BEGIN_UPGRADE_FILE = 1;
-  public static final long AFTER_UPGRADE_FILE = 2;
-  public static final long UPGRADE_SUCCESS = 3;
+  BEGIN_UPGRADE_FILE(1), AFTER_UPGRADE_FILE(2), UPGRADE_SUCCESS(3);
+
+  private final int checkStatusCode;
+
+  UpgradeCheckStatus(int checkStatusCode) {
+    this.checkStatusCode = checkStatusCode;
+  }
+
+  public int getCheckStatusCode() {
+    return checkStatusCode;
+  }
+
+  @Override
+  public String toString() {
+    return String.valueOf(checkStatusCode);
+  }
 }

server/src/main/java/org/apache/iotdb/db/engine/upgrade/UpgradeLog.java

@@ -0,0 +1,75 @@
+/*
+ * 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.engine.upgrade;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.utils.UpgradeUtils;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class UpgradeLog {
+
+  private static final Logger logger = LoggerFactory.getLogger(UpgradeLog.class);
+
+  private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+  private static final String UPGRADE_DIR = "upgrade";
+  private static final String UPGRADE_LOG_NAME = "upgrade.txt";
+  private static File upgradeLogPath = SystemFileFactory.INSTANCE
+      .getFile(SystemFileFactory.INSTANCE.getFile(config.getSystemDir(), UPGRADE_DIR),
+          UPGRADE_LOG_NAME);
+
+  public static boolean createUpgradeLog() {
+    try {
+      if (!upgradeLogPath.getParentFile().exists()) {
+        upgradeLogPath.getParentFile().mkdirs();
+      }
+      upgradeLogPath.createNewFile();
+      return true;
+    } catch (IOException e) {
+      logger.error("meet error when create upgrade log, file path:{}",
+          upgradeLogPath, e);
+      return false;
+    }
+  }
+
+  public static String getUpgradeLogPath() {
+    return upgradeLogPath.getAbsolutePath();
+  }
+
+  public static boolean writeUpgradeLogFile(String content) {
+    UpgradeUtils.getUpgradeLogLock().writeLock().lock();
+    try (BufferedWriter upgradeLogWriter = new BufferedWriter(
+        FSFactoryProducer.getFSFactory().getBufferedWriter(getUpgradeLogPath(), true))) {
+      upgradeLogWriter.write(content);
+      upgradeLogWriter.newLine();
+      return true;
+    } catch(IOException e) {
+      logger.error("write upgrade log file failed, the log file:{}", getUpgradeLogPath(), e);
+      return false;
+    } finally {
+      UpgradeUtils.getUpgradeLogLock().writeLock().unlock();
+    }
+  }
+}

server/src/main/java/org/apache/iotdb/db/engine/upgrade/UpgradeTask.java

@@ -18,9 +18,7 @@
  */
 package org.apache.iotdb.db.engine.upgrade;
 
-import java.io.File;
 import java.io.IOException;
-import java.io.RandomAccessFile;
 import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
 import org.apache.iotdb.db.service.UpgradeSevice;
 import org.apache.iotdb.db.utils.UpgradeUtils;
@@ -33,61 +31,26 @@ public class UpgradeTask implements Runnable {
 
   private final TsFileResource upgradeResource;
   private static final Logger logger = LoggerFactory.getLogger(UpgradeTask.class);
-  private static final String DOT_SEPERATOR = ",";
+  private static final String COMMA_SEPERATOR = ",";
+
 
   public UpgradeTask(TsFileResource upgradeResource) {
     this.upgradeResource = upgradeResource;
   }
 
-
-  private static boolean writeUpgradeLogFile(RandomAccessFile upgradeLog, String content,
-      long position) {
-    UpgradeUtils.getUpgradeLogLock().writeLock().lock();
-    try {
-      upgradeLog.seek(position);
-      upgradeLog.write(content.getBytes());
-      upgradeLog.write(System.getProperty("line.separator").getBytes());
-      return true;
-    } catch (IOException e) {
-      logger.error("write upgrade log file failed, the log file:{}", upgradeLog);
-      return false;
-    } finally {
-      UpgradeUtils.getUpgradeLogLock().writeLock().unlock();
-    }
-  }
-
-  private static long getUpgradeLogLength(RandomAccessFile upgradeLog) {
-    UpgradeUtils.getUpgradeLogLock().readLock().lock();
-    try {
-      return upgradeLog.length();
-    } catch (IOException e) {
-      logger.error("read upgrade log file failed, the log file:{}", upgradeLog);
-      return 0;
-    } finally {
-      UpgradeUtils.getUpgradeLogLock().readLock().unlock();
-    }
-  }
-
   @Override
   public void run() {
-    try (RandomAccessFile upgradeLogFile = new RandomAccessFile(UpgradeUtils.getUpgradeLogPath(),
-        "rw")) {
+    try {
       upgradeResource.getWriteQueryLock().readLock().lock();
       String tsfilePathBefore = upgradeResource.getFile().getAbsolutePath();
-      String tsfilePathAfter =
-          upgradeResource.getFile().getParentFile().getParent() + File.separator + "tmp"
-              + File.separator + "upgrade_" + upgradeResource
-              .getFile().getName();
+      String tsfilePathAfter = UpgradeUtils.getUpgradeFileName(upgradeResource.getFile());
 
-      long upgradePostion = getUpgradeLogLength(upgradeLogFile);
-      writeUpgradeLogFile(upgradeLogFile,
-          tsfilePathBefore + DOT_SEPERATOR + tsfilePathAfter + DOT_SEPERATOR
-              + UpgradeCheckStatus.BEGIN_UPGRADE_FILE, upgradePostion);
+      UpgradeLog.writeUpgradeLogFile(
+          tsfilePathBefore + COMMA_SEPERATOR + UpgradeCheckStatus.BEGIN_UPGRADE_FILE);
       try {
         UpgradeTool.upgradeOneTsfile(tsfilePathBefore, tsfilePathAfter);
-        writeUpgradeLogFile(upgradeLogFile,
-            tsfilePathBefore + DOT_SEPERATOR + tsfilePathAfter + DOT_SEPERATOR
-                + UpgradeCheckStatus.AFTER_UPGRADE_FILE, upgradePostion);
+        UpgradeLog.writeUpgradeLogFile(
+            tsfilePathBefore + COMMA_SEPERATOR + UpgradeCheckStatus.AFTER_UPGRADE_FILE);
       } catch (IOException e) {
         logger.error("generate upgrade file failed, the file to be upgraded:{}", tsfilePathBefore);
       } finally {
@@ -99,9 +62,8 @@ public void run() {
         FSFactoryProducer.getFSFactory()
             .moveFile(FSFactoryProducer.getFSFactory().getFile(tsfilePathAfter),
                 FSFactoryProducer.getFSFactory().getFile(tsfilePathBefore));
-        writeUpgradeLogFile(upgradeLogFile,
-            tsfilePathBefore + DOT_SEPERATOR + tsfilePathAfter + DOT_SEPERATOR
-                + UpgradeCheckStatus.UPGRADE_SUCCESS, upgradePostion);
+        UpgradeLog.writeUpgradeLogFile(
+            tsfilePathBefore + COMMA_SEPERATOR + UpgradeCheckStatus.UPGRADE_SUCCESS);
         FSFactoryProducer.getFSFactory().getFile(tsfilePathAfter).getParentFile().delete();
       } finally {
         upgradeResource.getWriteQueryLock().writeLock().unlock();

server/src/main/java/org/apache/iotdb/db/service/UpgradeSevice.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.upgrade.UpgradeLog;
 import org.apache.iotdb.db.engine.upgrade.UpgradeTask;
 import org.apache.iotdb.db.exception.StartupException;
 import org.apache.iotdb.db.exception.StorageEngineException;
@@ -55,7 +56,7 @@ public void start() throws StartupException {
     }
     upgradeThreadPool = Executors.newFixedThreadPool(updateThreadNum,
         r -> new Thread(r, "UpgradeThread-" + threadCnt.getAndIncrement()));
-    UpgradeUtils.createUpgradeLog();
+    UpgradeLog.createUpgradeLog();
     countUpgradeFiles();
     upgradeAll();
   }
@@ -80,11 +81,21 @@ public ServiceType getID() {
 
 
   public static void setCntUpgradeFileNum(int cntUpgradeFileNum) {
-    UpgradeSevice.cntUpgradeFileNum = cntUpgradeFileNum;
+    UpgradeUtils.getCntUpgradeFileLock().writeLock().lock();
+    try {
+      UpgradeSevice.cntUpgradeFileNum = cntUpgradeFileNum;
+    } finally {
+      UpgradeUtils.getCntUpgradeFileLock().writeLock().unlock();
+    }
   }
 
   public static int getCntUpgradeFileNum() {
-    return cntUpgradeFileNum;
+    UpgradeUtils.getCntUpgradeFileLock().readLock().lock();
+    try {
+      return cntUpgradeFileNum;
+    } finally {
+      UpgradeUtils.getCntUpgradeFileLock().readLock().unlock();
+    }
   }
 
   public void submitUpgradeTask(UpgradeTask upgradeTask) {

server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java

@@ -36,6 +36,7 @@
 import org.apache.iotdb.db.engine.modification.ModificationFile;
 import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
 import org.apache.iotdb.db.sync.conf.SyncSenderDescriptor;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -93,6 +94,9 @@ public void getCurrentLocalFiles(String dataDir) {
         dataDir + File.separatorChar + IoTDBConstant.SEQUENCE_FLODER_NAME)
         .listFiles();
     for (File sgFolder : allSGFolders) {
+      if (sgFolder.getName().equals(TsFileConstant.PATH_UPGRADE)){
+        continue;
+      }
       allSGs.add(sgFolder.getName());
       currentAllLocalFiles.putIfAbsent(sgFolder.getName(), new HashSet<>());
       File[] files = sgFolder.listFiles();