/*
 * 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.storageengine.dataregion.compaction.inner.sizetiered;

import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.file.SystemFileFactory;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.storageengine.buffer.ChunkCache;
import org.apache.iotdb.db.storageengine.buffer.TimeSeriesMetadataCache;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.recover.CompactionRecoverTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.subtask.FastCompactionTaskSummary;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.IDataBlockReader;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.reader.SeriesDataBlockReader;
import org.apache.iotdb.db.storageengine.dataregion.compaction.inner.AbstractInnerSpaceCompactionTest;
import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionConfigRestorer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionFileGeneratorUtils;
import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.db.utils.SchemaTestUtils;
import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.writer.TsFileOutput;

import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger.STR_DELETED_TARGET_FILES;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger.STR_SOURCE_FILES;
import static org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger.STR_TARGET_FILES;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
import static org.junit.Assert.assertEquals;

public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactionTest {

  ICompactionPerformer performer = new FastCompactionPerformer(false);

  @Before
  public void setUp() throws IOException, WriteProcessException, MetadataException {
    super.setUp();
  }

  @After
  public void tearDown() throws IOException, StorageEngineException {
    new CompactionConfigRestorer().restoreCompactionConfig();
    super.tearDown();
  }

  /** Target file uncompleted, source files and log exists */
  @Test
  public void testCompactionRecoverWithUncompletedTargetFileAndLog() throws Exception {
    TsFileManager tsFileManager =
        new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);

    TsFileResource targetTsFileResource =
        new TsFileResource(
            new File(
                SEQ_DIRS
                    + File.separator.concat(
                        0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 1
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
    File compactionLogFile =
        new File(
            seqResources.get(0).getTsFile().getParent()
                + File.separator
                + targetTsFileResource.getTsFile().getName()
                + INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));
    compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    deleteFileIfExists(targetTsFileResource.getTsFile());
    performer.setSourceFiles(new ArrayList<>(seqResources.subList(0, 3)));
    performer.setTargetFiles(Collections.singletonList(targetTsFileResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    compactionLogger.close();
    CompactionUtils.moveTargetFile(
        Collections.singletonList(targetTsFileResource), true, COMPACTION_TEST_SG);
    BufferedReader logReader = new BufferedReader(new FileReader(compactionLogFile));
    List<String> logs = new ArrayList<>();
    String line;
    while ((line = logReader.readLine()) != null) {
      logs.add(line);
    }
    logReader.close();
    BufferedWriter logStream =
        new BufferedWriter(
            new FileWriter(
                SystemFileFactory.INSTANCE.getFile(
                    tempSGDir.getPath(), COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX),
                false));
    for (int i = 0; i < logs.size() - 1; i++) {
      logStream.write(logs.get(i));
      logStream.newLine();
    }
    logStream.close();

    TsFileOutput out =
        FSFactoryProducer.getFileOutputFactory()
            .getTsFileOutput(targetTsFileResource.getTsFile().getPath(), true);
    out.truncate(((long) (targetTsFileResource.getTsFileSize() * 0.9)));
    out.close();

    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();
    path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  @Test
  public void testRecoverWithAllSourceFilesExisted() throws Exception {
    TsFileManager tsFileManager =
        new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);

    TsFileResource targetTsFileResource =
        new TsFileResource(
            new File(
                SEQ_DIRS
                    + File.separator.concat(
                        0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 1
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
    File compactionLogFile =
        new File(
            seqResources.get(0).getTsFile().getParent()
                + File.separator
                + targetTsFileResource.getTsFile().getName()
                + INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));
    compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    deleteFileIfExists(targetTsFileResource.getTsFile());
    performer.setSourceFiles(new ArrayList<>(seqResources.subList(0, 3)));
    performer.setTargetFiles(Collections.singletonList(targetTsFileResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    compactionLogger.close();
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();
    // all source file should still exist
    Assert.assertTrue(seqResources.get(0).getTsFile().exists());
    Assert.assertTrue(seqResources.get(1).getTsFile().exists());
    Assert.assertTrue(seqResources.get(2).getTsFile().exists());
    // tmp target file, target file and target resource file should be deleted
    Assert.assertFalse(targetTsFileResource.getTsFile().exists());
    Assert.assertFalse(
        new File(
                targetTsFileResource
                    .getTsFilePath()
                    .replace(
                        IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
                        TsFileConstant.TSFILE_SUFFIX))
            .exists());
    Assert.assertFalse(
        new File(targetTsFileResource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());

    path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  @Test
  public void testRecoverWithAllSourceFilesExistedAndTargetFileNotExist() throws Exception {
    TsFileManager tsFileManager =
        new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);

    TsFileResource targetTsFileResource =
        new TsFileResource(
            new File(
                SEQ_DIRS
                    + File.separator.concat(
                        0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 1
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
    File compactionLogFile =
        new File(
            seqResources.get(0).getTsFile().getParent()
                + File.separator
                + targetTsFileResource.getTsFile().getName()
                + INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));
    compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    deleteFileIfExists(targetTsFileResource.getTsFile());
    performer.setSourceFiles(new ArrayList<>(seqResources.subList(0, 3)));
    performer.setTargetFiles(Collections.singletonList(targetTsFileResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    // target file may not exist
    targetTsFileResource.remove();
    compactionLogger.close();
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();
    // all source file should still exist
    Assert.assertTrue(seqResources.get(0).getTsFile().exists());
    Assert.assertTrue(seqResources.get(1).getTsFile().exists());
    Assert.assertTrue(seqResources.get(2).getTsFile().exists());
    // tmp target file, target file and target resource file should be deleted
    Assert.assertFalse(targetTsFileResource.getTsFile().exists());
    Assert.assertFalse(
        new File(
                targetTsFileResource
                    .getTsFilePath()
                    .replace(
                        IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
                        TsFileConstant.TSFILE_SUFFIX))
            .exists());
    Assert.assertFalse(
        new File(targetTsFileResource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());

    path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  @Test
  public void testRecoverWithoutAllSourceFilesExisted() throws Exception {
    TsFileManager tsFileManager =
        new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);

    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));

    TsFileResource targetTsFileResource =
        TsFileNameGenerator.getInnerCompactionTargetFileResource(tmpSeqResources, true);
    File compactionLogFile =
        new File(
            seqResources.get(0).getTsFile().getParent()
                + File.separator
                + targetTsFileResource.getTsFile().getName()
                + INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    deleteFileIfExists(targetTsFileResource.getTsFile());
    performer.setSourceFiles(new ArrayList<>(seqResources.subList(0, 3)));
    performer.setTargetFiles(Collections.singletonList(targetTsFileResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    CompactionUtils.moveTargetFile(
        Collections.singletonList(targetTsFileResource), true, COMPACTION_TEST_SG);
    // delete one source file
    seqResources.get(0).remove();
    compactionLogger.close();
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();
    // all source files should be deleted
    Assert.assertFalse(seqResources.get(0).getTsFile().exists());
    Assert.assertFalse(seqResources.get(1).getTsFile().exists());
    Assert.assertFalse(seqResources.get(2).getTsFile().exists());
    // target file and target resource file should exist
    Assert.assertTrue(targetTsFileResource.getTsFile().exists());
    Assert.assertTrue(
        new File(targetTsFileResource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
    // tmp target file should be deleted
    Assert.assertFalse(
        new File(
                targetTsFileResource
                    .getTsFilePath()
                    .replace(
                        TsFileConstant.TSFILE_SUFFIX,
                        IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX))
            .exists());

    tsFileManager.add(targetTsFileResource, true);
    path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());

    tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true).subList(3, 6),
            new ArrayList<>(),
            true);
    count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  /**
   * All source files exist, each source file has compaction mods file which have been combined into
   * new mods file of the target file.
   */
  @Test
  public void testRecoverWithAllSourcesFileAndCompactonModFileExist() throws Exception {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    TsFileResource targetResource =
        TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
    File logFile =
        new File(
            targetResource.getTsFile().getPath()
                + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(logFile);
    compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
    performer.setSourceFiles(seqResources);
    performer.setTargetFiles(Collections.singletonList(targetResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    CompactionUtils.moveTargetFile(
        Collections.singletonList(targetResource), true, COMPACTION_TEST_SG);
    for (int i = 0; i < seqResources.size(); i++) {
      Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
      deleteMap.put(
          deviceIds[0] + "." + measurementSchemas[0].getMeasurementId(),
          new Pair<>(i * ptNum, i * ptNum + 10));
      CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), true);
      CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
    }
    CompactionUtils.combineModsInInnerCompaction(seqResources, targetResource);
    compactionLogger.close();

    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, logFile, true).doCompaction();
    // all source file should exist
    for (int i = 0; i < seqResources.size(); i++) {
      Assert.assertTrue(seqResources.get(i).getTsFile().exists());
      Assert.assertTrue(seqResources.get(i).resourceFileExists());
    }

    // tmp target file, target file and target resource file should be deleted
    Assert.assertFalse(targetResource.getTsFile().exists());
    Assert.assertFalse(
        new File(
                targetResource
                    .getTsFilePath()
                    .replace(
                        IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
                        TsFileConstant.TSFILE_SUFFIX))
            .exists());
    Assert.assertFalse(
        new File(targetResource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());

    // all compaction mods file of each source file should not exist
    for (int i = 0; i < seqResources.size(); i++) {
      Assert.assertFalse(seqResources.get(i).getCompactionModFile().exists());
    }

    // all mods file of each source file should exist
    for (int i = 0; i < seqResources.size(); i++) {
      seqResources.get(i).resetModFile();
      Assert.assertTrue(seqResources.get(i).getModFile().exists());
      Assert.assertEquals(1, seqResources.get(i).getModFile().getModifications().size());
    }

    // mods file of the target file should not exist
    Assert.assertFalse(targetResource.getModFile().exists());

    // compaction log file should not exist
    Assert.assertFalse(logFile.exists());

    Assert.assertTrue(tsFileManager.isAllowCompaction());
  }

  @Test
  public void testRecoverWithAllSourcesFileAndCompactonModFileExistAndTargetFileNotExist()
      throws Exception {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    TsFileResource targetResource =
        TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
    File logFile =
        new File(
            targetResource.getTsFile().getPath()
                + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(logFile);
    compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
    performer.setSourceFiles(seqResources);
    performer.setTargetFiles(Collections.singletonList(targetResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    // target file may not exist
    targetResource.remove();
    for (int i = 0; i < seqResources.size(); i++) {
      Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
      deleteMap.put(
          deviceIds[0] + "." + measurementSchemas[0].getMeasurementId(),
          new Pair<>(i * ptNum, i * ptNum + 10));
      CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), true);
      CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
    }
    compactionLogger.close();

    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, logFile, true).doCompaction();
    // all source file should exist
    for (int i = 0; i < seqResources.size(); i++) {
      Assert.assertTrue(seqResources.get(i).getTsFile().exists());
      Assert.assertTrue(seqResources.get(i).resourceFileExists());
    }

    // tmp target file, target file and target resource file should be deleted
    Assert.assertFalse(targetResource.getTsFile().exists());
    Assert.assertFalse(
        new File(
                targetResource
                    .getTsFilePath()
                    .replace(
                        IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
                        TsFileConstant.TSFILE_SUFFIX))
            .exists());
    Assert.assertFalse(
        new File(targetResource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());

    // all compaction mods file of each source file should not exist
    for (int i = 0; i < seqResources.size(); i++) {
      Assert.assertFalse(seqResources.get(i).getCompactionModFile().exists());
    }

    // all mods file of each source file should exist
    for (int i = 0; i < seqResources.size(); i++) {
      seqResources.get(i).resetModFile();
      Assert.assertTrue(seqResources.get(i).getModFile().exists());
      Assert.assertEquals(1, seqResources.get(i).getModFile().getModifications().size());
    }

    // mods file of the target file should not exist
    Assert.assertFalse(targetResource.getModFile().exists());

    // compaction log file should not exist
    Assert.assertFalse(logFile.exists());

    Assert.assertTrue(tsFileManager.isAllowCompaction());
  }

  /**
   * Some source files have been deleted, each source file has old mods file and new compaction mods
   * file.
   */
  @Test
  public void testRecoverWithoutAllSourceFilesExistAndModFiles() throws Exception {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    TsFileResource targetResource =
        TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
    File logFile =
        new File(
            targetResource.getTsFile().getPath()
                + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(logFile);
    compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
    performer.setSourceFiles(seqResources);
    performer.setTargetFiles(Collections.singletonList(targetResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    CompactionUtils.moveTargetFile(
        Collections.singletonList(targetResource), true, COMPACTION_TEST_SG);
    for (int i = 0; i < seqResources.size(); i++) {
      Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
      deleteMap.put(
          deviceIds[0] + "." + measurementSchemas[0].getMeasurementId(),
          new Pair<>(i * ptNum, i * ptNum + 10));
      CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), true);
      CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
    }
    CompactionUtils.combineModsInInnerCompaction(seqResources, targetResource);
    seqResources.get(0).remove();
    compactionLogger.close();

    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, logFile, true).doCompaction();
    // all source files should not exist
    for (int i = 0; i < seqResources.size(); i++) {
      Assert.assertFalse(seqResources.get(i).getTsFile().exists());
      Assert.assertFalse(seqResources.get(i).resourceFileExists());
    }

    // target file and target resource file should exist
    Assert.assertTrue(targetResource.getTsFile().exists());
    Assert.assertTrue(targetResource.resourceFileExists());

    // tmp target file should be deleted
    Assert.assertFalse(
        new File(
                targetResource
                    .getTsFilePath()
                    .replace(
                        TsFileConstant.TSFILE_SUFFIX,
                        IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX))
            .exists());

    // all compaction mods file and old mods file of each source file should not exist
    for (int i = 0; i < seqResources.size(); i++) {
      Assert.assertFalse(seqResources.get(i).getCompactionModFile().exists());
      Assert.assertFalse(seqResources.get(i).getModFile().exists());
    }

    // mods file of the target file should exist
    Assert.assertTrue(targetResource.getModFile().exists());

    // compaction log file should not exist
    Assert.assertFalse(logFile.exists());

    Assert.assertTrue(tsFileManager.isAllowCompaction());
  }

  /** compaction recover merge finished, delete one offset */
  @Test
  public void testRecoverCompleteTargetFileAndCompactionLog() throws Exception {
    TsFileManager tsFileManager =
        new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);

    TsFileResource targetTsFileResource =
        new TsFileResource(
            new File(
                SEQ_DIRS
                    + File.separator.concat(
                        0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 1
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
    File compactionLogFile =
        new File(
            seqResources.get(0).getTsFile().getParent()
                + File.separator
                + targetTsFileResource.getTsFile().getName()
                + INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));
    compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    deleteFileIfExists(targetTsFileResource.getTsFile());
    performer.setSourceFiles(new ArrayList<>(seqResources.subList(0, 3)));
    performer.setTargetFiles(Collections.singletonList(targetTsFileResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    compactionLogger.close();
    CompactionUtils.moveTargetFile(
        Collections.singletonList(targetTsFileResource), true, COMPACTION_TEST_SG);
    tsFileManager.add(targetTsFileResource, true);
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();
    path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    TimeSeriesMetadataCache.getInstance().clear();
    ChunkCache.getInstance().clear();
    tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true).subList(0, 5),
            new ArrayList<>(),
            true);
    count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  @Test
  public void testCompactionRecoverWithCompletedTargetFileAndLog() throws Exception {
    TsFileManager tsFileManager =
        new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);

    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));

    TsFileResource targetTsFileResource =
        TsFileNameGenerator.getInnerCompactionTargetFileResource(tmpSeqResources, true);

    File compactionLogFile =
        new File(
            seqResources.get(0).getTsFile().getParent()
                + File.separator
                + targetTsFileResource.getTsFile().getName()
                + INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    deleteFileIfExists(targetTsFileResource.getTsFile());
    performer.setSourceFiles(new ArrayList<>(seqResources.subList(0, 3)));
    performer.setTargetFiles(Collections.singletonList(targetTsFileResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    CompactionUtils.moveTargetFile(
        Collections.singletonList(targetTsFileResource), true, COMPACTION_TEST_SG);
    compactionLogger.close();
    for (TsFileResource resource : new ArrayList<>(seqResources.subList(0, 3))) {
      deleteFileIfExists(resource.getTsFile());
      tsFileManager.remove(resource, true);
    }
    tsFileManager.add(targetTsFileResource, true);
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();
    path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  /** compeleted target file, and not resource files, compaction log exists */
  @Test
  public void testCompactionRecoverWithCompletedTargetFile() throws Exception {
    TsFileManager tsFileManager =
        new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);

    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));

    TsFileResource targetTsFileResource =
        TsFileNameGenerator.getInnerCompactionTargetFileResource(tmpSeqResources, true);

    File compactionLogFile =
        new File(
            seqResources.get(0).getTsFile().getParent()
                + File.separator
                + targetTsFileResource.getTsFile().getName()
                + INNER_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    deleteFileIfExists(targetTsFileResource.getTsFile());
    performer.setSourceFiles(new ArrayList<>(seqResources.subList(0, 3)));
    performer.setTargetFiles(Collections.singletonList(targetTsFileResource));
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    compactionLogger.close();
    CompactionUtils.moveTargetFile(
        Collections.singletonList(targetTsFileResource), true, COMPACTION_TEST_SG);
    deleteFileIfExists(compactionLogFile);
    for (TsFileResource resource : new ArrayList<>(seqResources.subList(0, 3))) {
      tsFileManager.remove(resource, true);
      deleteFileIfExists(resource.getTsFile());
    }
    tsFileManager.add(targetTsFileResource, true);
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();
    path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    logger.warn("TsFiles in list is {}", tsFileManager.getTsFileList(true));
    tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  /** compaction recover merge start just log source file */
  @Test
  public void testCompactionMergeRecoverMergeStartSourceLog()
      throws IOException, MetadataException {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    CompactionLogger sizeTieredCompactionLogger =
        new CompactionLogger(
            new File(
                tempSGDir.getPath(),
                COMPACTION_TEST_SG + COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX));
    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));
    sizeTieredCompactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    sizeTieredCompactionLogger.close();
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  /** compaction recover merge start just log source file and sequence flag */
  @Test
  public void testCompactionMergeRecoverMergeStartSequenceLog()
      throws IOException, MetadataException {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    CompactionLogger sizeTieredCompactionLogger =
        new CompactionLogger(
            new File(tempSGDir.getPath(), COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX));
    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));
    sizeTieredCompactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    sizeTieredCompactionLogger.close();
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    tsFilesReader.close();
    closeTsFileSequenceReader();
    assertEquals(500, count);
  }

  /** compaction recover merge start target file logged */
  @Test
  public void testCompactionMergeRecoverMergeStart() throws IOException, MetadataException {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);
    CompactionLogger sizeTieredCompactionLogger =
        new CompactionLogger(
            new File(tempSGDir.getPath(), COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX));
    List<TsFileResource> tmpSeqResources = new ArrayList<>();
    tmpSeqResources.add(seqResources.get(0));
    tmpSeqResources.add(seqResources.get(1));
    tmpSeqResources.add(seqResources.get(2));
    sizeTieredCompactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
    TsFileResource targetTsFileResource =
        new TsFileResource(
            new File(
                SEQ_DIRS
                    + File.separator.concat(
                        0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 1
                            + IoTDBConstant.FILE_NAME_SEPARATOR
                            + 0
                            + IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
    sizeTieredCompactionLogger.logFiles(
        Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
    sizeTieredCompactionLogger.close();
    MeasurementPath path =
        SchemaTestUtils.getMeasurementPath(
            deviceIds[0]
                + TsFileConstant.PATH_SEPARATOR
                + measurementSchemas[0].getMeasurementId());
    IDataBlockReader tsFilesReader =
        new SeriesDataBlockReader(
            path,
            FragmentInstanceContext.createFragmentInstanceContextForCompaction(
                EnvironmentUtils.TEST_QUERY_CONTEXT.getQueryId()),
            tsFileManager.getTsFileList(true),
            new ArrayList<>(),
            true);
    int count = 0;
    while (tsFilesReader.hasNextBatch()) {
      TsBlock batchData = tsFilesReader.nextBatch();
      for (int i = 0, size = batchData.getPositionCount(); i < size; i++) {
        assertEquals(batchData.getTimeByIndex(i), batchData.getColumn(0).getDouble(i), 0.001);
        count++;
      }
    }
    assertEquals(500, count);
  }

  public void deleteFileIfExists(File file) {
    long waitingTime = 0l;
    while (file.exists()) {
      file.delete();
      System.gc();
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {

      }
      waitingTime += 100;
      if (waitingTime > 20_000) {
        System.out.println("fail to delete " + file);
        break;
      }
    }
  }

  @Test
  public void testWhenTargetFileShouldBeDeletedAfterCompactionAndSomeSourceFilesLost()
      throws Exception {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);

    // generate mods file, the target file should be deleted after compaction
    for (int device = 0; device < deviceNum; device++) {
      for (int measurement = 0; measurement < measurementNum; measurement++) {
        Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
        deleteMap.put(
            COMPACTION_TEST_SG
                + PATH_SEPARATOR
                + "device"
                + device
                + PATH_SEPARATOR
                + "sensor"
                + measurement,
            new Pair(Long.MIN_VALUE, Long.MAX_VALUE));
        seqResources.forEach(
            x -> {
              try {
                CompactionFileGeneratorUtils.generateMods(deleteMap, x, false);
              } catch (IllegalPathException | IOException e) {
                throw new RuntimeException(e);
              }
            });
      }
    }

    List<TsFileResource> targetResources =
        CompactionFileGeneratorUtils.getInnerCompactionTargetTsFileResources(seqResources, true);
    File compactionLogFile =
        new File(
            SEQ_DIRS,
            targetResources.get(0).getTsFile().getName()
                + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
    compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);

    ICompactionPerformer performer =
        new FastCompactionPerformer(seqResources, Collections.emptyList(), targetResources);
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    CompactionUtils.moveTargetFile(targetResources, true, COMPACTION_TEST_SG);
    CompactionUtils.combineModsInInnerCompaction(seqResources, targetResources.get(0));
    compactionLogger.logFile(targetResources.get(0), STR_DELETED_TARGET_FILES);
    compactionLogger.close();
    seqResources.get(0).remove();

    // recover compaction
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();

    Assert.assertTrue(tsFileManager.isAllowCompaction());

    // all source file should not exist
    for (TsFileResource resource : seqResources) {
      Assert.assertFalse(resource.getTsFile().exists());
      Assert.assertFalse(
          new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
      Assert.assertFalse(resource.getModFile().exists());
      Assert.assertFalse(resource.getCompactionModFile().exists());
    }
    // the target file should be deleted
    Assert.assertFalse(targetResources.get(0).getTsFile().exists());
    Assert.assertFalse(targetResources.get(0).resourceFileExists());
  }

  @Test
  public void testWhenTargetFileIsDeletedAfterCompactionAndSomeSourceFilesLost() throws Exception {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);

    // generate mods file, the target file should be deleted after compaction
    for (int device = 0; device < deviceNum; device++) {
      for (int measurement = 0; measurement < measurementNum; measurement++) {
        Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
        deleteMap.put(
            COMPACTION_TEST_SG
                + PATH_SEPARATOR
                + "device"
                + device
                + PATH_SEPARATOR
                + "sensor"
                + measurement,
            new Pair(Long.MIN_VALUE, Long.MAX_VALUE));
        seqResources.forEach(
            x -> {
              try {
                CompactionFileGeneratorUtils.generateMods(deleteMap, x, false);
              } catch (IllegalPathException | IOException e) {
                throw new RuntimeException(e);
              }
            });
      }
    }

    List<TsFileResource> targetResources =
        CompactionFileGeneratorUtils.getInnerCompactionTargetTsFileResources(seqResources, true);
    File compactionLogFile =
        new File(
            SEQ_DIRS,
            targetResources.get(0).getTsFile().getName()
                + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
    compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);

    ICompactionPerformer performer =
        new FastCompactionPerformer(seqResources, Collections.emptyList(), targetResources);
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    CompactionUtils.moveTargetFile(targetResources, true, COMPACTION_TEST_SG);
    CompactionUtils.combineModsInInnerCompaction(seqResources, targetResources.get(0));
    compactionLogger.logFile(targetResources.get(0), STR_DELETED_TARGET_FILES);
    compactionLogger.close();
    CompactionUtils.deleteTsFilesInDisk(seqResources, COMPACTION_TEST_SG);
    CompactionUtils.deleteModificationForSourceFile(seqResources, COMPACTION_TEST_SG);

    if (targetResources.get(0).isDeleted()) {
      targetResources.get(0).remove();
    }

    // recover compaction
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();

    Assert.assertTrue(tsFileManager.isAllowCompaction());

    // all source file should not exist
    for (TsFileResource resource : seqResources) {
      Assert.assertFalse(resource.getTsFile().exists());
      Assert.assertFalse(
          new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
      Assert.assertFalse(resource.getModFile().exists());
      Assert.assertFalse(resource.getCompactionModFile().exists());
    }
    // the target file should be deleted
    Assert.assertFalse(targetResources.get(0).getTsFile().exists());
    Assert.assertFalse(targetResources.get(0).resourceFileExists());
  }

  @Test
  public void testWhenTargetFileIsDeletedAfterCompactionAndAllSourceFilesExisted()
      throws Exception {
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);

    // generate mods file, the target file should be deleted after compaction
    for (int device = 0; device < deviceNum; device++) {
      for (int measurement = 0; measurement < measurementNum; measurement++) {
        Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
        deleteMap.put(
            COMPACTION_TEST_SG
                + PATH_SEPARATOR
                + "device"
                + device
                + PATH_SEPARATOR
                + "sensor"
                + measurement,
            new Pair(Long.MIN_VALUE, Long.MAX_VALUE));
        seqResources.forEach(
            x -> {
              try {
                CompactionFileGeneratorUtils.generateMods(deleteMap, x, false);
              } catch (IOException | IllegalPathException e) {
                throw new RuntimeException(e);
              }
            });
      }
    }

    List<TsFileResource> targetResources =
        CompactionFileGeneratorUtils.getInnerCompactionTargetTsFileResources(seqResources, true);
    File compactionLogFile =
        new File(
            SEQ_DIRS,
            targetResources.get(0).getTsFile().getName()
                + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
    CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
    compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
    compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
    compactionLogger.close();
    ICompactionPerformer performer =
        new FastCompactionPerformer(seqResources, Collections.emptyList(), targetResources);
    performer.setSummary(new FastCompactionTaskSummary());
    performer.perform();
    CompactionUtils.moveTargetFile(targetResources, true, COMPACTION_TEST_SG);
    CompactionUtils.combineModsInInnerCompaction(seqResources, targetResources.get(0));

    // recover compaction
    new CompactionRecoverTask(COMPACTION_TEST_SG, "0", tsFileManager, compactionLogFile, true)
        .doCompaction();

    Assert.assertTrue(tsFileManager.isAllowCompaction());

    // all source file should exist
    for (TsFileResource resource : seqResources) {
      Assert.assertTrue(resource.getTsFile().exists());
      Assert.assertTrue(
          new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
      Assert.assertTrue(resource.getModFile().exists());
      Assert.assertFalse(resource.getCompactionModFile().exists());
    }
    // tmp target file, target file and target resource file should be deleted after compaction
    for (TsFileResource resource : targetResources) {
      if (resource == null) {
        continue;
      }
      Assert.assertFalse(resource.getTsFile().exists());
      Assert.assertFalse(
          new File(
                  resource
                      .getTsFilePath()
                      .replace(
                          IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
                          TsFileConstant.TSFILE_SUFFIX))
              .exists());
      Assert.assertFalse(
          new File(
                  resource
                          .getTsFilePath()
                          .replace(
                              IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
                              TsFileConstant.TSFILE_SUFFIX)
                      + TsFileResource.RESOURCE_SUFFIX)
              .exists());
    }
  }

  private void closeTsFileSequenceReader() throws IOException {
    for (TsFileResource tsFileResource : seqResources) {
      FileReaderManager.getInstance().closeFileAndRemoveReader(tsFileResource.getTsFilePath());
    }
  }
}
