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

import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.RepairUnsortedFileCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionScheduleSummary;
import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionScheduleTaskManager;
import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionScheduler;
import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager;
import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTestFileWriter;
import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion;
import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileRepairStatus;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.metadata.AlignedChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.IChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.PlainDeviceID;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.read.common.Chunk;
import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.read.reader.IPointReader;
import org.apache.iotdb.tsfile.read.reader.chunk.AlignedChunkReader;

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

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class RepairUnsortedFileCompactionTest extends AbstractRepairDataTest {

  private boolean enableSeqSpaceCompaction =
      IoTDBDescriptor.getInstance().getConfig().isEnableSeqSpaceCompaction();
  private boolean enableUnSeqSpaceCompaction =
      IoTDBDescriptor.getInstance().getConfig().isEnableUnseqSpaceCompaction();
  private boolean enableCrossSpaceCompaction =
      IoTDBDescriptor.getInstance().getConfig().isEnableCrossSpaceCompaction();

  @Before
  public void setUp()
      throws IOException, WriteProcessException, MetadataException, InterruptedException {
    super.setUp();
    Thread.currentThread().setName("pool-1-IoTDB-Compaction-Worker-1");
    IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(true);
    IoTDBDescriptor.getInstance().getConfig().setEnableUnseqSpaceCompaction(true);
    IoTDBDescriptor.getInstance().getConfig().setEnableCrossSpaceCompaction(true);
    try {
      CompactionScheduleTaskManager.getInstance().start();
    } catch (StartupException e) {
      throw new RuntimeException(e);
    }
  }

  @After
  public void tearDown() throws IOException, StorageEngineException {
    IoTDBDescriptor.getInstance().getConfig().setEnableSeqSpaceCompaction(enableSeqSpaceCompaction);
    IoTDBDescriptor.getInstance()
        .getConfig()
        .setEnableUnseqSpaceCompaction(enableUnSeqSpaceCompaction);
    IoTDBDescriptor.getInstance()
        .getConfig()
        .setEnableCrossSpaceCompaction(enableCrossSpaceCompaction);
    super.tearDown();
  }

  @Test
  public void testRepairUnsortedDataBetweenPageWithNonAlignedSeries() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleNonAlignedSeriesToCurrentDevice(
          "s1",
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(5, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    Assert.assertFalse(TsFileResourceUtils.validateTsFileDataCorrectness(resource));
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, resource.isSeq(), 0);
    task.start();
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileDataCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
  }

  @Test
  public void testRepairUnsortedDataBetweenPageWithAlignedSeries() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(5, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    Assert.assertFalse(TsFileResourceUtils.validateTsFileDataCorrectness(resource));
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, resource.isSeq(), 0);
    task.start();
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileDataCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
  }

  @Test
  public void testRepairUnsortedDataInOnePageWithNonAlignedSeries() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleNonAlignedSeriesToCurrentDevice(
          "s1",
          new TimeRange[][][] {
            new TimeRange[][] {
              new TimeRange[] {new TimeRange(10, 20), new TimeRange(29, 30), new TimeRange(21, 25)}
            }
          },
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    Assert.assertFalse(TsFileResourceUtils.validateTsFileDataCorrectness(resource));
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, resource.isSeq(), 0);
    task.start();
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileDataCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
  }

  @Test
  public void testRepairUnsortedDataInOnePageWithMultiNonAlignedSeries() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      for (int i = 0; i < 1000; i++) {
        writer.generateSimpleNonAlignedSeriesToCurrentDevice(
            "s" + i,
            new TimeRange[][][] {
              new TimeRange[][] {
                new TimeRange[] {
                  new TimeRange(10, 20), new TimeRange(29, 30), new TimeRange(21, 25)
                }
              }
            },
            TSEncoding.PLAIN,
            CompressionType.LZ4);
      }
      writer.endChunkGroup();
      writer.endFile();
    }
    Assert.assertFalse(TsFileResourceUtils.validateTsFileDataCorrectness(resource));
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, resource.isSeq(), 0);
    task.start();
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileDataCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
  }

  @Test
  public void testRepairUnsortedDataInOnePageWithUnseqFile() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(false);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleNonAlignedSeriesToCurrentDevice(
          "s1",
          new TimeRange[][][] {
            new TimeRange[][] {
              new TimeRange[] {new TimeRange(10, 20), new TimeRange(29, 30), new TimeRange(21, 25)}
            }
          },
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    Assert.assertFalse(TsFileResourceUtils.validateTsFileDataCorrectness(resource));
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, resource.isSeq(), 0);
    task.start();
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileDataCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
  }

  @Test
  public void testRepairUnsortedDataInOnePageWithAlignedSeries() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][][] {
            new TimeRange[][] {
              new TimeRange[] {new TimeRange(10, 20), new TimeRange(29, 30), new TimeRange(21, 25)}
            }
          },
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    Assert.assertFalse(TsFileResourceUtils.validateTsFileDataCorrectness(resource));
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, resource.isSeq(), 0);
    task.start();
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileDataCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(
            tsFileManager.getOrCreateSequenceListByTimePartition(0)));
  }

  @Test
  public void testMarkFileAndRepairWithInnerSeqSpaceCompactionTask()
      throws IOException, InterruptedException {
    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][][] {
            new TimeRange[][] {
              new TimeRange[] {new TimeRange(10, 20), new TimeRange(29, 30), new TimeRange(21, 25)}
            }
          },
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(5, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    InnerSpaceCompactionTask task =
        new InnerSpaceCompactionTask(
            0,
            tsFileManager,
            Arrays.asList(seqResource1, seqResource2),
            true,
            new ReadChunkCompactionPerformer(),
            0);
    Assert.assertFalse(task.start());

    for (TsFileResource resource : tsFileManager.getTsFileList(true)) {
      Assert.assertEquals(resource.getTsFileRepairStatus(), TsFileRepairStatus.NEED_TO_REPAIR);
    }

    long initialFinishedCompactionTaskNum =
        CompactionTaskManager.getInstance().getFinishedTaskNum();
    CompactionScheduleSummary summary = new CompactionScheduleSummary();
    CompactionScheduler.scheduleCompaction(tsFileManager, 0, summary);
    Assert.assertEquals(2, summary.getSubmitSeqInnerSpaceCompactionTaskNum());

    int waitSecond = 20;
    while (CompactionTaskManager.getInstance().getFinishedTaskNum()
            - initialFinishedCompactionTaskNum
        < 2) {
      if (waitSecond == 0) {
        Assert.fail("Exceed the max time to wait repair compaction");
      }
      try {
        Thread.sleep(TimeUnit.SECONDS.toMillis(1));
        waitSecond--;
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(2, tsFileManager.getTsFileList(false).size());
    for (TsFileResource resource : tsFileManager.getTsFileList(false)) {
      TsFileResourceUtils.validateTsFileDataCorrectness(resource);
    }
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(
            tsFileManager.getOrCreateSequenceListByTimePartition(0)));
  }

  @Test
  public void testMarkFileAndRepairWithInnerUnSeqSpaceCompactionTask()
      throws IOException, InterruptedException {
    TsFileResource unSeqResource1 = createEmptyFileAndResource(false);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(unSeqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][][] {
            new TimeRange[][] {
              new TimeRange[] {new TimeRange(10, 20), new TimeRange(29, 30), new TimeRange(21, 25)}
            }
          },
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource unSeqResource2 = createEmptyFileAndResource(false);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(unSeqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(5, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    InnerSpaceCompactionTask task =
        new InnerSpaceCompactionTask(
            0,
            tsFileManager,
            Arrays.asList(unSeqResource1, unSeqResource2),
            false,
            new FastCompactionPerformer(false),
            0);
    Assert.assertFalse(task.start());

    for (TsFileResource resource : tsFileManager.getTsFileList(true)) {
      Assert.assertEquals(resource.getTsFileRepairStatus(), TsFileRepairStatus.NEED_TO_REPAIR);
    }

    long initialFinishedCompactionTaskNum =
        CompactionTaskManager.getInstance().getFinishedTaskNum();
    CompactionScheduleSummary summary = new CompactionScheduleSummary();
    CompactionScheduler.scheduleCompaction(tsFileManager, 0, summary);
    Assert.assertEquals(2, summary.getSubmitUnseqInnerSpaceCompactionTaskNum());

    int waitSecond = 20;
    while (CompactionTaskManager.getInstance().getFinishedTaskNum()
            - initialFinishedCompactionTaskNum
        < 2) {
      if (waitSecond == 0) {
        Assert.fail("Exceed the max time to wait repair compaction");
      }
      try {
        Thread.sleep(TimeUnit.SECONDS.toMillis(1));
        waitSecond--;
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(2, tsFileManager.getTsFileList(false).size());
    for (TsFileResource resource : tsFileManager.getTsFileList(false)) {
      TsFileResourceUtils.validateTsFileDataCorrectness(resource);
    }
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(
            tsFileManager.getOrCreateSequenceListByTimePartition(0)));
  }

  @Test
  public void testMarkFileAndRepairWithCrossSpaceCompactionTask()
      throws IOException, InterruptedException {
    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][][] {
            new TimeRange[][] {
              new TimeRange[] {new TimeRange(10, 20), new TimeRange(29, 30), new TimeRange(21, 25)}
            }
          },
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource unSeqResource1 = createEmptyFileAndResource(false);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(unSeqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(5, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    CrossSpaceCompactionTask task =
        new CrossSpaceCompactionTask(
            0,
            tsFileManager,
            Collections.singletonList(seqResource1),
            Collections.singletonList(unSeqResource1),
            new FastCompactionPerformer(true),
            0,
            0);
    Assert.assertFalse(task.start());

    for (TsFileResource resource : tsFileManager.getTsFileList(true)) {
      Assert.assertEquals(resource.getTsFileRepairStatus(), TsFileRepairStatus.NEED_TO_REPAIR);
    }

    long initialFinishedCompactionTaskNum =
        CompactionTaskManager.getInstance().getFinishedTaskNum();
    CompactionScheduleSummary summary = new CompactionScheduleSummary();
    CompactionScheduler.scheduleCompaction(tsFileManager, 0, summary);
    Assert.assertEquals(1, summary.getSubmitSeqInnerSpaceCompactionTaskNum());
    Assert.assertEquals(1, summary.getSubmitUnseqInnerSpaceCompactionTaskNum());

    int waitSecond = 20;
    while (CompactionTaskManager.getInstance().getFinishedTaskNum()
            - initialFinishedCompactionTaskNum
        < 2) {
      if (waitSecond == 0) {
        Assert.fail("Exceed the max time to wait repair compaction");
      }
      try {
        Thread.sleep(TimeUnit.SECONDS.toMillis(1));
        waitSecond--;
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
    Assert.assertEquals(0, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(2, tsFileManager.getTsFileList(false).size());
    for (TsFileResource resource : tsFileManager.getTsFileList(false)) {
      TsFileResourceUtils.validateTsFileDataCorrectness(resource);
    }
  }

  @Test
  public void testRepairOverlapBetweenFile() throws IOException {
    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(25, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(20, 30), new TimeRange(35, 40)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    seqResources.add(seqResource1);
    seqResources.add(seqResource2);

    tsFileManager.addAll(seqResources, true);
    Assert.assertFalse(TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(seqResources));

    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, seqResource2, true, false, 0);
    Assert.assertTrue(task.start());
    Assert.assertEquals(1, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileDataCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
    Assert.assertTrue(
        TsFileResourceUtils.validateTsFileResourceCorrectness(
            tsFileManager.getTsFileList(false).get(0)));
  }

  @Test
  public void testRepairOverlapBetweenFileWithModFile() throws IOException, IllegalPathException {
    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(25, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(20, 30), new TimeRange(35, 40)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    ModificationFile modFile = seqResource2.getModFile();
    Deletion writedModification =
        new Deletion(new PartialPath("root.testsg.d1.s1"), Long.MAX_VALUE, 15);
    modFile.write(writedModification);
    modFile.close();

    seqResources.add(seqResource1);
    seqResources.add(seqResource2);

    tsFileManager.addAll(seqResources, true);
    Assert.assertFalse(TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(seqResources));

    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, seqResource2, true, false, 0);
    Assert.assertTrue(task.start());
    Assert.assertEquals(1, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
    TsFileResource targetResource = tsFileManager.getTsFileList(false).get(0);
    Assert.assertTrue(TsFileResourceUtils.validateTsFileDataCorrectness(targetResource));
    Assert.assertTrue(TsFileResourceUtils.validateTsFileResourceCorrectness(targetResource));
    Assert.assertTrue(targetResource.modFileExists());
    Assert.assertEquals(1, targetResource.getModFile().getModifications().size());
    Deletion modification =
        (Deletion) targetResource.getModFile().getModifications().iterator().next();
    Assert.assertEquals(writedModification.getFileOffset(), modification.getFileOffset());
    Assert.assertEquals(writedModification.getEndTime(), modification.getEndTime());
  }

  @Test
  public void testScheduleRepairInternalUnsortedFile() throws IOException {
    DataRegion mockDataRegion = Mockito.mock(DataRegion.class);
    Mockito.when(mockDataRegion.getTsFileManager()).thenReturn(tsFileManager);
    Mockito.when(mockDataRegion.getDatabaseName()).thenReturn("root.testsg");
    Mockito.when(mockDataRegion.getDataRegionId()).thenReturn("0");
    Mockito.when(mockDataRegion.getTimePartitions()).thenReturn(Collections.singletonList(0L));

    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(15, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(40, 50), new TimeRange(55, 60)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    seqResources.add(seqResource1);
    seqResources.add(seqResource2);
    tsFileManager.addAll(seqResources, true);
    Assert.assertTrue(TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(seqResources));

    File tempDir = getEmptyRepairDataLogDir();

    CompactionScheduleTaskManager.getRepairTaskManagerInstance().markRepairTaskStart();
    UnsortedFileRepairTaskScheduler scheduler =
        new UnsortedFileRepairTaskScheduler(
            Collections.singletonList(mockDataRegion), false, tempDir);
    scheduler.run();
    Assert.assertEquals(1, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
  }

  @Test
  public void testScheduleRepairOverlapFile() throws IOException {
    DataRegion mockDataRegion = Mockito.mock(DataRegion.class);
    Mockito.when(mockDataRegion.getTsFileManager()).thenReturn(tsFileManager);
    Mockito.when(mockDataRegion.getDatabaseName()).thenReturn("root.testsg");
    Mockito.when(mockDataRegion.getDataRegionId()).thenReturn("0");
    Mockito.when(mockDataRegion.getTimePartitions()).thenReturn(Collections.singletonList(0L));

    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(25, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 50), new TimeRange(55, 60)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    seqResources.add(seqResource1);
    seqResources.add(seqResource2);
    tsFileManager.addAll(seqResources, true);
    Assert.assertFalse(TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(seqResources));

    File tempDir = getEmptyRepairDataLogDir();

    CompactionScheduleTaskManager.getRepairTaskManagerInstance().markRepairTaskStart();
    UnsortedFileRepairTaskScheduler scheduler =
        new UnsortedFileRepairTaskScheduler(
            Collections.singletonList(mockDataRegion), false, tempDir);
    scheduler.run();
    Assert.assertEquals(1, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(1, tsFileManager.getTsFileList(false).size());
  }

  @Test
  public void testScheduleRepairOverlapFileAndInternalUnsortedFile() throws IOException {
    DataRegion mockDataRegion = Mockito.mock(DataRegion.class);
    Mockito.when(mockDataRegion.getTsFileManager()).thenReturn(tsFileManager);
    Mockito.when(mockDataRegion.getDatabaseName()).thenReturn("root.testsg");
    Mockito.when(mockDataRegion.getDataRegionId()).thenReturn("0");
    Mockito.when(mockDataRegion.getTimePartitions()).thenReturn(Collections.singletonList(0L));

    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(15, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 50), new TimeRange(55, 60)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    TsFileResource seqResource3 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource3)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(40, 80)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    seqResources.add(seqResource1);
    seqResources.add(seqResource2);
    seqResources.add(seqResource3);
    tsFileManager.addAll(seqResources, true);
    Assert.assertFalse(TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(seqResources));

    File tempDir = getEmptyRepairDataLogDir();

    CompactionScheduleTaskManager.getRepairTaskManagerInstance().markRepairTaskStart();
    UnsortedFileRepairTaskScheduler scheduler =
        new UnsortedFileRepairTaskScheduler(
            Collections.singletonList(mockDataRegion), false, tempDir);
    scheduler.run();
    Assert.assertEquals(1, tsFileManager.getTsFileList(true).size());
    Assert.assertEquals(2, tsFileManager.getTsFileList(false).size());
  }

  @Test
  public void testRecoverRepairScheduleSkipRepairedTimePartitionAndMarkFile() throws IOException {
    DataRegion mockDataRegion = Mockito.mock(DataRegion.class);
    Mockito.when(mockDataRegion.getTsFileManager()).thenReturn(tsFileManager);
    Mockito.when(mockDataRegion.getDatabaseName()).thenReturn("root.testsg");
    Mockito.when(mockDataRegion.getDataRegionId()).thenReturn("0");
    Mockito.when(mockDataRegion.getTimePartitions()).thenReturn(Collections.singletonList(0L));

    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(15, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 50), new TimeRange(55, 60)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    TsFileResource seqResource3 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource3)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(40, 80)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    seqResource3.setTsFileRepairStatus(TsFileRepairStatus.CAN_NOT_REPAIR);
    seqResources.add(seqResource1);
    seqResources.add(seqResource2);
    seqResources.add(seqResource3);
    tsFileManager.addAll(seqResources, true);
    Assert.assertFalse(TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(seqResources));

    File tempDir = getEmptyRepairDataLogDir();
    try (RepairLogger logger = new RepairLogger(tempDir, false)) {
      logger.recordRepairTaskStartTimeIfLogFileEmpty(System.currentTimeMillis());
      RepairTimePartition timePartition =
          new RepairTimePartition(mockDataRegion, 0, System.currentTimeMillis());
      // record seqResource3 as cannot recover
      logger.recordRepairedTimePartition(timePartition);
    }
    // reset the repair status
    seqResource3.setTsFileRepairStatus(TsFileRepairStatus.NORMAL);

    CompactionScheduleTaskManager.getRepairTaskManagerInstance().markRepairTaskStart();
    UnsortedFileRepairTaskScheduler scheduler =
        new UnsortedFileRepairTaskScheduler(
            Collections.singletonList(mockDataRegion), true, tempDir);
    scheduler.run();
    Assert.assertEquals(3, tsFileManager.getTsFileList(true).size());
    // check whether the repair status is marked correctly
    Assert.assertEquals(TsFileRepairStatus.NEED_TO_REPAIR, seqResource3.getTsFileRepairStatus());
  }

  @Test
  public void testRecoverRepairScheduleSkipRepairedTimePartitionWithDeletedFile()
      throws IOException {
    DataRegion mockDataRegion = Mockito.mock(DataRegion.class);
    Mockito.when(mockDataRegion.getTsFileManager()).thenReturn(tsFileManager);
    Mockito.when(mockDataRegion.getDatabaseName()).thenReturn("root.testsg");
    Mockito.when(mockDataRegion.getDataRegionId()).thenReturn("0");
    Mockito.when(mockDataRegion.getTimePartitions()).thenReturn(Collections.singletonList(0L));

    TsFileResource seqResource1 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource1)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(15, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }

    TsFileResource seqResource2 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource2)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 50), new TimeRange(55, 60)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    TsFileResource seqResource3 = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(seqResource3)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDevice(
          Arrays.asList("s1", "s2"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(40, 80)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    seqResource3.setTsFileRepairStatus(TsFileRepairStatus.CAN_NOT_REPAIR);
    seqResources.add(seqResource1);
    seqResources.add(seqResource2);
    seqResources.add(seqResource3);
    tsFileManager.addAll(seqResources, true);
    Assert.assertFalse(TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(seqResources));

    File tempDir = getEmptyRepairDataLogDir();
    try (RepairLogger logger = new RepairLogger(tempDir, false)) {
      logger.recordRepairTaskStartTimeIfLogFileEmpty(System.currentTimeMillis());
      RepairTimePartition timePartition =
          new RepairTimePartition(mockDataRegion, 0, System.currentTimeMillis());
      // record seqResource3 as cannot recover
      logger.recordRepairedTimePartition(timePartition);
    }
    // reset the repair status
    seqResource3.setTsFileRepairStatus(TsFileRepairStatus.NORMAL);
    // resource3 is deleted
    tsFileManager.replace(
        Collections.singletonList(seqResource3),
        Collections.emptyList(),
        Collections.emptyList(),
        0);

    CompactionScheduleTaskManager.getRepairTaskManagerInstance().markRepairTaskStart();
    UnsortedFileRepairTaskScheduler scheduler =
        new UnsortedFileRepairTaskScheduler(
            Collections.singletonList(mockDataRegion), true, tempDir);
    scheduler.run();
    Assert.assertEquals(2, tsFileManager.getTsFileList(true).size());
  }

  @Test
  public void testTimePartitionFilterFiles() {
    DataRegion mockDataRegion = Mockito.mock(DataRegion.class);
    Mockito.when(mockDataRegion.getTsFileManager()).thenReturn(tsFileManager);
    Mockito.when(mockDataRegion.getTimePartitions()).thenReturn(Collections.singletonList(0L));

    TsFileResource seqResource1 = createEmptyFileAndResourceWithName("100-1-0-0.tsfile", 0, true);
    TsFileResource seqResource2 = createEmptyFileAndResourceWithName("200-3-0-0.tsfile", 0, true);
    TsFileResource seqResource3 = createEmptyFileAndResourceWithName("300-5-0-0.tsfile", 0, true);
    TsFileResource unseqResource1 =
        createEmptyFileAndResourceWithName("101-2-0-0.tsfile", 0, false);
    TsFileResource unseqResource2 =
        createEmptyFileAndResourceWithName("201-4-0-0.tsfile", 0, false);
    TsFileResource unseqResource3 =
        createEmptyFileAndResourceWithName("301-6-0-0.tsfile", 0, false);
    seqResources.add(seqResource1);
    seqResources.add(seqResource2);
    seqResources.add(seqResource3);
    unseqResources.add(unseqResource1);
    unseqResources.add(unseqResource2);
    unseqResources.add(unseqResource3);
    tsFileManager.addAll(seqResources, true);
    tsFileManager.addAll(unseqResources, false);

    RepairTimePartition timePartition = new RepairTimePartition(mockDataRegion, 0, 250);
    Assert.assertEquals(4, timePartition.getAllFileSnapshot().size());
    Assert.assertEquals(2, timePartition.getSeqFileSnapshot().size());
    Assert.assertEquals(2, timePartition.getUnSeqFileSnapshot().size());
  }

  @Test
  public void testEstimateRepairCompactionMemory() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleNonAlignedSeriesToCurrentDevice(
          "s1",
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20), new TimeRange(5, 30)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4);
      writer.endChunkGroup();
      writer.endFile();
    }
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, true, true, 0);
    Assert.assertTrue(task.getEstimatedMemoryCost() > 0);
    task = new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, true, false, 0);
    Assert.assertEquals(0, task.getEstimatedMemoryCost());
  }

  @Test
  public void testMergeAlignedSeriesPointWithSameTimestamp() throws IOException {
    TsFileResource resource = createEmptyFileAndResource(true);
    try (CompactionTestFileWriter writer = new CompactionTestFileWriter(resource)) {
      writer.startChunkGroup("d1");
      writer.generateSimpleAlignedSeriesToCurrentDeviceWithNullValue(
          Arrays.asList("s1", "s2", "s3"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4,
          Arrays.asList(true, false, false));
      writer.generateSimpleAlignedSeriesToCurrentDeviceWithNullValue(
          Arrays.asList("s1", "s2", "s3"),
          new TimeRange[][] {new TimeRange[] {new TimeRange(10, 20)}},
          TSEncoding.PLAIN,
          CompressionType.LZ4,
          Arrays.asList(false, true, true));
      writer.endChunkGroup();
      writer.endFile();
    }
    RepairUnsortedFileCompactionTask task =
        new RepairUnsortedFileCompactionTask(0, tsFileManager, resource, true, true, 0);
    Assert.assertTrue(task.start());
    TsFileResource target = tsFileManager.getTsFileList(false).get(0);
    try (TsFileSequenceReader reader = new TsFileSequenceReader(target.getTsFilePath())) {
      List<AlignedChunkMetadata> chunkMetadataList =
          reader.getAlignedChunkMetadata(new PlainDeviceID("root.testsg.d1"));
      for (AlignedChunkMetadata alignedChunkMetadata : chunkMetadataList) {
        ChunkMetadata timeChunkMetadata =
            (ChunkMetadata) alignedChunkMetadata.getTimeChunkMetadata();
        Chunk timeChunk = reader.readMemChunk(timeChunkMetadata);
        List<Chunk> valueChunks = new ArrayList<>();
        for (IChunkMetadata chunkMetadata : alignedChunkMetadata.getValueChunkMetadataList()) {
          Chunk valueChunk = reader.readMemChunk((ChunkMetadata) chunkMetadata);
          valueChunks.add(valueChunk);
        }
        AlignedChunkReader chunkReader = new AlignedChunkReader(timeChunk, valueChunks, null);
        while (chunkReader.hasNextSatisfiedPage()) {
          BatchData batchData = chunkReader.nextPageData();
          IPointReader pointReader = batchData.getBatchDataIterator();
          while (pointReader.hasNextTimeValuePair()) {
            TimeValuePair timeValuePair = pointReader.nextTimeValuePair();
            for (Object value : timeValuePair.getValues()) {
              if (value == null) {
                Assert.fail();
              }
            }
          }
        }
      }
    }
  }
}