/*
 * 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.s2graph.core.storage


import org.apache.s2graph.core.GraphExceptions.{NoStackException, FetchTimeoutException}
import org.apache.s2graph.core._
import org.apache.s2graph.core.mysqls.{Label, LabelMeta}
import org.apache.s2graph.core.parsers.WhereParser
import org.apache.s2graph.core.storage.serde.indexedge.wide.{IndexEdgeDeserializable, IndexEdgeSerializable}
import org.apache.s2graph.core.storage.serde.snapshotedge.wide.SnapshotEdgeDeserializable
import org.apache.s2graph.core.storage.serde.vertex.{VertexDeserializable, VertexSerializable}
import org.apache.s2graph.core.types._
import org.apache.s2graph.core.utils.{DeferCache, Extensions, logger}

import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.{Random, Try}
import java.util.concurrent.{Executors, TimeUnit}

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.hadoop.hbase.util.Bytes


abstract class Storage[Q, R](val graph: S2Graph,
                          val config: Config)(implicit ec: ExecutionContext) {
  import HBaseType._
  import S2Graph._

  val BackoffTimeout = graph.BackoffTimeout
  val MaxRetryNum = graph.MaxRetryNum
  val MaxBackOff = graph.MaxBackOff
  val FailProb = graph.FailProb
  val LockExpireDuration =  graph.LockExpireDuration
  val MaxSize = graph.MaxSize
  val ExpireAfterWrite = graph.ExpireAfterWrite
  val ExpireAfterAccess = graph.ExpireAfterAccess

  /** retry scheduler */
  val scheduledThreadPool = Executors.newSingleThreadScheduledExecutor()


  /**
   * Compatibility table
   * | label schema version | snapshot edge | index edge | vertex | note |
   * | v1 | serde.snapshotedge.wide | serde.indexedge.wide | serde.vertex | do not use this. this exist only for backward compatibility issue |
   * | v2 | serde.snapshotedge.wide | serde.indexedge.wide | serde.vertex | do not use this. this exist only for backward compatibility issue |
   * | v3 | serde.snapshotedge.tall | serde.indexedge.wide | serde.vertex | recommended with HBase. current stable schema |
   * | v4 | serde.snapshotedge.tall | serde.indexedge.tall | serde.vertex | experimental schema. use scanner instead of get |
   *
   */

  /**
   * create serializer that knows how to convert given snapshotEdge into kvs: Seq[SKeyValue]
   * so we can store this kvs.
   * @param snapshotEdge: snapshotEdge to serialize
   * @return serializer implementation for StorageSerializable which has toKeyValues return Seq[SKeyValue]
   */
  def snapshotEdgeSerializer(snapshotEdge: SnapshotEdge): Serializable[SnapshotEdge] = {
    snapshotEdge.schemaVer match {
//      case VERSION1 |
      case VERSION2 => new serde.snapshotedge.wide.SnapshotEdgeSerializable(snapshotEdge)
      case VERSION3 | VERSION4 => new serde.snapshotedge.tall.SnapshotEdgeSerializable(snapshotEdge)
      case _ => throw new RuntimeException(s"not supported version: ${snapshotEdge.schemaVer}")
    }
  }

  /**
   * create serializer that knows how to convert given indexEdge into kvs: Seq[SKeyValue]
   * @param indexEdge: indexEdge to serialize
   * @return serializer implementation
   */
  def indexEdgeSerializer(indexEdge: IndexEdge): Serializable[IndexEdge] = {
    indexEdge.schemaVer match {
//      case VERSION1
      case VERSION2 | VERSION3 => new IndexEdgeSerializable(indexEdge)
      case VERSION4 => new serde.indexedge.tall.IndexEdgeSerializable(indexEdge)
      case _ => throw new RuntimeException(s"not supported version: ${indexEdge.schemaVer}")

    }
  }

  /**
   * create serializer that knows how to convert given vertex into kvs: Seq[SKeyValue]
   * @param vertex: vertex to serialize
   * @return serializer implementation
   */
  def vertexSerializer(vertex: S2Vertex): Serializable[S2Vertex] = new VertexSerializable(vertex)

  /**
   * create deserializer that can parse stored CanSKeyValue into snapshotEdge.
   * note that each storage implementation should implement implicit type class
   * to convert storage dependent dataType into common SKeyValue type by implementing CanSKeyValue
   *
   * ex) Asynchbase use it's KeyValue class and CanSKeyValue object has implicit type conversion method.
   * if any storaage use different class to represent stored byte array,
   * then that storage implementation is responsible to provide implicit type conversion method on CanSKeyValue.
   * */

  val snapshotEdgeDeserializer: Deserializable[SnapshotEdge] = new serde.snapshotedge.tall.SnapshotEdgeDeserializable(graph)

  def snapshotEdgeDeserializer(schemaVer: String): Deserializable[SnapshotEdge] = snapshotEdgeDeserializer

  /** create deserializer that can parse stored CanSKeyValue into indexEdge. */
  val indexEdgeDeserializer: Deserializable[S2Edge] = new serde.indexedge.tall.IndexEdgeDeserializable(graph)

  def indexEdgeDeserializer(schemaVer: String) = new serde.indexedge.tall.IndexEdgeDeserializable(graph)

  /** create deserializer that can parser stored CanSKeyValue into vertex. */
  val vertexDeserializer: Deserializable[S2Vertex] = new VertexDeserializable(graph)


  /**
   * decide how to store given key values Seq[SKeyValue] into storage using storage's client.
   * note that this should be return true on all success.
   * we assumes that each storage implementation has client as member variable.
   *
   *
   * @param cluster: where this key values should be stored.
   * @param kvs: sequence of SKeyValue that need to be stored in storage.
   * @param withWait: flag to control wait ack from storage.
   *                  note that in AsynchbaseStorage(which support asynchronous operations), even with true,
   *                  it never block thread, but rather submit work and notified by event loop when storage send ack back.
   * @return ack message from storage.
   */
  def writeToStorage(cluster: String, kvs: Seq[SKeyValue], withWait: Boolean): Future[Boolean]

