/*
 * 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.carbondata.core.util;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.metadata.SegmentFileStore;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
import org.apache.carbondata.core.statusmanager.SegmentStatus;
import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.core.util.path.CarbonTablePath.DataFileUtil;

import org.apache.log4j.Logger;

/**
 *This util provide clean stale data methods for clean files command
 */
public class CleanFilesUtil {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(CleanFilesUtil.class.getName());

  /**
   * This method will clean all the stale segments for a table, delete the source folder after
   * copying the data to the trash and also remove the .segment files of the stale segments
   */
  public static void cleanStaleSegments(CarbonTable carbonTable)
      throws IOException {
    long timeStampForTrashFolder = CarbonUpdateUtil.readCurrentTime();
    List<String> staleSegmentFiles = new ArrayList<>();
    List<String> redundantSegmentFile = new ArrayList<>();
    getStaleSegmentFiles(carbonTable, staleSegmentFiles, redundantSegmentFile);
    for (String staleSegmentFile : staleSegmentFiles) {
      SegmentFileStore fileStore = new SegmentFileStore(carbonTable.getTablePath(),
          staleSegmentFile);
      SegmentFileStore.SegmentFile segmentFile = fileStore.getSegmentFile();
      if (segmentFile != null) {
        Map<String, SegmentFileStore.FolderDetails> locationMap = segmentFile.getLocationMap();
        if (locationMap != null) {
          if (locationMap.entrySet().iterator().next().getValue().isRelative()) {
            String segmentNumber = DataFileUtil.getSegmentNoFromSegmentFile(staleSegmentFile);
            CarbonFile segmentPath = FileFactory.getCarbonFile(CarbonTablePath.getSegmentPath(
                carbonTable.getTablePath(), segmentNumber));
            // copy the complete segment to the trash folder
            TrashUtil.copySegmentToTrash(segmentPath, TrashUtil.getCompleteTrashFolderPath(
                carbonTable.getTablePath(), timeStampForTrashFolder, segmentNumber));
            // Deleting the stale Segment folders and the segment file.
            try {
              CarbonUtil.deleteFoldersAndFiles(segmentPath);
              // delete the segment file as well
              FileFactory.deleteFile(CarbonTablePath.getSegmentFilePath(carbonTable.getTablePath(),
                  staleSegmentFile));
              List<String> deletedFiles = new ArrayList<>();
              deletedFiles.add(staleSegmentFile);
              for (String duplicateStaleSegmentFile : redundantSegmentFile) {
                if (DataFileUtil.getSegmentNoFromSegmentFile(duplicateStaleSegmentFile)
                    .equals(segmentNumber)) {
                  FileFactory.deleteFile(CarbonTablePath.getSegmentFilePath(carbonTable
                      .getTablePath(), duplicateStaleSegmentFile));
                  deletedFiles.add(duplicateStaleSegmentFile);
                }
              }
              LOGGER.info("Deleted the Segment :" + segmentPath.getName() + " after"
                  + " moving it to the trash folder");
              LOGGER.info("Deleted stale segment files: " + String.join(",", deletedFiles));
            } catch (IOException | InterruptedException e) {
              LOGGER.error("Unable to delete the segment: " + segmentPath + " from after moving" +
                  " it to the trash folder. Please delete them manually : " + e.getMessage(), e);
            }
          }
        }
      }
    }
  }

