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

import org.apache.drill.common.parser.LogicalExpressionParser;
import com.mapr.db.Admin;
import com.mapr.db.MapRDB;
import com.mapr.db.exceptions.DBException;
import com.mapr.db.index.IndexDesc;
import com.mapr.db.index.IndexDesc.MissingAndNullOrdering;
import com.mapr.db.index.IndexFieldDesc;
import org.apache.calcite.rel.RelFieldCollation;
import org.apache.calcite.rel.RelFieldCollation.NullDirection;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.drill.common.exceptions.DrillRuntimeException;
import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.common.types.Types;
import org.apache.drill.common.util.DrillFileUtils;
import org.apache.drill.exec.physical.base.AbstractDbGroupScan;
import org.apache.drill.exec.physical.base.GroupScan;
import org.apache.drill.exec.planner.common.DrillScanRelBase;
import org.apache.drill.exec.planner.logical.DrillTable;
import org.apache.drill.exec.planner.physical.ScanPrel;
import org.apache.drill.exec.store.dfs.DrillFileSystem;
import org.apache.drill.exec.store.dfs.FileSelection;
import org.apache.drill.exec.store.dfs.FileSystemPlugin;
import org.apache.drill.exec.store.mapr.db.MapRDBFormatMatcher;
import org.apache.drill.exec.store.mapr.db.MapRDBFormatPlugin;
import org.apache.drill.exec.store.mapr.db.MapRDBGroupScan;
import org.apache.drill.exec.store.mapr.db.json.FieldPathHelper;
import org.apache.drill.exec.util.ImpersonationUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
import org.ojai.FieldPath;

import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.google.common.collect.Maps;

public class MapRDBIndexDiscover extends IndexDiscoverBase implements IndexDiscover {

  public static final String DEFAULT_STRING_CAST_LEN_STR = "256";

  public MapRDBIndexDiscover(GroupScan inScan, DrillScanRelBase scanRel) {
    super((AbstractDbGroupScan) inScan, scanRel);
  }

  public MapRDBIndexDiscover(GroupScan inScan, ScanPrel scanRel) {
    super((AbstractDbGroupScan) inScan, scanRel);
  }

  @Override
  public IndexCollection getTableIndex(String tableName) {
    return getTableIndexFromMFS(tableName);
  }

  /**
   * For a given table name get the list of indexes defined on the table according to the visibility of
   * the indexes based on permissions.
   * @param tableName table name
   * @return an IndexCollection representing the list of indexes for that table
   */
  private IndexCollection getTableIndexFromMFS(String tableName) {
    try {
      Set<DrillIndexDescriptor> idxSet = new HashSet<>();
      Collection<IndexDesc> indexes = admin().getTableIndexes(new Path(tableName));
      if (indexes.size() == 0 ) {
        logger.error("No index returned from Admin.getTableIndexes for table {}", tableName);
        return null;
      }
      for (IndexDesc idx : indexes) {
        DrillIndexDescriptor hbaseIdx = buildIndexDescriptor(tableName, idx);
        if (hbaseIdx == null) {
          // not able to build a valid index based on the index info from MFS
          logger.error("Not able to build index for {}", idx.toString());
          continue;
        }
        idxSet.add(hbaseIdx);
      }
      if (idxSet.size() == 0) {
        logger.error("No index found for table {}.", tableName);
        return null;
      }
      return new DrillIndexCollection(getOriginalScanRel(), idxSet);
    } catch (DBException ex) {
      logger.error("Could not get table index from File system.", ex);
    }
    catch(InvalidIndexDefinitionException ex) {
      logger.error("Invalid index definition detected.", ex);
    }
    return null;
  }

  FileSelection deriveFSSelection(DrillFileSystem fs, IndexDescriptor idxDesc) throws IOException {
    String tableName = idxDesc.getTableName();
    String[] tablePath = tableName.split(DrillFileUtils.SEPARATOR);
    String tableParent = tableName.substring(0, tableName.lastIndexOf(DrillFileUtils.SEPARATOR));

    return FileSelection.create(fs, tableParent, tablePath[tablePath.length - 1], false);
  }

  @Override
  public DrillTable getNativeDrillTable(IndexDescriptor idxDescriptor) {

    try {
      final AbstractDbGroupScan origScan = getOriginalScan();
      if (!(origScan instanceof MapRDBGroupScan)) {
        return null;
      }
      MapRDBFormatPlugin maprFormatPlugin = ((MapRDBGroupScan) origScan).getFormatPlugin();
      FileSystemPlugin fsPlugin = (FileSystemPlugin) (origScan.getStoragePlugin());

      DrillFileSystem fs = ImpersonationUtil.createFileSystem(origScan.getUserName(), fsPlugin.getFsConf());
      MapRDBFormatMatcher matcher = (MapRDBFormatMatcher) (maprFormatPlugin.getMatcher());
      FileSelection fsSelection = deriveFSSelection(fs, idxDescriptor);
      return matcher.isReadableIndex(fs, fsSelection, fsPlugin, fsPlugin.getName(),
          origScan.getUserName(), idxDescriptor);

    } catch (Exception e) {
      logger.error("Failed to get native DrillTable.", e);
    }
    return null;
  }

