/*
 * 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.drill.exec.planner.common;

import com.fasterxml.jackson.annotation.JsonGetter;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonSetter;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.annotation.JsonTypeName;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.NamedType;
import com.fasterxml.jackson.databind.module.SimpleModule;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.logical.FormatPluginConfig;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.exec.ops.QueryContext;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.planner.logical.DrillTable;
import org.apache.drill.exec.planner.sql.DirectPlan;
import org.apache.drill.exec.record.MajorTypeSerDe;
import org.apache.drill.exec.store.StoragePlugin;
import org.apache.drill.exec.store.dfs.FileSystemPlugin;
import org.apache.drill.exec.store.dfs.FormatPlugin;
import org.apache.drill.exec.store.dfs.FormatSelection;
import org.apache.drill.exec.store.parquet.ParquetFormatConfig;
import org.apache.drill.exec.util.ImpersonationUtil;
import org.apache.drill.metastore.statistics.Histogram;
import org.apache.drill.metastore.statistics.TableStatisticsKind;
import org.apache.drill.metastore.statistics.ColumnStatisticsKind;
import org.apache.drill.metastore.statistics.StatisticsHolder;
import org.apache.drill.metastore.statistics.StatisticsKind;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * Wraps the stats table info including schema and tableName. Also materializes stats from storage
 * and keeps them in memory.
 */