  /**
   * This method will clean all the stale segments for partition table, delete the source folders
   * after copying the data to the trash and also remove the .segment files of the stale segments
   */
  public static void cleanStaleSegmentsForPartitionTable(CarbonTable carbonTable)
      throws IOException {
    long timeStampForTrashFolder = CarbonUpdateUtil.readCurrentTime();
    List<String> staleSegmentFiles = new ArrayList<>();
    List<String> redundantSegmentFile = new ArrayList<>();
    getStaleSegmentFiles(carbonTable, staleSegmentFiles, redundantSegmentFile);
    for (String staleSegmentFile : staleSegmentFiles) {
      String segmentNumber = DataFileUtil.getSegmentNoFromSegmentFile(staleSegmentFile);
      // for each segment we get the indexfile first, then we get the carbondata file. Move both
      // of those to trash folder
      SegmentFileStore fileStore = new SegmentFileStore(carbonTable.getTablePath(),
          staleSegmentFile);
      List<String> filesToProcess = fileStore.readIndexFiles(SegmentStatus.SUCCESS, true,
          FileFactory.getConfiguration());

      // get carbondata files from here
      Map<String, List<String>> indexFilesMap = fileStore.getIndexFilesMap();
      for (Map.Entry<String, List<String>> entry : indexFilesMap.entrySet()) {
        filesToProcess.addAll(entry.getValue());
      }
      // After all the files have been added to list, move them to the trash folder
      TrashUtil.copyFilesToTrash(filesToProcess, TrashUtil.getCompleteTrashFolderPath(
          carbonTable.getTablePath(), timeStampForTrashFolder, segmentNumber), segmentNumber);
      // After every file of that segment has been copied, need to delete those files.
      try {
        for (String file : filesToProcess) {
          FileFactory.deleteFile(file);
          LOGGER.info("Deleted file :" + file + " after moving it to the trash folder");
        }
        // Delete the segment file too
        FileFactory.deleteFile(CarbonTablePath.getSegmentFilePath(carbonTable.getTablePath(),
            staleSegmentFile));
        LOGGER.info("Deleted stale segment file after moving it to the trash folder :"
            + staleSegmentFile);
        // remove duplicate segment files if any
        for (String duplicateStaleSegmentFile : redundantSegmentFile) {
          if (DataFileUtil.getSegmentNoFromSegmentFile(duplicateStaleSegmentFile)
              .equals(segmentNumber)) {
            FileFactory.deleteFile(CarbonTablePath.getSegmentFilePath(carbonTable.getTablePath(),
                duplicateStaleSegmentFile));
            LOGGER.info("Deleted redundant segment file :" + duplicateStaleSegmentFile);
          }
        }
      } catch (IOException e) {
        LOGGER.error("Error while deleting the source data files. Please delete the files of" +
            " segment: " + segmentNumber + " manually.", e);
      }
    }
    // Delete the empty partition folders
    deleteEmptyPartitionFoldersRecursively(FileFactory.getCarbonFile(carbonTable.getTablePath()));
  }

  /**
   * This method will find all the stale segments by comparing the segment files in the
   * metadata directory with the segments in the table status file. Any segment which has entry
   * in the metadata folder and is not present in the table status file is considered as a
   * stale segment. Only comparing from tablestatus file, not checking tablestatus.history file
   */
  private static void getStaleSegmentFiles(CarbonTable carbonTable, List<String> staleSegmentFiles,
      List<String> redundantSegmentFile) {
    String segmentFilesLocation =
        CarbonTablePath.getSegmentFilesLocation(carbonTable.getTablePath());
    List<String> segmentFiles = Arrays.stream(FileFactory.getCarbonFile(segmentFilesLocation)
        .listFiles()).map(CarbonFile::getName).filter(segmentFileName -> segmentFileName
        .endsWith(CarbonTablePath.SEGMENT_EXT)).collect(Collectors.toList());
    // there are no segments present in the Metadata folder. Can return here
    if (segmentFiles.size() == 0) {
      return;
    }
    LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(carbonTable
        .getMetadataPath(), carbonTable.getTableStatusVersion());
    if (details == null || details.length == 0) {
      return;
    }
    Set<String> loadNameSet = Arrays.stream(details).map(LoadMetadataDetails::getLoadName)
        .collect(Collectors.toSet());
    // get all stale segment files, not include compaction segments.
    // during compaction, we don't add in-progress metadata entry into tablestatus file,
    // so here we don't known whether compaction segment is stale or not.
    List<String> staleSegments = segmentFiles.stream().filter(segmentFile -> {
      String segmentNo = DataFileUtil.getSegmentNoFromSegmentFile(segmentFile);
      return !loadNameSet.contains(segmentNo) && !segmentNo.contains(CarbonCommonConstants.POINT);
    }).collect(Collectors.toList());
    if (staleSegments.size() == 0) {
      return;
    }
    // sort the stale segment List
    Collections.sort(staleSegments);
    // in case of multiple segment files for a segment, add the segment with the largest
    // timestamp to staleSegmentFiles list and add the others to redundantsegmentfile list.
    for (int i = 0; i < staleSegments.size() - 1; i++) {
      if (!DataFileUtil.getSegmentNoFromSegmentFile(staleSegments.get(i)).equals(
          DataFileUtil.getSegmentNoFromSegmentFile(staleSegments.get(i + 1)))) {
        staleSegmentFiles.add(staleSegments.get(i));
      } else {
        redundantSegmentFile.add(staleSegments.get(i));
      }
    }
    // adding the last occurrence always
    staleSegmentFiles.add(staleSegments.get(staleSegments.size() - 1));
  }

  /**
   * This method will delete all the empty partition folders starting from the table path
   */
  public static void deleteEmptyPartitionFoldersRecursively(CarbonFile tablePath) {
    CarbonFile[] listOfFiles = tablePath.listFiles();
    if (listOfFiles.length == 0) {
      tablePath.delete();
      // if parent file folder also empty then delete that too.
      deleteEmptyPartitionFoldersRecursively(tablePath.getParentFile());
    } else {
      for (CarbonFile file: listOfFiles) {
        if (file.isDirectory() && file.getName().contains("=")) {
          deleteEmptyPartitionFoldersRecursively(file);
        }
      }
    }
  }
}
