package com.daumkakao.s2graph.core

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.HConnection
import org.apache.hadoop.hbase.client.HConnectionManager
import org.apache.hadoop.hbase.client.HTableInterface
import java.util.concurrent.Executors
import java.util.concurrent.ConcurrentHashMap
import org.slf4j.LoggerFactory

import scala.collection.mutable.SynchronizedQueue
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.filter.FilterList
import HBaseElement._
import org.apache.hadoop.hbase.client.Get
import org.apache.hadoop.hbase.client.Result
import scala.collection.JavaConversions._
import org.apache.hadoop.hbase.util.Bytes
import scala.collection.mutable.HashSet
import GraphUtil._
import org.apache.hadoop.hbase.filter.ColumnPaginationFilter
import scala.concurrent._
import scala.concurrent.duration._
import org.apache.hadoop.hbase.client.HTable
import scala.collection.mutable.ListBuffer
import scala.annotation.tailrec
import scala.collection.mutable.HashMap
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter
import org.apache.hadoop.hbase.Cell
import org.apache.hadoop.hbase.client.Delete
import Management._
import org.apache.hadoop.hbase.client.Row
import play.api.libs.json.Json
import KGraphExceptions._
import org.apache.hadoop.hbase.client.Mutation
import play.libs.Akka
import com.typesafe.config.Config
import scala.reflect.ClassTag
import org.hbase.async._
import com.stumbleupon.async.Deferred
import com.stumbleupon.async.Callback
import java.util.ArrayList

import scala.util.{Try, Success, Failure}

object GraphConstant {

  val vertexCf = "v".getBytes()
  val edgeCf = "e".getBytes()
  val updateCf = "u".getBytes()
  val ttsForActivity = 60 * 60 * 24 * 30
  val delimiter = "|"
  val seperator = ":"

  val writeBufferSize = 1024 * 1024 * 2

  val maxValidEdgeListSize = 10000

  //  implicit val ex = play.api.libs.concurrent.Execution.Implicits.defaultContext
  val queryLogger = play.api.Logger("query")
}
object GraphConnection {
  val logger = Graph.logger
  lazy val tablePool = Executors.newFixedThreadPool(1)
  lazy val connectionPool = Executors.newFixedThreadPool(1)
  val defaultConfigs = Map(
    "hbase.zookeeper.quorum" -> "localhost",
    "hbase.table.name" -> "s2graph",
    "phase" -> "dev")
  var config: Config = null

  def getOrElse[T: ClassTag](conf: com.typesafe.config.Config)(key: String, default: T): T = {
    if (conf.hasPath(key)) (default match {
      case _: String => conf.getString(key)
      case _: Int | _: Integer => conf.getInt(key)
      case _: Float | _: Double => conf.getDouble(key)
      case _: Boolean => conf.getBoolean(key)
      case _ => default
    }).asInstanceOf[T]
    else default
  }
  /**
   * requred: hbase.zookeeper.quorum
   * optional: all hbase. prefix configurations.
   */
  def toHBaseConfig(config: com.typesafe.config.Config) = {
    val configVals = for ((k, v) <- defaultConfigs) yield {
      val currentVal = getOrElse(config)(k, v)
      logger.debug(s"$k -> $currentVal")
      k -> currentVal
    }
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", configVals("hbase.zookeeper.quorum"))
    for (entry <- config.entrySet() if entry.getKey().startsWith("hbase.")) {
      val value = entry.getValue().unwrapped().toString
      conf.set(entry.getKey(), value)
    }
    conf
  }
  def apply(config: Config) = {
    this.config = config
    val hbaseConfig = toHBaseConfig(config)
    (hbaseConfig -> HConnectionManager.createConnection(hbaseConfig))
  }
}

object Graph {
  import GraphConstant._
  import GraphConnection._

  val logger = Edge.logger
  val conns = scala.collection.mutable.Map[String, HConnection]()
  val clients = scala.collection.mutable.Map[String, HBaseClient]()
  val emptyKVs = new ArrayList[KeyValue]()
  val emptyKVlist = new ArrayList[ArrayList[KeyValue]]();
  val emptyEdgeList = new ArrayList[ArrayList[(Edge, Double, QueryParam)]]
  val emptyEdges = new ArrayList[(Edge, Double, QueryParam)]

