server/src/main/java/org/apache/iotdb/db/utils/UpgradeUtils.java

/*
 * 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.utils;

import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
import org.apache.iotdb.db.engine.upgrade.UpgradeCheckStatus;
import org.apache.iotdb.db.engine.upgrade.UpgradeLog;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.apache.iotdb.tsfile.fileSystem.fsFactory.FSFactory;
import org.apache.iotdb.tsfile.v2.read.TsFileSequenceReaderForV2;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.nio.file.Files;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class UpgradeUtils {

  private static final Logger logger = LoggerFactory.getLogger(UpgradeUtils.class);
  private static final String COMMA_SEPERATOR = ",";
  private static final ReadWriteLock cntUpgradeFileLock = new ReentrantReadWriteLock();
  private static final ReadWriteLock upgradeLogLock = new ReentrantReadWriteLock();

  private static FSFactory fsFactory = FSFactoryProducer.getFSFactory();

  private static Map<String, Integer> upgradeRecoverMap = new HashMap<>();

  public static ReadWriteLock getCntUpgradeFileLock() {
    return cntUpgradeFileLock;
  }

  public static ReadWriteLock getUpgradeLogLock() {
    return upgradeLogLock;
  }

  /** judge whether a tsfile needs to be upgraded */
  public static boolean isNeedUpgrade(TsFileResource tsFileResource) {
    tsFileResource.readLock();
    // case the TsFile's length is equal to 0, the TsFile does not need to be upgraded
    try {
      if (tsFileResource.getTsFile().length() == 0) {
        return false;
      }
    } finally {
      tsFileResource.readUnlock();
    }
    tsFileResource.readLock();
    try (TsFileSequenceReaderForV2 tsFileSequenceReader =
        new TsFileSequenceReaderForV2(tsFileResource.getTsFile().getAbsolutePath())) {
      String versionNumber = tsFileSequenceReader.readVersionNumberV2();
      if (versionNumber.equals(TSFileConfig.VERSION_NUMBER_V2)
          || versionNumber.equals(TSFileConfig.VERSION_NUMBER_V1)) {
        return true;
      }
    } catch (IOException e) {
      logger.error(
          "meet error when judge whether file needs to be upgraded, the file's path:{}",
          tsFileResource.getTsFile().getAbsolutePath(),
          e);
    } finally {
      tsFileResource.readUnlock();
    }
    return false;
  }

  public static void moveUpgradedFiles(TsFileResource resource) throws IOException {
    List<TsFileResource> upgradedResources = resource.getUpgradedResources();
    for (TsFileResource upgradedResource : upgradedResources) {
      File upgradedFile = upgradedResource.getTsFile();
      long partition = upgradedResource.getTimePartition();
      String virtualStorageGroupDir = upgradedFile.getParentFile().getParentFile().getParent();
      File partitionDir = fsFactory.getFile(virtualStorageGroupDir, String.valueOf(partition));
      if (!partitionDir.exists()) {
        partitionDir.mkdir();
      }
      // move upgraded TsFile
      if (upgradedFile.exists()) {
        fsFactory.moveFile(upgradedFile, fsFactory.getFile(partitionDir, upgradedFile.getName()));
      }
      // get temp resource
      File tempResourceFile =
          fsFactory.getFile(upgradedResource.getTsFile().toPath() + TsFileResource.RESOURCE_SUFFIX);
      // move upgraded mods file
      File newModsFile =
          fsFactory.getFile(upgradedResource.getTsFile().toPath() + ModificationFile.FILE_SUFFIX);
      if (newModsFile.exists()) {
        fsFactory.moveFile(newModsFile, fsFactory.getFile(partitionDir, newModsFile.getName()));
      }
      // re-serialize upgraded resource to correct place
      upgradedResource.setFile(fsFactory.getFile(partitionDir, upgradedFile.getName()));
      if (fsFactory.getFile(partitionDir, newModsFile.getName()).exists()) {
        upgradedResource.getModFile();
      }
      upgradedResource.setStatus(TsFileResourceStatus.CLOSED);
      upgradedResource.serialize();
      // delete generated temp resource file
      Files.delete(tempResourceFile.toPath());
    }
  }

  public static boolean isUpgradedFileGenerated(String oldFileName) {
    return upgradeRecoverMap.containsKey(oldFileName)
        && upgradeRecoverMap.get(oldFileName)
            == UpgradeCheckStatus.AFTER_UPGRADE_FILE.getCheckStatusCode();
  }

  public static void clearUpgradeRecoverMap() {
    upgradeRecoverMap = null;
  }

  @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
  public static void recoverUpgrade() {
    if (FSFactoryProducer.getFSFactory().getFile(UpgradeLog.getUpgradeLogPath()).exists()) {
      try (BufferedReader upgradeLogReader =
          new BufferedReader(
              new FileReader(
                  FSFactoryProducer.getFSFactory().getFile(UpgradeLog.getUpgradeLogPath())))) {
        String line = null;
        while ((line = upgradeLogReader.readLine()) != null) {
          String oldFilePath = line.split(COMMA_SEPERATOR)[0];
          String oldFileName = new File(oldFilePath).getName();
          if (upgradeRecoverMap.containsKey(oldFileName)) {
            upgradeRecoverMap.put(oldFileName, upgradeRecoverMap.get(oldFileName) + 1);
          } else {
            upgradeRecoverMap.put(oldFileName, 1);
          }
        }
      } catch (IOException e) {
        logger.error(
            "meet error when recover upgrade process, file path:{}",
            UpgradeLog.getUpgradeLogPath(),
            e);
      } finally {
        FSFactoryProducer.getFSFactory().getFile(UpgradeLog.getUpgradeLogPath()).delete();
      }
    }
  }
}