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

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.orc.Reader;
import org.apache.orc.TypeDescription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Infer and track the evolution between the schema as stored in the file and
 * the schema that has been requested by the reader.
 */
public class SchemaEvolution {
  // indexed by reader column id
  private final TypeDescription[] readerFileTypes;
  // indexed by reader column id
  private final boolean[] readerIncluded;
  // the offset to the first column id ignoring any ACID columns
  private final int readerColumnOffset;
  // indexed by file column id
  private final boolean[] fileIncluded;
  private final TypeDescription fileSchema;
  private final TypeDescription readerSchema;
  private boolean hasConversion;
  private boolean isOnlyImplicitConversion;
  private final boolean isAcid;
  private final boolean isSchemaEvolutionCaseAware;
  /**
   * {@code true} if acid metadata columns should be decoded otherwise they will
   * be set to {@code null}.  {@link #acidEventFieldNames}.
   */
  private final boolean includeAcidColumns;

  // indexed by reader column id
  private final boolean[] ppdSafeConversion;

  // columns are indexed, not named between Reader & File schema
  private final boolean positionalColumns;

  private static final Logger LOG =
    LoggerFactory.getLogger(SchemaEvolution.class);
  private static final Pattern missingMetadataPattern =
    Pattern.compile("_col\\d+");


  public static class IllegalEvolutionException extends RuntimeException {
    public IllegalEvolutionException(String msg) {
      super(msg);
    }
  }
  public SchemaEvolution(TypeDescription fileSchema,
                         TypeDescription readerSchema,
                         Reader.Options options) {
    boolean allowMissingMetadata = options.getTolerateMissingSchema();
    boolean[] includedCols = options.getInclude();
    this.isSchemaEvolutionCaseAware=options.getIsSchemaEvolutionCaseAware();
    this.readerIncluded = includedCols == null ? null :
      Arrays.copyOf(includedCols, includedCols.length);
    this.fileIncluded = new boolean[fileSchema.getMaximumId() + 1];
    this.hasConversion = false;
    this.isOnlyImplicitConversion = true;
    this.fileSchema = fileSchema;
    this.isAcid = checkAcidSchema(fileSchema);
    boolean readerSchemaIsAcid = readerSchema == null ? false : checkAcidSchema(readerSchema);
    this.includeAcidColumns = options.getIncludeAcidColumns();
    this.readerColumnOffset = isAcid && !readerSchemaIsAcid ? acidEventFieldNames.size() : 0;
    if (readerSchema != null) {
      if (isAcid && !readerSchemaIsAcid) {
        this.readerSchema = createEventSchema(readerSchema);
      } else {
        this.readerSchema = readerSchema;
      }
      if (readerIncluded != null &&
          readerIncluded.length + readerColumnOffset !=
            this.readerSchema.getMaximumId() + 1) {
        throw new IllegalArgumentException("Include vector the wrong length: "
            + this.readerSchema.toJson() + " with include length "
            + readerIncluded.length);
      }
      this.readerFileTypes =
        new TypeDescription[this.readerSchema.getMaximumId() + 1];
      int positionalLevels = 0;
      if (options.getForcePositionalEvolution()) {
        positionalLevels = isAcid ? 2 : options.getPositionalEvolutionLevel();
      } else if (!hasColumnNames(isAcid? getBaseRow(fileSchema) : fileSchema)) {
        if (!this.fileSchema.equals(this.readerSchema)) {
          if (!allowMissingMetadata) {
            throw new RuntimeException("Found that schema metadata is missing"
                + " from file. This is likely caused by"
                + " a writer earlier than HIVE-4243. Will"
                + " not try to reconcile schemas");
          } else {
            LOG.warn("Column names are missing from this file. This is"
                + " caused by a writer earlier than HIVE-4243. The reader will"
                + " reconcile schemas based on index. File type: " +
                this.fileSchema + ", reader type: " + this.readerSchema);
            positionalLevels = isAcid ? 2 : options.getPositionalEvolutionLevel();
          }
        }
      }
      buildConversion(fileSchema, this.readerSchema, positionalLevels);
    } else {
      this.readerSchema = fileSchema;
      this.readerFileTypes =
        new TypeDescription[this.readerSchema.getMaximumId() + 1];
      if (readerIncluded != null &&
          readerIncluded.length + readerColumnOffset !=
            this.readerSchema.getMaximumId() + 1) {
        throw new IllegalArgumentException("Include vector the wrong length: "
            + this.readerSchema.toJson() + " with include length "
            + readerIncluded.length);
      }
      buildIdentityConversion(this.readerSchema);
    }
    this.positionalColumns = options.getForcePositionalEvolution();
    this.ppdSafeConversion = populatePpdSafeConversion();
  }

