You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@s2graph.apache.org by st...@apache.org on 2016/03/15 10:28:38 UTC
[22/27] incubator-s2graph git commit: [S2GRAPH-57]: Change package
names into org.apahce.s2graph.
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/PostProcess.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/PostProcess.scala b/s2core/src/main/scala/com/kakao/s2graph/core/PostProcess.scala
deleted file mode 100644
index f301d68..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/PostProcess.scala
+++ /dev/null
@@ -1,554 +0,0 @@
-package com.kakao.s2graph.core
-
-import com.kakao.s2graph.core.GraphExceptions.BadQueryException
-
-import com.kakao.s2graph.core.mysqls.{ColumnMeta, Label, ServiceColumn, LabelMeta}
-import com.kakao.s2graph.core.types.{InnerVal, InnerValLike}
-import com.kakao.s2graph.core.utils.logger
-import play.api.libs.json.{Json, _}
-import scala.collection.mutable.{ArrayBuffer, ListBuffer}
-
-object PostProcess extends JSONParser {
-
-
- type EDGE_VALUES = Map[String, JsValue]
- type ORDER_BY_VALUES = (Any, Any, Any, Any)
- type RAW_EDGE = (EDGE_VALUES, Double, ORDER_BY_VALUES)
-
- /**
- * Result Entity score field name
- */
- val emptyDegrees = Seq.empty[JsValue]
- val timeoutResults = Json.obj("size" -> 0, "degrees" -> Json.arr(), "results" -> Json.arr(), "isTimeout" -> true)
- val emptyResults = Json.obj("size" -> 0, "degrees" -> Json.arr(), "results" -> Json.arr(), "isEmpty" -> true)
- def badRequestResults(ex: => Exception) = ex match {
- case ex: BadQueryException => Json.obj("message" -> ex.msg)
- case _ => Json.obj("message" -> ex.getMessage)
- }
-
- val SCORE_FIELD_NAME = "scoreSum"
- val reservedColumns = Set("cacheRemain", "from", "to", "label", "direction", "_timestamp", "timestamp", "score", "props")
-
- def groupEdgeResult(queryRequestWithResultLs: Seq[QueryRequestWithResult], exclude: Seq[QueryRequestWithResult]) = {
- val excludeIds = resultInnerIds(exclude).map(innerId => innerId -> true).toMap
- // filterNot {case (edge, score) => edge.props.contains(LabelMeta.degreeSeq)}
- val groupedEdgesWithRank = (for {
- queryRequestWithResult <- queryRequestWithResultLs
- (queryRequest, queryResult) = QueryRequestWithResult.unapply(queryRequestWithResult).get
- edgeWithScore <- queryResult.edgeWithScoreLs
- (edge, score) = EdgeWithScore.unapply(edgeWithScore).get
- if !excludeIds.contains(toHashKey(edge, queryRequest.queryParam, queryRequest.query.filterOutFields))
- } yield {
- (queryRequest.queryParam, edge, score)
- }).groupBy {
- case (queryParam, edge, rank) if edge.labelWithDir.dir == GraphUtil.directions("in") =>
- (queryParam.label.srcColumn, queryParam.label.label, queryParam.label.tgtColumn, edge.tgtVertex.innerId, edge.isDegree)
- case (queryParam, edge, rank) =>
- (queryParam.label.tgtColumn, queryParam.label.label, queryParam.label.srcColumn, edge.tgtVertex.innerId, edge.isDegree)
- }
-
- val ret = for {
- ((tgtColumn, labelName, srcColumn, target, isDegreeEdge), edgesAndRanks) <- groupedEdgesWithRank if !isDegreeEdge
- edgesWithRanks = edgesAndRanks.groupBy(x => x._2.srcVertex).map(_._2.head)
- id <- innerValToJsValue(target, tgtColumn.columnType)
- } yield {
- Json.obj("name" -> tgtColumn.columnName, "id" -> id,
- SCORE_FIELD_NAME -> edgesWithRanks.map(_._3).sum,
- "label" -> labelName,
- "aggr" -> Json.obj(
- "name" -> srcColumn.columnName,
- "ids" -> edgesWithRanks.flatMap { case (queryParam, edge, rank) =>
- innerValToJsValue(edge.srcVertex.innerId, srcColumn.columnType)
- },
- "edges" -> edgesWithRanks.map { case (queryParam, edge, rank) =>
- Json.obj("id" -> innerValToJsValue(edge.srcVertex.innerId, srcColumn.columnType),
- "props" -> propsToJson(edge),
- "score" -> rank
- )
- }
- )
- )
- }
-
- ret.toList
- }
-
- def sortWithFormatted(jsons: Seq[JsObject],
- scoreField: String = "scoreSum",
- queryRequestWithResultLs: Seq[QueryRequestWithResult],
- decrease: Boolean = true): JsObject = {
- val ordering = if (decrease) -1 else 1
- val sortedJsons = jsons.sortBy { jsObject => (jsObject \ scoreField).as[Double] * ordering }
-
- if (queryRequestWithResultLs.isEmpty) Json.obj("size" -> sortedJsons.size, "results" -> sortedJsons)
- else Json.obj(
- "size" -> sortedJsons.size,
- "results" -> sortedJsons,
- "impressionId" -> queryRequestWithResultLs.head.queryRequest.query.impressionId()
- )
- }
-
- private def toHashKey(edge: Edge, queryParam: QueryParam, fields: Seq[String], delimiter: String = ","): Int = {
- val ls = for {
- field <- fields
- } yield {
- field match {
- case "from" | "_from" => edge.srcVertex.innerId.toIdString()
- case "to" | "_to" => edge.tgtVertex.innerId.toIdString()
- case "label" => edge.labelWithDir.labelId
- case "direction" => JsString(GraphUtil.fromDirection(edge.labelWithDir.dir))
- case "_timestamp" | "timestamp" => edge.ts
- case _ =>
- queryParam.label.metaPropsInvMap.get(field) match {
- case None => throw new RuntimeException(s"unknow column: $field")
- case Some(labelMeta) => edge.propsWithTs.get(labelMeta.seq) match {
- case None => labelMeta.defaultValue
- case Some(propVal) => propVal
- }
- }
- }
- }
- val ret = ls.hashCode()
- ret
- }
-
- def resultInnerIds(queryRequestWithResultLs: Seq[QueryRequestWithResult], isSrcVertex: Boolean = false): Seq[Int] = {
- for {
- queryRequestWithResult <- queryRequestWithResultLs
- (queryRequest, queryResult) = QueryRequestWithResult.unapply(queryRequestWithResult).get
- q = queryRequest.query
- edgeWithScore <- queryResult.edgeWithScoreLs
- (edge, score) = EdgeWithScore.unapply(edgeWithScore).get
- } yield toHashKey(edge, queryRequest.queryParam, q.filterOutFields)
- }
-
- def summarizeWithListExcludeFormatted(queryRequestWithResultLs: Seq[QueryRequestWithResult], exclude: Seq[QueryRequestWithResult]) = {
- val jsons = groupEdgeResult(queryRequestWithResultLs, exclude)
- sortWithFormatted(jsons, queryRequestWithResultLs = queryRequestWithResultLs, decrease = true)
- }
-
- def summarizeWithList(queryRequestWithResultLs: Seq[QueryRequestWithResult], exclude: Seq[QueryRequestWithResult]) = {
- val jsons = groupEdgeResult(queryRequestWithResultLs, exclude)
- sortWithFormatted(jsons, queryRequestWithResultLs = queryRequestWithResultLs)
- }
-
- def summarizeWithListFormatted(queryRequestWithResultLs: Seq[QueryRequestWithResult], exclude: Seq[QueryRequestWithResult]) = {
- val jsons = groupEdgeResult(queryRequestWithResultLs, exclude)
- sortWithFormatted(jsons, queryRequestWithResultLs = queryRequestWithResultLs)
- }
-
- def toSimpleVertexArrJson(queryRequestWithResultLs: Seq[QueryRequestWithResult]): JsValue = {
- toSimpleVertexArrJson(queryRequestWithResultLs, Seq.empty[QueryRequestWithResult])
- }
-
- private def orderBy(queryOption: QueryOption,
- rawEdges: ListBuffer[(EDGE_VALUES, Double, ORDER_BY_VALUES)]): ListBuffer[(EDGE_VALUES, Double, ORDER_BY_VALUES)] = {
- import com.kakao.s2graph.core.OrderingUtil._
-
- if (queryOption.withScore && queryOption.orderByColumns.nonEmpty) {
- val ascendingLs = queryOption.orderByColumns.map(_._2)
- rawEdges.sortBy(_._3)(TupleMultiOrdering[Any](ascendingLs))
- } else {
- rawEdges
- }
- }
-
- private def getColumnValue(keyWithJs: Map[String, JsValue], score: Double, edge: Edge, column: String): Any = {
- column match {
- case "score" => score
- case "timestamp" | "_timestamp" => edge.ts
- case _ =>
- keyWithJs.get(column) match {
- case None => keyWithJs.get("props").map { js => (js \ column).as[JsValue] }.get
- case Some(x) => x
- }
- }
- }
-
- private def buildReplaceJson(jsValue: JsValue)(mapper: JsValue => JsValue): JsValue = {
- def traverse(js: JsValue): JsValue = js match {
- case JsNull => mapper(JsNull)
- case JsUndefined() => mapper(JsUndefined(""))
- case JsNumber(v) => mapper(js)
- case JsString(v) => mapper(js)
- case JsBoolean(v) => mapper(js)
- case JsArray(elements) => JsArray(elements.map { t => traverse(mapper(t)) })
- case JsObject(values) => JsObject(values.map { case (k, v) => k -> traverse(mapper(v)) })
- }
-
- traverse(jsValue)
- }
-
- /** test query with filterOut is not working since it can not diffrentate filterOut */
- private def buildNextQuery(jsonQuery: JsValue, _cursors: Seq[Seq[String]]): JsValue = {
- val cursors = _cursors.flatten.iterator
-
- buildReplaceJson(jsonQuery) {
- case js@JsObject(fields) =>
- val isStep = fields.find { case (k, _) => k == "label" } // find label group
- if (isStep.isEmpty) js
- else {
- // TODO: Order not ensured
- val withCursor = js.fieldSet | Set("cursor" -> JsString(cursors.next))
- JsObject(withCursor.toSeq)
- }
- case js => js
- }
- }
-
- private def buildRawEdges(queryOption: QueryOption,
- queryRequestWithResultLs: Seq[QueryRequestWithResult],
- excludeIds: Map[Int, Boolean],
- scoreWeight: Double = 1.0): (ListBuffer[JsValue], ListBuffer[RAW_EDGE]) = {
- val degrees = ListBuffer[JsValue]()
- val rawEdges = ListBuffer[(EDGE_VALUES, Double, ORDER_BY_VALUES)]()
- val metaPropNamesMap = scala.collection.mutable.Map[String, Int]()
- for {
- queryRequestWithResult <- queryRequestWithResultLs
- } yield {
- val (queryRequest, _) = QueryRequestWithResult.unapply(queryRequestWithResult).get
- queryRequest.queryParam.label.metaPropNames.foreach { metaPropName =>
- metaPropNamesMap.put(metaPropName, metaPropNamesMap.getOrElse(metaPropName, 0) + 1)
- }
- }
- val propsExistInAll = metaPropNamesMap.filter(kv => kv._2 == queryRequestWithResultLs.length)
- val orderByColumns = queryOption.orderByColumns.filter { case (column, _) =>
- column match {
- case "from" | "to" | "label" | "score" | "timestamp" | "_timestamp" => true
- case _ =>
- propsExistInAll.contains(column)
-// //TODO??
-// false
-// queryParam.label.metaPropNames.contains(column)
- }
- }
-
- /** build result jsons */
- for {
- queryRequestWithResult <- queryRequestWithResultLs
- (queryRequest, queryResult) = QueryRequestWithResult.unapply(queryRequestWithResult).get
- queryParam = queryRequest.queryParam
- edgeWithScore <- queryResult.edgeWithScoreLs
- (edge, score) = EdgeWithScore.unapply(edgeWithScore).get
- hashKey = toHashKey(edge, queryRequest.queryParam, queryOption.filterOutFields)
- if !excludeIds.contains(hashKey)
- } {
-
- // edge to json
- val (srcColumn, _) = queryParam.label.srcTgtColumn(edge.labelWithDir.dir)
- val fromOpt = innerValToJsValue(edge.srcVertex.id.innerId, srcColumn.columnType)
- if (edge.isDegree && fromOpt.isDefined) {
- if (queryOption.limitOpt.isEmpty) {
- degrees += Json.obj(
- "from" -> fromOpt.get,
- "label" -> queryRequest.queryParam.label.label,
- "direction" -> GraphUtil.fromDirection(edge.labelWithDir.dir),
- LabelMeta.degree.name -> innerValToJsValue(edge.propsWithTs(LabelMeta.degreeSeq).innerVal, InnerVal.LONG)
- )
- }
- } else {
- val keyWithJs = edgeToJson(edge, score, queryRequest.query, queryRequest.queryParam)
- val orderByValues: (Any, Any, Any, Any) = orderByColumns.length match {
- case 0 =>
- (None, None, None, None)
- case 1 =>
- val it = orderByColumns.iterator
- val v1 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- (v1, None, None, None)
- case 2 =>
- val it = orderByColumns.iterator
- val v1 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- val v2 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- (v1, v2, None, None)
- case 3 =>
- val it = orderByColumns.iterator
- val v1 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- val v2 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- val v3 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- (v1, v2, v3, None)
- case _ =>
- val it = orderByColumns.iterator
- val v1 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- val v2 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- val v3 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- val v4 = getColumnValue(keyWithJs, score, edge, it.next()._1)
- (v1, v2, v3, v4)
- }
-
- val currentEdge = (keyWithJs, score * scoreWeight, orderByValues)
- rawEdges += currentEdge
- }
- }
- (degrees, rawEdges)
- }
-
- private def buildResultJsValue(queryOption: QueryOption,
- degrees: ListBuffer[JsValue],
- rawEdges: ListBuffer[RAW_EDGE]): JsValue = {
- if (queryOption.groupByColumns.isEmpty) {
- // ordering
- val filteredEdges = rawEdges.filter(t => t._2 >= queryOption.scoreThreshold)
-
- val edges = queryOption.limitOpt match {
- case None => orderBy(queryOption, filteredEdges).map(_._1)
- case Some(limit) => orderBy(queryOption, filteredEdges).map(_._1).take(limit)
- }
- val resultDegrees = if (queryOption.returnDegree) degrees else emptyDegrees
-
- Json.obj(
- "size" -> edges.size,
- "degrees" -> resultDegrees,
-// "queryNext" -> buildNextQuery(q.jsonQuery, q.cursorStrings()),
- "results" -> edges
- )
- } else {
- val grouped = rawEdges.groupBy { case (keyWithJs, _, _) =>
- val props = keyWithJs.get("props")
-
- for {
- column <- queryOption.groupByColumns
- value <- keyWithJs.get(column) match {
- case None => props.flatMap { js => (js \ column).asOpt[JsValue] }
- case Some(x) => Some(x)
- }
- } yield column -> value
- }
-
- val groupedEdgesWithScoreSum =
- for {
- (groupByKeyVals, groupedRawEdges) <- grouped
- scoreSum = groupedRawEdges.map(x => x._2).sum if scoreSum >= queryOption.scoreThreshold
- } yield {
- // ordering
- val edges = orderBy(queryOption, groupedRawEdges).map(_._1)
-
- //TODO: refactor this
- val js = if (queryOption.returnAgg)
- Json.obj(
- "groupBy" -> Json.toJson(groupByKeyVals.toMap),
- "scoreSum" -> scoreSum,
- "agg" -> edges
- )
- else
- Json.obj(
- "groupBy" -> Json.toJson(groupByKeyVals.toMap),
- "scoreSum" -> scoreSum,
- "agg" -> Json.arr()
- )
- (js, scoreSum)
- }
-
- val groupedSortedJsons = queryOption.limitOpt match {
- case None =>
- groupedEdgesWithScoreSum.toList.sortBy { case (jsVal, scoreSum) => scoreSum * -1 }.map(_._1)
- case Some(limit) =>
- groupedEdgesWithScoreSum.toList.sortBy { case (jsVal, scoreSum) => scoreSum * -1 }.map(_._1).take(limit)
- }
- val resultDegrees = if (queryOption.returnDegree) degrees else emptyDegrees
- Json.obj(
- "size" -> groupedSortedJsons.size,
- "degrees" -> resultDegrees,
-// "queryNext" -> buildNextQuery(q.jsonQuery, q.cursorStrings()),
- "results" -> groupedSortedJsons
- )
- }
- }
-
- def toSimpleVertexArrJsonMulti(queryOption: QueryOption,
- resultWithExcludeLs: Seq[(Seq[QueryRequestWithResult], Seq[QueryRequestWithResult])],
- excludes: Seq[QueryRequestWithResult]): JsValue = {
- val excludeIds = (Seq((Seq.empty, excludes)) ++ resultWithExcludeLs).foldLeft(Map.empty[Int, Boolean]) { case (acc, (result, excludes)) =>
- acc ++ resultInnerIds(excludes).map(hashKey => hashKey -> true).toMap
- }
-
- val (degrees, rawEdges) = (ListBuffer.empty[JsValue], ListBuffer.empty[RAW_EDGE])
- for {
- (result, localExclude) <- resultWithExcludeLs
- } {
- val newResult = result.map { queryRequestWithResult =>
- val (queryRequest, _) = QueryRequestWithResult.unapply(queryRequestWithResult).get
- val newQuery = queryRequest.query.copy(queryOption = queryOption)
- queryRequestWithResult.copy(queryRequest = queryRequest.copy(query = newQuery))
- }
- val (_degrees, _rawEdges) = buildRawEdges(queryOption, newResult, excludeIds)
- degrees ++= _degrees
- rawEdges ++= _rawEdges
- }
- buildResultJsValue(queryOption, degrees, rawEdges)
- }
-
- def toSimpleVertexArrJson(queryOption: QueryOption,
- queryRequestWithResultLs: Seq[QueryRequestWithResult],
- exclude: Seq[QueryRequestWithResult]): JsValue = {
- val excludeIds = resultInnerIds(exclude).map(innerId => innerId -> true).toMap
- val (degrees, rawEdges) = buildRawEdges(queryOption, queryRequestWithResultLs, excludeIds)
- buildResultJsValue(queryOption, degrees, rawEdges)
- }
-
-
- def toSimpleVertexArrJson(queryRequestWithResultLs: Seq[QueryRequestWithResult],
- exclude: Seq[QueryRequestWithResult]): JsValue = {
-
- queryRequestWithResultLs.headOption.map { queryRequestWithResult =>
- val (queryRequest, _) = QueryRequestWithResult.unapply(queryRequestWithResult).get
- val query = queryRequest.query
- val queryOption = query.queryOption
- val excludeIds = resultInnerIds(exclude).map(innerId => innerId -> true).toMap
- val (degrees, rawEdges) = buildRawEdges(queryOption, queryRequestWithResultLs, excludeIds)
- buildResultJsValue(queryOption, degrees, rawEdges)
- } getOrElse emptyResults
- }
-
- def verticesToJson(vertices: Iterable[Vertex]) = {
- Json.toJson(vertices.flatMap { v => vertexToJson(v) })
- }
-
- def propsToJson(edge: Edge, q: Query, queryParam: QueryParam): Map[String, JsValue] = {
- for {
- (seq, innerValWithTs) <- edge.propsWithTs if LabelMeta.isValidSeq(seq)
- labelMeta <- queryParam.label.metaPropsMap.get(seq)
- jsValue <- innerValToJsValue(innerValWithTs.innerVal, labelMeta.dataType)
- } yield labelMeta.name -> jsValue
- }
-
- private def edgeParent(parentEdges: Seq[EdgeWithScore], q: Query, queryParam: QueryParam): JsValue = {
- if (parentEdges.isEmpty) {
- JsNull
- } else {
- val parents = for {
- parent <- parentEdges
- (parentEdge, parentScore) = EdgeWithScore.unapply(parent).get
- parentQueryParam = QueryParam(parentEdge.labelWithDir)
- parents = edgeParent(parentEdge.parentEdges, q, parentQueryParam) if parents != JsNull
- } yield {
- val originalEdge = parentEdge.originalEdgeOpt.getOrElse(parentEdge)
- val edgeJson = edgeToJsonInner(originalEdge, parentScore, q, parentQueryParam) + ("parents" -> parents)
- Json.toJson(edgeJson)
- }
-
- Json.toJson(parents)
- }
- }
-
- /** TODO */
- def edgeToJsonInner(edge: Edge, score: Double, q: Query, queryParam: QueryParam): Map[String, JsValue] = {
- val (srcColumn, tgtColumn) = queryParam.label.srcTgtColumn(edge.labelWithDir.dir)
-
- val kvMapOpt = for {
- from <- innerValToJsValue(edge.srcVertex.id.innerId, srcColumn.columnType)
- to <- innerValToJsValue(edge.tgtVertex.id.innerId, tgtColumn.columnType)
- } yield {
- val targetColumns = if (q.selectColumnsSet.isEmpty) reservedColumns else (reservedColumns & q.selectColumnsSet) + "props"
- val _propsMap = queryParam.label.metaPropsDefaultMapInner ++ propsToJson(edge, q, queryParam)
- val propsMap = if (q.selectColumnsSet.nonEmpty) _propsMap.filterKeys(q.selectColumnsSet) else _propsMap
-
- val kvMap = targetColumns.foldLeft(Map.empty[String, JsValue]) { (map, column) =>
- val jsValue = column match {
- case "cacheRemain" => JsNumber(queryParam.cacheTTLInMillis - (System.currentTimeMillis() - queryParam.timestamp))
- case "from" => from
- case "to" => to
- case "label" => JsString(queryParam.label.label)
- case "direction" => JsString(GraphUtil.fromDirection(edge.labelWithDir.dir))
- case "_timestamp" | "timestamp" => JsNumber(edge.ts)
- case "score" => JsNumber(score)
- case "props" if propsMap.nonEmpty => Json.toJson(propsMap)
- case _ => JsNull
- }
-
- if (jsValue == JsNull) map else map + (column -> jsValue)
- }
- kvMap
- }
-
- kvMapOpt.getOrElse(Map.empty)
- }
-
- def edgeToJson(edge: Edge, score: Double, q: Query, queryParam: QueryParam): Map[String, JsValue] = {
- val kvs = edgeToJsonInner(edge, score, q, queryParam)
- if (kvs.nonEmpty && q.returnTree) kvs + ("parents" -> Json.toJson(edgeParent(edge.parentEdges, q, queryParam)))
- else kvs
- }
-
- def vertexToJson(vertex: Vertex): Option[JsObject] = {
- val serviceColumn = ServiceColumn.findById(vertex.id.colId)
-
- for {
- id <- innerValToJsValue(vertex.innerId, serviceColumn.columnType)
- } yield {
- Json.obj("serviceName" -> serviceColumn.service.serviceName,
- "columnName" -> serviceColumn.columnName,
- "id" -> id, "props" -> propsToJson(vertex),
- "timestamp" -> vertex.ts,
- // "belongsTo" -> vertex.belongLabelIds)
- "belongsTo" -> vertex.belongLabelIds.flatMap(Label.findByIdOpt(_).map(_.label)))
- }
- }
-
- private def keysToName(seqsToNames: Map[Int, String], props: Map[Int, InnerValLike]) = {
- for {
- (seq, value) <- props
- name <- seqsToNames.get(seq)
- } yield (name, value)
- }
-
- private def propsToJson(vertex: Vertex) = {
- val serviceColumn = vertex.serviceColumn
- val props = for {
- (propKey, innerVal) <- vertex.props
- columnMeta <- ColumnMeta.findByIdAndSeq(serviceColumn.id.get, propKey.toByte, useCache = true)
- jsValue <- innerValToJsValue(innerVal, columnMeta.dataType)
- } yield {
- (columnMeta.name -> jsValue)
- }
- props.toMap
- }
-
- @deprecated(message = "deprecated", since = "0.2")
- def propsToJson(edge: Edge) = {
- for {
- (seq, v) <- edge.propsWithTs if LabelMeta.isValidSeq(seq)
- metaProp <- edge.label.metaPropsMap.get(seq)
- jsValue <- innerValToJsValue(v.innerVal, metaProp.dataType)
- } yield {
- (metaProp.name, jsValue)
- }
- }
-
- @deprecated(message = "deprecated", since = "0.2")
- def summarizeWithListExclude(queryRequestWithResultLs: Seq[QueryRequestWithResult], exclude: Seq[QueryRequestWithResult]): JsObject = {
- val excludeIds = resultInnerIds(exclude).map(innerId => innerId -> true).toMap
-
-
- val groupedEdgesWithRank = (for {
- queryRequestWithResult <- queryRequestWithResultLs
- (queryRequest, queryResult) = QueryRequestWithResult.unapply(queryRequestWithResult).get
- edgeWithScore <- queryResult.edgeWithScoreLs
- (edge, score) = EdgeWithScore.unapply(edgeWithScore).get
- if !excludeIds.contains(toHashKey(edge, queryRequest.queryParam, queryRequest.query.filterOutFields))
- } yield {
- (edge, score)
- }).groupBy { case (edge, score) =>
- (edge.label.tgtColumn, edge.label.srcColumn, edge.tgtVertex.innerId)
- }
-
- val jsons = for {
- ((tgtColumn, srcColumn, target), edgesAndRanks) <- groupedEdgesWithRank
- (edges, ranks) = edgesAndRanks.groupBy(x => x._1.srcVertex).map(_._2.head).unzip
- tgtId <- innerValToJsValue(target, tgtColumn.columnType)
- } yield {
- Json.obj(tgtColumn.columnName -> tgtId,
- s"${srcColumn.columnName}s" ->
- edges.flatMap(edge => innerValToJsValue(edge.srcVertex.innerId, srcColumn.columnType)), "scoreSum" -> ranks.sum)
- }
- val sortedJsons = jsons.toList.sortBy { jsObj => (jsObj \ "scoreSum").as[Double] }.reverse
- if (queryRequestWithResultLs.isEmpty) {
- Json.obj("size" -> sortedJsons.size, "results" -> sortedJsons)
- } else {
- Json.obj("size" -> sortedJsons.size, "results" -> sortedJsons,
- "impressionId" -> queryRequestWithResultLs.head.queryRequest.query.impressionId())
- }
-
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/QueryParam.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/QueryParam.scala b/s2core/src/main/scala/com/kakao/s2graph/core/QueryParam.scala
deleted file mode 100644
index 0effa07..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/QueryParam.scala
+++ /dev/null
@@ -1,583 +0,0 @@
-package com.kakao.s2graph.core
-
-import com.google.common.hash.Hashing
-import com.kakao.s2graph.core.mysqls._
-import com.kakao.s2graph.core.parsers.{Where, WhereParser}
-import com.kakao.s2graph.core.types._
-import org.apache.hadoop.hbase.util.Bytes
-import org.hbase.async.ColumnRangeFilter
-import play.api.libs.json.{JsNull, JsNumber, JsValue, Json}
-
-import scala.util.hashing.MurmurHash3
-import scala.util.{Success, Try}
-
-object Query {
- val initialScore = 1.0
- lazy val empty = Query()
-
- def toQuery(srcVertices: Seq[Vertex], queryParam: QueryParam) = Query(srcVertices, Vector(Step(List(queryParam))))
-
- object DuplicatePolicy extends Enumeration {
- type DuplicatePolicy = Value
- val First, Sum, CountSum, Raw = Value
-
- def apply(policy: String): Value = {
- policy match {
- case "sum" => Query.DuplicatePolicy.Sum
- case "countSum" => Query.DuplicatePolicy.CountSum
- case "raw" => Query.DuplicatePolicy.Raw
- case _ => DuplicatePolicy.First
- }
- }
- }
-}
-
-case class MultiQuery(queries: Seq[Query],
- weights: Seq[Double],
- queryOption: QueryOption,
- jsonQuery: JsValue = JsNull)
-
-case class QueryOption(removeCycle: Boolean = false,
- selectColumns: Seq[String] = Seq.empty,
- groupByColumns: Seq[String] = Seq.empty,
- orderByColumns: Seq[(String, Boolean)] = Seq.empty,
- filterOutQuery: Option[Query] = None,
- filterOutFields: Seq[String] = Seq(LabelMeta.to.name),
- withScore: Boolean = true,
- returnTree: Boolean = false,
- limitOpt: Option[Int] = None,
- returnAgg: Boolean = true,
- scoreThreshold: Double = Double.MinValue,
- returnDegree: Boolean = true)
-
-
-case class Query(vertices: Seq[Vertex] = Seq.empty[Vertex],
- steps: IndexedSeq[Step] = Vector.empty[Step],
- queryOption: QueryOption = QueryOption(),
- jsonQuery: JsValue = JsNull) {
-
- val removeCycle = queryOption.removeCycle
- val selectColumns = queryOption.selectColumns
-// val groupBy = queryOption.groupBy
- val groupByColumns = queryOption.groupByColumns
- val orderByColumns = queryOption.orderByColumns
- val filterOutQuery = queryOption.filterOutQuery
- val filterOutFields = queryOption.filterOutFields
- val withScore = queryOption.withScore
- val returnTree = queryOption.returnTree
- val limitOpt = queryOption.limitOpt
- val returnAgg = queryOption.returnAgg
- val returnDegree = queryOption.returnDegree
-
- def cacheKeyBytes: Array[Byte] = {
- val selectBytes = Bytes.toBytes(queryOption.selectColumns.toString)
- val groupBytes = Bytes.toBytes(queryOption.groupByColumns.toString)
- val orderByBytes = Bytes.toBytes(queryOption.orderByColumns.toString)
- val filterOutBytes = queryOption.filterOutQuery.map(_.cacheKeyBytes).getOrElse(Array.empty[Byte])
- val returnTreeBytes = Bytes.toBytes(queryOption.returnTree)
-
- Seq(selectBytes, groupBytes, orderByBytes, filterOutBytes, returnTreeBytes).foldLeft(Array.empty[Byte])(Bytes.add)
- }
-
- lazy val selectColumnsSet = queryOption.selectColumns.map { c =>
- if (c == "_from") "from"
- else if (c == "_to") "to"
- else c
- }.toSet
-
- /** return logical query id without considering parameter values */
- def templateId(): JsValue = {
- Json.toJson(for {
- step <- steps
- queryParam <- step.queryParams.sortBy(_.labelWithDir.labelId)
- } yield {
- Json.obj("label" -> queryParam.label.label, "direction" -> GraphUtil.fromDirection(queryParam.labelWithDir.dir))
- })
- }
-
- def impressionId(): JsNumber = {
- val hash = MurmurHash3.stringHash(templateId().toString())
- JsNumber(hash)
- }
-
- def cursorStrings(): Seq[Seq[String]] = {
- //Don`t know how to replace all cursor keys in json
- steps.map { step =>
- step.queryParams.map { queryParam =>
- queryParam.cursorOpt.getOrElse("")
- }
- }
- }
-}
-
-object EdgeTransformer {
- val DefaultTransformField = Json.arr("_to")
- val DefaultTransformFieldAsList = Json.arr("_to").as[List[String]]
- val DefaultJson = Json.arr(DefaultTransformField)
-}
-
-/**
- * TODO: step wise outputFields should be used with nextStepLimit, nextStepThreshold.
- * @param jsValue
- */
-case class EdgeTransformer(queryParam: QueryParam, jsValue: JsValue) {
- val Delimiter = "\\$"
- val targets = jsValue.asOpt[List[Vector[String]]].toList
- val fieldsLs = for {
- target <- targets
- fields <- target
- } yield fields
- val isDefault = fieldsLs.size == 1 && fieldsLs.head.size == 1 && (fieldsLs.head.head == "_to" || fieldsLs.head.head == "to")
-
- def toHashKeyBytes: Array[Byte] = if (isDefault) Array.empty[Byte] else Bytes.toBytes(jsValue.toString)
-
- def replace(fmt: String,
- values: Seq[InnerValLike],
- nextStepOpt: Option[Step]): Seq[InnerValLike] = {
-
- val tokens = fmt.split(Delimiter)
- val _values = values.padTo(tokens.length, InnerVal.withStr("", queryParam.label.schemaVersion))
- val mergedStr = tokens.zip(_values).map { case (prefix, innerVal) => prefix + innerVal.toString }.mkString
- // logger.error(s"${tokens.toList}, ${values}, $mergedStr")
- // println(s"${tokens.toList}, ${values}, $mergedStr")
- nextStepOpt match {
- case None =>
- val columnType =
- if (queryParam.labelWithDir.dir == GraphUtil.directions("out")) queryParam.label.tgtColumnType
- else queryParam.label.srcColumnType
-
- if (columnType == InnerVal.STRING) Seq(InnerVal.withStr(mergedStr, queryParam.label.schemaVersion))
- else Nil
- case Some(nextStep) =>
- val nextQueryParamsValid = nextStep.queryParams.filter { qParam =>
- if (qParam.labelWithDir.dir == GraphUtil.directions("out")) qParam.label.srcColumnType == "string"
- else qParam.label.tgtColumnType == "string"
- }
- for {
- nextQueryParam <- nextQueryParamsValid
- } yield {
- InnerVal.withStr(mergedStr, nextQueryParam.label.schemaVersion)
- }
- }
- }
-
- def toInnerValOpt(edge: Edge, fieldName: String): Option[InnerValLike] = {
- fieldName match {
- case LabelMeta.to.name => Option(edge.tgtVertex.innerId)
- case LabelMeta.from.name => Option(edge.srcVertex.innerId)
- case _ =>
- for {
- labelMeta <- queryParam.label.metaPropsInvMap.get(fieldName)
- value <- edge.propsWithTs.get(labelMeta.seq)
- } yield value.innerVal
- }
- }
-
- def transform(edge: Edge, nextStepOpt: Option[Step]): Seq[Edge] = {
- if (isDefault) Seq(edge)
- else {
- val edges = for {
- fields <- fieldsLs
- innerVal <- {
- if (fields.size == 1) {
- val fieldName = fields.head
- toInnerValOpt(edge, fieldName).toSeq
- } else {
- val fmt +: fieldNames = fields
- replace(fmt, fieldNames.flatMap(fieldName => toInnerValOpt(edge, fieldName)), nextStepOpt)
- }
- }
- } yield edge.updateTgtVertex(innerVal).copy(originalEdgeOpt = Option(edge))
-
-
- edges
- }
- }
-}
-
-object Step {
- val Delimiter = "|"
-}
-
-case class Step(queryParams: List[QueryParam],
- labelWeights: Map[Int, Double] = Map.empty,
- // scoreThreshold: Double = 0.0,
- nextStepScoreThreshold: Double = 0.0,
- nextStepLimit: Int = -1,
- cacheTTL: Long = -1) {
-
- lazy val excludes = queryParams.filter(_.exclude)
- lazy val includes = queryParams.filterNot(_.exclude)
- lazy val excludeIds = excludes.map(x => x.labelWithDir.labelId -> true).toMap
-
- def toCacheKey(lss: Seq[Long]): Long = Hashing.murmur3_128().hashBytes(toCacheKeyRaw(lss)).asLong()
-// MurmurHash3.bytesHash(toCacheKeyRaw(lss))
-
- def toCacheKeyRaw(lss: Seq[Long]): Array[Byte] = {
- var bytes = Array.empty[Byte]
- lss.sorted.foreach { h => bytes = Bytes.add(bytes, Bytes.toBytes(h)) }
- bytes
- }
-}
-
-case class VertexParam(vertices: Seq[Vertex]) {
- var filters: Option[Map[Byte, InnerValLike]] = None
-
- def has(what: Option[Map[Byte, InnerValLike]]): VertexParam = {
- what match {
- case None => this
- case Some(w) => has(w)
- }
- }
-
- def has(what: Map[Byte, InnerValLike]): VertexParam = {
- this.filters = Some(what)
- this
- }
-
-}
-
-//object RankParam {
-// def apply(labelId: Int, keyAndWeights: Seq[(Byte, Double)]) = {
-// new RankParam(labelId, keyAndWeights)
-// }
-//}
-
-case class RankParam(labelId: Int, var keySeqAndWeights: Seq[(Byte, Double)] = Seq.empty[(Byte, Double)]) {
- // empty => Count
- lazy val rankKeysWeightsMap = keySeqAndWeights.toMap
-
- def defaultKey() = {
- this.keySeqAndWeights = List((LabelMeta.countSeq, 1.0))
- this
- }
-
- def toHashKeyBytes(): Array[Byte] = {
- var bytes = Array.empty[Byte]
- keySeqAndWeights.map { case (key, weight) =>
- bytes = Bytes.add(bytes, Array.fill(1)(key), Bytes.toBytes(weight))
- }
- bytes
- }
-}
-
-object QueryParam {
- lazy val Empty = QueryParam(LabelWithDirection(0, 0))
- lazy val DefaultThreshold = Double.MinValue
- val Delimiter = ","
-}
-
-case class QueryParam(labelWithDir: LabelWithDirection, timestamp: Long = System.currentTimeMillis()) {
-
- import HBaseSerializable._
- import Query.DuplicatePolicy
- import Query.DuplicatePolicy._
-
- lazy val label = Label.findById(labelWithDir.labelId)
- val DefaultKey = LabelIndex.DefaultSeq
- val fullKey = DefaultKey
-
- var labelOrderSeq = fullKey
-
- var sample = -1
- var limit = 10
- var offset = 0
- var rank = new RankParam(labelWithDir.labelId, List(LabelMeta.countSeq -> 1))
-
- var duration: Option[(Long, Long)] = None
- var isInverted: Boolean = false
-
- var columnRangeFilter: ColumnRangeFilter = null
-
- var hasFilters: Map[Byte, InnerValLike] = Map.empty[Byte, InnerValLike]
- var where: Try[Where] = Success(WhereParser.success)
- var whereRawOpt: Option[String] = None
- var duplicatePolicy = DuplicatePolicy.First
- var rpcTimeoutInMillis = 1000
- var maxAttempt = 2
- var includeDegree = false
- var tgtVertexInnerIdOpt: Option[InnerValLike] = None
- var cacheTTLInMillis: Long = -1L
- var threshold = QueryParam.DefaultThreshold
- var timeDecay: Option[TimeDecay] = None
- var transformer: EdgeTransformer = EdgeTransformer(this, EdgeTransformer.DefaultJson)
- var scorePropagateOp: String = "multiply"
- var exclude = false
- var include = false
- var shouldNormalize= false
- var cursorOpt: Option[String] = None
-
- var columnRangeFilterMinBytes = Array.empty[Byte]
- var columnRangeFilterMaxBytes = Array.empty[Byte]
-
- lazy val srcColumnWithDir = label.srcColumnWithDir(labelWithDir.dir)
- lazy val tgtColumnWithDir = label.tgtColumnWithDir(labelWithDir.dir)
-
- def toBytes(idxSeq: Byte, offset: Int, limit: Int, isInverted: Boolean): Array[Byte] = {
- val front = Array[Byte](idxSeq, if (isInverted) 1.toByte else 0.toByte)
- Bytes.add(front, Bytes.toBytes((offset.toLong << 32 | limit)))
- }
-
- /**
- * consider only I/O specific parameters.
- * properties that is used on Graph.filterEdges should not be considered.
- * @param bytes
- * @return
- */
- def toCacheKey(bytes: Array[Byte]): Long = {
- val hashBytes = toCacheKeyRaw(bytes)
- Hashing.murmur3_128().hashBytes(hashBytes).asLong()
-// MurmurHash3.bytesHash(hashBytes)
- }
-
- def toCacheKeyRaw(bytes: Array[Byte]): Array[Byte] = {
- val transformBytes = transformer.toHashKeyBytes
- //TODO: change this to binrary format.
- val whereBytes = Bytes.toBytes(whereRawOpt.getOrElse(""))
- val durationBytes = duration.map { case (min, max) =>
- val minTs = min / cacheTTLInMillis
- val maxTs = max / cacheTTLInMillis
- Bytes.add(Bytes.toBytes(minTs), Bytes.toBytes(maxTs))
- } getOrElse Array.empty[Byte]
-// Bytes.toBytes(duration.toString)
- val conditionBytes = Bytes.add(transformBytes, whereBytes, durationBytes)
- Bytes.add(Bytes.add(bytes, labelWithDir.bytes, toBytes(labelOrderSeq, offset, limit, isInverted)), rank.toHashKeyBytes(),
- Bytes.add(columnRangeFilterMinBytes, columnRangeFilterMaxBytes, conditionBytes))
- }
-
- def isInverted(isInverted: Boolean): QueryParam = {
- this.isInverted = isInverted
- this
- }
-
- def labelOrderSeq(labelOrderSeq: Byte): QueryParam = {
- this.labelOrderSeq = labelOrderSeq
- this
- }
-
- def sample(n: Int): QueryParam = {
- this.sample = n
- this
- }
-
- def limit(offset: Int, limit: Int): QueryParam = {
- /** since degree info is located on first always */
- if (offset == 0 && this.columnRangeFilter == null) {
- this.limit = limit + 1
- this.offset = offset
- } else {
- this.limit = limit
- this.offset = offset + 1
- }
- // this.columnPaginationFilter = new ColumnPaginationFilter(this.limit, this.offset)
- this
- }
-
- def interval(fromTo: Option[(Seq[(Byte, InnerValLike)], Seq[(Byte, InnerValLike)])]): QueryParam = {
- fromTo match {
- case Some((from, to)) => interval(from, to)
- case _ => this
- }
- }
-
- def interval(from: Seq[(Byte, InnerValLike)], to: Seq[(Byte, InnerValLike)]): QueryParam = {
- // val len = label.orderTypes.size.toByte
- // val len = label.extraIndicesMap(labelOrderSeq).sortKeyTypes.size.toByte
- // logger.error(s"indicesMap: ${label.indicesMap(labelOrderSeq)}")
- val len = label.indicesMap(labelOrderSeq).sortKeyTypes.size.toByte
-
- val minMetaByte = InnerVal.minMetaByte
- // val maxMetaByte = InnerVal.maxMetaByte
- val maxMetaByte = -1.toByte
- val toVal = Bytes.add(propsToBytes(to), Array.fill(1)(minMetaByte))
- //FIXME
- val fromVal = Bytes.add(propsToBytes(from), Array.fill(10)(maxMetaByte))
- toVal(0) = len
- fromVal(0) = len
- val maxBytes = fromVal
- val minBytes = toVal
- this.columnRangeFilterMaxBytes = maxBytes
- this.columnRangeFilterMinBytes = minBytes
- val rangeFilter = new ColumnRangeFilter(minBytes, true, maxBytes, true)
- this.columnRangeFilter = rangeFilter
- this
- }
-
- def duration(minMaxTs: Option[(Long, Long)]): QueryParam = {
- minMaxTs match {
- case Some((minTs, maxTs)) => duration(minTs, maxTs)
- case _ => this
- }
- }
-
- def duration(minTs: Long, maxTs: Long): QueryParam = {
- this.duration = Some((minTs, maxTs))
- this
- }
-
- def rank(r: RankParam): QueryParam = {
- this.rank = r
- this
- }
-
- def exclude(filterOut: Boolean): QueryParam = {
- this.exclude = filterOut
- this
- }
-
- def include(filterIn: Boolean): QueryParam = {
- this.include = filterIn
- this
- }
-
- def has(hasFilters: Map[Byte, InnerValLike]): QueryParam = {
- this.hasFilters = hasFilters
- this
- }
-
- def where(whereTry: Try[Where]): QueryParam = {
- this.where = whereTry
- this
- }
-
- def duplicatePolicy(policy: Option[DuplicatePolicy]): QueryParam = {
- this.duplicatePolicy = policy.getOrElse(DuplicatePolicy.First)
- this
- }
-
- def rpcTimeout(millis: Int): QueryParam = {
- this.rpcTimeoutInMillis = millis
- this
- }
-
- def maxAttempt(attempt: Int): QueryParam = {
- this.maxAttempt = attempt
- this
- }
-
- def includeDegree(includeDegree: Boolean): QueryParam = {
- this.includeDegree = includeDegree
- this
- }
-
- def tgtVertexInnerIdOpt(other: Option[InnerValLike]): QueryParam = {
- this.tgtVertexInnerIdOpt = other
- this
- }
-
- def cacheTTLInMillis(other: Long): QueryParam = {
- this.cacheTTLInMillis = other
- this
- }
-
- def timeDecay(other: Option[TimeDecay]): QueryParam = {
- this.timeDecay = other
- this
- }
-
- def threshold(other: Double): QueryParam = {
- this.threshold = other
- this
- }
-
- def transformer(other: Option[JsValue]): QueryParam = {
- other match {
- case Some(js) => this.transformer = EdgeTransformer(this, js)
- case None =>
- }
- this
- }
-
- def scorePropagateOp(scorePropagateOp: String): QueryParam = {
- this.scorePropagateOp = scorePropagateOp
- this
- }
-
- def shouldNormalize(shouldNormalize: Boolean): QueryParam = {
- this.shouldNormalize = shouldNormalize
- this
- }
-
- def whereRawOpt(sqlOpt: Option[String]): QueryParam = {
- this.whereRawOpt = sqlOpt
- this
- }
-
- def cursorOpt(cursorOpt: Option[String]): QueryParam = {
- this.cursorOpt = cursorOpt
- this
- }
-
- def isSnapshotEdge = tgtVertexInnerIdOpt.isDefined
-
- override def toString = {
- List(label.label, labelOrderSeq, offset, limit, rank,
- duration, isInverted, exclude, include, hasFilters).mkString("\t")
- // duration, isInverted, exclude, include, hasFilters, outputFields).mkString("\t")
- }
-
- //
- // def buildGetRequest(srcVertex: Vertex) = {
- // val (srcColumn, tgtColumn) = label.srcTgtColumn(labelWithDir.dir)
- // val (srcInnerId, tgtInnerId) = tgtVertexInnerIdOpt match {
- // case Some(tgtVertexInnerId) => // _to is given.
- // /** we use toInvertedEdgeHashLike so dont need to swap src, tgt */
- // val src = InnerVal.convertVersion(srcVertex.innerId, srcColumn.columnType, label.schemaVersion)
- // val tgt = InnerVal.convertVersion(tgtVertexInnerId, tgtColumn.columnType, label.schemaVersion)
- // (src, tgt)
- // case None =>
- // val src = InnerVal.convertVersion(srcVertex.innerId, srcColumn.columnType, label.schemaVersion)
- // (src, src)
- // }
- //
- // val (srcVId, tgtVId) = (SourceVertexId(srcColumn.id.get, srcInnerId), TargetVertexId(tgtColumn.id.get, tgtInnerId))
- // val (srcV, tgtV) = (Vertex(srcVId), Vertex(tgtVId))
- // val edge = Edge(srcV, tgtV, labelWithDir)
- //
- // val get = if (tgtVertexInnerIdOpt.isDefined) {
- // val snapshotEdge = edge.toInvertedEdgeHashLike
- // val kv = snapshotEdge.kvs.head
- // new GetRequest(label.hbaseTableName.getBytes, kv.row, edgeCf, kv.qualifier)
- // } else {
- // val indexedEdgeOpt = edge.edgesWithIndex.find(e => e.labelIndexSeq == labelOrderSeq)
- // assert(indexedEdgeOpt.isDefined)
- // val indexedEdge = indexedEdgeOpt.get
- // val kv = indexedEdge.kvs.head
- // val table = label.hbaseTableName.getBytes
- // //kv.table //
- // val rowKey = kv.row // indexedEdge.rowKey.bytes
- // val cf = edgeCf
- // new GetRequest(table, rowKey, cf)
- // }
- //
- // val (minTs, maxTs) = duration.getOrElse((0L, Long.MaxValue))
- //
- // get.maxVersions(1)
- // get.setFailfast(true)
- // get.setMaxResultsPerColumnFamily(limit)
- // get.setRowOffsetPerColumnFamily(offset)
- // get.setMinTimestamp(minTs)
- // get.setMaxTimestamp(maxTs)
- // get.setTimeout(rpcTimeoutInMillis)
- // if (columnRangeFilter != null) get.setFilter(columnRangeFilter)
- // // get.setMaxAttempt(maxAttempt.toByte)
- // // get.setRpcTimeout(rpcTimeoutInMillis)
- //
- // // if (columnRangeFilter != null) get.filter(columnRangeFilter)
- // // logger.debug(s"Get: $get, $offset, $limit")
- //
- // get
- // }
-}
-
-case class TimeDecay(initial: Double = 1.0,
- lambda: Double = 0.1,
- timeUnit: Double = 60 * 60 * 24,
- labelMetaSeq: Byte = LabelMeta.timeStampSeq) {
- def decay(diff: Double): Double = {
- //FIXME
- val ret = initial * Math.pow(1.0 - lambda, diff / timeUnit)
- // logger.debug(s"$initial, $lambda, $timeUnit, $diff, ${diff / timeUnit}, $ret")
- ret
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/QueryResult.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/QueryResult.scala b/s2core/src/main/scala/com/kakao/s2graph/core/QueryResult.scala
deleted file mode 100644
index 02d9736..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/QueryResult.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-package com.kakao.s2graph.core
-
-import com.kakao.s2graph.core.mysqls.LabelMeta
-import com.kakao.s2graph.core.types.{InnerVal, InnerValLikeWithTs}
-
-import scala.collection.Seq
-
-object QueryResult {
- def fromVertices(query: Query): Seq[QueryRequestWithResult] = {
- if (query.steps.isEmpty || query.steps.head.queryParams.isEmpty) {
- Seq.empty
- } else {
- val queryParam = query.steps.head.queryParams.head
- val label = queryParam.label
- val currentTs = System.currentTimeMillis()
- val propsWithTs = Map(LabelMeta.timeStampSeq ->
- InnerValLikeWithTs(InnerVal.withLong(currentTs, label.schemaVersion), currentTs))
- for {
- vertex <- query.vertices
- } yield {
- val edge = Edge(vertex, vertex, queryParam.labelWithDir, propsWithTs = propsWithTs)
- val edgeWithScore = EdgeWithScore(edge, Graph.DefaultScore)
- QueryRequestWithResult(QueryRequest(query, -1, vertex, queryParam),
- QueryResult(edgeWithScoreLs = Seq(edgeWithScore)))
- }
- }
- }
-}
-case class QueryRequestWithResult(queryRequest: QueryRequest, queryResult: QueryResult)
-
-case class QueryRequest(query: Query,
- stepIdx: Int,
- vertex: Vertex,
- queryParam: QueryParam)
-
-
-case class QueryResult(edgeWithScoreLs: Seq[EdgeWithScore] = Nil,
- tailCursor: Array[Byte] = Array.empty,
- timestamp: Long = System.currentTimeMillis(),
- isFailure: Boolean = false)
-
-case class EdgeWithScore(edge: Edge, score: Double)
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/Vertex.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/Vertex.scala b/s2core/src/main/scala/com/kakao/s2graph/core/Vertex.scala
deleted file mode 100644
index c2b86c2..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/Vertex.scala
+++ /dev/null
@@ -1,107 +0,0 @@
-package com.kakao.s2graph.core
-
-
-import com.kakao.s2graph.core.mysqls._
-
-//import com.kakao.s2graph.core.models._
-
-import com.kakao.s2graph.core.types._
-import play.api.libs.json.Json
-
-/**
- */
-case class Vertex(id: VertexId,
- ts: Long = System.currentTimeMillis(),
- props: Map[Int, InnerValLike] = Map.empty[Int, InnerValLike],
- op: Byte = 0,
- belongLabelIds: Seq[Int] = Seq.empty) extends GraphElement {
-
- val innerId = id.innerId
-
- def schemaVer = serviceColumn.schemaVersion
-
- def serviceColumn = ServiceColumn.findById(id.colId)
-
- def service = Service.findById(serviceColumn.serviceId)
-
- lazy val (hbaseZkAddr, hbaseTableName) = (service.cluster, service.hTableName)
-
- def defaultProps = Map(ColumnMeta.lastModifiedAtColumnSeq.toInt -> InnerVal.withLong(ts, schemaVer))
-
- // lazy val kvs = Graph.client.vertexSerializer(this).toKeyValues
-
- /** TODO: make this as configurable */
- override def serviceName = service.serviceName
-
- override def isAsync = false
-
- override def queueKey = Seq(ts.toString, serviceName).mkString("|")
-
- override def queuePartitionKey = id.innerId.toString
-
- def propsWithName = for {
- (seq, v) <- props
- meta <- ColumnMeta.findByIdAndSeq(id.colId, seq.toByte)
- } yield (meta.name -> v.toString)
-
- // /** only used by bulk loader */
- // def buildPuts(): List[Put] = {
- // // logger.error(s"put: $this => $rowKey")
- //// val put = new Put(rowKey.bytes)
- //// for ((q, v) <- qualifiersWithValues) {
- //// put.addColumn(vertexCf, q, ts, v)
- //// }
- //// List(put)
- // val kv = kvs.head
- // val put = new Put(kv.row)
- // kvs.map { kv =>
- // put.addColumn(kv.cf, kv.qualifier, kv.timestamp, kv.value)
- // }
- // List(put)
- // }
-
- def toEdgeVertex() = Vertex(SourceVertexId(id.colId, innerId), ts, props, op)
-
-
- override def hashCode() = {
- val hash = id.hashCode()
- // logger.debug(s"Vertex.hashCode: $this -> $hash")
- hash
- }
-
- override def equals(obj: Any) = {
- obj match {
- case otherVertex: Vertex =>
- val ret = id == otherVertex.id
- // logger.debug(s"Vertex.equals: $this, $obj => $ret")
- ret
- case _ => false
- }
- }
-
- def withProps(newProps: Map[Int, InnerValLike]) = Vertex(id, ts, newProps, op)
-
- def toLogString(): String = {
- val (serviceName, columnName) =
- if (!id.storeColId) ("", "")
- else (serviceColumn.service.serviceName, serviceColumn.columnName)
-
- if (propsWithName.nonEmpty)
- Seq(ts, GraphUtil.fromOp(op), "v", id.innerId, serviceName, columnName, Json.toJson(propsWithName)).mkString("\t")
- else
- Seq(ts, GraphUtil.fromOp(op), "v", id.innerId, serviceName, columnName).mkString("\t")
- }
-}
-
-object Vertex {
-
- def toPropKey(labelId: Int): Int = Byte.MaxValue + labelId
-
- def toLabelId(propKey: Int): Int = propKey - Byte.MaxValue
-
- def isLabelId(propKey: Int): Boolean = propKey > Byte.MaxValue
-
- // val emptyVertex = Vertex(new CompositeId(CompositeId.defaultColId, CompositeId.defaultInnerId, false, true),
- // System.currentTimeMillis())
- def fromString(s: String): Option[Vertex] = Graph.toVertex(s)
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Bucket.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Bucket.scala b/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Bucket.scala
deleted file mode 100644
index edc6147..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Bucket.scala
+++ /dev/null
@@ -1,66 +0,0 @@
-package com.kakao.s2graph.core.mysqls
-
-/**
- * Created by shon on 8/5/15.
- */
-
-import scalikejdbc._
-
-object Bucket extends Model[Bucket] {
-
- val rangeDelimiter = "~"
- val INVALID_BUCKET_EXCEPTION = new RuntimeException("invalid bucket.")
-
- def apply(rs: WrappedResultSet): Bucket = {
- Bucket(rs.intOpt("id"),
- rs.int("experiment_id"),
- rs.string("modular"),
- rs.string("http_verb"),
- rs.string("api_path"),
- rs.string("request_body"),
- rs.int("timeout"),
- rs.string("impression_id"),
- rs.boolean("is_graph_query"),
- rs.boolean("is_empty"))
- }
-
- def finds(experimentId: Int)(implicit session: DBSession = AutoSession): List[Bucket] = {
- val cacheKey = "experimentId=" + experimentId
- withCaches(cacheKey) {
- sql"""select * from buckets where experiment_id = $experimentId"""
- .map { rs => Bucket(rs) }.list().apply()
- }
- }
-
- def toRange(str: String): Option[(Int, Int)] = {
- val range = str.split(rangeDelimiter)
- if (range.length == 2) Option(range.head.toInt, range.last.toInt)
- else None
- }
-
- def findByImpressionId(impressionId: String, useCache: Boolean = true)(implicit session: DBSession = AutoSession): Option[Bucket] = {
- val cacheKey = "impressionId=" + impressionId
- val sql = sql"""select * from buckets where impression_id=$impressionId"""
- .map { rs => Bucket(rs)}
- if (useCache) {
- withCache(cacheKey) {
- sql.single().apply()
- }
- } else {
- sql.single().apply()
- }
- }
-}
-
-case class Bucket(id: Option[Int],
- experimentId: Int,
- modular: String,
- httpVerb: String, apiPath: String,
- requestBody: String, timeout: Int, impressionId: String,
- isGraphQuery: Boolean = true,
- isEmpty: Boolean = false) {
-
- import Bucket._
-
- lazy val rangeOpt = toRange(modular)
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/ColumnMeta.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/ColumnMeta.scala b/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/ColumnMeta.scala
deleted file mode 100644
index 1b6d55f..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/ColumnMeta.scala
+++ /dev/null
@@ -1,110 +0,0 @@
-package com.kakao.s2graph.core.mysqls
-
-import play.api.libs.json.Json
-import scalikejdbc._
-
-object ColumnMeta extends Model[ColumnMeta] {
-
- val timeStampSeq = 0.toByte
- val countSeq = -1.toByte
- val lastModifiedAtColumnSeq = 0.toByte
- val lastModifiedAtColumn = ColumnMeta(Some(0), 0, "lastModifiedAt", lastModifiedAtColumnSeq, "long")
- val maxValue = Byte.MaxValue
-
- def apply(rs: WrappedResultSet): ColumnMeta = {
- ColumnMeta(Some(rs.int("id")), rs.int("column_id"), rs.string("name"), rs.byte("seq"), rs.string("data_type").toLowerCase())
- }
-
- def findById(id: Int)(implicit session: DBSession = AutoSession) = {
- // val cacheKey = s"id=$id"
- val cacheKey = "id=" + id
- withCache(cacheKey) {
- sql"""select * from column_metas where id = ${id}""".map { rs => ColumnMeta(rs) }.single.apply
- }.get
- }
-
- def findAllByColumn(columnId: Int, useCache: Boolean = true)(implicit session: DBSession = AutoSession) = {
- // val cacheKey = s"columnId=$columnId"
- val cacheKey = "columnId=" + columnId
- if (useCache) {
- withCaches(cacheKey)( sql"""select *from column_metas where column_id = ${columnId} order by seq ASC"""
- .map { rs => ColumnMeta(rs) }.list.apply())
- } else {
- sql"""select * from column_metas where column_id = ${columnId} order by seq ASC"""
- .map { rs => ColumnMeta(rs) }.list.apply()
- }
- }
-
- def findByName(columnId: Int, name: String)(implicit session: DBSession = AutoSession) = {
- // val cacheKey = s"columnId=$columnId:name=$name"
- val cacheKey = "columnId=" + columnId + ":name=" + name
- withCache(cacheKey)( sql"""select * from column_metas where column_id = ${columnId} and name = ${name}"""
- .map { rs => ColumnMeta(rs) }.single.apply())
- }
-
- def insert(columnId: Int, name: String, dataType: String)(implicit session: DBSession = AutoSession) = {
- val ls = findAllByColumn(columnId, false)
- val seq = ls.size + 1
- if (seq <= maxValue) {
- sql"""insert into column_metas(column_id, name, seq, data_type)
- select ${columnId}, ${name}, ${seq}, ${dataType}"""
- .updateAndReturnGeneratedKey.apply()
- }
- }
-
- def findOrInsert(columnId: Int, name: String, dataType: String)(implicit session: DBSession = AutoSession): ColumnMeta = {
- findByName(columnId, name) match {
- case Some(c) => c
- case None =>
- insert(columnId, name, dataType)
- expireCache(s"columnId=$columnId:name=$name")
- findByName(columnId, name).get
- }
- }
-
- def findByIdAndSeq(columnId: Int, seq: Byte, useCache: Boolean = true)(implicit session: DBSession = AutoSession) = {
- val cacheKey = "columnId=" + columnId + ":seq=" + seq
- lazy val columnMetaOpt = sql"""
- select * from column_metas where column_id = ${columnId} and seq = ${seq}
- """.map { rs => ColumnMeta(rs) }.single.apply()
-
- if (useCache) withCache(cacheKey)(columnMetaOpt)
- else columnMetaOpt
- }
-
- def delete(id: Int)(implicit session: DBSession = AutoSession) = {
- val columnMeta = findById(id)
- val (columnId, name) = (columnMeta.columnId, columnMeta.name)
- sql"""delete from column_metas where id = ${id}""".execute.apply()
- val cacheKeys = List(s"id=$id", s"columnId=$columnId:name=$name", s"colunmId=$columnId")
- cacheKeys.foreach { key =>
- expireCache(key)
- expireCaches(key)
- }
- }
-
- def findAll()(implicit session: DBSession = AutoSession) = {
- val ls = sql"""select * from column_metas""".map { rs => ColumnMeta(rs) }.list().apply()
-
- putsToCache(ls.map { x =>
- val cacheKey = s"id=${x.id.get}"
- (cacheKey -> x)
- })
- putsToCache(ls.map { x =>
- val cacheKey = s"columnId=${x.columnId}:name=${x.name}"
- (cacheKey -> x)
- })
- putsToCache(ls.map { x =>
- val cacheKey = s"columnId=${x.columnId}:seq=${x.seq}"
- (cacheKey -> x)
- })
- putsToCaches(ls.groupBy(x => x.columnId).map { case (columnId, ls) =>
- val cacheKey = s"columnId=${columnId}"
- (cacheKey -> ls)
- }.toList)
- }
-}
-
-case class ColumnMeta(id: Option[Int], columnId: Int, name: String, seq: Byte, dataType: String) {
- lazy val toJson = Json.obj("name" -> name, "dataType" -> dataType)
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Experiment.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Experiment.scala b/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Experiment.scala
deleted file mode 100644
index 46b92ab..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Experiment.scala
+++ /dev/null
@@ -1,78 +0,0 @@
-package com.kakao.s2graph.core.mysqls
-
-import com.kakao.s2graph.core.GraphUtil
-import com.kakao.s2graph.core.utils.logger
-import scalikejdbc._
-
-import scala.util.Random
-
-object Experiment extends Model[Experiment] {
- val impressionKey = "S2-Impression-Id"
-
- def apply(rs: WrappedResultSet): Experiment = {
- Experiment(rs.intOpt("id"),
- rs.int("service_id"),
- rs.string("name"),
- rs.string("description"),
- rs.string("experiment_type"),
- rs.int("total_modular"))
- }
-
- def finds(serviceId: Int)(implicit session: DBSession = AutoSession): List[Experiment] = {
- val cacheKey = "serviceId=" + serviceId
- withCaches(cacheKey) {
- sql"""select * from experiments where service_id = ${serviceId}"""
- .map { rs => Experiment(rs) }.list().apply()
- }
- }
-
- def findBy(serviceId: Int, name: String)(implicit session: DBSession = AutoSession): Option[Experiment] = {
- val cacheKey = "serviceId=" + serviceId + ":name=" + name
- withCache(cacheKey) {
- sql"""select * from experiments where service_id = ${serviceId} and name = ${name}"""
- .map { rs => Experiment(rs) }.single.apply
- }
- }
-
- def findById(id: Int)(implicit session: DBSession = AutoSession): Option[Experiment] = {
- val cacheKey = "id=" + id
- withCache(cacheKey)(
- sql"""select * from experiments where id = ${id}"""
- .map { rs => Experiment(rs) }.single.apply
- )
- }
-}
-
-case class Experiment(id: Option[Int],
- serviceId: Int,
- name: String,
- description: String,
- experimentType: String,
- totalModular: Int) {
-
- def buckets = Bucket.finds(id.get)
-
- def rangeBuckets = for {
- bucket <- buckets
- range <- bucket.rangeOpt
- } yield range -> bucket
-
-
- def findBucket(uuid: String, impIdOpt: Option[String] = None): Option[Bucket] = {
- impIdOpt match {
- case Some(impId) => Bucket.findByImpressionId(impId)
- case None =>
- val seed = experimentType match {
- case "u" => (GraphUtil.murmur3(uuid) % totalModular) + 1
- case _ => Random.nextInt(totalModular) + 1
- }
- findBucket(seed)
- }
- }
-
- def findBucket(uuidMod: Int): Option[Bucket] = {
- rangeBuckets.find { case ((from, to), bucket) =>
- from <= uuidMod && uuidMod <= to
- }.map(_._2)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Label.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Label.scala b/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Label.scala
deleted file mode 100644
index 005a01e..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/Label.scala
+++ /dev/null
@@ -1,372 +0,0 @@
-package com.kakao.s2graph.core.mysqls
-
-import com.kakao.s2graph.core.GraphExceptions.ModelNotFoundException
-import com.kakao.s2graph.core.Management.JsonModel.{Index, Prop}
-import com.kakao.s2graph.core.utils.logger
-import com.kakao.s2graph.core.{GraphUtil, JSONParser, Management}
-import play.api.libs.json.Json
-import scalikejdbc._
-
-object Label extends Model[Label] {
-
- val maxHBaseTableNames = 2
-
- def apply(rs: WrappedResultSet): Label = {
- Label(Option(rs.int("id")), rs.string("label"),
- rs.int("src_service_id"), rs.string("src_column_name"), rs.string("src_column_type"),
- rs.int("tgt_service_id"), rs.string("tgt_column_name"), rs.string("tgt_column_type"),
- rs.boolean("is_directed"), rs.string("service_name"), rs.int("service_id"), rs.string("consistency_level"),
- rs.string("hbase_table_name"), rs.intOpt("hbase_table_ttl"), rs.string("schema_version"), rs.boolean("is_async"), rs.string("compressionAlgorithm"))
- }
-
- def deleteAll(label: Label)(implicit session: DBSession) = {
- val id = label.id
- LabelMeta.findAllByLabelId(id.get, false).foreach { x => LabelMeta.delete(x.id.get) }
- LabelIndex.findByLabelIdAll(id.get, false).foreach { x => LabelIndex.delete(x.id.get) }
- Label.delete(id.get)
- }
-
- def findByName(labelName: String, useCache: Boolean = true)(implicit session: DBSession = AutoSession): Option[Label] = {
- val cacheKey = "label=" + labelName
- lazy val labelOpt =
- sql"""
- select *
- from labels
- where label = ${labelName}""".map { rs => Label(rs) }.single.apply()
-
- if (useCache) withCache(cacheKey)(labelOpt)
- else labelOpt
- }
-
- def insert(label: String,
- srcServiceId: Int,
- srcColumnName: String,
- srcColumnType: String,
- tgtServiceId: Int,
- tgtColumnName: String,
- tgtColumnType: String,
- isDirected: Boolean,
- serviceName: String,
- serviceId: Int,
- consistencyLevel: String,
- hTableName: String,
- hTableTTL: Option[Int],
- schemaVersion: String,
- isAsync: Boolean,
- compressionAlgorithm: String)(implicit session: DBSession = AutoSession) = {
- sql"""
- insert into labels(label,
- src_service_id, src_column_name, src_column_type,
- tgt_service_id, tgt_column_name, tgt_column_type,
- is_directed, service_name, service_id, consistency_level, hbase_table_name, hbase_table_ttl, schema_version, is_async, compressionAlgorithm)
- values (${label},
- ${srcServiceId}, ${srcColumnName}, ${srcColumnType},
- ${tgtServiceId}, ${tgtColumnName}, ${tgtColumnType},
- ${isDirected}, ${serviceName}, ${serviceId}, ${consistencyLevel}, ${hTableName}, ${hTableTTL},
- ${schemaVersion}, ${isAsync}, ${compressionAlgorithm})
- """
- .updateAndReturnGeneratedKey.apply()
- }
-
- def findByIdOpt(id: Int)(implicit session: DBSession = AutoSession): Option[Label] = {
- val cacheKey = "id=" + id
- withCache(cacheKey)(
- sql"""
- select *
- from labels
- where id = ${id}"""
- .map { rs => Label(rs) }.single.apply())
- }
-
- def findById(id: Int)(implicit session: DBSession = AutoSession): Label = {
- val cacheKey = "id=" + id
- withCache(cacheKey)(
- sql"""
- select *
- from labels
- where id = ${id}"""
- .map { rs => Label(rs) }.single.apply()).get
- }
-
- def findByTgtColumnId(columnId: Int)(implicit session: DBSession = AutoSession): List[Label] = {
- val cacheKey = "tgtColumnId=" + columnId
- val col = ServiceColumn.findById(columnId)
- withCaches(cacheKey)(
- sql"""
- select *
- from labels
- where tgt_column_name = ${col.columnName}
- and service_id = ${col.serviceId}
- """.map { rs => Label(rs) }.list().apply())
- }
-
- def findBySrcColumnId(columnId: Int)(implicit session: DBSession = AutoSession): List[Label] = {
- val cacheKey = "srcColumnId=" + columnId
- val col = ServiceColumn.findById(columnId)
- withCaches(cacheKey)(
- sql"""
- select *
- from labels
- where src_column_name = ${col.columnName}
- and service_id = ${col.serviceId}
- """.map { rs => Label(rs) }.list().apply())
- }
-
- def findBySrcServiceId(serviceId: Int)(implicit session: DBSession = AutoSession): List[Label] = {
- val cacheKey = "srcServiceId=" + serviceId
- withCaches(cacheKey)(
- sql"""select * from labels where src_service_id = ${serviceId}""".map { rs => Label(rs) }.list().apply
- )
- }
-
- def findByTgtServiceId(serviceId: Int)(implicit session: DBSession = AutoSession): List[Label] = {
- val cacheKey = "tgtServiceId=" + serviceId
- withCaches(cacheKey)(
- sql"""select * from labels where tgt_service_id = ${serviceId}""".map { rs => Label(rs) }.list().apply
- )
- }
-
- def insertAll(labelName: String, srcServiceName: String, srcColumnName: String, srcColumnType: String,
- tgtServiceName: String, tgtColumnName: String, tgtColumnType: String,
- isDirected: Boolean = true,
- serviceName: String,
- indices: Seq[Index],
- metaProps: Seq[Prop],
- consistencyLevel: String,
- hTableName: Option[String],
- hTableTTL: Option[Int],
- schemaVersion: String,
- isAsync: Boolean,
- compressionAlgorithm: String)(implicit session: DBSession = AutoSession): Label = {
-
- val srcServiceOpt = Service.findByName(srcServiceName, useCache = false)
- val tgtServiceOpt = Service.findByName(tgtServiceName, useCache = false)
- val serviceOpt = Service.findByName(serviceName, useCache = false)
- if (srcServiceOpt.isEmpty) throw new RuntimeException(s"source service $srcServiceName is not created.")
- if (tgtServiceOpt.isEmpty) throw new RuntimeException(s"target service $tgtServiceName is not created.")
- if (serviceOpt.isEmpty) throw new RuntimeException(s"service $serviceName is not created.")
-
- val newLabel = for {
- srcService <- srcServiceOpt
- tgtService <- tgtServiceOpt
- service <- serviceOpt
- } yield {
- val srcServiceId = srcService.id.get
- val tgtServiceId = tgtService.id.get
- val serviceId = service.id.get
-
- /** insert serviceColumn */
- val srcCol = ServiceColumn.findOrInsert(srcServiceId, srcColumnName, Some(srcColumnType), schemaVersion)
- val tgtCol = ServiceColumn.findOrInsert(tgtServiceId, tgtColumnName, Some(tgtColumnType), schemaVersion)
-
- if (srcCol.columnType != srcColumnType) throw new RuntimeException(s"source service column type not matched ${srcCol.columnType} != ${srcColumnType}")
- if (tgtCol.columnType != tgtColumnType) throw new RuntimeException(s"target service column type not matched ${tgtCol.columnType} != ${tgtColumnType}")
-
- /** create label */
- Label.findByName(labelName, useCache = false).getOrElse {
-
- val createdId = insert(labelName, srcServiceId, srcColumnName, srcColumnType,
- tgtServiceId, tgtColumnName, tgtColumnType, isDirected, serviceName, serviceId, consistencyLevel,
- hTableName.getOrElse(service.hTableName), hTableTTL.orElse(service.hTableTTL), schemaVersion, isAsync, compressionAlgorithm).toInt
-
- val labelMetaMap = metaProps.map { case Prop(propName, defaultValue, dataType) =>
- val labelMeta = LabelMeta.findOrInsert(createdId, propName, defaultValue, dataType)
- (propName -> labelMeta.seq)
- }.toMap ++ LabelMeta.reservedMetas.map (labelMeta => labelMeta.name -> labelMeta.seq).toMap
-
- if (indices.isEmpty) {
- // make default index with _PK, _timestamp, 0
- LabelIndex.findOrInsert(createdId, LabelIndex.DefaultName, LabelIndex.DefaultMetaSeqs.toList, "none")
- } else {
- indices.foreach { index =>
- val metaSeq = index.propNames.map { name => labelMetaMap(name) }
- LabelIndex.findOrInsert(createdId, index.name, metaSeq.toList, "none")
- }
- }
-
- val cacheKeys = List(s"id=$createdId", s"label=$labelName")
- val ret = findByName(labelName, useCache = false).get
- putsToCache(cacheKeys.map(k => k -> ret))
- ret
- }
- }
-
- newLabel.getOrElse(throw new RuntimeException("failed to create label"))
- }
-
- def findAll()(implicit session: DBSession = AutoSession) = {
- val ls = sql"""select * from labels""".map { rs => Label(rs) }.list().apply()
- putsToCache(ls.map { x =>
- val cacheKey = s"id=${x.id.get}"
- (cacheKey -> x)
- })
- putsToCache(ls.map { x =>
- val cacheKey = s"label=${x.label}"
- (cacheKey -> x)
- })
- }
-
- def updateName(oldName: String, newName: String)(implicit session: DBSession = AutoSession) = {
- logger.info(s"rename label: $oldName -> $newName")
- sql"""update labels set label = ${newName} where label = ${oldName}""".update.apply()
- }
-
- def updateHTableName(labelName: String, newHTableName: String)(implicit session: DBSession = AutoSession) = {
- logger.info(s"update HTable of label $labelName to $newHTableName")
- val cnt = sql"""update labels set hbase_table_name = $newHTableName where label = $labelName""".update().apply()
- val label = Label.findByName(labelName, useCache = false).get
-
- val cacheKeys = List(s"id=${label.id}", s"label=${label.label}")
- cacheKeys.foreach { key =>
- expireCache(key)
- expireCaches(key)
- }
- cnt
- }
-
- def delete(id: Int)(implicit session: DBSession = AutoSession) = {
- val label = findById(id)
- logger.info(s"delete label: $label")
- val cnt = sql"""delete from labels where id = ${label.id.get}""".update().apply()
- val cacheKeys = List(s"id=$id", s"label=${label.label}")
- cacheKeys.foreach { key =>
- expireCache(key)
- expireCaches(key)
- }
- cnt
- }
-}
-
-case class Label(id: Option[Int], label: String,
- srcServiceId: Int, srcColumnName: String, srcColumnType: String,
- tgtServiceId: Int, tgtColumnName: String, tgtColumnType: String,
- isDirected: Boolean = true, serviceName: String, serviceId: Int, consistencyLevel: String = "strong",
- hTableName: String, hTableTTL: Option[Int],
- schemaVersion: String, isAsync: Boolean = false,
- compressionAlgorithm: String) extends JSONParser {
- def metas = LabelMeta.findAllByLabelId(id.get)
-
- def metaSeqsToNames = metas.map(x => (x.seq, x.name)) toMap
-
- // lazy val firstHBaseTableName = hbaseTableName.split(",").headOption.getOrElse(Config.HBASE_TABLE_NAME)
- lazy val srcService = Service.findById(srcServiceId)
- lazy val tgtService = Service.findById(tgtServiceId)
- lazy val service = Service.findById(serviceId)
- /**
- * TODO
- * change this to apply hbase table from target serviceName
- */
- // lazy val (hbaseZkAddr, hbaseTableName) = (service.cluster, service.tableName.split(",").headOption.getOrElse(Config.HBASE_TABLE_NAME))
- // lazy val (hbaseZkAddr, hbaseTableName) = (Config.HBASE_ZOOKEEPER_QUORUM, hTableName.split(",").headOption.getOrElse(Config.HBASE_TABLE_NAME))
- // lazy val (hbaseZkAddr, hbaseTableName) = (service.cluster, hTableName.split(",").headOption.getOrElse(GraphConnection.getConfVal("hbase.table.name")))
- lazy val (hbaseZkAddr, hbaseTableName) = (service.cluster, hTableName.split(",").head)
-
- lazy val srcColumn = ServiceColumn.find(srcServiceId, srcColumnName).getOrElse(throw ModelNotFoundException("Source column not found"))
- lazy val tgtColumn = ServiceColumn.find(tgtServiceId, tgtColumnName).getOrElse(throw ModelNotFoundException("Target column not found"))
-
- lazy val direction = if (isDirected) "out" else "undirected"
- lazy val defaultIndex = LabelIndex.findByLabelIdAndSeq(id.get, LabelIndex.DefaultSeq)
-
- //TODO: Make sure this is correct
- lazy val indices = LabelIndex.findByLabelIdAll(id.get, useCache = true)
- lazy val indicesMap = indices.map(idx => (idx.seq, idx)) toMap
- lazy val indexSeqsMap = indices.map(idx => (idx.metaSeqs, idx)) toMap
- lazy val indexNameMap = indices.map(idx => (idx.name, idx)) toMap
- lazy val extraIndices = indices.filter(idx => defaultIndex.isDefined && idx.id.get != defaultIndex.get.id.get)
- // indices filterNot (_.id.get == defaultIndex.get.id.get)
- lazy val extraIndicesMap = extraIndices.map(idx => (idx.seq, idx)) toMap
-
- lazy val metaProps = LabelMeta.reservedMetas.map { m =>
- if (m == LabelMeta.to) m.copy(dataType = tgtColumnType)
- else if (m == LabelMeta.from) m.copy(dataType = srcColumnType)
- else m
- } ::: LabelMeta.findAllByLabelId(id.get, useCache = true)
-
- lazy val metaPropsInner = LabelMeta.reservedMetasInner.map { m =>
- if (m == LabelMeta.to) m.copy(dataType = tgtColumnType)
- else if (m == LabelMeta.from) m.copy(dataType = srcColumnType)
- else m
- } ::: LabelMeta.findAllByLabelId(id.get, useCache = true)
-
- lazy val metaPropsMap = metaProps.map(x => (x.seq, x)).toMap
- lazy val metaPropsInvMap = metaProps.map(x => (x.name, x)).toMap
- lazy val metaPropNames = metaProps.map(x => x.name)
- lazy val metaPropNamesMap = metaProps.map(x => (x.seq, x.name)) toMap
-
- /** this is used only by edgeToProps */
- lazy val metaPropsDefaultMap = (for {
- prop <- metaProps if LabelMeta.isValidSeq(prop.seq)
- jsValue <- innerValToJsValue(toInnerVal(prop.defaultValue, prop.dataType, schemaVersion), prop.dataType)
- } yield prop.name -> jsValue).toMap
-
- lazy val metaPropsDefaultMapInner = (for {
- prop <- metaPropsInner if LabelMeta.isValidSeq(prop.seq)
- jsValue <- innerValToJsValue(toInnerVal(prop.defaultValue, prop.dataType, schemaVersion), prop.dataType)
- } yield prop.name -> jsValue).toMap
-
- def srcColumnWithDir(dir: Int) = {
- if (dir == GraphUtil.directions("out")) srcColumn else tgtColumn
- }
-
- def tgtColumnWithDir(dir: Int) = {
- if (dir == GraphUtil.directions("out")) tgtColumn else srcColumn
- }
-
- def srcTgtColumn(dir: Int) =
- if (isDirected) {
- (srcColumnWithDir(dir), tgtColumnWithDir(dir))
- } else {
- if (dir == GraphUtil.directions("in")) {
- (tgtColumn, srcColumn)
- } else {
- (srcColumn, tgtColumn)
- }
- }
-
- def init() = {
- metas
- metaSeqsToNames
- service
- srcColumn
- tgtColumn
- defaultIndex
- indices
- metaProps
- }
-
- // def srcColumnInnerVal(jsValue: JsValue) = {
- // jsValueToInnerVal(jsValue, srcColumnType, version)
- // }
- // def tgtColumnInnerVal(jsValue: JsValue) = {
- // jsValueToInnerVal(jsValue, tgtColumnType, version)
- // }
-
- override def toString(): String = {
- val orderByKeys = LabelMeta.findAllByLabelId(id.get)
- super.toString() + orderByKeys.toString()
- }
-
- // def findLabelIndexSeq(scoring: List[(Byte, Double)]): Byte = {
- // if (scoring.isEmpty) LabelIndex.defaultSeq
- // else {
- // LabelIndex.findByLabelIdAndSeqs(id.get, scoring.map(_._1).sorted).map(_.seq).getOrElse(LabelIndex.defaultSeq)
- //
- //// LabelIndex.findByLabelIdAndSeqs(id.get, scoring.map(_._1).sorted).map(_.seq).getOrElse(LabelIndex.defaultSeq)
- // }
- // }
- lazy val toJson = Json.obj("labelName" -> label,
- "from" -> srcColumn.toJson, "to" -> tgtColumn.toJson,
- "isDirected" -> isDirected,
- "serviceName" -> serviceName,
- "consistencyLevel" -> consistencyLevel,
- "schemaVersion" -> schemaVersion,
- "isAsync" -> isAsync,
- "compressionAlgorithm" -> compressionAlgorithm,
- "defaultIndex" -> defaultIndex.map(x => x.toJson),
- "extraIndex" -> extraIndices.map(exIdx => exIdx.toJson),
- "metaProps" -> metaProps.filter { labelMeta => LabelMeta.isValidSeqForAdmin(labelMeta.seq) }.map(_.toJson)
- )
-
-
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-s2graph/blob/b6fe32fc/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/LabelIndex.scala
----------------------------------------------------------------------
diff --git a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/LabelIndex.scala b/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/LabelIndex.scala
deleted file mode 100644
index 47f4a2a..0000000
--- a/s2core/src/main/scala/com/kakao/s2graph/core/mysqls/LabelIndex.scala
+++ /dev/null
@@ -1,143 +0,0 @@
-package com.kakao.s2graph.core.mysqls
-
-/**
- * Created by shon on 6/3/15.
- */
-
-import play.api.libs.json.Json
-import scalikejdbc._
-
-object LabelIndex extends Model[LabelIndex] {
- val DefaultName = "_PK"
- val DefaultMetaSeqs = Seq(LabelMeta.timeStampSeq)
- val DefaultSeq = 1.toByte
- val MaxOrderSeq = 7
-
- def apply(rs: WrappedResultSet): LabelIndex = {
- LabelIndex(rs.intOpt("id"), rs.int("label_id"), rs.string("name"), rs.byte("seq"),
- rs.string("meta_seqs").split(",").filter(_ != "").map(s => s.toByte).toList match {
- case metaSeqsList => metaSeqsList
- },
- rs.string("formulars"))
- }
-
- def findById(id: Int)(implicit session: DBSession = AutoSession) = {
- val cacheKey = "id=" + id
- withCache(cacheKey) {
- sql"""select * from label_indices where id = ${id}""".map { rs => LabelIndex(rs) }.single.apply
- }.get
- }
-
- def findByLabelIdAll(labelId: Int, useCache: Boolean = true)(implicit session: DBSession = AutoSession) = {
- val cacheKey = "labelId=" + labelId
- if (useCache) {
- withCaches(cacheKey)( sql"""
- select * from label_indices where label_id = ${labelId} and seq > 0 order by seq ASC
- """.map { rs => LabelIndex(rs) }.list.apply)
- } else {
- sql"""
- select * from label_indices where label_id = ${labelId} and seq > 0 order by seq ASC
- """.map { rs => LabelIndex(rs) }.list.apply
- }
- }
-
- def insert(labelId: Int, indexName: String, seq: Byte, metaSeqs: List[Byte], formulars: String)(implicit session: DBSession = AutoSession): Long = {
- sql"""
- insert into label_indices(label_id, name, seq, meta_seqs, formulars)
- values (${labelId}, ${indexName}, ${seq}, ${metaSeqs.mkString(",")}, ${formulars})
- """
- .updateAndReturnGeneratedKey.apply()
- }
-
- def findOrInsert(labelId: Int, indexName: String, metaSeqs: List[Byte], formulars: String)(implicit session: DBSession = AutoSession): LabelIndex = {
- findByLabelIdAndSeqs(labelId, metaSeqs) match {
- case Some(s) => s
- case None =>
- val orders = findByLabelIdAll(labelId, false)
- val seq = (orders.size + 1).toByte
- assert(seq <= MaxOrderSeq)
- val createdId = insert(labelId, indexName, seq, metaSeqs, formulars)
- val cacheKeys = List(s"labelId=$labelId:seq=$seq",
- s"labelId=$labelId:seqs=$metaSeqs", s"labelId=$labelId:seq=$seq", s"id=$createdId")
- cacheKeys.foreach { key =>
- expireCache(key)
- expireCaches(key)
- }
-
- findByLabelIdAndSeq(labelId, seq).get
- }
- }
-
- def findByLabelIdAndSeqs(labelId: Int, seqs: List[Byte])(implicit session: DBSession = AutoSession): Option[LabelIndex] = {
- val cacheKey = "labelId=" + labelId + ":seqs=" + seqs.mkString(",")
- withCache(cacheKey) {
- sql"""
- select * from label_indices where label_id = ${labelId} and meta_seqs = ${seqs.mkString(",")}
- """.map { rs => LabelIndex(rs) }.single.apply
- }
- }
-
- def findByLabelIdAndSeq(labelId: Int, seq: Byte, useCache: Boolean = true)(implicit session: DBSession = AutoSession): Option[LabelIndex] = {
- // val cacheKey = s"labelId=$labelId:seq=$seq"
- val cacheKey = "labelId=" + labelId + ":seq=" + seq
- if (useCache) {
- withCache(cacheKey)( sql"""
- select * from label_indices where label_id = ${labelId} and seq = ${seq}
- """.map { rs => LabelIndex(rs) }.single.apply)
- } else {
- sql"""
- select * from label_indices where label_id = ${labelId} and seq = ${seq}
- """.map { rs => LabelIndex(rs) }.single.apply
- }
- }
-
- def delete(id: Int)(implicit session: DBSession = AutoSession) = {
- val labelIndex = findById(id)
- val seqs = labelIndex.metaSeqs.mkString(",")
- val (labelId, seq) = (labelIndex.labelId, labelIndex.seq)
- sql"""delete from label_indices where id = ${id}""".execute.apply()
-
- val cacheKeys = List(s"id=$id", s"labelId=$labelId", s"labelId=$labelId:seq=$seq", s"labelId=$labelId:seqs=$seqs")
- cacheKeys.foreach { key =>
- expireCache(key)
- expireCaches(key)
- }
- }
-
- def findAll()(implicit session: DBSession = AutoSession) = {
- val ls = sql"""select * from label_indices""".map { rs => LabelIndex(rs) }.list.apply
- putsToCache(ls.map { x =>
- var cacheKey = s"id=${x.id.get}"
- (cacheKey -> x)
- })
- putsToCache(ls.map { x =>
- var cacheKey = s"labelId=${x.labelId}:seq=${x.seq}}"
- (cacheKey -> x)
- })
- putsToCache(ls.map { x =>
- var cacheKey = s"labelId=${x.labelId}:seqs=${x.metaSeqs.mkString(",")}"
- (cacheKey -> x)
- })
- putsToCaches(ls.groupBy(x => x.labelId).map { case (labelId, ls) =>
- val cacheKey = s"labelId=${labelId}"
- (cacheKey -> ls)
- }.toList)
- }
-}
-
-/**
- * formular
- * ex1): w1, w2, w3
- * ex2): 1.5 * w1^2 + 3.4 * (w1 * w2), w2, w1
- */
-
-case class LabelIndex(id: Option[Int], labelId: Int, name: String, seq: Byte, metaSeqs: Seq[Byte], formulars: String) {
- lazy val label = Label.findById(labelId)
- lazy val metas = label.metaPropsMap
- lazy val sortKeyTypes = metaSeqs.flatMap(metaSeq => label.metaPropsMap.get(metaSeq))
- lazy val propNames = sortKeyTypes.map { labelMeta => labelMeta.name }
- lazy val toJson = Json.obj(
- "name" -> name,
- "propNames" -> sortKeyTypes.map(x => x.name)
- )
-}