  var executionContext: ExecutionContext = null
  var config: com.typesafe.config.Config = null
  var hbaseConfig: org.apache.hadoop.conf.Configuration = null
  var storageExceptionCount = 0L
  var singleGetTimeout = 1000 millis
  //  implicit val ex = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(16))
  def apply(config: com.typesafe.config.Config)(implicit ex: ExecutionContext) = {
    this.config = config
    val (hbaseConfig, conn) = GraphConnection.apply(config)
    this.hbaseConfig = hbaseConfig
    Model.apply(config)
    this.executionContext = ex
    this.singleGetTimeout = getOrElse(config)("hbase.client.operation.timeout", 1000 millis)
    val zkQuorum = hbaseConfig.get("hbase.zookeeper.quorum")
    conns += (zkQuorum -> conn)
    //    clients += (zkQuorum -> new HBaseClient(zkQuorum, "/hbase", Executors.newCachedThreadPool(), 8))
    clients += (zkQuorum -> new HBaseClient(zkQuorum))

  }
  def getClient(zkQuorum: String) = {
    clients.get(zkQuorum) match {
      case None =>
        val client = new HBaseClient(zkQuorum)
        clients += (zkQuorum -> client)
        client
      //        throw new RuntimeException(s"connection to $zkQuorum is not established.")
      case Some(c) => c
    }
  }
  def getConn(zkQuorum: String) = {
    conns.get(zkQuorum) match {
      case None =>
        val conn = HConnectionManager.createConnection(this.hbaseConfig)
        conns += (zkQuorum -> conn)
        conn
      //        throw new RuntimeException(s"connection to $zkQuorum is not established.")
      case Some(c) => c
    }
  }
//
//  def withWriteTable[T](connName: String, tName: String)(op: HTableInterface => T)(fallback: => T): T = {
//    //    Logger.debug(s"withWriteTable: $connName, $tName")
//    try {
//      val conn = getConn(connName)
//      if (!conn.isMasterRunning()) throw new RuntimeException(s"master is not running. $connName")
//      val table = conn.getTable(tName, tablePool)
//
//      table.setAutoFlush(false, false)
//      table.setWriteBufferSize(writeBufferSize)
//      try {
//        op(table)
//      } catch {
//        case e: Exception =>
//          Logger.error(s"Write Operation to table ($connName), ($tName) is failed: ${e.getMessage}", e)
//          fallback
//          throw e
//      } finally {
//        table.close()
//      }
//    } catch {
//      case e: Exception =>
//        Logger.error(s"withWriteTable ($connName) ($tName) is failed: ${e.getMessage}", e)
//        fallback
//        throw e
//    }
//  }
//
//  def withReadTable[T](connName: String, tName: String)(op: HTableInterface => T)(fallback: => T): T = {
//    //    queryLogger.debug(s"withReadTable: $connName, $tName")
//    try {
//      val conn = getConn(connName)
//      if (!conn.isMasterRunning()) throw new RuntimeException(s"master is not running. $connName")
//      val table = conn.getTable(tName, tablePool)
//
//      table.setAutoFlush(false, false)
//      table.setWriteBufferSize(writeBufferSize)
//      try {
//        op(table)
//      } catch {
//        case e: Exception =>
//          queryLogger.error(s"Read Operation to table ($connName) ($tName) is failed: ${e.getMessage}", e)
//          fallback
//      } finally {
//        table.close()
//      }
//    } catch {
//      case e: Exception =>
//        queryLogger.error(s"withReadTable ($connName) ($tName) is failed: ${e.getMessage}", e)
//        fallback
//    }
//  }
//
//  // to catch exception on htable.get
//  def htableGet(table: HTableInterface)(get: Get) = {
//    try {
//      table.get(get)
//    } catch {
//      case e: Throwable =>
//        // if hbase is throw any exception, simply return empty result
//        queryLogger.error(s"hbaseGet: $get, $e", e)
//        new Result()
//    }
//  }