  @Deprecated
  public SchemaEvolution(TypeDescription fileSchema, boolean[] readerIncluded) {
    this(fileSchema, null, readerIncluded);
  }

  @Deprecated
  public SchemaEvolution(TypeDescription fileSchema,
                         TypeDescription readerSchema,
                         boolean[] readerIncluded) {
    this(fileSchema, readerSchema,
        new Reader.Options(new Configuration())
            .include(readerIncluded));
  }

  // Return true iff all fields have names like _col[0-9]+
  private boolean hasColumnNames(TypeDescription fileSchema) {
    if (fileSchema.getCategory() != TypeDescription.Category.STRUCT) {
      return true;
    }
    for (String fieldName : fileSchema.getFieldNames()) {
      if (!missingMetadataPattern.matcher(fieldName).matches()) {
        return true;
      }
    }
    return false;
  }

  public TypeDescription getReaderSchema() {
    return readerSchema;
  }

  /**
   * Returns the non-ACID (aka base) reader type description.
   *
   * @return the reader type ignoring the ACID rowid columns, if any
   */
  public TypeDescription getReaderBaseSchema() {
    return isAcid ? getBaseRow(readerSchema) : readerSchema;
  }

  /**
   * Does the file include ACID columns?
   * @return is this an ACID file?
   */
  boolean isAcid() {
    return isAcid;
  }

  /**
   * Is there Schema Evolution data type conversion?
   * @return
   */
  public boolean hasConversion() {
    return hasConversion;
  }

  /**
   * When there Schema Evolution data type conversion i.e. hasConversion() returns true,
   * is the conversion only the implicit kind?
   *
   * (see aaa).
   * @return
   */
  public boolean isOnlyImplicitConversion() {
    return isOnlyImplicitConversion;
  }

  public TypeDescription getFileSchema() {
    return fileSchema;
  }

  public TypeDescription getFileType(TypeDescription readerType) {
    return getFileType(readerType.getId());
  }

  /**
   * Get the file type by reader type id.
   * @param id reader column id
   * @return
   */
  public TypeDescription getFileType(int id) {
    return readerFileTypes[id];
  }

  /**
   * Get whether each column is included from the reader's point of view.
   * @return a boolean array indexed by reader column id
   */
  public boolean[] getReaderIncluded() {
    return readerIncluded;
  }

  /**
   * Get whether each column is included from the file's point of view.
   * @return a boolean array indexed by file column id
   */
  public boolean[] getFileIncluded() {
    return fileIncluded;
  }

  /**
   * Get whether the columns are handled via position or name
   */
  public boolean getPositionalColumns() {
    return this.positionalColumns;
  }

  /**
   * Determine if there is implicit conversion from a file to reader type.
   *
   * Implicit conversions are:
   *   Small to larger integer (e.g. INT to LONG)
   *   FLOAT to DOUBLE
   *   Some String Family conversions.
   *
   * NOTE: This check is independent of the PPD conversion checks.
   * @return
   */
  private boolean typesAreImplicitConversion(final TypeDescription fileType,
      final TypeDescription readerType) {
    switch (fileType.getCategory()) {
    case BYTE:
        if (readerType.getCategory().equals(TypeDescription.Category.SHORT) ||
            readerType.getCategory().equals(TypeDescription.Category.INT) ||
            readerType.getCategory().equals(TypeDescription.Category.LONG)) {
          return true;
        }
        break;
    case SHORT:
        if (readerType.getCategory().equals(TypeDescription.Category.INT) ||
            readerType.getCategory().equals(TypeDescription.Category.LONG)) {
          return true;
        }
        break;
    case INT:
        if (readerType.getCategory().equals(TypeDescription.Category.LONG)) {
          return true;
        }
        break;
    case FLOAT:
        if (readerType.getCategory().equals(TypeDescription.Category.DOUBLE)) {
          return true;
        }
        break;
    case CHAR:
    case VARCHAR:
        if (readerType.getCategory().equals(TypeDescription.Category.STRING)) {
          return true;
        }
        if (readerType.getCategory().equals(TypeDescription.Category.CHAR) ||
            readerType.getCategory().equals(TypeDescription.Category.VARCHAR)) {
          return (fileType.getMaxLength() <= readerType.getMaxLength());
        }
        break;
    default:
        break;
    }
    return false;
  }