public class DrillStatsTable {
  private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillStatsTable.class);
  // All the statistics versions till date
  public enum STATS_VERSION {V0, V1}
  // The current version
  public static final STATS_VERSION CURRENT_VERSION = STATS_VERSION.V1;
  // 10 histogram buckets (TODO: can make this configurable later)
  public static final int NUM_HISTOGRAM_BUCKETS = 10;

  private final FileSystem fs;
  private final Path tablePath;
  private final String schemaName;
  private final String tableName;
  private final Map<SchemaPath, Long> ndv = new HashMap<>();
  private final Map<SchemaPath, Histogram> histogram = new HashMap<>();
  private double rowCount = -1;
  private final Map<SchemaPath, Long> nnRowCount = new HashMap<>();
  private boolean materialized = false;
  private DrillTable table;
  private TableStatistics statistics = null;

  public DrillStatsTable(DrillTable table, String schemaName, String tableName, Path tablePath, FileSystem fs) {
    this.schemaName = schemaName;
    this.tableName = tableName;
    this.tablePath = tablePath;
    this.fs = ImpersonationUtil.createFileSystem(ImpersonationUtil.getProcessUserName(), fs.getConf());
    this.table = table;
  }

  public DrillStatsTable(TableStatistics statistics) {
    this.statistics = statistics;
    this.schemaName = null;
    this.tableName = null;
    this.tablePath = null;
    this.fs = null;
    materializeFromStatistics();
  }

  public String getSchemaName() {
    return schemaName;
  }

  public String getTableName() {
    return tableName;
  }

  /*
   * Returns whether statistics have materialized or not i.e. were the table statistics successfully read
   * from the persistent store?
   */
  public boolean isMaterialized() { return materialized; }
  /**
   * Get the approximate number of distinct values of given column. If stats are not present for the
   * given column, a null is returned.
   *
   * Note: returned data may not be accurate. Accuracy depends on whether the table data has changed
   * after the stats are computed.
   *
   * @param col - column for which approximate count distinct is desired
   * @return approximate count distinct of the column, if available. NULL otherwise.
   */
  public Double getNdv(SchemaPath col) {
    // Stats might not have materialized because of errors.
    if (!materialized) {
      return null;
    }
    Long ndvCol = ndv.get(col);
    // Ndv estimation techniques like HLL may over-estimate, hence cap it at rowCount
    if (ndvCol != null) {
      return (double) Math.min(ndvCol, rowCount);
    }
    return null;
  }

  public Set<SchemaPath> getColumns() {
    return ndv.keySet();
  }

  /**
   * Get row count of the table. Returns null if stats are not present.
   *
   * Note: returned data may not be accurate. Accuracy depends on whether the table data has
   * changed after the stats are computed.
   *
   * @return rowcount for the table, if available. NULL otherwise.
   */
  public Double getRowCount() {
    // Stats might not have materialized because of errors.
    if (!materialized) {
      return null;
    }
    return rowCount > 0 ? rowCount : null;
  }

  /**
   * Get non-null rowcount for the column If stats are not present for the given column, a null is returned.
   *
   * Note: returned data may not be accurate. Accuracy depends on whether the table data has changed
   * after the stats are computed.
   *
   * @param col - column for which non-null rowcount is desired
   * @return non-null rowcount of the column, if available. NULL otherwise.
   */
  public Double getNNRowCount(SchemaPath col) {
    // Stats might not have materialized because of errors.
    if (!materialized) {
      return null;
    }
    Long nnRowCntCol = nnRowCount.get(col);
    // Cap it at row count (just in case)
    if (nnRowCntCol != null) {
      return Math.min(nnRowCntCol, rowCount);
    }
    return null;
  }

  /**
   * Get the histogram of a given column. If stats are not present for the given column,
   * a null is returned.
   * <p>
   * Note: returned data may not be accurate. Accuracy depends on whether the table data has changed after the
   * stats are computed.
   *
   * @param column path to the column whose histogram should be obtained
   * @return Histogram for this column
   */
  public Histogram getHistogram(SchemaPath column) {
    // Stats might not have materialized because of errors.
    if (!materialized) {
      return null;
    }
    return histogram.get(column);
  }

  /**
   * Read the stats from storage and keep them in memory.
   */
  public void materialize() {
    try {
      // for the case when tablePath is not set, or it does not exists, statistics cannot be read
      if (materialized || tablePath == null || !fs.exists(tablePath)) {
        return;
      }
      // Deserialize statistics from JSON
      this.statistics = readStatistics(table, tablePath);
      // Handle based on the statistics version read from the file
      materializeFromStatistics();
    } catch (FileNotFoundException ex) {
      logger.debug(String.format("Did not find statistics file %s", tablePath.toString()), ex);
    } catch (IOException ex) {
      logger.debug(String.format("Error trying to read statistics table %s", tablePath.toString()), ex);
    }
  }

  private void materializeFromStatistics() {
    if (statistics instanceof Statistics_v0) {
      // Do nothing
    } else if (statistics instanceof Statistics_v1) {
      for (DirectoryStatistics_v1 ds : ((Statistics_v1) statistics).getDirectoryStatistics()) {
        for (ColumnStatistics_v1 cs : ds.getColumnStatistics()) {
          ndv.put(cs.getName(), cs.getNdv());
          nnRowCount.put(cs.getName(), (long) cs.getNonNullCount());
          rowCount = Math.max(rowCount, cs.getCount());

          // get the histogram for this column
          Histogram hist = cs.getHistogram();
          histogram.put(cs.getName(), hist);
        }
      }
    }
    if (statistics != null) { // See stats are available before setting materialized
      materialized = true;
    }
  }

  /* Each change to the format SHOULD increment the default and/or the max values of the option
   * exec.statistics.capability_version
   */
  @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY,
      property = "statistics_version")
  @JsonSubTypes({
      @JsonSubTypes.Type(value = DrillStatsTable.Statistics_v1.class, name="v0"),
      @JsonSubTypes.Type(value = DrillStatsTable.Statistics_v1.class, name="v1")
  })
  public static abstract class TableStatistics {
    @JsonIgnore public abstract List<? extends DirectoryStatistics> getDirectoryStatistics();
  }

  public static abstract class DirectoryStatistics {
  }

  public static abstract class ColumnStatistics {
  }

  @JsonTypeName("v0")
  public static class Statistics_v0 extends TableStatistics {
    @JsonProperty ("directories") List<DirectoryStatistics_v0> directoryStatistics;
    // Default constructor required for deserializer
    public Statistics_v0 () { }
    @Override
    @JsonGetter ("directories")
    public List<DirectoryStatistics_v0> getDirectoryStatistics() {
      return directoryStatistics;
    }
    @JsonSetter ("directories")
    public void setDirectoryStatistics(List<DirectoryStatistics_v0> directoryStatistics) {
      this.directoryStatistics = directoryStatistics;
    }
  }

  public static class DirectoryStatistics_v0 extends DirectoryStatistics {
    @JsonProperty private double computed;
    // Default constructor required for deserializer
    public DirectoryStatistics_v0() { }
    @JsonGetter ("computed")
    public double getComputedTime() {
      return this.computed;
    }
    @JsonSetter ("computed")
    public void setComputedTime(double computed) {
      this.computed = computed;
    }
  }

  /**
   * Struct which contains the statistics for the entire directory structure
   */
  @JsonTypeName("v1")
  public static class Statistics_v1 extends TableStatistics {
    @JsonProperty ("directories")
    List<DirectoryStatistics_v1> directoryStatistics;
    // Default constructor required for deserializer
    public Statistics_v1 () { }
    @Override
    @JsonGetter ("directories")
    public List<DirectoryStatistics_v1> getDirectoryStatistics() {
      return directoryStatistics;
    }
    @JsonSetter ("directories")
    public void setDirectoryStatistics(List<DirectoryStatistics_v1> directoryStatistics) {
      this.directoryStatistics = directoryStatistics;
    }
  }

  public static class DirectoryStatistics_v1 extends DirectoryStatistics {
    @JsonProperty private String computed;
    @JsonProperty ("columns") private List<ColumnStatistics_v1> columnStatistics;
    // Default constructor required for deserializer
    public DirectoryStatistics_v1() { }
    @JsonGetter ("computed")
    public String getComputedTime() {
      return this.computed;
    }
    @JsonSetter ("computed")
    public void setComputedTime(String computed) {
      this.computed = computed;
    }
    @JsonGetter ("columns")
    public List<ColumnStatistics_v1> getColumnStatistics() {
      return this.columnStatistics;
    }
    @JsonSetter ("columns")
    public void setColumnStatistics(List<ColumnStatistics_v1> columnStatistics) {
      this.columnStatistics = columnStatistics;
    }
  }

  public static class ColumnStatistics_v1 extends ColumnStatistics {
    @JsonProperty ("column") private SchemaPath name = null;
    @JsonProperty ("majortype")   private TypeProtos.MajorType type = null;
    @JsonProperty ("schema") private long schema = 0;
    @JsonProperty ("rowcount") private long count = 0;
    @JsonProperty ("nonnullrowcount") private long nonNullCount = 0;
    @JsonProperty ("ndv") private long ndv = 0;
    @JsonProperty ("avgwidth") private double width = 0;
    @JsonProperty ("histogram") private Histogram histogram = null;

    public ColumnStatistics_v1() {}
    @JsonGetter ("column")
    public SchemaPath getName() { return this.name; }
    @JsonSetter ("column")
    public void setName(SchemaPath name) {
      this.name = name;
    }
    @JsonGetter ("majortype")
    public TypeProtos.MajorType getType() { return this.type; }
    @JsonSetter ("type")
    public void setType(TypeProtos.MajorType type) {
      this.type = type;
    }
    @JsonGetter ("schema")
    public long getSchema() {
      return this.schema;
    }
    @JsonSetter ("schema")
    public void setSchema(long schema) {
      this.schema = schema;
    }
    @JsonGetter ("rowcount")
    public double getCount() {
      return this.count;
    }
    @JsonSetter ("rowcount")
    public void setCount(long count) {
      this.count = count;
    }
    @JsonGetter ("nonnullrowcount")
    public double getNonNullCount() {
      return this.nonNullCount;
    }
    @JsonSetter ("nonnullrowcount")
    public void setNonNullCount(long nonNullCount) {
      this.nonNullCount = nonNullCount;
    }
    @JsonGetter ("ndv")
    public long getNdv() {
      return this.ndv;
    }
    @JsonSetter ("ndv")
    public void setNdv(long ndv) { this.ndv = ndv; }
    @JsonGetter ("avgwidth")
    public double getAvgWidth() {
      return this.width;
    }
    @JsonSetter ("avgwidth")
    public void setAvgWidth(double width) { this.width = width; }
    @JsonGetter("histogram")
    public Histogram getHistogram() { return this.histogram; }
    @JsonSetter("histogram")
    public void setHistogram(Histogram histogram) {
      this.histogram = histogram;
    }
    @JsonIgnore
    public void buildHistogram(byte[] tdigest_bytearray) {
      int num_buckets = (int) Math.min(ndv, DrillStatsTable.NUM_HISTOGRAM_BUCKETS);
      this.histogram = HistogramUtils.buildHistogramFromTDigest(tdigest_bytearray, this.getType(),
              num_buckets, nonNullCount);
    }
  }

  private TableStatistics readStatistics(DrillTable drillTable, Path path) throws IOException {
    final Object selection = drillTable.getSelection();

    if (selection instanceof FormatSelection) {
      StoragePlugin storagePlugin =  drillTable.getPlugin();
      FormatSelection formatSelection = (FormatSelection) selection;
      FormatPluginConfig formatConfig = formatSelection.getFormat();

      if (storagePlugin instanceof FileSystemPlugin
          && (formatConfig instanceof ParquetFormatConfig)) {
        FormatPlugin fmtPlugin = storagePlugin.getFormatPlugin(formatConfig);
        if (fmtPlugin.supportsStatistics()) {
          return fmtPlugin.readStatistics(fs, path);
        }
      }
    }
    return null;
  }

  public static TableStatistics generateDirectoryStructure(String dirComputedTime,
      List<ColumnStatistics> columnStatisticsList) {
    // TODO: Split up columnStatisticsList() based on directory names. We assume only
    // one directory right now but this WILL change in the future
    // HashMap<String, Boolean> dirNames = new HashMap<String, Boolean>();
    Statistics_v1 statistics = new Statistics_v1();
    List<DirectoryStatistics_v1> dirStats = new ArrayList<>();
    List<ColumnStatistics_v1> columnStatisticsV1s = new ArrayList<>();
    // Create dirStats
    DirectoryStatistics_v1 dirStat = new DirectoryStatistics_v1();
    // Add columnStats corresponding to this dirStats
    for (ColumnStatistics colStats : columnStatisticsList) {
      columnStatisticsV1s.add((ColumnStatistics_v1) colStats);
    }
    dirStat.setComputedTime(dirComputedTime);
    dirStat.setColumnStatistics(columnStatisticsV1s);
    // Add this dirStats to the list of dirStats
    dirStats.add(dirStat);
    // Add list of dirStats to tableStats
    statistics.setDirectoryStatistics(dirStats);
    return statistics;
  }

  public static PhysicalPlan direct(QueryContext context, boolean outcome, String message, Object... values) {
    return DirectPlan.createDirectPlan(context, outcome, String.format(message, values));
  }

  /* Helper function to generate error - statistics not supported on non-parquet tables */
  public static PhysicalPlan notSupported(QueryContext context, String tbl) {
    return direct(context, false, "Table %s is not supported by ANALYZE."
        + " Support is currently limited to directory-based Parquet tables.", tbl);
  }

  public static PhysicalPlan notRequired(QueryContext context, String tbl) {
    return direct(context, false, "Table %s has not changed since last ANALYZE!", tbl);
  }

  /**
   * This method returns the statistics (de)serializer which can be used to (de)/serialize the
   * {@link TableStatistics} from/to JSON
   */
  public static ObjectMapper getMapper() {
    ObjectMapper mapper = new ObjectMapper();
    SimpleModule deModule = new SimpleModule("StatisticsSerDeModule")
        .addSerializer(TypeProtos.MajorType.class, new MajorTypeSerDe.Se())
        .addDeserializer(TypeProtos.MajorType.class, new MajorTypeSerDe.De())
        .addDeserializer(SchemaPath.class, new SchemaPath.De());
    mapper.registerModule(deModule);
    mapper.registerSubtypes(new NamedType(NumericEquiDepthHistogram.class, "numeric-equi-depth"));
    return mapper;
  }

  /**
   * Returns list of {@link StatisticsKind} and statistics values obtained from specified {@link DrillStatsTable}.
   *
   * @param statsProvider the source of statistics
   * @return list of {@link StatisticsKind} and statistics values
   */
  public static List<StatisticsHolder<?>> getEstimatedTableStats(DrillStatsTable statsProvider) {
    if (statsProvider != null && statsProvider.isMaterialized()) {
      List<StatisticsHolder<?>> tableStatistics = Arrays.asList(
          new StatisticsHolder<>(statsProvider.getRowCount(), TableStatisticsKind.EST_ROW_COUNT),
          new StatisticsHolder<>(Boolean.TRUE, TableStatisticsKind.HAS_DESCRIPTIVE_STATISTICS));
      return tableStatistics;
    }
    return Collections.emptyList();
  }

  /**
   * Returns list of {@link StatisticsKind} and statistics values obtained from specified {@link DrillStatsTable} for specified column.
   *
   * @param statsProvider the source of statistics
   * @param fieldName     name of the columns whose statistics should be obtained
   * @return list of {@link StatisticsKind} and statistics values
   */
  public static List<StatisticsHolder<?>> getEstimatedColumnStats(DrillStatsTable statsProvider, SchemaPath fieldName) {
    if (statsProvider != null && statsProvider.isMaterialized()) {
      List<StatisticsHolder<?>> statisticsValues = new ArrayList<>();
      Double ndv = statsProvider.getNdv(fieldName);
      if (ndv != null) {
        statisticsValues.add(new StatisticsHolder<>(ndv, ColumnStatisticsKind.NDV));
      }
      Double nonNullCount = statsProvider.getNNRowCount(fieldName);
      if (nonNullCount != null) {
        statisticsValues.add(new StatisticsHolder<>(nonNullCount, ColumnStatisticsKind.NON_NULL_COUNT));
      }
      Histogram histogram = statsProvider.getHistogram(fieldName);
      if (histogram != null) {
        statisticsValues.add(new StatisticsHolder<>(histogram, ColumnStatisticsKind.HISTOGRAM));
      }
      Double rowcount = statsProvider.getRowCount();
      if (rowcount != null) {
        statisticsValues.add(new StatisticsHolder<>(rowcount, ColumnStatisticsKind.ROWCOUNT));
      }
      return statisticsValues;
    }
    return Collections.emptyList();
  }
}