content: /*
   * 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.spark.sql.secondaryindex.load;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.locks.CarbonLockUtil;
  import org.apache.carbondata.core.locks.ICarbonLock;
  import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
  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.processing.loading.model.CarbonLoadModel;
  import org.apache.carbondata.processing.util.CarbonLoaderUtil;
  
  import org.apache.log4j.Logger;
  import org.apache.spark.sql.index.CarbonIndexUtil;
  import org.apache.spark.sql.secondaryindex.command.ErrorMessage;
  
  public class CarbonInternalLoaderUtil {
  
    private static final Logger LOGGER = LogServiceFactory.getLogService(
        CarbonInternalLoaderUtil.class.getName());
  
    public static List<String> getListOfValidSlices(LoadMetadataDetails[] details) {
      List<String> activeSlices = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      for (LoadMetadataDetails oneLoad : details) {
        // External added segments are not loaded to SI
        if (isaBoolean(oneLoad)) {
          activeSlices.add(oneLoad.getLoadName());
        }
      }
      return activeSlices;
    }
  
    private static boolean isaBoolean(LoadMetadataDetails oneLoad) {
      return oneLoad.getPath() == null && SegmentStatus.SUCCESS.equals(oneLoad.getSegmentStatus())
          || SegmentStatus.LOAD_PARTIAL_SUCCESS.equals(oneLoad.getSegmentStatus())
          || SegmentStatus.MARKED_FOR_UPDATE.equals(oneLoad.getSegmentStatus());
    }
  
    /**
     * This method will return the mapping of valid segments to segment load start time
     */
    public static Map<String, Long> getSegmentToLoadStartTimeMapping(LoadMetadataDetails[] details) {
      Map<String, Long> segmentToLoadStartTimeMap = new HashMap<>(details.length);
      for (LoadMetadataDetails oneLoad : details) {
        // valid segments will only have Success status
        if (SegmentStatus.SUCCESS.equals(oneLoad.getSegmentStatus())
            || SegmentStatus.LOAD_PARTIAL_SUCCESS.equals(oneLoad.getSegmentStatus())) {
          segmentToLoadStartTimeMap.put(oneLoad.getLoadName(), oneLoad.getLoadStartTime());
        }
      }
      return segmentToLoadStartTimeMap;
    }
  
    /**
     * This API will write the load level metadata for the load management module inorder to
     * manage the load and query execution management smoothly.
     *
     * @return boolean which determines whether status update is done or not.
     */
    public static boolean recordLoadMetadata(List<LoadMetadataDetails> newLoadMetadataDetails,
        List<String> validSegments, CarbonTable carbonTable, List<CarbonTable> indexCarbonTables,
        CarbonLoadModel loadModel, String tableName, SegmentStatus segmentStatus,
        boolean isRebuiltSegments) {
      String databaseName = loadModel.getDatabaseName();
      boolean status = false;
      String metaDataFilepath = carbonTable.getMetadataPath();
      AbsoluteTableIdentifier absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier();
      SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier,
          loadModel.getLatestTableStatusWriteVersion());
      ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
      try {
        int retryCount = CarbonLockUtil
            .getLockProperty(CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK,
                CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK_DEFAULT);
        int maxTimeout = CarbonLockUtil
            .getLockProperty(CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK,
                CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT);
        if (carbonLock.lockWithRetries(retryCount, maxTimeout)) {
          LOGGER.info("Acquired lock for table" + databaseName + "." + tableName
              + " for table status update");
  
          if (isSegmentsAlreadyCompactedForNewMetaDataDetails(indexCarbonTables, tableName,
              newLoadMetadataDetails)) {
            return false;
          }
  
          loadModel.setLatestTableStatusWriteVersion(String.valueOf(System.currentTimeMillis()));
          String tblStatusVersion = loadModel.getLatestTableStatusWriteVersion();
          if (segmentStatus == SegmentStatus.INSERT_IN_PROGRESS
              || segmentStatus == SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS || isRebuiltSegments) {
            tblStatusVersion = carbonTable.getTableStatusVersion();
          }
  
          LoadMetadataDetails[] currentLoadMetadataDetails = SegmentStatusManager.readLoadMetadata(
              metaDataFilepath, tblStatusVersion);
  
          List<LoadMetadataDetails> updatedLoadMetadataDetails = new ArrayList<>(
              CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
  
          // check which load needs to be overwritten which are in in progress state
          boolean found = false;
          for (int i = 0; i < currentLoadMetadataDetails.length; i++) {
            for (LoadMetadataDetails newLoadMetadataDetail : newLoadMetadataDetails) {
              if (currentLoadMetadataDetails[i].getLoadName().equals(
                  newLoadMetadataDetail.getLoadName())) {
                currentLoadMetadataDetails[i] = newLoadMetadataDetail;
                found = true;
                break;
              }
            }
            updatedLoadMetadataDetails.add(currentLoadMetadataDetails[i]);
          }
  
          // check if newLoadMetadataDetail has segments which are not in  currentLoadMetaDetails
          // and add them to the updatedLoadMetadataDetails
          boolean foundNext = false;
          for (int i = 0; i < newLoadMetadataDetails.size(); i++) {
            foundNext = false;
            for (int j = 0; j < currentLoadMetadataDetails.length; j++) {
              if (newLoadMetadataDetails.get(i).getLoadName().equals(
                  currentLoadMetadataDetails[j].getLoadName())) {
                foundNext = true;
                break;
              }
              if (j == currentLoadMetadataDetails.length - 1 && !foundNext) {
                // if not found in the list then add it
                updatedLoadMetadataDetails.add(newLoadMetadataDetails.get(i));
                found = true;
              }
            }
          }
  
          // when data load is done for first time, add all the details
          if (currentLoadMetadataDetails.length == 0 || !found) {
            updatedLoadMetadataDetails.addAll(newLoadMetadataDetails);
          }
  
          List<String> indexTables = CarbonIndexUtil.getSecondaryIndexes(carbonTable);
          if (!indexTables.isEmpty()) {
            List<LoadMetadataDetails> newSegmentDetailsListForIndexTable = new ArrayList<>(
                validSegments.size());
            for (String segmentId : validSegments) {
              LoadMetadataDetails newSegmentDetailsObject = new LoadMetadataDetails();
              newSegmentDetailsObject.setLoadName(segmentId);
              newSegmentDetailsListForIndexTable.add(newSegmentDetailsObject);
            }
            for (CarbonTable indexTable : indexCarbonTables) {
              List<LoadMetadataDetails> indexTableDetailsList =
                  CarbonIndexUtil.getTableStatusDetailsForIndexTable(updatedLoadMetadataDetails,
                      indexTable, newSegmentDetailsListForIndexTable);
  
              SegmentStatusManager.writeLoadDetailsIntoFile(
                  CarbonTablePath.getTableStatusFilePath(indexTable.getTablePath(),
                      loadModel.getLatestTableStatusWriteVersion()),
                  indexTableDetailsList.toArray(new LoadMetadataDetails[0]));
            }
          } else if (carbonTable.isIndexTable()) {
            SegmentStatusManager.writeLoadDetailsIntoFile(
                CarbonTablePath.getTableStatusFilePath(carbonTable.getTablePath(),
                    loadModel.getLatestTableStatusWriteVersion()),
                updatedLoadMetadataDetails.toArray(new LoadMetadataDetails[0]));
          }
          status = true;
        } else {
          LOGGER.error(
              "Not able to acquire the lock for Table status update for table " + databaseName + "."
                  + tableName);
          throw new RuntimeException(
              "Not able to acquire the lock for Table status updation for table " + databaseName + "."
                  + tableName);
        }
      } catch (IOException e) {
        LOGGER.error(
            "Not able to acquire the lock for Table status update for table " + databaseName + "."
                + tableName);
      } finally {
        if (carbonLock.unlock()) {
          LOGGER.info("Table unlocked successfully after table status update" + databaseName + "."
              + tableName);
        } else {
          LOGGER.error("Unable to unlock Table lock for table" + databaseName + "." + tableName
              + " during table status update");
        }
      }
      return status;
    }
  
    /**
     * This method read the details of SI table and check whether new metadata details are already
     * compacted, if it is, then already compaction for SI is completed and updating with new segment
     * status is useless, this can happen in case of updating the status of index while loading
     * segments for failed segments, so do not update anything, just exit gracefully
     */
    private static boolean isSegmentsAlreadyCompactedForNewMetaDataDetails(
        List<CarbonTable> indexTables, String indexTableName,
        List<LoadMetadataDetails> newLoadMetadataDetails) {
      CarbonTable currentIndexTable = null;
      for (CarbonTable indexTable : indexTables) {
        if (indexTable.getTableName().equalsIgnoreCase(indexTableName)) {
          currentIndexTable = indexTable;
          break;
        }
      }
      boolean isIndexTableSegmentsCompacted = false;
      if (null != currentIndexTable) {
        LoadMetadataDetails[] existingLoadMetaDataDetails = SegmentStatusManager.readLoadMetadata(
            currentIndexTable.getMetadataPath(), currentIndexTable.getTableStatusVersion());
        for (LoadMetadataDetails existingLoadMetaDataDetail : existingLoadMetaDataDetails) {
          for (LoadMetadataDetails newLoadMetadataDetail : newLoadMetadataDetails) {
            if (existingLoadMetaDataDetail.getLoadName().equalsIgnoreCase(
                newLoadMetadataDetail.getLoadName())
                && existingLoadMetaDataDetail.getSegmentStatus() == SegmentStatus.COMPACTED) {
              isIndexTableSegmentsCompacted = true;
              break;
            }
          }
          if (isIndexTableSegmentsCompacted) {
            break;
          }
        }
        return isIndexTableSegmentsCompacted;
      } else {
        return false;
      }
    }
  
    /**
     * method to update table status in case of IUD Update Delta Compaction.
     */
    public static boolean updateLoadMetadataWithMergeStatus(CarbonTable indexCarbonTable,
        String[] loadsToMerge, String mergedLoadNumber, Map<String, String> segmentToLoadStartTimeMap,
        long mergeLoadStartTime, SegmentStatus segmentStatus, long newLoadStartTime,
        List<String> rebuiltSegments) throws IOException {
      boolean tableStatusUpdateStatus = false;
      List<String> loadMergeList = new ArrayList<>(Arrays.asList(loadsToMerge));
      SegmentStatusManager segmentStatusManager = new SegmentStatusManager(
          indexCarbonTable.getAbsoluteTableIdentifier(), indexCarbonTable.getTableStatusVersion());
  
      ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
  
      try {
        if (carbonLock.lockWithRetries()) {
          LOGGER.info("Acquired lock for the table " + indexCarbonTable.getDatabaseName() + "."
              + indexCarbonTable.getTableName() + " for table status update ");
          LoadMetadataDetails[] loadDetails = SegmentStatusManager.readLoadMetadata(
              indexCarbonTable.getMetadataPath(), indexCarbonTable.getTableStatusVersion());
  
          long modificationOrDeletionTimeStamp = CarbonUpdateUtil.readCurrentTime();
          for (LoadMetadataDetails loadDetail : loadDetails) {
            // check if this segment is merged.
            if (loadMergeList.contains(loadDetail.getLoadName()) || loadMergeList.contains(
                loadDetail.getMergedLoadName())) {
              // if the compacted load is deleted after the start of the compaction process,
              // then need to discard the compaction process and treat it as failed compaction.
              if (loadDetail.getSegmentStatus() == SegmentStatus.MARKED_FOR_DELETE) {
                LOGGER.error("Compaction is aborted as the segment " + loadDetail.getLoadName()
                    + " is deleted after the compaction is started.");
                return false;
              }
              loadDetail.setSegmentStatus(SegmentStatus.COMPACTED);
              loadDetail.setModificationOrDeletionTimestamp(modificationOrDeletionTimeStamp);
              loadDetail.setMergedLoadName(mergedLoadNumber);
            }
          }
  
          // create entry for merged one.
          LoadMetadataDetails loadMetadataDetails = new LoadMetadataDetails();
          loadMetadataDetails.setSegmentStatus(segmentStatus);
          long loadEndDate = CarbonUpdateUtil.readCurrentTime();
          loadMetadataDetails.setLoadEndTime(loadEndDate);
          loadMetadataDetails.setLoadName(mergedLoadNumber);
          loadMetadataDetails.setSegmentFile(SegmentFileStore.genSegmentFileName(mergedLoadNumber,
              String.valueOf(segmentToLoadStartTimeMap.get(mergedLoadNumber)))
              + CarbonTablePath.SEGMENT_EXT);
          CarbonLoaderUtil.addDataIndexSizeIntoMetaEntry(loadMetadataDetails, mergedLoadNumber,
              indexCarbonTable);
          if (rebuiltSegments.contains(loadMetadataDetails.getLoadName())) {
            loadMetadataDetails.setLoadStartTime(newLoadStartTime);
          } else {
            loadMetadataDetails.setLoadStartTime(mergeLoadStartTime);
          }
  
          // put the merged folder entry
          for (int i = 0; i < loadDetails.length; i++) {
            if (loadDetails[i].getLoadName().equals(loadMetadataDetails.getLoadName())) {
              loadDetails[i] = loadMetadataDetails;
            }
          }
  
          // if this is a major compaction then set the segment as major compaction.
          List<LoadMetadataDetails> updatedDetailsList = new ArrayList<>(Arrays.asList(loadDetails));
  
          SegmentStatusManager.writeLoadDetailsIntoFile(
              CarbonTablePath.getTableStatusFilePath(indexCarbonTable.getTablePath(),
                  indexCarbonTable.getTableStatusVersion()),
              updatedDetailsList.toArray(new LoadMetadataDetails[0]));
          tableStatusUpdateStatus = true;
        } else {
          LOGGER.error(
              "Could not able to obtain lock for table" + indexCarbonTable.getDatabaseName() + "."
                  + indexCarbonTable.getTableName() + "for table status update");
        }
      } finally {
        if (carbonLock.unlock()) {
          LOGGER.info("Table unlocked successfully after table status update" + indexCarbonTable
              .getDatabaseName() + "." + indexCarbonTable.getTableName());
        } else {
          LOGGER.error(
              "Unable to unlock Table lock for table" + indexCarbonTable.getDatabaseName() + "."
                  + indexCarbonTable.getTableName() + " during table status update");
        }
      }
      return tableStatusUpdateStatus;
    }
  
    public static boolean checkMainTableSegEqualToSiSeg(
        LoadMetadataDetails[] mainTableLoadMetadataDetails,
        LoadMetadataDetails[] siTableLoadMetadataDetails) throws ErrorMessage {
      return checkMainTableSegEqualToSiSeg(mainTableLoadMetadataDetails, siTableLoadMetadataDetails,
          false);
    }
  
    /**
     * Method to check if main table and SI have same number of valid segments or not
     */
    public static boolean checkMainTableSegEqualToSiSeg(
        LoadMetadataDetails[] mainTableLoadMetadataDetails,
        LoadMetadataDetails[] siTableLoadMetadataDetails, boolean isRegisterIndex)
        throws ErrorMessage {
      List<String> mainTableSegmentsList = getListOfValidSlices(mainTableLoadMetadataDetails);
      List<String> indexList = getListOfValidSlices(siTableLoadMetadataDetails);
      Collections.sort(mainTableSegmentsList);
      Collections.sort(indexList);
      // In the case when number of SI segments are more than the main table segments do nothing
      // and proceed to process the segments. Return False in case if main table segments are more
      // than SI Segments
      if (indexList.size() < mainTableSegmentsList.size()) {
        return false;
      } else if ((indexList.size() > mainTableSegmentsList.size()) && isRegisterIndex) {
        throw new ErrorMessage("Cannot register index, as the number of Secondary index table "
            + "segments are more than the main table segments. Try Drop and recreate SI.");
      }
      // There can be cases when the number of segments in the main table are less than the index
      // table. In this case mapping all the segments in main table to SI table.
      // Return False if a segment in main table is not in index table
      HashSet<String> indexTableSet = new HashSet<String>();
      for (int i = 0; i < indexList.size(); i++) {
        indexTableSet.add(indexList.get(i));
      }
      for (int i = 0; i < mainTableSegmentsList.size(); i++) {
        if (!indexTableSet.contains(mainTableSegmentsList.get(i))) {
          return false;
        }
      }
      return true;
    }
  
    /**
     * Method to check if main table has in progress load and same segment not present in SI
     */
    public static boolean checkInProgLoadInMainTableAndSi(CarbonTable carbonTable,
        LoadMetadataDetails[] mainTableLoadMetadataDetails,
        LoadMetadataDetails[] siTableLoadMetadataDetails) {
      List<String> allSiSlices = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      for (LoadMetadataDetails oneLoad : siTableLoadMetadataDetails) {
        allSiSlices.add(oneLoad.getLoadName());
      }
  
      if (mainTableLoadMetadataDetails.length != 0) {
        for (LoadMetadataDetails loadDetail : mainTableLoadMetadataDetails) {
          // if load in progress and check if same load is present in SI.
          if (SegmentStatusManager.isLoadInProgress(carbonTable.getAbsoluteTableIdentifier(),
              loadDetail.getLoadName())) {
            if (!allSiSlices.contains(loadDetail.getLoadName())) {
              return false;
            }
          }
        }
      }
      return true;
    }
  }
  