  /**
   * Check if column is safe for ppd evaluation
   * @param colId reader column id
   * @return true if the specified column is safe for ppd evaluation else false
   */
  public boolean isPPDSafeConversion(final int colId) {
    if (hasConversion()) {
      return !(colId < 0 || colId >= ppdSafeConversion.length) &&
          ppdSafeConversion[colId];
    }

    // when there is no schema evolution PPD is safe
    return true;
  }

  private boolean[] populatePpdSafeConversion() {
    if (fileSchema == null || readerSchema == null || readerFileTypes == null) {
      return null;
    }

    boolean[] result = new boolean[readerSchema.getMaximumId() + 1];
    boolean safePpd = validatePPDConversion(fileSchema, readerSchema);
    result[readerSchema.getId()] = safePpd;
    return populatePpdSafeConversionForChildern(result,
        readerSchema.getChildren());
  }

  /**
   * Recursion to check the conversion of nested field.
   *
   * @param ppdSafeConversion boolean array to specify which column are safe.
   * @param children reader schema children.
   *
   * @return boolean array to represent list of column safe or not.
   */
  private boolean[] populatePpdSafeConversionForChildern(
                        boolean[] ppdSafeConversion,
                        List<TypeDescription> children) {
    boolean safePpd;
    if (children != null) {
      for (TypeDescription child : children) {
        TypeDescription fileType = getFileType(child.getId());
        safePpd = validatePPDConversion(fileType, child);
        ppdSafeConversion[child.getId()] = safePpd;
        populatePpdSafeConversionForChildern(ppdSafeConversion,
            child.getChildren());
      }
    }
    return  ppdSafeConversion;
  }

  private boolean validatePPDConversion(final TypeDescription fileType,
      final TypeDescription readerType) {
    if (fileType == null) {
      return false;
    }
    if (fileType.getCategory().isPrimitive()) {
      if (fileType.getCategory().equals(readerType.getCategory())) {
        // for decimals alone do equality check to not mess up with precision change
        return !(fileType.getCategory() == TypeDescription.Category.DECIMAL &&
            !fileType.equals(readerType));
      }

      // only integer and string evolutions are safe
      // byte -> short -> int -> long
      // string <-> char <-> varchar
      // NOTE: Float to double evolution is not safe as floats are stored as doubles in ORC's
      // internal index, but when doing predicate evaluation for queries like "select * from
      // orc_float where f = 74.72" the constant on the filter is converted from string -> double
      // so the precisions will be different and the comparison will fail.
      // Soon, we should convert all sargs that compare equality between floats or
      // doubles to range predicates.

      // Similarly string -> char and varchar -> char and vice versa is not possible, as ORC stores
      // char with padded spaces in its internal index.
      switch (fileType.getCategory()) {
        case BYTE:
          if (readerType.getCategory().equals(TypeDescription.Category.SHORT) ||
              readerType.getCategory().equals(TypeDescription.Category.INT) ||
              readerType.getCategory().equals(TypeDescription.Category.LONG)) {
            return true;
          }
          break;
        case SHORT:
          if (readerType.getCategory().equals(TypeDescription.Category.INT) ||
              readerType.getCategory().equals(TypeDescription.Category.LONG)) {
            return true;
          }
          break;
        case INT:
          if (readerType.getCategory().equals(TypeDescription.Category.LONG)) {
            return true;
          }
          break;
        case STRING:
          if (readerType.getCategory().equals(TypeDescription.Category.VARCHAR)) {
            return true;
          }
          break;
        case VARCHAR:
          if (readerType.getCategory().equals(TypeDescription.Category.STRING)) {
            return true;
          }
          break;
        default:
          break;
      }
    }
    return false;
  }

