/*
 * 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.atlas.query

import java.util
import java.util.concurrent.atomic.AtomicInteger

import org.apache.atlas.AtlasException
import org.apache.atlas.query.Expressions.{PathExpression, SelectExpression}
import org.apache.atlas.repository.Constants
import org.apache.atlas.typesystem.types.DataTypes.{ArrayType, PrimitiveType, TypeCategory}
import org.apache.atlas.typesystem.types._

object TypeUtils {
    val typSystem = TypeSystem.getInstance()

    def numericTypes : Seq[PrimitiveType[_]] = Seq(DataTypes.BYTE_TYPE,
        DataTypes.SHORT_TYPE,
        DataTypes.INT_TYPE,
        DataTypes.FLOAT_TYPE,
        DataTypes.LONG_TYPE,
        DataTypes.DOUBLE_TYPE,
        DataTypes.BIGINTEGER_TYPE,
        DataTypes.BIGDECIMAL_TYPE)

    def combinedType(typ1 : IDataType[_], typ2 : IDataType[_]) : PrimitiveType[_] = {
        val typ1Idx =  if (numericTypes.contains(typ1))  Some(numericTypes.indexOf(typ1)) else None
        val typ2Idx =  if (numericTypes.contains(typ2))  Some(numericTypes.indexOf(typ2)) else None

        if ( typ1Idx.isDefined && typ2Idx.isDefined ) {
            val rIdx = math.max(typ1Idx.get, typ2Idx.get)

            if ( (typ1 == DataTypes.FLOAT_TYPE && typ2 == DataTypes.LONG_TYPE) ||
                (typ1 == DataTypes.LONG_TYPE && typ2 == DataTypes.FLOAT_TYPE) ) {
                return DataTypes.DOUBLE_TYPE
            }
            return numericTypes(rIdx)
        }

        throw new AtlasException(s"Cannot combine types: ${typ1.getName} and ${typ2.getName}")
    }

    var tempStructCounter : AtomicInteger = new AtomicInteger(0)
    val TEMP_STRUCT_NAME_PREFIX = "__tempQueryResultStruct"
    def createStructType(selectExprs : List[Expressions.AliasExpression]) : StructType = {
        val aDefs = new Array[AttributeDefinition](selectExprs.size)
        selectExprs.zipWithIndex.foreach { t =>
            val (e,i) = t
            aDefs(i) = new AttributeDefinition(e.alias,e.dataType.getName, Multiplicity.OPTIONAL, false, null)
        }
        return typSystem.defineQueryResultType(s"${TEMP_STRUCT_NAME_PREFIX}${tempStructCounter.getAndIncrement}",
            null,
            aDefs:_*);
    }

    object ResultWithPathStruct {
      val pathAttrName = "path"
      val resultAttrName = "result"
      val pathAttrType = DataTypes.arrayTypeName(typSystem.getIdType.getStructType)

      val pathAttr = new AttributeDefinition(pathAttrName, pathAttrType, Multiplicity.COLLECTION, false, null)

      def createType(pE : PathExpression, resultType : IDataType[_]) : StructType = {
        val resultAttr = new AttributeDefinition(resultAttrName, resultType.getName, Multiplicity.REQUIRED, false, null)
        val typName = s"${TEMP_STRUCT_NAME_PREFIX}${tempStructCounter.getAndIncrement}"
        val m : java.util.HashMap[String, IDataType[_]] = new util.HashMap[String, IDataType[_]]()
        if ( pE.child.isInstanceOf[SelectExpression]) {
          m.put(pE.child.dataType.getName, pE.child.dataType)
        }
        typSystem.defineQueryResultType(typName, m, pathAttr, resultAttr);
      }
    }

  /**
   * Structure representing the Closure Graph.
   * Returns:
   * 1. A map of vertexId -> vertex Info(these are the attributes requested in the query)
   * 2. A edges map: each entry is a mapping from an vertexId to the List of adjacent vertexIds.
   *
   * '''The Vertex Map doesn't contain all the vertices in the Graph. Only the ones for which Attributes are
   * available.''' These are the vertices that represent the EntityType whose Closure was requested. For e.g. for
   * Table Lineage the ''vertex map'' will contain information  about Tables, but not about ''Load Process'' vertices
   * that connect Tables.
   */
  object GraphResultStruct {
    val SRC_PREFIX = "src"
    val DEST_PREFIX = "dest"

    val verticesAttrName = "vertices"
    val edgesAttrName = "edges"
    val vertexIdAttrName = "vertexId"

    lazy val edgesAttrType = typSystem.defineMapType(DataTypes.STRING_TYPE,
      typSystem.defineArrayType(DataTypes.STRING_TYPE))

    def createType(resultWithPathType: StructType): StructType = {
      val resultType = resultWithPathType.fieldMapping().fields.get(ResultWithPathStruct.resultAttrName).dataType()

      val verticesAttrType = typSystem.defineMapType(DataTypes.STRING_TYPE,
        vertexType(resultType.asInstanceOf[StructType]))
      val typName = s"${TEMP_STRUCT_NAME_PREFIX}${tempStructCounter.getAndIncrement}"
      val verticesAttr = new AttributeDefinition(verticesAttrName, verticesAttrType.getName,
        Multiplicity.REQUIRED, false, null)
      val edgesAttr = new AttributeDefinition(edgesAttrName, edgesAttrType.getName, Multiplicity.REQUIRED, false, null)

      val m: java.util.HashMap[String, IDataType[_]] = new util.HashMap[String, IDataType[_]]()
      m.put(resultWithPathType.getName, resultWithPathType)
      m.put(resultType.getName, resultType)
      m.put(edgesAttrType.getName, edgesAttrType)
      m.put(verticesAttrType.getName, verticesAttrType)
      typSystem.defineQueryResultType(typName, m, verticesAttr, edgesAttr)
    }

    private def vertexType(resultType: StructType): StructType = {

      import scala.collection.JavaConverters._

      var attrs: List[AttributeDefinition] =
        resultType.fieldMapping.fields.asScala.filter(_._1.startsWith(s"${SRC_PREFIX}_")).mapValues { aInfo =>

        new AttributeDefinition(aInfo.name.substring(s"${SRC_PREFIX}_".length), aInfo.dataType.getName,
          aInfo.multiplicity, aInfo.isComposite, aInfo.reverseAttributeName)
      }.values.toList

      attrs = new AttributeDefinition(vertexIdAttrName, typSystem.getIdType.getStructType.name,
        Multiplicity.REQUIRED, false, null) :: attrs

      return typSystem.defineQueryResultType(s"${TEMP_STRUCT_NAME_PREFIX}${tempStructCounter.getAndIncrement}",
        null,
        attrs: _*)
    }
  }

    def fieldMapping(iDataType: IDataType[_]) : Option[FieldMapping] = iDataType match {
        case c : ClassType => Some(c.fieldMapping())
        case t : TraitType => Some(t.fieldMapping())
        case s : StructType => Some(s.fieldMapping())
        case _ => None
    }

    def hasFields(iDataType: IDataType[_]) : Boolean = {
        fieldMapping(iDataType).isDefined
    }

    import scala.language.existentials
    case class FieldInfo(dataType : IDataType[_],
                         attrInfo : AttributeInfo,
                         reverseDataType : IDataType[_] = null,
                          traitName : String = null) {
        def isReverse = reverseDataType != null
        override  def toString : String = {
            if ( traitName != null ) {
              s"""FieldInfo("${dataType.getName}", "$traitName")"""
            }
            else if ( reverseDataType == null ) {
                s"""FieldInfo("${dataType.getName}", "${attrInfo.name}")"""
            } else {
                s"""FieldInfo("${dataType.getName}", "${attrInfo.name}", "${reverseDataType.getName}")"""
            }
        }
    }

    val FIELD_QUALIFIER = "(.*?)(->.*)?".r

    /**
     * Given a ComposedType `t` and a name resolve using the following rules:
     * - if `id` is a field in `t` resolve to the field
     * - if `id` is the name of a  Struct|Class|Trait Type and it has a field that is of type `t` then return that type
     *
     * For e.g.
     * 1. if we have types Table(name : String, cols : List[Column]), Column(name : String) then
     * `resolveReference(Table, "cols")` resolves to type Column. So a query can be "Table.cols"
     * 2. But if we have Table(name : String), Column(name : String, tbl : Table) then "Table.Column" will resolve
     * to type Column
     *
     * This way the language will support navigation even if the relationship is one-sided.
     *
     * @param typ
     * @param id
     * @return
     */
    def resolveReference(typ : IDataType[_], id : String) : Option[FieldInfo] = {

        val fMap = fieldMapping(typ)
        if ( fMap.isDefined) {

            if (fMap.get.fields.containsKey(id)) {
                return Some(FieldInfo(typ,fMap.get.fields.get(id)))
            }

            val systemField = Constants.getAttributeInfoForSystemAttributes(id)
            if (systemField != null) {
              return Some(FieldInfo(systemField.dataType(), systemField))
            }

            try {
              val FIELD_QUALIFIER(clsNm, rest) = id
                val idTyp = typSystem.getDataType(classOf[IDataType[_]], clsNm)
                val idTypFMap = fieldMapping(idTyp)

                if (rest != null ) {
                  val attrNm = rest.substring(2)

                  if (idTypFMap.get.fields.containsKey(attrNm)) {
                    return Some(FieldInfo(typ,idTypFMap.get.fields.get(attrNm), idTyp))
                  }
                }

                if (idTypFMap.isDefined) {
                    import scala.collection.JavaConversions._
                    val fields: Seq[AttributeInfo] = idTypFMap.get.fields.values().filter { aInfo =>
                        aInfo.dataType() == typ ||
                            ( aInfo.dataType().getTypeCategory == TypeCategory.ARRAY &&
                                aInfo.dataType().asInstanceOf[ArrayType].getElemType == typ
                                )
                    }.toSeq
                    if (fields.size == 1) {
                        return Some(FieldInfo(typ, fields(0), idTyp))
                    }
                    /*
                     * is there only 1 array field of this type?
                     * If yes resolve to it.
                     * @todo: allow user to specify the relationship to follow by further qualifying the type. for e.g.
                     *   field("LoadProcess.inputTables")
                     */
                    val aFields = fields.filter { aInfo => aInfo.dataType().getTypeCategory == TypeCategory.ARRAY}
                    if (aFields.size == 1) {
                        return Some(FieldInfo(typ, aFields(0), idTyp))
                    }
                }
            } catch {
                case _ : AtlasException => None
            }
        }
        None
    }

    def resolveAsClassType(id : String) : Option[ClassType] = {
        try {
            Some(typSystem.getDataType(classOf[ClassType], id))
        } catch {
            case _ : AtlasException => None
        }
    }

    def resolveAsTraitType(id : String) : Option[TraitType] = {
        try {
            Some(typSystem.getDataType(classOf[TraitType], id))
        } catch {
            case _ : AtlasException => None
        }
    }
}