  private SchemaPath fieldName2SchemaPath(String fieldName) {
    if (fieldName.contains(":")) {
      fieldName = fieldName.split(":")[1];
    }
    if (fieldName.contains(".")) {
      return FieldPathHelper.fieldPath2SchemaPath(FieldPath.parseFrom(fieldName));
    }
    return SchemaPath.getSimplePath(fieldName);
  }

  String getDrillTypeStr(String maprdbTypeStr) {
    String typeStr = maprdbTypeStr.toUpperCase();
    String[] typeTokens = typeStr.split("[)(]");
    String typeData = DEFAULT_STRING_CAST_LEN_STR;
    if(typeTokens.length > 1) {
      typeStr = typeTokens[0];
      typeData = typeTokens[1];
    }
    switch(typeStr){
      case "STRING":
        // set default width since it is not specified
        return "VARCHAR("+typeData+")";
      case "LONG":
        return "BIGINT";
      case "INT":
      case "INTEGER":
        return "INT";
      case "FLOAT":
        return "FLOAT4";
      case "DOUBLE":
        return "FLOAT8";
      case "INTERVAL_YEAR_MONTH":
        return "INTERVALYEAR";
      case "INTERVAL_DAY_TIME":
        return "INTERVALDAY";
      case "BOOLEAN":
        return "BIT";
      case "BINARY":
        return "VARBINARY";
      case "ANY":
      case "DECIMAL":
        return null;
      default: return typeStr;
    }

  }

  TypeProtos.MajorType getDrillType(String typeStr) {
    switch(typeStr){
      case "VARCHAR":
      case "CHAR":
      case "STRING":
        // set default width since it is not specified
        return
            Types.required(TypeProtos.MinorType.VARCHAR).toBuilder().setWidth(
                getOriginalScanRel().getCluster().getTypeFactory().createSqlType(SqlTypeName.VARCHAR).getPrecision()).build();
      case "LONG":
      case "BIGINT":
        return Types.required(TypeProtos.MinorType.BIGINT);
      case "INT":
      case "INTEGER":
        return Types.required(TypeProtos.MinorType.INT);
      case "FLOAT":
        return Types.required(TypeProtos.MinorType.FLOAT4);
      case "DOUBLE":
        return Types.required(TypeProtos.MinorType.FLOAT8);
      case "INTERVAL_YEAR_MONTH":
        return Types.required(TypeProtos.MinorType.INTERVALYEAR);
      case "INTERVAL_DAY_TIME":
        return Types.required(TypeProtos.MinorType.INTERVALDAY);
      case "BOOLEAN":
        return Types.required(TypeProtos.MinorType.BIT);
      case "BINARY":
        return Types.required(TypeProtos.MinorType.VARBINARY).toBuilder().build();
      case "ANY":
      case "DECIMAL":
        return null;
      default: return Types.required(TypeProtos.MinorType.valueOf(typeStr));
    }
  }

  private LogicalExpression castFunctionSQLSyntax(String field, String type) throws InvalidIndexDefinitionException {
    // get castTypeStr so we can construct SQL syntax string before MapRDB could provide such syntax
    String castTypeStr = getDrillTypeStr(type);
    if(castTypeStr == null) {  // no cast
      throw new InvalidIndexDefinitionException("cast function type not recognized: " + type + "for field " + field);
    }
    try {
      String castFunc = String.format("cast( %s as %s)", field, castTypeStr);
      return LogicalExpressionParser.parse(castFunc);
    } catch (Exception ex) {
      logger.error("parse failed: {}", ex);
    }
    return null;
  }

  private LogicalExpression getIndexExpression(IndexFieldDesc desc) throws InvalidIndexDefinitionException {
    final String fieldName = desc.getFieldPath().asPathString();
    final String functionDef = desc.getFunctionName();
    if ((functionDef != null)) {  // this is a function
      String[] tokens = functionDef.split("\\s+");
      if (tokens[0].equalsIgnoreCase("cast")) {
        if (tokens.length != 3) {
          throw new InvalidIndexDefinitionException("cast function definition not recognized: " + functionDef);
        }
        LogicalExpression idxExpr = castFunctionSQLSyntax(fieldName, tokens[2]);
        if (idxExpr == null) {
          throw new InvalidIndexDefinitionException("got null expression for function definition: " + functionDef);
        }
        return idxExpr;
      } else {
        throw new InvalidIndexDefinitionException("function definition is not supported for indexing: " + functionDef);
      }
    }
    // else it is a schemaPath
    return fieldName2SchemaPath(fieldName);
  }

