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

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.common.logging.impl.StatisticLevel;

import static org.apache.carbondata.core.util.CarbonUtil.printLine;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

/**
 * Class will be used to record and log the query statistics
 */
public class DriverQueryStatisticsRecorderImpl implements QueryStatisticsRecorder {

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

  /**
   * singleton QueryStatisticsRecorder for driver
   */
  private Map<String, List<QueryStatistic>> queryStatisticsMap;

  /**
   * lock for log statistics table
   */
  private static final Object lock = new Object();

  private DriverQueryStatisticsRecorderImpl() {
    // use ConcurrentHashMap, it is thread-safe
    queryStatisticsMap = new ConcurrentHashMap<String, List<QueryStatistic>>();
  }

  private static DriverQueryStatisticsRecorderImpl carbonLoadStatisticsImplInstance =
      new DriverQueryStatisticsRecorderImpl();

  public static DriverQueryStatisticsRecorderImpl getInstance() {
    return carbonLoadStatisticsImplInstance;
  }

  public void recordStatistics(QueryStatistic statistic) {

  }

  public void logStatistics() {

  }

  public TaskStatistics statisticsForTask(long taskId, long startTime) {
    return null;
  }

  public void logStatisticsForTask(TaskStatistics task) {

  }

  /**
   * Below method will be used to add the statistics
   *
   * @param statistic
   */
  public void recordStatisticsForDriver(QueryStatistic statistic, String queryId) {
    if (null != queryId) {
      synchronized (lock) {
        // refresh query Statistics Map
        if (queryStatisticsMap.get(queryId) != null) {
          queryStatisticsMap.get(queryId).add(statistic);
        } else {
          List<QueryStatistic> newQueryStatistics = new ArrayList<QueryStatistic>();
          newQueryStatistics.add(statistic);
          queryStatisticsMap.put(queryId, newQueryStatistics);
        }
      }
    }
  }

  /**
   * Below method will be used to show statistic log as table
   */
  public void logStatisticsAsTableDriver() {
    synchronized (lock) {
      Iterator<Map.Entry<String, List<QueryStatistic>>> entries =
              queryStatisticsMap.entrySet().iterator();
      while (entries.hasNext()) {
        Map.Entry<String, List<QueryStatistic>> entry = entries.next();
        String queryId = entry.getKey();
        // clear the unknown query statistics
        if (StringUtils.isEmpty(queryId)) {
          entries.remove();
        } else {
          // clear the timeout query statistics
          long interval = System.nanoTime() - Long.parseLong(queryId);
          if (interval > QueryStatisticsConstants.CLEAR_STATISTICS_TIMEOUT) {
            entries.remove();
          } else {
            // print sql_parse_t,load_meta_t,block_allocation_t,block_identification_t
            // or just print block_allocation_t,block_identification_t
            if (entry.getValue().size() >= 2) {
              String tableInfo = collectDriverStatistics(entry.getValue(), queryId);
              if (null != tableInfo) {
                LOGGER.log(StatisticLevel.STATISTIC, tableInfo);
                // clear the statistics that has been printed
                entries.remove();
              }
            }
          }
        }
      }
    }
  }