  /**
   * Should we read the given reader column?
   * @param readerId the id of column in the extended reader schema
   * @return true if the column should be read
   */
  public boolean includeReaderColumn(int readerId) {
    if(readerId == 0) {
      //always want top level struct - everything is its child
      return true;
    }
    if(isAcid) {
      if(readerId < readerColumnOffset) {
        return includeAcidColumns;
      }
      return readerIncluded == null ||
          readerIncluded[readerId - readerColumnOffset];
    }
    return readerIncluded == null || readerIncluded[readerId];
  }

  /**
   * Build the mapping from the file type to the reader type. For pre-HIVE-4243
   * ORC files, the top level structure is matched using position within the
   * row. Otherwise, structs fields are matched by name.
   * @param fileType the type in the file
   * @param readerType the type in the reader
   * @param positionalLevels the number of structure levels that must be
   *                         mapped by position rather than field name. Pre
   *                         HIVE-4243 files have either 1 or 2 levels matched
   *                         positionally depending on whether they are ACID.
   */
  void buildConversion(TypeDescription fileType,
                       TypeDescription readerType,
                       int positionalLevels) {
    // if the column isn't included, don't map it
    if (!includeReaderColumn(readerType.getId())) {
      return;
    }
    boolean isOk = true;
    // check the easy case first
    if (fileType.getCategory() == readerType.getCategory()) {
      switch (readerType.getCategory()) {
        case BOOLEAN:
        case BYTE:
        case SHORT:
        case INT:
        case LONG:
        case DOUBLE:
        case FLOAT:
        case STRING:
        case TIMESTAMP:
        case TIMESTAMP_INSTANT:
        case BINARY:
        case DATE:
          // these are always a match
          break;
        case CHAR:
        case VARCHAR:
          // We do conversion when same CHAR/VARCHAR type but different
          // maxLength.
          if (fileType.getMaxLength() != readerType.getMaxLength()) {
            hasConversion = true;
            if (!typesAreImplicitConversion(fileType, readerType)) {
              isOnlyImplicitConversion = false;
            }
          }
          break;
        case DECIMAL:
          // We do conversion when same DECIMAL type but different
          // precision/scale.
          if (fileType.getPrecision() != readerType.getPrecision() ||
              fileType.getScale() != readerType.getScale()) {
            hasConversion = true;
            isOnlyImplicitConversion = false;
          }
          break;
        case UNION:
        case MAP:
        case LIST: {
          // these must be an exact match
          List<TypeDescription> fileChildren = fileType.getChildren();
          List<TypeDescription> readerChildren = readerType.getChildren();
          if (fileChildren.size() == readerChildren.size()) {
            for(int i=0; i < fileChildren.size(); ++i) {
              buildConversion(fileChildren.get(i),
                              readerChildren.get(i), positionalLevels - 1);
            }
          } else {
            isOk = false;
          }
          break;
        }
        case STRUCT: {
          List<TypeDescription> readerChildren = readerType.getChildren();
          List<TypeDescription> fileChildren = fileType.getChildren();
          if (fileChildren.size() != readerChildren.size()) {
            hasConversion = true;
            // UNDONE: Does LLAP detect fewer columns and NULL them out????
            isOnlyImplicitConversion = false;
          }

          if (positionalLevels == 0) {
            List<String> readerFieldNames = readerType.getFieldNames();
            List<String> fileFieldNames = fileType.getFieldNames();

            final Map<String, TypeDescription> fileTypesIdx;
            if (isSchemaEvolutionCaseAware) {
              fileTypesIdx = new HashMap<>();
            } else {
              fileTypesIdx = new CaseInsensitiveMap<TypeDescription>();
            }
            for (int i = 0; i < fileFieldNames.size(); i++) {
              final String fileFieldName = fileFieldNames.get(i);
              fileTypesIdx.put(fileFieldName, fileChildren.get(i));
            }

            for (int i = 0; i < readerFieldNames.size(); i++) {
              final String readerFieldName = readerFieldNames.get(i);
              TypeDescription readerField = readerChildren.get(i);

              TypeDescription fileField = fileTypesIdx.get(readerFieldName);
              if (fileField == null) {
                continue;
              }

              buildConversion(fileField, readerField, 0);
            }
          } else {
            int jointSize = Math.min(fileChildren.size(),
                                     readerChildren.size());
            for (int i = 0; i < jointSize; ++i) {
              buildConversion(fileChildren.get(i), readerChildren.get(i),
                  positionalLevels - 1);
            }
          }
          break;
        }
        default:
          throw new IllegalArgumentException("Unknown type " + readerType);
      }
    } else {
      /*
       * Check for the few cases where will not convert....
       */

      isOk = ConvertTreeReaderFactory.canConvert(fileType, readerType);
      hasConversion = true;
      if (!typesAreImplicitConversion(fileType, readerType)) {
        isOnlyImplicitConversion = false;
      }
    }
    if (isOk) {
      readerFileTypes[readerType.getId()] = fileType;
      fileIncluded[fileType.getId()] = true;
    } else {
      throw new IllegalEvolutionException(
          String.format("ORC does not support type conversion from file" +
                        " type %s (%d) to reader type %s (%d)",
                        fileType.toString(), fileType.getId(),
                        readerType.toString(), readerType.getId()));
    }
  }