  private List<LogicalExpression> field2SchemaPath(Collection<IndexFieldDesc> descCollection)
      throws InvalidIndexDefinitionException {
    List<LogicalExpression> listSchema = new ArrayList<>();
    for (IndexFieldDesc field : descCollection) {
        listSchema.add(getIndexExpression(field));
    }
    return listSchema;
  }

  private List<RelFieldCollation> getFieldCollations(IndexDesc desc, Collection<IndexFieldDesc> descCollection) {
    List<RelFieldCollation> fieldCollations = new ArrayList<>();
    int i = 0;
    for (IndexFieldDesc field : descCollection) {
      RelFieldCollation.Direction direction = (field.getSortOrder() == IndexFieldDesc.Order.Asc) ?
          RelFieldCollation.Direction.ASCENDING : (field.getSortOrder() == IndexFieldDesc.Order.Desc ?
              RelFieldCollation.Direction.DESCENDING : null);
      if (direction != null) {
        // assume null direction of NULLS UNSPECIFIED for now until MapR-DB adds that to the APIs
        RelFieldCollation.NullDirection nulldir =
            direction == RelFieldCollation.Direction.ASCENDING ? NullDirection.LAST :
            (direction == RelFieldCollation.Direction.DESCENDING ?
                NullDirection.FIRST : NullDirection.UNSPECIFIED);
        RelFieldCollation c = new RelFieldCollation(i++, direction, nulldir);
        fieldCollations.add(c);
      } else {
        // if the direction is not present for a field, no need to examine remaining fields
        break;
      }
    }
    return fieldCollations;
  }

  private CollationContext buildCollationContext(List<LogicalExpression> indexFields,
      List<RelFieldCollation> indexFieldCollations) {
    assert indexFieldCollations.size() <= indexFields.size();
    Map<LogicalExpression, RelFieldCollation> collationMap = Maps.newHashMap();
    for (int i = 0; i < indexFieldCollations.size(); i++) {
      collationMap.put(indexFields.get(i), indexFieldCollations.get(i));
    }
    return new CollationContext(collationMap, indexFieldCollations);
  }

  private DrillIndexDescriptor buildIndexDescriptor(String tableName, IndexDesc desc)
      throws InvalidIndexDefinitionException {
    if (desc.isExternal()) {
      // External index is not currently supported
      return null;
    }

    IndexDescriptor.IndexType idxType = IndexDescriptor.IndexType.NATIVE_SECONDARY_INDEX;
    List<LogicalExpression> indexFields = field2SchemaPath(desc.getIndexedFields());
    List<LogicalExpression> coveringFields = field2SchemaPath(desc.getIncludedFields());
    coveringFields.add(SchemaPath.getSimplePath("_id"));
    CollationContext collationContext = null;
    if (!desc.isHashed()) { // hash index has no collation property
      List<RelFieldCollation> indexFieldCollations = getFieldCollations(desc, desc.getIndexedFields());
      collationContext = buildCollationContext(indexFields, indexFieldCollations);
    }

    DrillIndexDescriptor idx = new MapRDBIndexDescriptor (
        indexFields,
        collationContext,
        coveringFields,
        null,
        desc.getIndexName(),
        tableName,
        idxType,
        desc,
        this.getOriginalScan(),
        desc.getMissingAndNullOrdering() == MissingAndNullOrdering.MissingAndNullFirst ? NullDirection.FIRST :
            (desc.getMissingAndNullOrdering() == MissingAndNullOrdering.MissingAndNullLast ?
                NullDirection.LAST : NullDirection.UNSPECIFIED));

    String storageName = this.getOriginalScan().getStoragePlugin().getName();
    materializeIndex(storageName, idx);
    return idx;
  }

  @SuppressWarnings("deprecation")
  private Admin admin() {
    assert getOriginalScan() instanceof MapRDBGroupScan;

    final MapRDBGroupScan dbGroupScan = (MapRDBGroupScan) getOriginalScan();
    final UserGroupInformation currentUser = ImpersonationUtil.createProxyUgi(dbGroupScan.getUserName());
    final Configuration conf = dbGroupScan.getFormatPlugin().getFsConf();

    final Admin admin;
    try {
      admin = currentUser.doAs((PrivilegedExceptionAction<Admin>) () -> MapRDB.getAdmin(conf));
    } catch (Exception e) {
      throw new DrillRuntimeException("Failed to get Admin instance for user: " + currentUser.getUserName(), e);
    }
    return admin;
  }
}