  //  def withTimeout[T](op: => Future[T], fallback: => T)(implicit timeout: Duration): Future[T] = {
  //    //    val timeoutFuture = play.api.libs.concurrent.Promise.timeout(fallback, timeout)
  //    //    Future.firstCompletedOf(Seq(op, timeoutFuture))
  //    TimeoutFuture(op, fallback)(executionContext, timeout)
  //    //    val timeoutFuture = akka.pattern.after(timeout.toMillis millis, using = Akka.system.scheduler) { Future { fallback } }
  //    //    Future.firstCompletedOf(Seq(op, timeoutFuture))
  //  }
  def withTimeout[T](zkQuorum: String, op: => Future[T], fallback: => T)(implicit timeout: Duration): Future[T] = {
    try {
      val client = getClient(zkQuorum)
      //      TimeoutFuture(op, fallback)(this.executionContext, timeout)
      op
    } catch {
      case e: Throwable =>
        queryLogger.error(s"withTimeout: $e", e)
        Future { fallback }(this.executionContext)
    }
  }

//  def save(zkQuorum: String, tableName: String, mutations: Seq[Mutation]): Unit = {
//    //    Logger.debug(s"save to $zkQuorum, $tableName, ${mutations.size} mutations.")
//    if (mutations.isEmpty) {}
//    else {
//      withWriteTable[Unit](zkQuorum, tableName) { table =>
//        val ret: Array[Object] = Array.fill(mutations.size)(null)
//        table.batch(mutations, ret)
//        // TODO: retry
//        for (r <- ret if r == null) {
//          Logger.error(s"save failed after batch.")
//        }
//      } {
//        Logger.error(s"save mutation failed.")
//      }
//    }
//  }
  def defferedToFuture[A](d: Deferred[A])(fallback: A): Future[A] = {
    val promise = Promise[A]

    d.addBoth(new Callback[Unit, A] {
      def call(arg: A) = arg match {
        case e: Throwable =>
          queryLogger.error(s"deferred return throwable: $e", e)
          promise.success(fallback)
        case _ => promise.success(arg)
      }
    })

    promise.future
  }
  def deferredToFutureWithoutFallback[T](d: Deferred[T]) = {
    val promise = Promise[T]
    d.addBoth(new Callback[Unit, T]{
      def call(arg: T) = arg match {
        case e: Throwable =>
          queryLogger.error(s"deferred return throwable: $e", e)
          promise.failure(e)
        case _ => promise.success(arg)
      }
    })
    promise.future
  }
  def deferredCallbackWithFallback[T, R](d: Deferred[T])(f: T => R, fallback: => R) = {
    d.addCallback(new Callback[R, T] {
      def call(args: T): R = {
        f(args)
      }
    }).addErrback(new Callback[R, Exception] {
      def call(e: Exception): R = {
        queryLogger.error(s"Exception on deferred: $e", e)
        fallback
      }
    })
  }
  def writeAsync(zkQuorum: String, rpcs: Seq[HBaseRpc]) = {
    if (rpcs.isEmpty) {}
    else {
      try {
        val client = getClient(zkQuorum)
        val futures = rpcs.map { rpc =>
          val deferred = rpc match {
            case d: DeleteRequest => client.delete(d)
            case p: PutRequest => client.put(p)
//            case i: AtomicIncrementRequest => client.atomicIncrement(i)
          }
          deferredToFutureWithoutFallback(deferred)
        }
//        Future.sequence(futures)
      } catch {
        case e: Throwable =>
          queryLogger.error(s"writeAsync failed. $e", e)
      }
    }
  }
  /**
   * Edge
   */
//  def mutateEdge(edge: Edge): Unit = {
//    save(edge.label.hbaseZkAddr, edge.label.hbaseTableName, edge.buildPutsAll())
//  }
  def mutateEdge(edge: Edge) = {
    writeAsync(edge.label.hbaseZkAddr, edge.buildPutsAll())
  }
//  def mutateEdges(edges: Seq[Edge], mutateInPlace: Boolean = false): Unit = {
//
//    val edgesPerTable = edges.groupBy { edge => (edge.label.hbaseZkAddr, edge.label.hbaseTableName) }
//    for (((zkQuorum, tableName), edges) <- edgesPerTable) {
//      /**
//       * delete/update/increment can`t be batched.
//       */
//      val (batches, others) = edges.partition(e => e.canBeBatched)
//      save(zkQuorum, tableName, batches.flatMap(_.buildPutsAll))
//      others.foreach(other => save(zkQuorum, tableName, other.buildPutsAll))
//    }
//    val verticesPerTable = edges.flatMap { edge => List(edge.srcForVertex, edge.tgtForVertex) }.groupBy { v => (v.hbaseZkAddr, v.hbaseTableName) }
//    for (((zkQuorum, tableName), vertices) <- verticesPerTable) {
//      save(zkQuorum, tableName, vertices.flatMap(v => v.buildPuts))
//    }
//  }
  def mutateEdges(edges: Seq[Edge]) = {
    val edgesPerTable = edges.groupBy { edge => edge.label.hbaseZkAddr }
    for ((zkQuorum, edges) <- edgesPerTable) {
      val (batches, others) = edges.partition(e => e.canBeBatched)
      writeAsync(zkQuorum, batches.flatMap(_.buildPutsAll))
      others.foreach(other => writeAsync(zkQuorum, other.buildPutsAll))
    }
    val verticesPerTable = edges.flatMap { edge => List(edge.srcForVertex, edge.tgtForVertex) }.groupBy { v => v.hbaseZkAddr }
    for ((zkQuorum, vertices) <- verticesPerTable) {
      writeAsync(zkQuorum, vertices.flatMap(v => v.buildPutsAsync))
    }
  }
  //select
  /**
   *
   */
  def getEdgesAsync(q: Query): Future[Seq[Iterable[(Edge, Double)]]] = {
    implicit val ex = this.executionContext
    // not sure this is right. make sure refactor this after.
    try {
      if (q.steps.isEmpty) {
        // TODO: this should be get vertex query.
        Future { q.vertices.map(v => List.empty[(Edge, Double)]) }
      } else {
        val stepLen = q.steps.length
        var step = q.steps.head

        var seedEdgesFuture = getEdgesAsyncWithRankForFistStep(q.vertices, q, 0)

        for (i <- (1 until stepLen)) {
          seedEdgesFuture = getEdgesAsyncWithRank(seedEdgesFuture, q, i)
        }

        seedEdgesFuture
      }
    } catch {
      case e: Throwable =>
        queryLogger.error(s"getEdgesAsync: $e", e)
        Future { q.vertices.map(v => List.empty[(Edge, Double)]) }
    }
  }