  /**
   * Below method will parse queryStatisticsMap and put time into table
   */
  public String collectDriverStatistics(List<QueryStatistic> statisticsList, String queryId) {
    String sql_parse_time = "";
    String load_meta_time = "";
    String load_blocks_time = "";
    String block_allocation_time = "";
    String block_identification_time = "";
    long driver_part_time_tmp = 0L;
    long driver_part_time_tmp2 = 0L;
    long load_blocks_time_tmp = 0L;
    String splitChar = " ";
    try {
      // get statistic time from the QueryStatistic
      for (QueryStatistic statistic : statisticsList) {
        switch (statistic.getMessage()) {
          case QueryStatisticsConstants.SQL_PARSE:
            sql_parse_time += statistic.getTimeTaken() + splitChar;
            driver_part_time_tmp += statistic.getTimeTaken();
            break;
          case QueryStatisticsConstants.LOAD_META:
            load_meta_time += statistic.getTimeTaken() + splitChar;
            driver_part_time_tmp += statistic.getTimeTaken();
            break;
          case QueryStatisticsConstants.LOAD_BLOCKS_DRIVER:
            // multi segments will generate multi load_blocks_time
            load_blocks_time_tmp += statistic.getTimeTaken();
            driver_part_time_tmp += statistic.getTimeTaken();
            driver_part_time_tmp2 += statistic.getTimeTaken();
            break;
          case QueryStatisticsConstants.BLOCK_ALLOCATION:
            block_allocation_time += statistic.getTimeTaken() + splitChar;
            driver_part_time_tmp += statistic.getTimeTaken();
            driver_part_time_tmp2 += statistic.getTimeTaken();
            break;
          case QueryStatisticsConstants.BLOCK_IDENTIFICATION:
            block_identification_time += statistic.getTimeTaken() + splitChar;
            driver_part_time_tmp += statistic.getTimeTaken();
            driver_part_time_tmp2 += statistic.getTimeTaken();
            break;
          default:
            break;
        }
      }
      load_blocks_time = load_blocks_time_tmp + splitChar;
      String driver_part_time = driver_part_time_tmp + splitChar;
      // structure the query statistics info table
      StringBuilder tableInfo = new StringBuilder();
      int len1 = 8;
      int len2 = 20;
      int len3 = 21;
      int len4 = 24;
      String line = "+" + printLine("-", len1) + "+" + printLine("-", len2) + "+" +
          printLine("-", len3) + "+" + printLine("-", len4) + "+";
      String line2 = "|" + printLine(" ", len1) + "+" + printLine("-", len2) + "+" +
          printLine(" ", len3) + "+" + printLine("-", len4) + "+";
      // table header
      tableInfo.append(line).append("\n");
      tableInfo.append("|" + printLine(" ", (len1 - "Module".length())) + "Module" + "|" +
          printLine(" ", (len2 - "Operation Step".length())) + "Operation Step" + "|" +
          printLine(" ", (len3 - "Total Query Cost".length())) + "Total Query Cost" + "|" +
          printLine(" ", (len4 - "Query Cost".length())) + "Query Cost" + "|" + "\n");
      tableInfo.append(line).append("\n");
      // print sql_parse_t,load_meta_t,block_allocation_t,block_identification_t
      if (!StringUtils.isEmpty(sql_parse_time) &&
          !StringUtils.isEmpty(load_meta_time) &&
          !StringUtils.isEmpty(block_allocation_time) &&
          !StringUtils.isEmpty(block_identification_time)) {
        tableInfo.append("|" + printLine(" ", len1) + "|" +
            printLine(" ", (len2 - "SQL parse".length())) + "SQL parse" + "|" +
            printLine(" ", len3) + "|" +
            printLine(" ", (len4 - sql_parse_time.length())) + sql_parse_time + "|" + "\n");
        tableInfo.append(line2).append("\n");
        tableInfo.append("|" + printLine(" ", (len1 - "Driver".length())) + "Driver" + "|" +
            printLine(" ", (len2 - "Load meta data".length())) + "Load meta data" + "|" +
            printLine(" ", (len3 - driver_part_time.length())) + driver_part_time + "|" +
            printLine(" ", (len4 - load_meta_time.length())) +
            load_meta_time + "|" + "\n");
        tableInfo.append(line2).append("\n");
        tableInfo.append("|" + printLine(" ", (len1 - "Part".length())) + "Part" + "|" +
                printLine(" ", (len2 - "Load blocks driver".length())) +
                "Load blocks driver" + "|" +
                printLine(" ", len3) + "|" +
                printLine(" ", (len4 - load_blocks_time.length())) +
                load_blocks_time + "|" + "\n");
        tableInfo.append(line2).append("\n");
        tableInfo.append("|" + printLine(" ", len1) + "|" +
            printLine(" ", (len2 - "Block allocation".length())) + "Block allocation" + "|" +
            printLine(" ", len3) + "|" +
            printLine(" ", (len4 - block_allocation_time.length())) +
            block_allocation_time + "|" + "\n");
        tableInfo.append(line2).append("\n");
        tableInfo.append("|" +
            printLine(" ", len1) + "|" +
            printLine(" ", (len2 - "Block identification".length())) +
            "Block identification" + "|" +
            printLine(" ", len3) + "|" +
            printLine(" ", (len4 - block_identification_time.length())) +
            block_identification_time + "|" + "\n");
        tableInfo.append(line).append("\n");

        // show query statistic as "query id" + "table"
        return "Print query statistic for query id: " + queryId + "\n" + tableInfo.toString();
      } else if (!StringUtils.isEmpty(block_allocation_time) &&
          !StringUtils.isEmpty(block_identification_time)) {
        // when we can't get sql parse time, we only print the last two
        driver_part_time = driver_part_time_tmp2 + splitChar;
        tableInfo.append("|" + printLine(" ", (len1 - "Driver".length())) + "Driver" + "|" +
                printLine(" ", (len2 - "Load blocks driver".length())) +
                "Load blocks driver" + "|" +
                printLine(" ", len3) + "|" +
                printLine(" ", (len4 - load_blocks_time.length())) +
                load_blocks_time + "|" + "\n");
        tableInfo.append(line2).append("\n");
        tableInfo.append("|" + printLine(" ", (len1 - "Part".length())) + "Part" + "|" +
            printLine(" ", (len2 - "Block allocation".length())) + "Block allocation" + "|" +
            printLine(" ", (len3 - driver_part_time.length())) + driver_part_time + "|" +
            printLine(" ", (len4 - block_allocation_time.length())) +
            block_allocation_time + "|" + "\n");
        tableInfo.append(line2).append("\n");
        tableInfo.append("|" +
            printLine(" ", len1) + "|" +
            printLine(" ", (len2 - "Block identification".length())) +
            "Block identification" + "|" +
            printLine(" ", len3) + "|" +
            printLine(" ", (len4 - block_identification_time.length())) +
            block_identification_time + "|" + "\n");
        tableInfo.append(line).append("\n");

        // show query statistic as "query id" + "table"
        return "Print query statistic for query id: " + queryId + "\n" + tableInfo.toString();
      }

      return null;
    } catch (Exception ex) {
      return "Put statistics into table failed, catch exception: " + ex.getMessage();
    }
  }
}