//  def writeToStorage(kv: SKeyValue, withWait: Boolean): Future[Boolean]

  /**
   * fetch SnapshotEdge for given request from storage.
   * also storage datatype should be converted into SKeyValue.
   * note that return type is Sequence rather than single SKeyValue for simplicity,
   * even though there is assertions sequence.length == 1.
   * @param request
   * @return
   */
  def fetchSnapshotEdgeKeyValues(request: QueryRequest): Future[Seq[SKeyValue]]

  /**
   * write requestKeyValue into storage if the current value in storage that is stored matches.
   * note that we only use SnapshotEdge as place for lock, so this method only change SnapshotEdge.
   *
   * Most important thing is this have to be 'atomic' operation.
   * When this operation is mutating requestKeyValue's snapshotEdge, then other thread need to be
   * either blocked or failed on write-write conflict case.
   *
   * Also while this method is still running, then fetchSnapshotEdgeKeyValues should be synchronized to
   * prevent wrong data for read.
   *
   * Best is use storage's concurrency control(either pessimistic or optimistic) such as transaction,
   * compareAndSet to synchronize.
   *
   * for example, AsynchbaseStorage use HBase's CheckAndSet atomic operation to guarantee 'atomicity'.
   * for storage that does not support concurrency control, then storage implementation
   * itself can maintain manual locks that synchronize read(fetchSnapshotEdgeKeyValues)
   * and write(writeLock).
   * @param requestKeyValue
   * @param expectedOpt
   * @return
   */
  def writeLock(requestKeyValue: SKeyValue, expectedOpt: Option[SKeyValue]): Future[Boolean]

  /**
   * build proper request which is specific into storage to call fetchIndexEdgeKeyValues or fetchSnapshotEdgeKeyValues.
   * for example, Asynchbase use GetRequest, Scanner so this method is responsible to build
   * client request(GetRequest, Scanner) based on user provided query.
    *
    * @param queryRequest
   * @return
    */
  protected def buildRequest(queryRequest: QueryRequest, edge: S2Edge): Q

  /**
   * fetch IndexEdges for given queryParam in queryRequest.
   * this expect previous step starting score to propagate score into next step.
   * also parentEdges is necessary to return full bfs tree when query require it.
   *
   * note that return type is general type.
   * for example, currently we wanted to use Asynchbase
   * so single I/O return type should be Deferred[T].
   *
   * if we use native hbase client, then this return type can be Future[T] or just T.
    *
    * @param queryRequest
   * @param isInnerCall
   * @param parentEdges
   * @return
   */
  def fetch(queryRequest: QueryRequest,
            isInnerCall: Boolean,
            parentEdges: Seq[EdgeWithScore]): R

  /**
   * responsible to fire parallel fetch call into storage and create future that will return merged result.
   *
   * @param queryRequests
   * @param prevStepEdges
   * @return
   */
  def fetches(queryRequests: Seq[QueryRequest],
              prevStepEdges: Map[VertexId, Seq[EdgeWithScore]]): Future[Seq[StepResult]]

  /**
   * fetch Vertex for given request from storage.
    *
    * @param request
   * @return
   */
  def fetchVertexKeyValues(request: QueryRequest): Future[Seq[SKeyValue]]

  /**
   * decide how to apply given edges(indexProps values + Map(_count -> countVal)) into storage.
    *
    * @param edges
   * @param withWait
   * @return
   */
  def incrementCounts(edges: Seq[S2Edge], withWait: Boolean): Future[Seq[(Boolean, Long, Long)]]

  /**
   * this method need to be called when client shutdown. this is responsible to cleanUp the resources
   * such as client into storage.
   */
  def flush(): Unit = {
  }

  def fetchEdgesAll(): Future[Seq[S2Edge]]

  def fetchVerticesAll(): Future[Seq[S2Vertex]]

  /**
   * create table on storage.
   * if storage implementation does not support namespace or table, then there is nothing to be done
    *
    * @param zkAddr
   * @param tableName
   * @param cfs
   * @param regionMultiplier
   * @param ttl
   * @param compressionAlgorithm
   */
  def createTable(zkAddr: String,
                  tableName: String,
                  cfs: List[String],
                  regionMultiplier: Int,
                  ttl: Option[Int],
                  compressionAlgorithm: String,
                  replicationScopeOpt: Option[Int] = None,
                  totalRegionCount: Option[Int] = None): Unit

  def truncateTable(zkAddr: String, tableNameStr: String): Unit = {}

  def deleteTable(zkAddr: String, tableNameStr: String): Unit = {}

  def shutdown(): Unit

  /** Public Interface */
  def getVertices(vertices: Seq[S2Vertex]): Future[Seq[S2Vertex]] = {
    def fromResult(kvs: Seq[SKeyValue],
                   version: String): Option[S2Vertex] = {
      if (kvs.isEmpty) None
      else vertexDeserializer.fromKeyValues(kvs, None)
//        .map(S2Vertex(graph, _))
    }

    val futures = vertices.map { vertex =>
      val queryParam = QueryParam.Empty
      val q = Query.toQuery(Seq(vertex), Seq(queryParam))
      val queryRequest = QueryRequest(q, stepIdx = -1, vertex, queryParam)

      fetchVertexKeyValues(queryRequest).map { kvs =>
        fromResult(kvs, vertex.serviceColumn.schemaVersion)
      } recoverWith { case ex: Throwable =>
        Future.successful(None)
      }
    }

    Future.sequence(futures).map { result => result.toList.flatten }
  }
  def mutateStrongEdges(_edges: Seq[S2Edge], withWait: Boolean): Future[Seq[Boolean]] = {

    val edgeWithIdxs = _edges.zipWithIndex
    val grouped = edgeWithIdxs.groupBy { case (edge, idx) =>
      (edge.innerLabel, edge.srcVertex.innerId, edge.tgtVertex.innerId)
    } toSeq

    val mutateEdges = grouped.map { case ((_, _, _), edgeGroup) =>
      val edges = edgeGroup.map(_._1)
      val idxs = edgeGroup.map(_._2)
      // After deleteAll, process others
      val mutateEdgeFutures = edges.toList match {
        case head :: tail =>
          val edgeFuture = mutateEdgesInner(edges, checkConsistency = true , withWait)

          //TODO: decide what we will do on failure on vertex put
          val puts = buildVertexPutsAsync(head)
          val vertexFuture = writeToStorage(head.innerLabel.hbaseZkAddr, puts, withWait)
          Seq(edgeFuture, vertexFuture)
        case Nil => Nil
      }

      val composed = for {
//        deleteRet <- Future.sequence(deleteAllFutures)
        mutateRet <- Future.sequence(mutateEdgeFutures)
      } yield mutateRet

      composed.map(_.forall(identity)).map { ret => idxs.map( idx => idx -> ret) }
    }

    Future.sequence(mutateEdges).map { squashedRets =>
      squashedRets.flatten.sortBy { case (idx, ret) => idx }.map(_._2)
    }
  }

  def mutateVertex(vertex: S2Vertex, withWait: Boolean): Future[Boolean] = {
    if (vertex.op == GraphUtil.operations("delete")) {
      writeToStorage(vertex.hbaseZkAddr,
        vertexSerializer(vertex).toKeyValues.map(_.copy(operation = SKeyValue.Delete)), withWait)
    } else if (vertex.op == GraphUtil.operations("deleteAll")) {
      logger.info(s"deleteAll for vertex is truncated. $vertex")
      Future.successful(true) // Ignore withWait parameter, because deleteAll operation may takes long time
    } else {
      writeToStorage(vertex.hbaseZkAddr, buildPutsAll(vertex), withWait)
    }
  }

  def mutateVertices(vertices: Seq[S2Vertex],
                     withWait: Boolean = false): Future[Seq[Boolean]] = {
    val futures = vertices.map { vertex => mutateVertex(vertex, withWait) }
    Future.sequence(futures)
  }


  def mutateEdgesInner(edges: Seq[S2Edge],
                       checkConsistency: Boolean,
                       withWait: Boolean): Future[Boolean] = {
    assert(edges.nonEmpty)
    // TODO:: remove after code review: unreachable code
    if (!checkConsistency) {

      val zkQuorum = edges.head.innerLabel.hbaseZkAddr
      val futures = edges.map { edge =>
        val (_, edgeUpdate) = S2Edge.buildOperation(None, Seq(edge))

        val (bufferIncr, nonBufferIncr) = increments(edgeUpdate.deepCopy)
        val mutations =
          indexedEdgeMutations(edgeUpdate.deepCopy) ++ snapshotEdgeMutations(edgeUpdate.deepCopy) ++ nonBufferIncr

        if (bufferIncr.nonEmpty) writeToStorage(zkQuorum, bufferIncr, withWait = false)

        writeToStorage(zkQuorum, mutations, withWait)
      }
      Future.sequence(futures).map { rets => rets.forall(identity) }
    } else {
      fetchSnapshotEdgeInner(edges.head).flatMap { case (queryParam, snapshotEdgeOpt, kvOpt) =>
        retry(1)(edges, 0, snapshotEdgeOpt)
      }
    }
  }

  def exponentialBackOff(tryNum: Int) = {
    // time slot is divided by 10 ms
    val slot = 10
    Random.nextInt(Math.min(BackoffTimeout, slot * Math.pow(2, tryNum)).toInt)
  }

  def retry(tryNum: Int)(edges: Seq[S2Edge], statusCode: Byte, fetchedSnapshotEdgeOpt: Option[S2Edge]): Future[Boolean] = {
    if (tryNum >= MaxRetryNum) {
      edges.foreach { edge =>
        logger.error(s"commit failed after $MaxRetryNum\n${edge.toLogString}")
      }

      Future.successful(false)
    } else {
      val future = commitUpdate(edges, statusCode, fetchedSnapshotEdgeOpt)
      future.onSuccess {
        case success =>
          logger.debug(s"Finished. [$tryNum]\n${edges.head.toLogString}\n")
      }
      future recoverWith {
        case FetchTimeoutException(retryEdge) =>
          logger.info(s"[Try: $tryNum], Fetch fail.\n${retryEdge}")
          /* fetch failed. re-fetch should be done */
          fetchSnapshotEdgeInner(edges.head).flatMap { case (queryParam, snapshotEdgeOpt, kvOpt) =>
            retry(tryNum + 1)(edges, statusCode, snapshotEdgeOpt)
          }

        case PartialFailureException(retryEdge, failedStatusCode, faileReason) =>
          val status = failedStatusCode match {
            case 0 => "AcquireLock failed."
            case 1 => "Mutation failed."
            case 2 => "Increment failed."
            case 3 => "ReleaseLock failed."
            case 4 => "Unknown"
          }
          logger.info(s"[Try: $tryNum], [Status: $status] partial fail.\n${retryEdge.toLogString}\nFailReason: ${faileReason}")

          /* retry logic */
          val promise = Promise[Boolean]
          val backOff = exponentialBackOff(tryNum)
          scheduledThreadPool.schedule(new Runnable {
            override def run(): Unit = {
              val future = if (failedStatusCode == 0) {
                // acquire Lock failed. other is mutating so this thead need to re-fetch snapshotEdge.
                /* fetch failed. re-fetch should be done */
                fetchSnapshotEdgeInner(edges.head).flatMap { case (queryParam, snapshotEdgeOpt, kvOpt) =>
                  retry(tryNum + 1)(edges, statusCode, snapshotEdgeOpt)
                }
              } else {
                // partial failure occur while self locked and mutating.
                //            assert(fetchedSnapshotEdgeOpt.nonEmpty)
                retry(tryNum + 1)(edges, failedStatusCode, fetchedSnapshotEdgeOpt)
              }
              promise.completeWith(future)
            }

          }, backOff, TimeUnit.MILLISECONDS)
          promise.future

        case ex: Exception =>
          logger.error("Unknown exception", ex)
          Future.successful(false)
      }
    }
  }

  protected def commitUpdate(edges: Seq[S2Edge],
                             statusCode: Byte,
                             fetchedSnapshotEdgeOpt: Option[S2Edge]): Future[Boolean] = {
//    Future.failed(new PartialFailureException(edges.head, 0, "ahahah"))
    assert(edges.nonEmpty)
//    assert(statusCode == 0 || fetchedSnapshotEdgeOpt.isDefined)

    statusCode match {
      case 0 =>
        fetchedSnapshotEdgeOpt match {
          case None =>
          /*
           * no one has never mutated this SN.
           * (squashedEdge, edgeMutate) = Edge.buildOperation(None, edges)
           * pendingE = squashedEdge.copy(statusCode = 1, lockTs = now, version = squashedEdge.ts + 1)
           * lock = (squashedEdge, pendingE)
           * releaseLock = (edgeMutate.newSnapshotEdge, None)
           */
            val (squashedEdge, edgeMutate) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, edges)

            assert(edgeMutate.newSnapshotEdge.isDefined)

            val lockTs = Option(System.currentTimeMillis())
            val pendingEdge = squashedEdge.copy(statusCode = 1, lockTs = lockTs, version = squashedEdge.ts + 1)
            val lockSnapshotEdge = squashedEdge.toSnapshotEdge.copy(pendingEdgeOpt = Option(pendingEdge))
            val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge.get.copy(statusCode = 0,
              pendingEdgeOpt = None, version = lockSnapshotEdge.version + 1)

            commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)

          case Some(snapshotEdge) =>
            snapshotEdge.pendingEdgeOpt match {
              case None =>
                /*
                 * others finished commit on this SN. but there is no contention.
                 * (squashedEdge, edgeMutate) = Edge.buildOperation(snapshotEdgeOpt, edges)
                 * pendingE = squashedEdge.copy(statusCode = 1, lockTs = now, version = snapshotEdge.version + 1) ?
                 * lock = (snapshotEdge, pendingE)
                 * releaseLock = (edgeMutate.newSnapshotEdge, None)
                 */
                val (squashedEdge, edgeMutate) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, edges)
                if (edgeMutate.newSnapshotEdge.isEmpty) {
                  logger.debug(s"drop this requests: \n${edges.map(_.toLogString).mkString("\n")}")
                  Future.successful(true)
                } else {
                  val lockTs = Option(System.currentTimeMillis())
                  val pendingEdge = squashedEdge.copy(statusCode = 1, lockTs = lockTs, version = snapshotEdge.version + 1)
                  val lockSnapshotEdge = snapshotEdge.toSnapshotEdge.copy(pendingEdgeOpt = Option(pendingEdge))
                  val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge.get.copy(statusCode = 0,
                    pendingEdgeOpt = None, version = lockSnapshotEdge.version + 1)
                  commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)
                }
              case Some(pendingEdge) =>
                val isLockExpired = pendingEdge.lockTs.get + LockExpireDuration < System.currentTimeMillis()
                if (isLockExpired) {
                  /*
                   * if pendingEdge.ts == snapshotEdge.ts =>
                   *    (squashedEdge, edgeMutate) = Edge.buildOperation(None, Seq(pendingEdge))
                   * else =>
                   *    (squashedEdge, edgeMutate) = Edge.buildOperation(snapshotEdgeOpt, Seq(pendingEdge))
                   * pendingE = squashedEdge.copy(statusCode = 1, lockTs = now, version = snapshotEdge.version + 1)
                   * lock = (snapshotEdge, pendingE)
                   * releaseLock = (edgeMutate.newSnapshotEdge, None)
                   */
                  logger.debug(s"${pendingEdge.toLogString} has been expired.")
                  val (squashedEdge, edgeMutate) =
                    if (pendingEdge.ts == snapshotEdge.ts) S2Edge.buildOperation(None, pendingEdge +: edges)
                    else S2Edge.buildOperation(fetchedSnapshotEdgeOpt, pendingEdge +: edges)

                  val lockTs = Option(System.currentTimeMillis())
                  val newPendingEdge = squashedEdge.copy(statusCode = 1, lockTs = lockTs, version = snapshotEdge.version + 1)
                  val lockSnapshotEdge = snapshotEdge.toSnapshotEdge.copy(pendingEdgeOpt = Option(newPendingEdge))
                  val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge.get.copy(statusCode = 0,
                    pendingEdgeOpt = None, version = lockSnapshotEdge.version + 1)

                  commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)
                } else {
                  /*
                   * others finished commit on this SN and there is currently contention.
                   * this can't be proceed so retry from re-fetch.
                   * throw EX
                   */
                  val (squashedEdge, _) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, edges)
                  Future.failed(new PartialFailureException(squashedEdge, 0, s"others[${pendingEdge.ts}] is mutating. me[${squashedEdge.ts}]"))
                }
            }

        }
      case _ =>

        /*
         * statusCode > 0 which means self locked and there has been partial failure either on mutate, increment, releaseLock
         */

        /*
         * this succeed to lock this SN. keep doing on commit process.
         * if SN.isEmpty =>
         * no one never succed to commit on this SN.
         * this is first mutation try on this SN.
         * (squashedEdge, edgeMutate) = Edge.buildOperation(SN, edges)
         * else =>
         * assert(SN.pengingEdgeOpt.isEmpty) no-fetch after acquire lock when self retrying.
         * there has been success commit on this SN.
         * (squashedEdge, edgeMutate) = Edge.buildOperation(SN, edges)
         * releaseLock = (edgeMutate.newSnapshotEdge, None)
         */
        val _edges =
          if (fetchedSnapshotEdgeOpt.isDefined && fetchedSnapshotEdgeOpt.get.pendingEdgeOpt.isDefined) fetchedSnapshotEdgeOpt.get.pendingEdgeOpt.get +: edges
          else edges
        val (squashedEdge, edgeMutate) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, _edges)
        val newVersion = fetchedSnapshotEdgeOpt.map(_.version).getOrElse(squashedEdge.ts) + 2
        val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge match {
          case None => squashedEdge.toSnapshotEdge.copy(statusCode = 0, pendingEdgeOpt = None, version = newVersion)
          case Some(newSnapshotEdge) => newSnapshotEdge.copy(statusCode = 0, pendingEdgeOpt = None, version = newVersion)
        }
        // lockSnapshotEdge will be ignored.
        commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, releaseLockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)
    }
  }
  /**
   * orchestrate commit process.
   * we separate into 4 step to avoid duplicating each step over and over.
    *
    * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge.
   * @param fetchedSnapshotEdgeOpt: fetched snapshotEdge from storage before commit process begin.
   * @param lockSnapshotEdge: lockEdge that hold necessary data to lock this snapshotEdge for this thread.
   * @param releaseLockSnapshotEdge: releaseLockEdge that will remove lock by storing new final merged states
   *                               all from current request edges and fetched snapshotEdge.
   * @param edgeMutate: mutations for indexEdge and snapshotEdge.
   * @return
   */
  protected def commitProcess(statusCode: Byte,
                              squashedEdge: S2Edge,
                              fetchedSnapshotEdgeOpt:Option[S2Edge],
                              lockSnapshotEdge: SnapshotEdge,
                              releaseLockSnapshotEdge: SnapshotEdge,
                              edgeMutate: EdgeMutate): Future[Boolean] = {
    for {
      locked <- acquireLock(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge)
      mutated <- commitIndexEdgeMutations(locked, statusCode, squashedEdge, edgeMutate)
      incremented <- commitIndexEdgeDegreeMutations(mutated, statusCode, squashedEdge, edgeMutate)
      lockReleased <- releaseLock(incremented, statusCode, squashedEdge, releaseLockSnapshotEdge)
    } yield lockReleased
  }

  case class PartialFailureException(edge: S2Edge, statusCode: Byte, failReason: String) extends NoStackException(failReason)

  protected def debug(ret: Boolean, phase: String, snapshotEdge: SnapshotEdge) = {
    val msg = Seq(s"[$ret] [$phase]", s"${snapshotEdge.toLogString()}").mkString("\n")
    logger.debug(msg)
  }

  protected def debug(ret: Boolean, phase: String, snapshotEdge: SnapshotEdge, edgeMutate: EdgeMutate) = {
    val msg = Seq(s"[$ret] [$phase]", s"${snapshotEdge.toLogString()}",
      s"${edgeMutate.toLogString}").mkString("\n")
    logger.debug(msg)
  }

  /**
   * try to acquire lock on storage for this given snapshotEdge(lockEdge).
    *
    * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param fetchedSnapshotEdgeOpt: fetched snapshot edge from storage.
   * @param lockEdge: lockEdge to build RPC request(compareAndSet) into Storage.
   * @return
   */
  protected def acquireLock(statusCode: Byte,
                            squashedEdge: S2Edge,
                            fetchedSnapshotEdgeOpt: Option[S2Edge],
                            lockEdge: SnapshotEdge): Future[Boolean] = {
    if (statusCode >= 1) {
      logger.debug(s"skip acquireLock: [$statusCode]\n${squashedEdge.toLogString}")
      Future.successful(true)
    } else {
      val p = Random.nextDouble()
      if (p < FailProb) {
        Future.failed(new PartialFailureException(squashedEdge, 0, s"$p"))
      } else {
        val lockEdgePut = snapshotEdgeSerializer(lockEdge).toKeyValues.head
        val oldPut = fetchedSnapshotEdgeOpt.map(e => snapshotEdgeSerializer(e.toSnapshotEdge).toKeyValues.head)
        writeLock(lockEdgePut, oldPut).recoverWith { case ex: Exception =>
          logger.error(s"AcquireLock RPC Failed.")
          throw new PartialFailureException(squashedEdge, 0, "AcquireLock RPC Failed")
        }.map { ret =>
          if (ret) {
            val log = Seq(
              "\n",
              "=" * 50,
              s"[Success]: acquireLock",
              s"[RequestEdge]: ${squashedEdge.toLogString}",
              s"[LockEdge]: ${lockEdge.toLogString()}",
              s"[PendingEdge]: ${lockEdge.pendingEdgeOpt.map(_.toLogString).getOrElse("")}",
              "=" * 50, "\n").mkString("\n")

            logger.debug(log)
            //            debug(ret, "acquireLock", edge.toSnapshotEdge)
          } else {
            throw new PartialFailureException(squashedEdge, 0, "hbase fail.")
          }
          true
        }
      }
    }
  }


  /**
   * change this snapshot's state on storage from locked into committed by
   * storing new merged states on storage. merge state come from releaseLockEdge.
   * note that releaseLock return Future.failed on predicate failure.
    *
    * @param predicate: indicate if this releaseLock phase should be proceed or not.
   * @param statusCode: releaseLock do not use statusCode, only for debug.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param releaseLockEdge: final merged states if all process goes well.
   * @return
   */
  protected def releaseLock(predicate: Boolean,
                            statusCode: Byte,
                            squashedEdge: S2Edge,
                            releaseLockEdge: SnapshotEdge): Future[Boolean] = {
    if (!predicate) {
      Future.failed(new PartialFailureException(squashedEdge, 3, "predicate failed."))
    } else {
      val p = Random.nextDouble()
      if (p < FailProb) Future.failed(new PartialFailureException(squashedEdge, 3, s"$p"))
      else {
        val releaseLockEdgePuts = snapshotEdgeSerializer(releaseLockEdge).toKeyValues
        writeToStorage(squashedEdge.innerLabel.hbaseZkAddr, releaseLockEdgePuts, withWait = true).recoverWith {
          case ex: Exception =>
            logger.error(s"ReleaseLock RPC Failed.")
            throw new PartialFailureException(squashedEdge, 3, "ReleaseLock RPC Failed")
        }.map { ret =>
          if (ret) {
            debug(ret, "releaseLock", squashedEdge.toSnapshotEdge)
          } else {
            val msg = Seq("\nFATAL ERROR\n",
              "=" * 50,
              squashedEdge.toLogString,
              releaseLockEdgePuts,
              "=" * 50,
              "\n"
            )
            logger.error(msg.mkString("\n"))
            //          error(ret, "releaseLock", edge.toSnapshotEdge)
            throw new PartialFailureException(squashedEdge, 3, "hbase fail.")
          }
          true
        }
      }
    }
  }

  /**
   *
   * @param predicate: indicate if this commitIndexEdgeMutations phase should be proceed or not.
   * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param edgeMutate: actual collection of mutations. note that edgeMutate contains snapshotEdge mutations,
   *                  but in here, we only use indexEdge's mutations.
   * @return
   */
  protected def commitIndexEdgeMutations(predicate: Boolean,
                       statusCode: Byte,
                       squashedEdge: S2Edge,
                       edgeMutate: EdgeMutate): Future[Boolean] = {
    if (!predicate) Future.failed(new PartialFailureException(squashedEdge, 1, "predicate failed."))
    else {
      if (statusCode >= 2) {
        logger.debug(s"skip mutate: [$statusCode]\n${squashedEdge.toLogString}")
        Future.successful(true)
      } else {
        val p = Random.nextDouble()
        if (p < FailProb) Future.failed(new PartialFailureException(squashedEdge, 1, s"$p"))
        else
          writeToStorage(squashedEdge.innerLabel.hbaseZkAddr, indexedEdgeMutations(edgeMutate), withWait = true).map { ret =>
            if (ret) {
              debug(ret, "mutate", squashedEdge.toSnapshotEdge, edgeMutate)
            } else {
              throw new PartialFailureException(squashedEdge, 1, "hbase fail.")
            }
            true
          }
      }
    }
  }

  /**
   *
   * @param predicate: indicate if this commitIndexEdgeMutations phase should be proceed or not.
   * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param edgeMutate: actual collection of mutations. note that edgeMutate contains snapshotEdge mutations,
   *                  but in here, we only use indexEdge's degree mutations.
   * @return
   */
  protected def commitIndexEdgeDegreeMutations(predicate: Boolean,
                          statusCode: Byte,
                          squashedEdge: S2Edge,
                          edgeMutate: EdgeMutate): Future[Boolean] = {

    def _write(kvs: Seq[SKeyValue], withWait: Boolean): Future[Boolean] = {
      writeToStorage(squashedEdge.innerLabel.hbaseZkAddr, kvs, withWait = withWait).map { ret =>
        if (ret) {
          debug(ret, "increment", squashedEdge.toSnapshotEdge, edgeMutate)
        } else {
          throw new PartialFailureException(squashedEdge, 2, "hbase fail.")
        }
        true
      }
    }

    if (!predicate) Future.failed(new PartialFailureException(squashedEdge, 2, "predicate failed."))
    if (statusCode >= 3) {
      logger.debug(s"skip increment: [$statusCode]\n${squashedEdge.toLogString}")
      Future.successful(true)
    } else {
      val p = Random.nextDouble()
      if (p < FailProb) Future.failed(new PartialFailureException(squashedEdge, 2, s"$p"))
      else {
        val (bufferIncr, nonBufferIncr) = increments(edgeMutate.deepCopy)

        if (bufferIncr.nonEmpty) _write(bufferIncr, withWait = false)
        _write(nonBufferIncr, withWait = true)
      }
    }
  }

  /** end of methods for consistency */

  def mutateLog(snapshotEdgeOpt: Option[S2Edge], edges: Seq[S2Edge],
                newEdge: S2Edge, edgeMutate: EdgeMutate) =
    Seq("----------------------------------------------",
      s"SnapshotEdge: ${snapshotEdgeOpt.map(_.toLogString)}",
      s"requestEdges: ${edges.map(_.toLogString).mkString("\n")}",
      s"newEdge: ${newEdge.toLogString}",
      s"mutation: \n${edgeMutate.toLogString}",
      "----------------------------------------------").mkString("\n")


  /** Delete All */
  def deleteAllFetchedEdgesAsyncOld(stepInnerResult: StepResult,
                                              requestTs: Long,
                                              retryNum: Int): Future[Boolean] = {
    if (stepInnerResult.isEmpty) Future.successful(true)
    else {
      val head = stepInnerResult.edgeWithScores.head
      val zkQuorum = head.edge.innerLabel.hbaseZkAddr
      val futures = for {
        edgeWithScore <- stepInnerResult.edgeWithScores
      } yield {
          val edge = edgeWithScore.edge
          val score = edgeWithScore.score

          val edgeSnapshot = edge.copyEdge(propsWithTs = S2Edge.propsToState(edge.updatePropsWithTs()))
          val reversedSnapshotEdgeMutations = snapshotEdgeSerializer(edgeSnapshot.toSnapshotEdge).toKeyValues.map(_.copy(operation = SKeyValue.Put))

          val edgeForward = edge.copyEdge(propsWithTs = S2Edge.propsToState(edge.updatePropsWithTs()))
          val forwardIndexedEdgeMutations = edgeForward.edgesWithIndex.flatMap { indexEdge =>
            indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Delete)) ++
              buildIncrementsAsync(indexEdge, -1L)
          }

          /* reverted direction */
          val edgeRevert = edge.copyEdge(propsWithTs = S2Edge.propsToState(edge.updatePropsWithTs()))
          val reversedIndexedEdgesMutations = edgeRevert.duplicateEdge.edgesWithIndex.flatMap { indexEdge =>
            indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Delete)) ++
              buildIncrementsAsync(indexEdge, -1L)
          }

          val mutations = reversedIndexedEdgesMutations ++ reversedSnapshotEdgeMutations ++ forwardIndexedEdgeMutations

          writeToStorage(zkQuorum, mutations, withWait = true)
        }

      Future.sequence(futures).map { rets => rets.forall(identity) }
    }
  }

  /** End Of Delete All */


  /** Parsing Logic: parse from kv from Storage into Edge */
  def toEdge[K: CanSKeyValue](kv: K,
                              queryRequest: QueryRequest,
                              cacheElementOpt: Option[S2Edge],
                              parentEdges: Seq[EdgeWithScore]): Option[S2Edge] = {
    logger.debug(s"toEdge: $kv")

    try {
      val queryOption = queryRequest.query.queryOption
      val queryParam = queryRequest.queryParam
      val schemaVer = queryParam.label.schemaVersion
      val indexEdgeOpt = indexEdgeDeserializer(schemaVer).fromKeyValues(Seq(kv), cacheElementOpt)
      if (!queryOption.returnTree) indexEdgeOpt.map(indexEdge => indexEdge.copy(parentEdges = parentEdges))
      else indexEdgeOpt
    } catch {
      case ex: Exception =>
        logger.error(s"Fail on toEdge: ${kv.toString}, ${queryRequest}", ex)
        None
    }
  }

  def toSnapshotEdge[K: CanSKeyValue](kv: K,
                                      queryRequest: QueryRequest,
                                      cacheElementOpt: Option[SnapshotEdge] = None,
                                      isInnerCall: Boolean,
                                      parentEdges: Seq[EdgeWithScore]): Option[S2Edge] = {
//        logger.debug(s"SnapshottoEdge: $kv")
    val queryParam = queryRequest.queryParam
    val schemaVer = queryParam.label.schemaVersion
    val snapshotEdgeOpt = snapshotEdgeDeserializer(schemaVer).fromKeyValues(Seq(kv), cacheElementOpt)

    if (isInnerCall) {
      snapshotEdgeOpt.flatMap { snapshotEdge =>
        val edge = snapshotEdge.toEdge.copy(parentEdges = parentEdges)
        if (queryParam.where.map(_.filter(edge)).getOrElse(true)) Option(edge)
        else None
      }
    } else {
      snapshotEdgeOpt.flatMap { snapshotEdge =>
        if (snapshotEdge.allPropsDeleted) None
        else {
          val edge = snapshotEdge.toEdge.copy(parentEdges = parentEdges)
          if (queryParam.where.map(_.filter(edge)).getOrElse(true)) Option(edge)
          else None
        }
      }
    }
  }

  val dummyCursor: Array[Byte] = Array.empty

  def toEdges[K: CanSKeyValue](kvs: Seq[K],
                               queryRequest: QueryRequest,
                               prevScore: Double = 1.0,
                               isInnerCall: Boolean,
                               parentEdges: Seq[EdgeWithScore],
                               startOffset: Int = 0,
                               len: Int = Int.MaxValue): StepResult = {

    val toSKeyValue = implicitly[CanSKeyValue[K]].toSKeyValue _

    if (kvs.isEmpty) StepResult.Empty.copy(cursors = Seq(dummyCursor))
    else {
      val queryOption = queryRequest.query.queryOption
      val queryParam = queryRequest.queryParam
      val labelWeight = queryRequest.labelWeight
      val nextStepOpt = queryRequest.nextStepOpt
      val where = queryParam.where.get
      val label = queryParam.label
      val isDefaultTransformer = queryParam.edgeTransformer.isDefault
      val first = kvs.head
      val kv = first
      val schemaVer = queryParam.label.schemaVersion
      val cacheElementOpt =
        if (queryParam.isSnapshotEdge) None
        else indexEdgeDeserializer(schemaVer).fromKeyValues(Seq(kv), None)

      val (degreeEdges, keyValues) = cacheElementOpt match {
        case None => (Nil, kvs)
        case Some(cacheElement) =>
          val head = cacheElement
          if (!head.isDegree) (Nil, kvs)
          else (Seq(EdgeWithScore(head, 1.0, label)), kvs.tail)
      }

      val lastCursor: Seq[Array[Byte]] = Seq(if (keyValues.nonEmpty) toSKeyValue(keyValues(keyValues.length - 1)).row else dummyCursor)

      if (!queryOption.ignorePrevStepCache) {
        val edgeWithScores = for {
          (kv, idx) <- keyValues.zipWithIndex if idx >= startOffset && idx < startOffset + len
          edge <- (if (queryParam.isSnapshotEdge) toSnapshotEdge(kv, queryRequest, None, isInnerCall, parentEdges) else toEdge(kv, queryRequest, cacheElementOpt, parentEdges)).toSeq
          if where == WhereParser.success || where.filter(edge)
          convertedEdge <- if (isDefaultTransformer) Seq(edge) else convertEdges(queryParam, edge, nextStepOpt)
        } yield {
            val score = edge.rank(queryParam.rank)
            EdgeWithScore(convertedEdge, score, label)
          }
        StepResult(edgeWithScores = edgeWithScores, grouped = Nil, degreeEdges = degreeEdges, cursors = lastCursor)
      } else {
        val degreeScore = 0.0

        val edgeWithScores = for {
          (kv, idx) <- keyValues.zipWithIndex if idx >= startOffset && idx < startOffset + len
          edge <- (if (queryParam.isSnapshotEdge) toSnapshotEdge(kv, queryRequest, None, isInnerCall, parentEdges) else toEdge(kv, queryRequest, cacheElementOpt, parentEdges)).toSeq
          if where == WhereParser.success || where.filter(edge)
          convertedEdge <- if (isDefaultTransformer) Seq(edge) else convertEdges(queryParam, edge, nextStepOpt)
        } yield {
            val edgeScore = edge.rank(queryParam.rank)
            val score = queryParam.scorePropagateOp match {
              case "plus" => edgeScore + prevScore
              case "divide" =>
                if ((prevScore + queryParam.scorePropagateShrinkage) == 0) 0
                else edgeScore / (prevScore + queryParam.scorePropagateShrinkage)
              case _ => edgeScore * prevScore
            }
            val tsVal = processTimeDecay(queryParam, edge)
            val newScore = degreeScore + score
            EdgeWithScore(convertedEdge.copy(parentEdges = parentEdges), score = newScore * labelWeight * tsVal, label = label)
          }

        val sampled =
          if (queryRequest.queryParam.sample >= 0) sample(queryRequest, edgeWithScores, queryRequest.queryParam.sample)
          else edgeWithScores

        val normalized = if (queryParam.shouldNormalize) normalize(sampled) else sampled

        StepResult(edgeWithScores = normalized, grouped = Nil, degreeEdges = degreeEdges, cursors = lastCursor)
      }
    }
  }

  /** End Of Parse Logic */

  protected def toRequestEdge(queryRequest: QueryRequest, parentEdges: Seq[EdgeWithScore]): S2Edge = {
    val srcVertex = queryRequest.vertex
    val queryParam = queryRequest.queryParam
    val tgtVertexIdOpt = queryParam.tgtVertexInnerIdOpt
    val label = queryParam.label
    val labelWithDir = queryParam.labelWithDir
    val (srcColumn, tgtColumn) = label.srcTgtColumn(labelWithDir.dir)
    val propsWithTs = label.EmptyPropsWithTs

    tgtVertexIdOpt match {
      case Some(tgtVertexId) => // _to is given.
        /* we use toSnapshotEdge so dont need to swap src, tgt */
        val src = srcVertex.innerId
        val tgt = tgtVertexId
        val (srcVId, tgtVId) = (SourceVertexId(srcColumn, src), TargetVertexId(tgtColumn, tgt))
        val (srcV, tgtV) = (graph.newVertex(srcVId), graph.newVertex(tgtVId))

        graph.newEdge(srcV, tgtV, label, labelWithDir.dir, propsWithTs = propsWithTs)
      case None =>
        val src = srcVertex.innerId
        val srcVId = SourceVertexId(srcColumn, src)
        val srcV = graph.newVertex(srcVId)

        graph.newEdge(srcV, srcV, label, labelWithDir.dir, propsWithTs = propsWithTs, parentEdges = parentEdges)
    }
  }

  protected def fetchSnapshotEdgeInner(edge: S2Edge): Future[(QueryParam, Option[S2Edge], Option[SKeyValue])] = {
    /* TODO: Fix this. currently fetchSnapshotEdge should not use future cache
      * so use empty cacheKey.
      * */
    val queryParam = QueryParam(labelName = edge.innerLabel.label,
      direction = GraphUtil.fromDirection(edge.labelWithDir.dir),
      tgtVertexIdOpt = Option(edge.tgtVertex.innerIdVal),
      cacheTTLInMillis = -1)
    val q = Query.toQuery(Seq(edge.srcVertex), Seq(queryParam))
    val queryRequest = QueryRequest(q, 0, edge.srcVertex, queryParam)
    //    val q = Query.toQuery(Seq(edge.srcVertex), queryParam)


    fetchSnapshotEdgeKeyValues(queryRequest).map { kvs =>
      val (edgeOpt, kvOpt) =
        if (kvs.isEmpty) (None, None)
        else {
          val snapshotEdgeOpt = toSnapshotEdge(kvs.head, queryRequest, isInnerCall = true, parentEdges = Nil)
          val _kvOpt = kvs.headOption
          (snapshotEdgeOpt, _kvOpt)
        }
      (queryParam, edgeOpt, kvOpt)
    } recoverWith { case ex: Throwable =>
      logger.error(s"fetchQueryParam failed. fallback return.", ex)
      throw new FetchTimeoutException(s"${edge.toLogString}")
    }
  }


  /** end of query */

  /** Mutation Builder */


  /** EdgeMutate */
  def indexedEdgeMutations(edgeMutate: EdgeMutate): Seq[SKeyValue] = {
    // skip sampling for delete operation
    val deleteMutations = edgeMutate.edgesToDeleteWithIndexOpt.flatMap { indexEdge =>
      indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Delete, durability = indexEdge.label.durability))
    }

    val insertMutations = edgeMutate.edgesToInsertWithIndexOpt.flatMap { indexEdge =>
      indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Put, durability = indexEdge.label.durability))
    }

    deleteMutations ++ insertMutations
  }

  def snapshotEdgeMutations(edgeMutate: EdgeMutate): Seq[SKeyValue] =
    edgeMutate.newSnapshotEdge.map(e => snapshotEdgeSerializer(e).toKeyValues.map(_.copy(durability = e.label.durability))).getOrElse(Nil)

  def increments(edgeMutate: EdgeMutate): (Seq[SKeyValue], Seq[SKeyValue]) = {
    (edgeMutate.edgesToDeleteWithIndexOptForDegree.isEmpty, edgeMutate.edgesToInsertWithIndexOptForDegree.isEmpty) match {
      case (true, true) =>
        /* when there is no need to update. shouldUpdate == false */
        Nil -> Nil

      case (true, false) =>
        /* no edges to delete but there is new edges to insert so increase degree by 1 */
        val (buffer, nonBuffer) = EdgeMutate.partitionBufferedIncrement(edgeMutate.edgesToInsertWithIndexOptForDegree)
        buffer.flatMap(buildIncrementsAsync(_)) -> nonBuffer.flatMap(buildIncrementsAsync(_))

      case (false, true) =>
        /* no edges to insert but there is old edges to delete so decrease degree by 1 */
        val (buffer, nonBuffer) = EdgeMutate.partitionBufferedIncrement(edgeMutate.edgesToDeleteWithIndexOptForDegree)
        buffer.flatMap(buildIncrementsAsync(_, -1)) -> nonBuffer.flatMap(buildIncrementsAsync(_, -1))

      case (false, false) =>
        /* update on existing edges so no change on degree */
        Nil -> Nil
    }
  }

  /** IndexEdge */
  def buildIncrementsAsync(indexedEdge: IndexEdge, amount: Long = 1L): Seq[SKeyValue] = {
    val newProps = indexedEdge.updatePropsWithTs()
    newProps.put(LabelMeta.degree.name, new S2Property(indexedEdge.toEdge, LabelMeta.degree, LabelMeta.degree.name, amount, indexedEdge.ts))
    val _indexedEdge = indexedEdge.copy(propsWithTs = newProps)
    indexEdgeSerializer(_indexedEdge).toKeyValues.map(_.copy(operation = SKeyValue.Increment, durability = _indexedEdge.label.durability))
  }

  def buildIncrementsCountAsync(indexedEdge: IndexEdge, amount: Long = 1L): Seq[SKeyValue] = {
    val newProps = indexedEdge.updatePropsWithTs()
    newProps.put(LabelMeta.degree.name, new S2Property(indexedEdge.toEdge, LabelMeta.degree, LabelMeta.degree.name, amount, indexedEdge.ts))
    val _indexedEdge = indexedEdge.copy(propsWithTs = newProps)
    indexEdgeSerializer(_indexedEdge).toKeyValues.map(_.copy(operation = SKeyValue.Increment, durability = _indexedEdge.label.durability))
  }

  //TODO: ServiceColumn do not have durability property yet.
  def buildDeleteBelongsToId(vertex: S2Vertex): Seq[SKeyValue] = {
    val kvs = vertexSerializer(vertex).toKeyValues
    val kv = kvs.head
    vertex.belongLabelIds.map { id =>
      kv.copy(qualifier = Bytes.toBytes(S2Vertex.toPropKey(id)), operation = SKeyValue.Delete)
    }
  }

  def buildVertexPutsAsync(edge: S2Edge): Seq[SKeyValue] = {
    val storeVertex = edge.innerLabel.extraOptions.get("storeVertex").map(_.as[Boolean]).getOrElse(false)

    if (storeVertex) {
      if (edge.op == GraphUtil.operations("delete"))
        buildDeleteBelongsToId(edge.srcForVertex) ++ buildDeleteBelongsToId(edge.tgtForVertex)
      else
        vertexSerializer(edge.srcForVertex).toKeyValues ++ vertexSerializer(edge.tgtForVertex).toKeyValues
    } else {
      Seq.empty
    }
  }

  def buildDegreePuts(edge: S2Edge, degreeVal: Long): Seq[SKeyValue] = {
    edge.propertyInner(LabelMeta.degree.name, degreeVal, edge.ts)
    val kvs = edge.edgesWithIndexValid.flatMap { indexEdge =>
      indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Put, durability = indexEdge.label.durability))
    }

    kvs
  }

  def buildPutsAll(vertex: S2Vertex): Seq[SKeyValue] = {
    vertex.op match {
      case d: Byte if d == GraphUtil.operations("delete") => vertexSerializer(vertex).toKeyValues.map(_.copy(operation = SKeyValue.Delete))
      case _ => vertexSerializer(vertex).toKeyValues.map(_.copy(operation = SKeyValue.Put))
    }
  }

  def info: Map[String, String] = Map("className" -> this.getClass.getSimpleName)


}