  //only for testcase.
  def getEdgesSync(q: Query): Seq[Iterable[(Edge, Double)]] = {
    Await.result(getEdgesAsync(q), 10 seconds)
  }
  //only for testcase.
//  def getEdgeSync(srcVertex: Vertex, tgtVertex: Vertex, label: Label, dir: Int) = {
//    val rowKey = EdgeRowKey(srcVertex.id, LabelWithDirection(label.id.get, dir), LabelIndex.defaultSeq, isInverted = true)
//    val qualifier = EdgeQualifierInverted(tgtVertex.id)
//    val get = new Get(rowKey.bytes)
//    get.addColumn(edgeCf, qualifier.bytes)
//    withReadTable(label.hbaseZkAddr, label.hbaseTableName) { table =>
//      //      table.get(get)
//      Edge.toEdges(htableGet(table)(get))
//    } {
//      List.empty[Edge]
//    }
//  }

  def getEdge(srcVertex: Vertex, tgtVertex: Vertex, label: Label, dir: Int): Future[Iterable[Edge]] = {
    implicit val ex = this.executionContext
    val rowKey = EdgeRowKey(srcVertex.id, LabelWithDirection(label.id.get, dir), label.defaultIndex.get.seq, isInverted = true)

    val qualifier = EdgeQualifierInverted(tgtVertex.id)
    val client = getClient(label.hbaseZkAddr)
    val getRequest = new GetRequest(label.hbaseTableName.getBytes(), rowKey.bytes, edgeCf, qualifier.bytes)
    defferedToFuture(client.get(getRequest))(emptyKVs).map { kvs =>
      for {
        kv <- kvs
        edge <- Edge.toEdge(kv, QueryParam(LabelWithDirection(label.id.get, dir.toByte)))
      } yield edge
    }
  }

  def buildGetRequests(srcVertices: Seq[Vertex], params: List[QueryParam]): Seq[Iterable[(GetRequest, QueryParam)]] = {
    srcVertices.map { vertex =>
      params.map { param =>
        (param.buildGetRequest(vertex), param)
      }
    }
  }
  //  def buildGets(srcVertices: Seq[Vertex], params: List[QueryParam]): Seq[Iterable[(Get, QueryParam)]] = {
  //    srcVertices.map { vertex =>
  //      params.map { param =>
  //        (param.buildGet(vertex), param)
  //      }
  //    }
  //  }