  void buildIdentityConversion(TypeDescription readerType) {
    int id = readerType.getId();
    if (!includeReaderColumn(id)) {
      return;
    }
    if (readerFileTypes[id] != null) {
      throw new RuntimeException("reader to file type entry already assigned");
    }
    readerFileTypes[id] = readerType;
    fileIncluded[id] = true;
    List<TypeDescription> children = readerType.getChildren();
    if (children != null) {
      for (TypeDescription child : children) {
        buildIdentityConversion(child);
      }
    }
  }

  public static boolean checkAcidSchema(TypeDescription type) {
    if (type.getCategory().equals(TypeDescription.Category.STRUCT)) {
      List<String> rootFields = type.getFieldNames();
      if (rootFields.size() != acidEventFieldNames.size()) {
        return false;
      }
      for (int i = 0; i < rootFields.size(); i++) {
        if (!acidEventFieldNames.get(i).equalsIgnoreCase(rootFields.get(i))) {
          return false;
        }
      }
      return true;
    }
    return false;
  }

  /**
   * @param typeDescr
   * @return ORC types for the ACID event based on the row's type description
   */
  public static TypeDescription createEventSchema(TypeDescription typeDescr) {
    TypeDescription result = TypeDescription.createStruct()
        .addField("operation", TypeDescription.createInt())
        .addField("originalTransaction", TypeDescription.createLong())
        .addField("bucket", TypeDescription.createInt())
        .addField("rowId", TypeDescription.createLong())
        .addField("currentTransaction", TypeDescription.createLong())
        .addField("row", typeDescr.clone());
    return result;
  }

  /**
   * Get the underlying base row from an ACID event struct.
   * @param typeDescription the ACID event schema.
   * @return the subtype for the real row
   */
  public static TypeDescription getBaseRow(TypeDescription typeDescription) {
    final int ACID_ROW_OFFSET = 5;
    return typeDescription.getChildren().get(ACID_ROW_OFFSET);
  }

  private static final List<String> acidEventFieldNames=
    new ArrayList<String>();

  static {
    acidEventFieldNames.add("operation");
    acidEventFieldNames.add("originalTransaction");
    acidEventFieldNames.add("bucket");
    acidEventFieldNames.add("rowId");
    acidEventFieldNames.add("currentTransaction");
    acidEventFieldNames.add("row");
  }

  private static class CaseInsensitiveMap<V> extends HashMap<String,V> {
    @Override
    public V put(String key, V value) {
      return super.put(key.toLowerCase(), value);
    }

    @Override
    public V get(Object key) {
      return this.get((String) key);
    }

    // not @Override as key to be of type Object
    public V get(String key) {
      return super.get(key.toLowerCase());
    }
  }
}