/*
 * 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.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.List;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.annotations.InterfaceStability;
import org.apache.carbondata.core.util.CarbonUtil;

@InterfaceAudience.Developer("stats")
@InterfaceStability.Evolving
public class TaskStatistics implements Serializable {

  private static final Column[] columns = {
      new Column("query_id", "query id"),
      new Column("task_id", "spark task id"),
      new Column("start_time", "start time"),
      new Column("total_time", QueryStatisticsConstants.EXECUTOR_PART, true),
      new Column("load_blocks_time", QueryStatisticsConstants.LOAD_BLOCKS_EXECUTOR, true),
      new Column("load_dictionary_time", QueryStatisticsConstants.LOAD_DICTIONARY, true),
      new Column("carbon_scan_time", QueryStatisticsConstants.SCAN_BLOCKlET_TIME),
      new Column("carbon_IO_time", QueryStatisticsConstants.READ_BLOCKlET_TIME),
      new Column("scan_blocks_num", QueryStatisticsConstants.SCAN_BLOCKS_NUM),
      new Column("total_blocklets", QueryStatisticsConstants.TOTAL_BLOCKLET_NUM),
      new Column("scanned_blocklets", QueryStatisticsConstants.BLOCKLET_SCANNED_NUM),
      new Column("valid_blocklets", QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM),
      new Column("total_pages", QueryStatisticsConstants.TOTAL_PAGE_SCANNED),
      new Column("scanned_pages", QueryStatisticsConstants.PAGE_SCANNED),
      new Column("valid_pages", QueryStatisticsConstants.VALID_PAGE_SCANNED),
      new Column("result_size", QueryStatisticsConstants.RESULT_SIZE),
      new Column("key_column_filling_time", QueryStatisticsConstants.KEY_COLUMN_FILLING_TIME),
      new Column("measure_filling_time", QueryStatisticsConstants.MEASURE_FILLING_TIME),
      new Column("page_uncompress_time", QueryStatisticsConstants.PAGE_UNCOMPRESS_TIME),
      new Column("result_preparation_time", QueryStatisticsConstants.RESULT_PREP_TIME)
  };

  private static final int numOfColumns = columns.length;

  private String queryId;

  private long[] values = new long[numOfColumns];

  private long fileSize;

  private String[] files;

  TaskStatistics(String queryId, long taskId) {
    this.queryId = queryId;
    this.values[1] = taskId;
  }

  public TaskStatistics(String queryId, long[] values, long fileSize, String[] files) {
    this.queryId = queryId;
    this.values = values;
    this.fileSize = fileSize;
    this.files = files;
  }

  public TaskStatistics build(long startTime, List<QueryStatistic> queryStatistics) {
    this.values[2] = startTime;
    for (QueryStatistic statistic : queryStatistics) {
      if (statistic.getMessage() != null) {
        for (int columnIndex = 3; columnIndex <= numOfColumns - 1; columnIndex++) {
          if (columns[columnIndex].comment.equals(statistic.getMessage())) {
            if (columns[columnIndex].isDuration) {
              values[columnIndex] += statistic.getTimeTaken();
            } else {
              values[columnIndex] += statistic.getCount();
            }
            break;
          }
        }
      }
    }
    return this;
  }

  @Override
  public String toString() {
    StringBuilder builder = new StringBuilder();
    printStatisticTable(Collections.singletonList(this), builder, "");
    return builder.toString();
  }

  public static void printStatisticTable(List<TaskStatistics> stats, StringBuilder builder,
      String indent) {
    int numOfRows = stats.size();
    int numOfColumns = columns.length;

    // header as string[]
    String[] header = new String[numOfColumns];
    for (int columnIndex = 0; columnIndex < numOfColumns; columnIndex++) {
      header[columnIndex] = columns[columnIndex].name;
    }

    // convert rows to string[][]
    String[][] rows = new String[numOfRows][];
    for (int rowIndex = 0; rowIndex < numOfRows; rowIndex++) {
      rows[rowIndex] = stats.get(rowIndex).convertValueToString();
    }

    CarbonUtil.logTable(builder, header, rows, indent);
  }

  private String[] convertValueToString() {
    String[] valueStrings = new String[numOfColumns];
    valueStrings[0] = queryId;
    for (int i = 1; i < numOfColumns; i++) {
      if (columns[i].isDuration) {
        valueStrings[i] = values[i] + "ms";
      } else {
        valueStrings[i] = String.valueOf(values[i]);
      }
    }
    valueStrings[2] = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(values[2]);
    return valueStrings;
  }

  private static class Column {
    String name;
    String comment;
    boolean isDuration;

    Column(String name, String comment) {
      this.name = name;
      this.comment = comment;
      this.isDuration = false;
    }

    Column(String name, String comment, boolean isDuration) {
      this(name, comment);
      this.isDuration = isDuration;
    }
  }

  public String getQueryId() {
    return queryId;
  }

  public long[] getValues() {
    return values;
  }

  public long getFileSize() {
    return fileSize;
  }

  public String[] getFiles() {
    return files;
  }
}