  def singleGet(table: Array[Byte], rowKey: Array[Byte], cf: Array[Byte], offset: Int, limit: Int,
    minTs: Long, maxTs: Long,
    maxAttempt: Int, rpcTimeoutInMillis: Int,
    columnRangeFilter: ColumnRangeFilter) = {
    val get = new GetRequest(table, rowKey, cf)
    get.maxVersions(1)
    get.setFailfast(true)
    get.setMaxResultsPerColumnFamily(limit)
    get.setRowOffsetPerColumnFamily(offset)
    get.setMinTimestamp(minTs)
    get.setMaxTimestamp(maxTs)
    get.setMaxAttempt(maxAttempt.toByte)
    get.setRpcTimeout(rpcTimeoutInMillis)
    if (columnRangeFilter != null) get.filter(columnRangeFilter)
    get
  }
  def convertEdge(edge: Edge, labelOutputFields: Map[Int, Byte]): Option[Edge] = {
    labelOutputFields.get(edge.labelWithDir.labelId) match {
      case None => Some(edge)
      case Some(outputField) =>
        edge.propsWithTs.get(outputField) match {
          case None => None
          case Some(outputFieldVal) =>
            Some(edge.updateTgtVertex(outputFieldVal.innerVal))
        }
    }
  }

  def filterEdges(edgesFuture: Future[ArrayList[ArrayList[(Edge, Double, QueryParam)]]],
    q: Query,
    stepIdx: Int,
    alreadyVisited: Map[(LabelWithDirection, Vertex), Boolean] = Map.empty[(LabelWithDirection, Vertex), Boolean]): Future[Seq[Iterable[(Edge, Double)]]] = {
    implicit val ex = Graph.executionContext
    edgesFuture.map { edgesByVertices =>
      val step = q.steps(stepIdx)
      val labelOutputFields = step.queryParams.map { qParam => qParam.outputField.map(outputField => qParam.labelWithDir.labelId -> outputField) }.flatten.toMap
      val excludeLabelWithDirs = (for (queryParam <- step.queryParams if queryParam.exclude) yield queryParam.labelWithDir).toSet
      val includeLabelWithDirOpt = (for (queryParam <- step.queryParams if queryParam.include) yield queryParam.labelWithDir).toSet.headOption
      val seen = new HashMap[(Vertex, LabelWithDirection, Vertex), Double]
      //      val seen = new HashSet[(LabelWithDirection, Vertex)]
      //      val excludeFromTos = new HashSet[(CompositeId, CompositeId)]
      //      val includeFromTos = new HashSet[(CompositeId, CompositeId)]
      val excludeFromTos = new HashSet[(Vertex, Vertex)]
      val includeFromTos = new HashSet[(Vertex, Vertex)]

      val hasIncludeLabel = includeLabelWithDirOpt.isDefined
      //      if (hasIncludeLabel) {
      for {
        edgesWithScore <- edgesByVertices
        (edge, score, queryParam) <- edgesWithScore
      } {
        //        Logger.debug(s"${edge.toStringRaw}")
        if (hasIncludeLabel && edge.labelWithDir == includeLabelWithDirOpt.get) {
          //          includeFromTos += (edge.srcVertex.id -> edge.tgtVertex.id)
          includeFromTos += (edge.srcVertex -> edge.tgtVertex)
        }
        if (excludeLabelWithDirs.contains(edge.labelWithDir)) {
          //          excludeFromTos += (edge.srcVertex.id -> edge.tgtVertex.id)
          excludeFromTos += (edge.srcVertex -> edge.tgtVertex)
        }
      }
      //      }

      //      for {
      //        edgesWithScore <- edgesByVertices
      //        (edge, score) <- edgesWithScore if excludeLabelWithDirs.contains(edge.labelWithDir)
      //      } {
      //        excludeFromTos += (edge.srcVertex -> edge.tgtVertex)
      //      }

      //      Logger.debug(s"${excludeFromTos.mkString("\n")}")
      //      Logger.debug(s"$includeFromTos")

      val convertedEdges = for {
        edgesWithScore <- edgesByVertices
      } yield {
        for {
          //          (edge, score) <- edgesWithScore if !excludeFromTos.contains((edge.srcVertex.id -> edge.tgtVertex.id))
          //          if (!hasIncludeLabel || includeFromTos.contains((edge.srcVertex.id -> edge.tgtVertex.id)))
          (edge, score, queryParam) <- edgesWithScore
          fromTo = (edge.srcVertex -> edge.tgtVertex)
          if !excludeFromTos.contains(fromTo)
          if (!hasIncludeLabel || includeFromTos.contains(fromTo))
          convertedEdge <- convertEdge(edge, labelOutputFields)
          key = (convertedEdge.labelWithDir, convertedEdge.tgtVertex)
          //          if !seen.contains(key)
          if filterDuplicates(seen, edge, score, queryParam)
          if !(q.removeCycle && alreadyVisited.contains(key))
        } yield {
          //          seen += key

          (convertedEdge, score)
        }
      }
      for {
        edgesWithScore <- convertedEdges
      } yield {
        for {
          (edge, score) <- edgesWithScore
          key = (edge.srcVertex, edge.labelWithDir, edge.tgtVertex)
          aggregatedScore = seen.getOrElse(key, score)
        } yield {
          (edge, aggregatedScore)
        }
      }
    }
  }
  private def filterDuplicates(seen: HashMap[(Vertex, LabelWithDirection, Vertex), Double], edge: Edge, score: Double, queryParam: QueryParam) = {
    val key = (edge.srcVertex, edge.labelWithDir, edge.tgtVertex)
    val newScore = queryParam.duplicatePolicy match {
      case Query.DuplicatePolicy.CountSum => 1.0
      case _ => score
    }
    seen.get(key) match {
      case None =>
        seen += (key -> newScore)
        true
      case Some(oldScore) =>

        queryParam.duplicatePolicy match {
          case Query.DuplicatePolicy.First =>
            // use first occurrence`s score 
            false
          case Query.DuplicatePolicy.Raw =>
            // TODO: assumes all duplicate vertices will have same score 
            seen += (key -> newScore)
            true
          case _ =>
            // aggregate score
            seen += (key -> (oldScore + newScore))
            false
        }
    }
  }


