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.carbondata.core.util;
  
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  
  import org.apache.carbondata.common.logging.LogServiceFactory;
  
  import org.apache.log4j.Logger;
  
  /**
   * A util which provide methods used to record time information during data loading.
   */
  public class CarbonLoadStatisticsImpl implements LoadStatistics {
    private CarbonLoadStatisticsImpl() {
  
    }
  
    private static CarbonLoadStatisticsImpl carbonLoadStatisticsImplInstance =
            new CarbonLoadStatisticsImpl();
  
    public static CarbonLoadStatisticsImpl getInstance() {
      return carbonLoadStatisticsImplInstance;
    }
  
    private static final Logger LOGGER =
            LogServiceFactory.getLogService(CarbonLoadStatisticsImpl.class.getName());
  
    /*
     *We only care about the earliest start time(EST) and the latest end time(LET) of different
     *threads, who does the same thing, LET - EST is the cost time of doing one thing using
     *multiple thread.
   */
    private long loadCsvFilesToDfStartTime = 0;
    private long loadCsvFilesToDfCostTime = 0;
    private long dicShuffleAndWriteFileTotalStartTime = 0;
  
    //LRU cache load one time
    private double lruCacheLoadTime = 0;
  
    //Generate surrogate keys total time for each partition:
    private ConcurrentHashMap<String, Long[]> parDictionaryValuesTotalTimeMap =
            new ConcurrentHashMap<String, Long[]>();
    private ConcurrentHashMap<String, Long[]> parCsvInputStepTimeMap =
            new ConcurrentHashMap<String, Long[]>();
    private ConcurrentHashMap<String, Long[]> parGeneratingDictionaryValuesTimeMap =
            new ConcurrentHashMap<String, Long[]>();
  
    //Sort rows step total time for each partition:
    private ConcurrentHashMap<String, Long[]> parSortRowsStepTotalTimeMap =
            new ConcurrentHashMap<String, Long[]>();
  
    //MDK generate total time for each partition:
    private ConcurrentHashMap<String, Long[]> parMdkGenerateTotalTimeMap =
            new ConcurrentHashMap<String, Long[]>();
    private ConcurrentHashMap<String, Long[]> parDictionaryValue2MdkAdd2FileTime =
            new ConcurrentHashMap<String, Long[]>();
  
    //Node block process information
    private ConcurrentHashMap<String, Integer> hostBlockMap =
            new ConcurrentHashMap<String, Integer>();
  
    //Partition block process information
    private ConcurrentHashMap<String, Integer> partitionBlockMap =
            new ConcurrentHashMap<String, Integer>();
  
    private long totalRecords = 0;
    private double totalTime = 0;
  
    @Override
    public void initPartitionInfo(String PartitionId) {
      parDictionaryValuesTotalTimeMap.put(PartitionId, new Long[2]);
      parCsvInputStepTimeMap.put(PartitionId, new Long[2]);
      parSortRowsStepTotalTimeMap.put(PartitionId, new Long[2]);
      parGeneratingDictionaryValuesTimeMap.put(PartitionId, new Long[2]);
      parMdkGenerateTotalTimeMap.put(PartitionId, new Long[2]);
      parDictionaryValue2MdkAdd2FileTime.put(PartitionId, new Long[2]);
    }
  
    //Record the time
    public void recordDicShuffleAndWriteTime() {
      long dicShuffleAndWriteTimePoint = System.currentTimeMillis();
      if (0 == dicShuffleAndWriteFileTotalStartTime) {
        dicShuffleAndWriteFileTotalStartTime = dicShuffleAndWriteTimePoint;
      }
    }
  
    public void recordLoadCsvFilesToDfTime() {
      long loadCsvFilesToDfTimePoint = System.currentTimeMillis();
      if (0 == loadCsvFilesToDfStartTime) {
        loadCsvFilesToDfStartTime = loadCsvFilesToDfTimePoint;
      }
      if (loadCsvFilesToDfTimePoint - loadCsvFilesToDfStartTime > loadCsvFilesToDfCostTime) {
        loadCsvFilesToDfCostTime = loadCsvFilesToDfTimePoint - loadCsvFilesToDfStartTime;
      }
    }
  
    public double getLruCacheLoadTime() {
      return lruCacheLoadTime;
    }
  
    private void recordStatistics(Map<String, Long[]> statMap, String partitionID, Long value) {
      if (null != statMap.get(partitionID)) {
        if (null == statMap.get(partitionID)[0]) {
          statMap.get(partitionID)[0] = value;
        }
        if (null == statMap.get(partitionID)[1] ||
            value - statMap.get(partitionID)[0] > statMap.get(partitionID)[1]) {
          statMap.get(partitionID)[1] = value - statMap.get(partitionID)[0];
        }
      }
    }
  
    public void recordDictionaryValuesTotalTime(String partitionID,
        Long dictionaryValuesTotalTimeTimePoint) {
      recordStatistics(parDictionaryValuesTotalTimeMap, partitionID,
          dictionaryValuesTotalTimeTimePoint);
    }
  
    public void recordCsvInputStepTime(String partitionID, Long csvInputStepTimePoint) {
      recordStatistics(parCsvInputStepTimeMap, partitionID, csvInputStepTimePoint);
    }
  
    public void recordLruCacheLoadTime(double lruCacheLoadTime) {
      this.lruCacheLoadTime = lruCacheLoadTime;
    }
  
    public void recordGeneratingDictionaryValuesTime(String partitionID,
        Long generatingDictionaryValuesTimePoint) {
      recordStatistics(parGeneratingDictionaryValuesTimeMap, partitionID,
          generatingDictionaryValuesTimePoint);
    }
  
    public void recordSortRowsStepTotalTime(String partitionID, Long sortRowsStepTotalTimePoint) {
      recordStatistics(parSortRowsStepTotalTimeMap, partitionID, sortRowsStepTotalTimePoint);
    }
  
    public void recordMdkGenerateTotalTime(String partitionID, Long mdkGenerateTotalTimePoint) {
      recordStatistics(parMdkGenerateTotalTimeMap, partitionID, mdkGenerateTotalTimePoint);
    }
  
    public void recordDictionaryValue2MdkAdd2FileTime(String partitionID,
        Long dictionaryValue2MdkAdd2FileTimePoint) {
      recordStatistics(parDictionaryValue2MdkAdd2FileTime, partitionID,
          dictionaryValue2MdkAdd2FileTimePoint);
    }
  
    //Record the node blocks information map
    public void recordHostBlockMap(String host, Integer numBlocks) {
      hostBlockMap.put(host, numBlocks);
    }
  
    //Record the partition blocks information map
    public void recordPartitionBlockMap(String partitionID, Integer numBlocks) {
      partitionBlockMap.put(partitionID, numBlocks);
    }
  
    public void recordTotalRecords(long totalRecords) {
      this.totalRecords = totalRecords;
    }
  
    private double getLoadCsvFilesToDfTime() {
      return loadCsvFilesToDfCostTime / 1000.0;
    }
  
    private double getDictionaryValuesTotalTime(String partitionID) {
      return parDictionaryValuesTotalTimeMap.get(partitionID)[1] / 1000.0;
    }
  
    private double getCsvInputStepTime(String partitionID) {
      return parCsvInputStepTimeMap.get(partitionID)[1] / 1000.0;
    }
  
    private double getGeneratingDictionaryValuesTime(String partitionID) {
      return parGeneratingDictionaryValuesTimeMap.get(partitionID)[1] / 1000.0;
    }
  
    private double getSortRowsStepTotalTime(String partitionID) {
      return parSortRowsStepTotalTimeMap.get(partitionID)[1] / 1000.0;
    }
  
    private double getDictionaryValue2MdkAdd2FileTime(String partitionID) {
      return parDictionaryValue2MdkAdd2FileTime.get(partitionID)[1] / 1000.0;
    }
  
    //Get the hostBlockMap
    private ConcurrentHashMap<String, Integer> getHostBlockMap() {
      return hostBlockMap;
    }
  
    //Get the partitionBlockMap
    private ConcurrentHashMap<String, Integer> getPartitionBlockMap() {
      return partitionBlockMap;
    }
  
    //Speed calculate
    private long getTotalRecords() {
      return this.totalRecords;
    }
  
    private int getLoadSpeed() {
      return (int)(totalRecords / totalTime);
    }
  
    private int getReadCSVSpeed(String partitionID) {
      return (int)(totalRecords / getCsvInputStepTime(partitionID));
    }
  
    private int getGenSurKeySpeed(String partitionID) {
      return (int)(totalRecords / getGeneratingDictionaryValuesTime(partitionID));
    }
  
    private int getSortKeySpeed(String partitionID) {
      return (int)(totalRecords / getSortRowsStepTotalTime(partitionID));
    }
  
    private int getMDKSpeed(String partitionID) {
      return (int)(totalRecords / getDictionaryValue2MdkAdd2FileTime(partitionID));
    }
  
    private double getTotalTime(String partitionID) {
      this.totalTime = getLoadCsvFilesToDfTime() +
          getLruCacheLoadTime() + getDictionaryValuesTotalTime(partitionID) +
          getDictionaryValue2MdkAdd2FileTime(partitionID);
      return totalTime;
    }
  
    //Print the statistics information
    private void printDicGenStatisticsInfo() {
      double loadCsvFilesToDfTime = getLoadCsvFilesToDfTime();
      LOGGER.info("STAGE 1 ->Load csv to DataFrame and generate" +
              " block distinct values: " + loadCsvFilesToDfTime + "(s)");
    }
  
    private void printLruCacheLoadTimeInfo() {
      LOGGER.info("STAGE 2 ->LRU cache load: " + getLruCacheLoadTime() + "(s)");
    }
  
    private void printDictionaryValuesGenStatisticsInfo(String partitionID) {
      double dictionaryValuesTotalTime = getDictionaryValuesTotalTime(partitionID);
      LOGGER.info("STAGE 3 ->Total cost of gen dictionary values, sort and write to temp files: "
              + dictionaryValuesTotalTime + "(s)");
      double csvInputStepTime = getCsvInputStepTime(partitionID);
      double generatingDictionaryValuesTime = getGeneratingDictionaryValuesTime(partitionID);
      LOGGER.info("STAGE 3.1 ->  |_read csv file: " + csvInputStepTime + "(s)");
      LOGGER.info("STAGE 3.2 ->  |_transform to surrogate key: "
              + generatingDictionaryValuesTime + "(s)");
    }
  
    private void printSortRowsStepStatisticsInfo(String partitionID) {
      double sortRowsStepTotalTime = getSortRowsStepTotalTime(partitionID);
      LOGGER.info("STAGE 3.3 ->  |_sort rows and write to temp file: "
              + sortRowsStepTotalTime + "(s)");
    }
  
    private void printGenMdkStatisticsInfo(String partitionID) {
      double dictionaryValue2MdkAdd2FileTime = getDictionaryValue2MdkAdd2FileTime(partitionID);
      LOGGER.info("STAGE 4 ->Transform to MDK, compress and write fact files: "
              + dictionaryValue2MdkAdd2FileTime + "(s)");
    }
  
    //Print the node blocks information
    private void printHostBlockMapInfo() {
      LOGGER.info("========== BLOCK_INFO ==========");
      if (getHostBlockMap().size() > 0) {
        for (String host: getHostBlockMap().keySet()) {
          LOGGER.info("BLOCK_INFO ->Node host: " + host);
          LOGGER.info("BLOCK_INFO ->The block count in this node: " + getHostBlockMap().get(host));
        }
      } else if (getPartitionBlockMap().size() > 0) {
        for (String parID: getPartitionBlockMap().keySet()) {
          LOGGER.info("BLOCK_INFO ->Partition ID: " + parID);
          LOGGER.info("BLOCK_INFO ->The block count in this partition: " +
                  getPartitionBlockMap().get(parID));
        }
      }
    }
  
    //Print the speed information
    private void printLoadSpeedInfo(String partitionID) {
      LOGGER.info("===============Load_Speed_Info===============");
      LOGGER.info("Total Num of Records Processed: " + getTotalRecords());
      LOGGER.info("Total Time Cost: " + getTotalTime(partitionID) + "(s)");
      LOGGER.info("Total Load Speed: " + getLoadSpeed() + "records/s");
      LOGGER.info("Read CSV Speed: " + getReadCSVSpeed(partitionID) + " records/s");
      LOGGER.info("Generate Surrogate Key Speed: " + getGenSurKeySpeed(partitionID) + " records/s");
      LOGGER.info("Sort Key/Write Temp Files Speed: " + getSortKeySpeed(partitionID) + " records/s");
      LOGGER.info("MDK Step Speed: " + getMDKSpeed(partitionID) + " records/s");
      LOGGER.info("=============================================");
    }
  
    public void printStatisticsInfo(String partitionID) {
      try {
        LOGGER.info("========== TIME_STATISTICS PartitionID: " + partitionID + "==========");
        printDicGenStatisticsInfo();
        printLruCacheLoadTimeInfo();
        printDictionaryValuesGenStatisticsInfo(partitionID);
        printSortRowsStepStatisticsInfo(partitionID);
        printGenMdkStatisticsInfo(partitionID);
        printHostBlockMapInfo();
        printLoadSpeedInfo(partitionID);
      } catch (Exception e) {
        LOGGER.error("Can't print Statistics Information");
      } finally {
        resetLoadStatistics();
      }
    }
  
    //Reset the load statistics values
    private void resetLoadStatistics() {
      loadCsvFilesToDfStartTime = 0;
      loadCsvFilesToDfCostTime = 0;
      dicShuffleAndWriteFileTotalStartTime = 0;
      lruCacheLoadTime = 0;
      totalRecords = 0;
      totalTime = 0;
      parDictionaryValuesTotalTimeMap.clear();
      parCsvInputStepTimeMap.clear();
      parSortRowsStepTotalTimeMap.clear();
      parGeneratingDictionaryValuesTimeMap.clear();
      parMdkGenerateTotalTimeMap.clear();
      parDictionaryValue2MdkAdd2FileTime.clear();
    }
  
  }
  