  def getEdgesAsyncWithRankForFistStep(srcVertices: Seq[Vertex], q: Query, stepIdx: Int): Future[Seq[Iterable[(Edge, Double)]]] = {
    val step = q.steps(stepIdx)
    val uniqSrcVertices = srcVertices.groupBy(v => v.id).map { kv => (kv._2.head, 1.0) }
    getEdgesAsyncWithRank(uniqSrcVertices.toSeq, q, stepIdx)
  }


  private def getEdgesAsyncWithRank(prevEdges: Seq[(Edge, Double)], q: Query, stepIdx: Int): Future[Seq[Iterable[(Edge, Double)]]] = {
    val step = q.steps(stepIdx)
    val alreadyVisited = prevEdges.map {
      case (edge, score) =>
        (edge.labelWithDir, if (edge.labelWithDir.dir == GraphUtil.directions("out")) edge.tgtVertex else edge.srcVertex) -> true
    }.toMap

    val srcVertices = prevEdges.map { case (edge, score) => (edge.tgtVertex -> score) }
    //    val getsAll = buildGets(srcVertices.map(_._1), step.queryParams).zip(srcVertices.map(_._2))
    val getsAll = buildGetRequests(srcVertices.map(_._1), step.queryParams).zip(srcVertices.map(_._2))

    implicit val ex = executionContext
    val deffered = getsAll.flatMap { //by verticies
      case (getsWithQueryParams, prevScore) =>
        getsWithQueryParams.map { //by labels
          case (get, queryParam) =>
            try {
              val client = getClient(queryParam.label.hbaseZkAddr)
              deferredCallbackWithFallback(client.get(get))({ kvs =>
                val edges = for {
                  kv <- kvs
                  edge <- Edge.toEdge(kv, queryParam)
                } yield {
                  (edge, edge.rank(queryParam.rank) * prevScore, queryParam)
                }
                new ArrayList(edges)
              }, emptyEdges)
            } catch {
              case e @ (_: Throwable | _: Exception) =>
                queryLogger.error(s"Exception: $e", e)
                Deferred.fromResult(emptyEdges)
            }
        }
    }
    val grouped = Deferred.group(deffered)
    filterEdges(defferedToFuture(grouped)(emptyEdgeList), q, stepIdx, alreadyVisited)
  }
  private def getEdgesAsyncWithRank(srcVertices: Seq[(Vertex, Double)], q: Query, stepIdx: Int): Future[Seq[Iterable[(Edge, Double)]]] = {
    val step = q.steps(stepIdx)
    //    val getsAll = buildGets(srcVertices.map(_._1), step.queryParams).zip(srcVertices.map(_._2))
    val getsAll = buildGetRequests(srcVertices.map(_._1), step.queryParams).zip(srcVertices.map(_._2))
    implicit val ex = executionContext
    val deffered = getsAll.flatMap { //by verticies
      case (getsWithQueryParams, prevScore) =>
        getsWithQueryParams.map { //by labels
          case (get, queryParam) =>
            try {
              val client = getClient(queryParam.label.hbaseZkAddr)
              deferredCallbackWithFallback(client.get(get))({ kvs =>
                val edges = for {
                  kv <- kvs
                  edge <- Edge.toEdge(kv, queryParam)
                } yield {
                  (edge, edge.rank(queryParam.rank) * prevScore, queryParam)
                }
                new ArrayList(edges)
              }, emptyEdges)
            } catch {
              case e @ (_: Throwable | _: Exception) =>
                queryLogger.error(s"Exception: $e", e)
                Deferred.fromResult(emptyEdges)

            }
        }
    }
    val grouped = Deferred.group(deffered)

    filterEdges(defferedToFuture(grouped)(emptyEdgeList), q, stepIdx)
  }

  def getEdgesAsyncWithRank(srcEdgesFuture: Future[Seq[Iterable[(Edge, Double)]]], q: Query, stepIdx: Int): Future[Seq[Iterable[(Edge, Double)]]] = {
    implicit val ex = executionContext
    for {
      srcEdges <- srcEdgesFuture
      edgesWithScore = srcEdges.flatten
      ret <- getEdgesAsyncWithRank(edgesWithScore, q, stepIdx)
      //      verticeWithRanks = edgesWithScore.map(t => (t._1.tgtVertex, t._2)).toSeq
      //      ret <- getEdgesAsyncWithRank(verticeWithRanks, step)
    } yield {
      ret
    }
  }

  def getVerticesAsync(vertices: Seq[Vertex]): Future[Seq[Vertex]] = {
    // TODO: vertex needs meta for hbase table.
    //    play.api.Logger.error(s"$vertices")
    implicit val ex = executionContext
    val futures = vertices.map { vertex =>
      withTimeout[Option[Vertex]](vertex.hbaseZkAddr, {
        //          val get = vertex.buildGet
        val client = getClient(vertex.hbaseZkAddr)
        //        val get = vertex.buildGetRequest()
        val get = vertex.buildGet
        defferedToFuture(client.get(get))(emptyKVs).map { kvs =>
          Vertex(kvs)
        }
        //          Logger.error(s"$get")
      }, { None })(singleGetTimeout)

    }
    Future.sequence(futures).map { result => result.toList.flatten }
  }
  /**
   * Vertex
   */

//  def mutateVertex(vertex: Vertex) = {
//    mutateVertices(List(vertex))
//  }
  def mutateVertex(vertex: Vertex) = {
    mutateVertices(List(vertex))
  }
//  def mutateVertices(vertices: Seq[Vertex]) = {
//    for (((zkQuorum, tableName, op), vertices) <- vertices.groupBy(v => (v.hbaseZkAddr, v.hbaseTableName, v.op))) {
//      if (op == GraphUtil.operations("delete") || op == GraphUtil.operations("deleteAll")) deleteVertexAll(vertices)
//      else save(zkQuorum, tableName, vertices.flatMap(v => v.buildPutsAll()))
//    }
//  }
  def mutateVertices(vertices: Seq[Vertex]) = {
    for (((zkQuorum, op), vertices) <- vertices.groupBy(v => (v.hbaseZkAddr, v.op))) {
      if (op == GraphUtil.operations("delete") || op == GraphUtil.operations("deleteAll")) deleteVertexAll(vertices)
      else writeAsync(zkQuorum, vertices.flatMap(v => v.buildPutsAll()))
    }
  }
  // delete only vertices
  def deleteVertices(vertices: Seq[Vertex]) = {
    for ((zkQuorum, vs) <- vertices.groupBy(v => v.hbaseZkAddr)) {
      writeAsync(zkQuorum, vs.flatMap(v => v.buildDeleteAsync()))
    }
  }
  /**
   * O(E), maynot feasible
   */
  def deleteVertexAll(vertices: Seq[Vertex]): Unit = {
    for {
      vertex <- vertices
      label <- (Label.findBySrcColumnId(vertex.id.colId) ++ Label.findByTgtColumnId(vertex.id.colId)).groupBy(_.id.get).map { _._2.head }
    } {
      deleteVertexAllAsync(vertex.toEdgeVertex, label)
    }
    deleteVertices(vertices)
  }

  private def deleteVertexAllAsync(srcVertex: Vertex, label: Label): Future[Boolean] = {
    implicit val ex = Graph.executionContext
    val qParams = for (dir <- List(0, 1)) yield {
      val labelWithDir = LabelWithDirection(label.id.get, dir)
      QueryParam(labelWithDir).limit(0, maxValidEdgeListSize * 5)
    }
    val step = Step(qParams)
    val q = Query(List(srcVertex), List(step), true)
    val seen = new HashMap[(CompositeId, LabelWithDirection), Boolean]
    for {
      edgesByVertex <- getEdgesAsync(q)
    } yield {
      val fetchedEdges = for {
        edges <- edgesByVertex
        (edge, score) <- edges if edge.ts <= srcVertex.ts && !seen.containsKey((edge.tgtVertex.id, edge.labelWithDir))
      } yield {
        val labelWithDir = if (label.isDirected) edge.labelWithDir.updateDir(2) else edge.labelWithDir
        val edgeToDelete = Edge(edge.srcVertex, edge.tgtVertex, labelWithDir, GraphUtil.operations("delete"), srcVertex.ts, edge.version + Edge.incrementVersion, edge.propsWithTs)
        seen += ((edgeToDelete.tgtVertex.id, edgeToDelete.labelWithDir) -> true)

        // reverse or not? => reverse.
        // delete edge or real delete operation? => ?
        //          play.api.Logger.debug(s"EdgeToDelete: $edgeToDelete")
        //        (edge.label.hbaseTableName, edgeToDelete.relatedEdges.flatMap(e => e.buildPutsAll()))
        edge
      }
      for ((zkQuorum, edges) <- fetchedEdges.groupBy(e => e.label.hbaseZkAddr)) {
        writeAsync(zkQuorum, edges.flatMap(_.buildPutsAll))
      }
      true
    }
  }
  // select
  def getVertex(vertex: Vertex): Future[Option[Vertex]] = {
    implicit val ex = executionContext
    val client = getClient(vertex.hbaseZkAddr)
    defferedToFuture(client.get(vertex.buildGet))(emptyKVs).map { kvs =>
      Vertex(kvs)
    }
  }
  /**
   * Bulk
   */

  /**
   * when how from to what direction (meta info key:value)
   * ex) timestamp insert shon sol talk_friend directed/undirected propperties
   * ex) timestamp insert shon talk_user_id propperties
   *
   */

  def toGraphElement(s: String): Option[GraphElement] = {
    val parts = GraphUtil.split(s)
    try {
      val logType = parts(2)
      val element = if (logType == "edge" | logType == "e") {
        toEdge(parts)
      } else if (logType == "vertex" | logType == "v") {
        toVertex(parts)
      } else {
        throw new KGraphExceptions.JsonParseException("log type is not exist in log.")
      }
      element
    } catch {
      case e: Throwable =>
        logger.error(s"toGraphElement: $s => $e", e)
        None
    }
  }

  def bulkMutates(elements: Iterable[GraphElement], mutateInPlace: Boolean = false) = {
    val vertices = new ListBuffer[Vertex]
    val edges = new ListBuffer[Edge]
    for (e <- elements) {
      e match {
        case edge: Edge => edges += edge
        case vertex: Vertex => vertices += vertex
        case _ => throw new Exception("GraphElement should be either vertex or edge.")
      }
    }
    mutateVertices(vertices)
    mutateEdges(edges)
  }
  def toVertex(s: String): Option[Vertex] = {
    toVertex(GraphUtil.split(s))
  }
  def toEdge(s: String): Option[Edge] = {
    toEdge(GraphUtil.split(s))
  }
  //"1418342849000\tu\te\t3286249\t71770\ttalk_friend\t{\"is_hidden\":false}"
  //{"from":1,"to":101,"label":"graph_test","props":{"time":-1, "weight":10},"timestamp":1417616431},
  def toEdge(parts: Array[String]): Option[Edge] = {
    try {
      val (ts, operation, logType, srcId, tgtId, label) = (parts(0), parts(1), parts(2), parts(3), parts(4), parts(5))
      val props = if (parts.length >= 7) parts(6) else "{}"
      // use db field is_directed.
      val direction = ""
      val edge = Management.toEdge(ts.toLong, operation, srcId, tgtId, label, direction, props)
      //      Logger.debug(s"toEdge: $edge")
      Some(edge)
    } catch {
      case e: Throwable =>
        logger.error(s"toEdge: $e", e)
        throw e
    }
  }
  //"1418342850000\ti\tv\t168756793\ttalk_user_id\t{\"country_iso\":\"KR\"}"
  def toVertex(parts: Array[String]): Option[Vertex] = {
    try {
      val (ts, operation, logType, srcId, serviceName, colName) = (parts(0), parts(1), parts(2), parts(3), parts(4), parts(5))
      val props = if (parts.length >= 7) parts(6) else "{}"
      Some(Management.toVertex(ts.toLong, operation, srcId, serviceName, colName, props))
    } catch {
      case e: Throwable =>
        logger.error(s"toVertex: $e", e)
        throw e
    }
  }

}