| /* |
| * 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.rest |
| |
| |
| import java.util.concurrent.{Callable, TimeUnit} |
| |
| import com.google.common.cache.CacheBuilder |
| import org.apache.s2graph.core.GraphExceptions.{BadQueryException, ModelNotFoundException} |
| import org.apache.s2graph.core.JSONParser._ |
| import org.apache.s2graph.core._ |
| import org.apache.s2graph.core.mysqls._ |
| import org.apache.s2graph.core.parsers.{Where, WhereParser} |
| import org.apache.s2graph.core.types._ |
| import play.api.libs.json._ |
| |
| |
| import scala.util.{Random, Failure, Success, Try} |
| |
| object TemplateHelper { |
| val findVar = """\"?\$\{(.*?)\}\"?""".r |
| val num = """(next_minute|next_day|next_hour|next_week|now)?\s*(-?\s*[0-9]+)?\s*(minute|hour|day|week)?""".r |
| val randIntRegex = """randint\((.*,.*)\)""".r |
| |
| val minute: Long = 60 * 1000L |
| val hour = 60 * minute |
| val day = 24 * hour |
| val week = 7 * day |
| |
| def calculate(now: Long, n: Int, unit: String): Long = { |
| val duration = unit match { |
| case "minute" | "MINUTE" => n * minute |
| case "hour" | "HOUR" => n * hour |
| case "day" | "DAY" => n * day |
| case "week" | "WEEK" => n * week |
| case _ => n |
| } |
| |
| duration + now |
| } |
| |
| def randInt(s: String): Long = { |
| val tokens = s.split(",").map(_.trim) |
| if (tokens.length != 2) throw new RuntimeException(s"TemplateHelper.randint has wrong format. $s") |
| val (from, to) = try { |
| (tokens.head.toInt, tokens.last.toInt) |
| } catch { |
| case e: Exception => throw new RuntimeException(s"TemplateHelper.randint has wrong format. $s") |
| } |
| if (from > to) throw new RuntimeException(s"TemplateHelper.randint has wrong format. $s") |
| val diff = to - from |
| val r = Random.nextInt(diff + 1) |
| assert(diff >= 0 && diff < Int.MaxValue && from + r < Int.MaxValue) |
| from + r |
| } |
| |
| def replaceVariable(now: Long, body: String): String = { |
| findVar.replaceAllIn(body, m => { |
| val matched = m group 1 |
| randIntRegex.findFirstMatchIn(matched) match { |
| case None => |
| num.replaceSomeIn(matched, m => { |
| val (_pivot, n, unit) = (m.group(1), m.group(2), m.group(3)) |
| val ts = _pivot match { |
| case null => now |
| case "now" | "NOW" => now |
| case "next_minute" | "NEXT_MINUTE" => now / minute * minute + minute |
| case "next_week" | "NEXT_WEEK" => now / week * week + week |
| case "next_day" | "NEXT_DAY" => now / day * day + day |
| case "next_hour" | "NEXT_HOUR" => now / hour * hour + hour |
| } |
| |
| if (_pivot == null && n == null && unit == null) None |
| else if (n == null || unit == null) Option(ts.toString) |
| else Option(calculate(ts, n.replaceAll(" ", "").toInt, unit).toString) |
| }) |
| case Some(m) => |
| val range = m group 1 |
| randInt(range).toString |
| } |
| }) |
| } |
| } |
| |
| object RequestParser { |
| type ExperimentParam = (JsObject, String, String, String, Option[String]) |
| val defaultLimit = 100 |
| |
| def toJsValues(jsValue: JsValue): List[JsValue] = { |
| jsValue match { |
| case obj: JsObject => List(obj) |
| case arr: JsArray => arr.as[List[JsValue]] |
| case _ => List.empty[JsValue] |
| } |
| } |
| |
| def jsToStr(js: JsValue): String = js match { |
| case JsString(s) => s |
| case _ => js.toString() |
| } |
| |
| } |
| |
| class RequestParser(graph: S2Graph) { |
| |
| import Management.JsonModel._ |
| import RequestParser._ |
| |
| val config = graph.config |
| val hardLimit = config.getInt("query.hardlimit") |
| val maxLimit = Int.MaxValue - 1 |
| val DefaultRpcTimeout = config.getInt("hbase.rpc.timeout") |
| val DefaultMaxAttempt = config.getInt("hbase.client.retries.number") |
| val DefaultCluster = config.getString("hbase.zookeeper.quorum") |
| val DefaultCompressionAlgorithm = config.getString("hbase.table.compression.algorithm") |
| val DefaultPhase = config.getString("phase") |
| val parserCache = CacheBuilder.newBuilder() |
| .expireAfterAccess(10000, TimeUnit.MILLISECONDS) |
| .expireAfterWrite(10000, TimeUnit.MILLISECONDS) |
| .maximumSize(10000) |
| .initialCapacity(1000) |
| .build[String, Try[Where]] |
| |
| private def extractScoring(label: Label, value: JsValue): Option[Seq[(LabelMeta, Double)]] = { |
| val ret = for { |
| js <- parseOption[JsObject](value, "scoring") |
| } yield { |
| for { |
| (k, v) <- js.fields |
| labelMata <- label.metaPropsInvMap.get(k) |
| } yield { |
| val value = v match { |
| case n: JsNumber => n.as[Double] |
| case _ => throw new Exception("scoring weight should be double.") |
| } |
| (labelMata, value) |
| } |
| } |
| |
| ret |
| } |
| |
| def extractInterval(label: Label, jsValue: JsValue): Option[(Seq[(String, JsValue)], Seq[(String, JsValue)])] = { |
| def extractKv(js: JsValue): Seq[(String, JsValue)] = js match { |
| case JsObject(obj) => obj.toSeq |
| case JsArray(arr) => arr.flatMap { |
| case JsObject(obj) => obj.toSeq |
| case _ => throw new RuntimeException(s"cannot support json type $js") |
| } |
| case _ => throw new RuntimeException(s"cannot support json type: $js") |
| } |
| |
| val ret = for { |
| js <- parseOption[JsObject](jsValue, "interval") |
| fromJs <- (js \ "from").asOpt[JsValue] |
| toJs <- (js \ "to").asOpt[JsValue] |
| } yield { |
| val from = extractKv(fromJs) |
| val to = extractKv(toJs) |
| (from, to) |
| } |
| |
| ret |
| } |
| |
| def extractDuration(label: Label, jsValue: JsValue) = { |
| for { |
| js <- parseOption[JsObject](jsValue, "duration") |
| } yield { |
| val minTs = (js \ "from").get match { |
| case JsString(s) => TemplateHelper.replaceVariable(System.currentTimeMillis(), s).toLong |
| case JsNumber(n) => n.toLong |
| case _ => Long.MinValue |
| } |
| |
| val maxTs = (js \ "to").get match { |
| case JsString(s) => TemplateHelper.replaceVariable(System.currentTimeMillis(), s).toLong |
| case JsNumber(n) => n.toLong |
| case _ => Long.MaxValue |
| } |
| |
| if (minTs > maxTs) { |
| throw new BadQueryException("Duration error. Timestamp of From cannot be larger than To.") |
| } |
| |
| (minTs, maxTs) |
| } |
| } |
| |
| def extractHas(label: Label, jsValue: JsValue) = { |
| val ret = for { |
| js <- parseOption[JsObject](jsValue, "has") |
| } yield { |
| for { |
| (k, v) <- js.fields |
| labelMeta <- LabelMeta.findByName(label.id.get, k) |
| value <- jsValueToInnerVal(v, labelMeta.dataType, label.schemaVersion) |
| } yield { |
| labelMeta.name -> value |
| } |
| } |
| ret.map(_.toMap).getOrElse(Map.empty[String, InnerValLike]) |
| } |
| |
| def extractWhere(label: Label, whereClauseOpt: Option[String]): Try[Where] = { |
| whereClauseOpt match { |
| case None => Success(WhereParser.success) |
| case Some(where) => |
| val whereParserKey = s"${label.label}_${where}" |
| |
| parserCache.get(whereParserKey, new Callable[Try[Where]] { |
| override def call(): Try[Where] = { |
| val _where = TemplateHelper.replaceVariable(System.currentTimeMillis(), where) |
| |
| WhereParser(label).parse(_where) match { |
| case s@Success(_) => s |
| case Failure(ex) => throw BadQueryException(ex.getMessage, ex) |
| } |
| } |
| }) |
| |
| } |
| } |
| |
| def extractGroupBy(value: Option[JsValue]): GroupBy = value.map { |
| case obj: JsObject => |
| val keys = (obj \ "keys").asOpt[Seq[String]].getOrElse(Nil) |
| val groupByLimit = (obj \ "limit").asOpt[Int].getOrElse(hardLimit) |
| val minShouldMatchOpt = (obj \ "minimumShouldMatch").asOpt[JsObject].map { o => |
| val prop = (o \ "prop").asOpt[String].getOrElse("to") |
| val count = (o \ "count").asOpt[Int].getOrElse(0) |
| val terms = (o \ "terms").asOpt[Set[JsValue]].getOrElse(Set.empty).map { |
| case JsString(s) => s |
| case JsNumber(n) => n |
| case _ => throw new RuntimeException("not supported data type") |
| }.map(_.asInstanceOf[Any]) |
| |
| MinShouldMatchParam(prop, count, terms) |
| } |
| |
| GroupBy(keys, groupByLimit, minShouldMatch = minShouldMatchOpt) |
| case arr: JsArray => |
| val keys = arr.asOpt[Seq[String]].getOrElse(Nil) |
| GroupBy(keys) |
| case _ => GroupBy.Empty |
| }.getOrElse(GroupBy.Empty) |
| |
| def toVertices(labelName: String, direction: String, ids: Seq[JsValue]): Seq[S2Vertex] = { |
| val vertices = for { |
| label <- Label.findByName(labelName).toSeq |
| serviceColumn = if (direction == "out") label.srcColumn else label.tgtColumn |
| id <- ids |
| innerId <- jsValueToInnerVal(id, serviceColumn.columnType, label.schemaVersion) |
| } yield { |
| graph.newVertex(SourceVertexId(serviceColumn, innerId), System.currentTimeMillis()) |
| } |
| |
| vertices |
| } |
| |
| def toMultiQuery(jsValue: JsValue, impIdOpt: Option[String]): MultiQuery = { |
| val globalQueryOption = toQueryOption(jsValue, impIdOpt) |
| val queries = for { |
| queryJson <- (jsValue \ "queries").asOpt[Seq[JsValue]].getOrElse(Seq.empty) |
| } yield { |
| val innerQuery = toQuery(queryJson, impIdOpt = impIdOpt) |
| val queryOption = innerQuery.queryOption |
| |
| if (queryOption.groupBy.keys.nonEmpty) throw new BadQueryException("Group by option is not allowed in multiple queries.") |
| if (queryOption.orderByKeys.nonEmpty) throw new BadQueryException("Order by option is not allowed in multiple queries.") |
| |
| if (globalQueryOption.withScore) innerQuery.copy(queryOption = innerQuery.queryOption.copy(withScore = false)) |
| else innerQuery |
| // val innerQuery3 = |
| // if (globalQueryOption.groupBy.keys.nonEmpty) innerQuery2.copy(queryOption = innerQuery2.queryOption.copy(groupBy = GroupBy.Empty)) |
| // else innerQuery2 |
| |
| } |
| val weights = (jsValue \ "weights").asOpt[Seq[Double]].getOrElse(queries.map(_ => 1.0)) |
| MultiQuery(queries = queries, weights = weights, queryOption = globalQueryOption) |
| } |
| |
| |
| def toQueryOption(jsValue: JsValue, impIdOpt: Option[String]): QueryOption = { |
| val filterOutFields = (jsValue \ "filterOutFields").asOpt[List[String]].getOrElse(List(LabelMeta.to.name)) |
| val filterOutQuery = (jsValue \ "filterOut").asOpt[JsValue].map { v => |
| toQuery(v, impIdOpt = impIdOpt) |
| }.map { q => |
| q.copy(queryOption = q.queryOption.copy(filterOutFields = filterOutFields, selectColumns = filterOutFields)) |
| } |
| val removeCycle = (jsValue \ "removeCycle").asOpt[Boolean].getOrElse(true) |
| val selectColumns = (jsValue \ "select").asOpt[List[String]].getOrElse(List.empty) |
| |
| val groupBy = extractGroupBy((jsValue \ "groupBy").asOpt[JsValue]) |
| |
| val orderByColumns: List[(String, Boolean)] = (jsValue \ "orderBy").asOpt[List[JsObject]].map { jsLs => |
| for { |
| js <- jsLs |
| (column, orderJs) <- js.fields |
| } yield { |
| val ascending = orderJs.as[String].toUpperCase match { |
| case "ASC" => true |
| case "DESC" => false |
| } |
| column -> ascending |
| } |
| }.getOrElse(Nil) |
| val withScore = (jsValue \ "withScore").asOpt[Boolean].getOrElse(true) |
| val returnTree = (jsValue \ "returnTree").asOpt[Boolean].getOrElse(false) |
| //TODO: Refactor this |
| val limitOpt = (jsValue \ "limit").asOpt[Int] |
| val returnAgg = (jsValue \ "returnAgg").asOpt[Boolean].getOrElse(true) |
| val scoreThreshold = (jsValue \ "scoreThreshold").asOpt[Double].getOrElse(Double.MinValue) |
| val returnDegree = (jsValue \ "returnDegree").asOpt[Boolean].getOrElse(true) |
| val ignorePrevStepCache = (jsValue \ "ignorePrevStepCache").asOpt[Boolean].getOrElse(false) |
| val shouldPropagateScore = (jsValue \ "shouldPropagateScore").asOpt[Boolean].getOrElse(true) |
| |
| QueryOption(removeCycle = removeCycle, |
| selectColumns = selectColumns, |
| groupBy = groupBy, |
| orderByColumns = orderByColumns, |
| filterOutQuery = filterOutQuery, |
| filterOutFields = filterOutFields, |
| withScore = withScore, |
| returnTree = returnTree, |
| limitOpt = limitOpt, |
| returnAgg = returnAgg, |
| scoreThreshold = scoreThreshold, |
| returnDegree = returnDegree, |
| impIdOpt = impIdOpt, |
| ignorePrevStepCache, |
| shouldPropagateScore |
| ) |
| } |
| |
| def toQuery(jsValue: JsValue, impIdOpt: Option[String]): Query = { |
| try { |
| val vertices = for { |
| value <- (jsValue \ "srcVertices").asOpt[Seq[JsValue]].getOrElse(Nil) |
| serviceName <- (value \ "serviceName").asOpt[String].toSeq |
| columnName <- (value \ "columnName").asOpt[String].toSeq |
| idJson = (value \ "id").asOpt[JsValue].map(Seq(_)).getOrElse(Nil) |
| idsJson = (value \ "ids").asOpt[Seq[JsValue]].getOrElse(Nil) |
| id <- (idJson ++ idsJson).flatMap(jsValueToAny(_).toSeq).distinct |
| } yield graph.toVertex(serviceName, columnName, id) |
| |
| if (vertices.isEmpty) throw BadQueryException("srcVertices`s id is empty") |
| val steps = parse[Vector[JsValue]](jsValue, "steps") |
| val queryOption = toQueryOption(jsValue, impIdOpt) |
| |
| val querySteps = |
| steps.zipWithIndex.map { case (step, stepIdx) => |
| val labelWeights = step match { |
| case obj: JsObject => |
| val converted = for { |
| (k, v) <- (obj \ "weights").asOpt[JsObject].getOrElse(Json.obj()).fields |
| l <- Label.findByName(k) |
| } yield { |
| l.id.get -> v.toString().toDouble |
| } |
| converted.toMap |
| case _ => Map.empty[Int, Double] |
| } |
| val queryParamJsVals = step match { |
| case arr: JsArray => arr.as[List[JsValue]] |
| case obj: JsObject => (obj \ "step").as[List[JsValue]] |
| case _ => List.empty[JsValue] |
| } |
| val nextStepScoreThreshold = step match { |
| case obj: JsObject => (obj \ "nextStepThreshold").asOpt[Double].getOrElse(QueryParam.DefaultThreshold) |
| case _ => QueryParam.DefaultThreshold |
| } |
| val nextStepLimit = step match { |
| case obj: JsObject => (obj \ "nextStepLimit").asOpt[Int].getOrElse(-1) |
| case _ => -1 |
| } |
| val cacheTTL = step match { |
| case obj: JsObject => (obj \ "cacheTTL").asOpt[Int].getOrElse(-1) |
| case _ => -1 |
| } |
| val queryParams = |
| for { |
| labelGroup <- queryParamJsVals |
| queryParam <- parseQueryParam(labelGroup, queryOption) |
| } yield { |
| val (_, columnName) = |
| if (queryParam.dir == GraphUtil.directions("out")) { |
| (queryParam.label.srcService.serviceName, queryParam.label.srcColumnName) |
| } else { |
| (queryParam.label.tgtService.serviceName, queryParam.label.tgtColumnName) |
| } |
| //FIXME: |
| if (stepIdx == 0 && vertices.nonEmpty && !vertices.exists(v => v.columnName == columnName)) { |
| throw BadQueryException("srcVertices contains incompatiable serviceName or columnName with first step.") |
| } |
| |
| queryParam |
| } |
| |
| |
| val groupBy = extractGroupBy((step \ "groupBy").asOpt[JsValue]) |
| |
| Step(queryParams = queryParams, |
| labelWeights = labelWeights, |
| nextStepScoreThreshold = nextStepScoreThreshold, |
| nextStepLimit = nextStepLimit, |
| cacheTTL = cacheTTL, |
| groupBy = groupBy) |
| |
| } |
| |
| Query(vertices, querySteps, queryOption) |
| } catch { |
| case e: BadQueryException => |
| throw e |
| case e: ModelNotFoundException => |
| throw BadQueryException(e.getMessage, e) |
| case e: Exception => |
| throw BadQueryException(s"$jsValue, $e", e) |
| } |
| } |
| |
| private def parseQueryParam(labelGroup: JsValue, queryOption: QueryOption): Option[QueryParam] = { |
| for { |
| labelName <- parseOption[String](labelGroup, "label") |
| } yield { |
| val label = Label.findByName(labelName).getOrElse(throw BadQueryException(s"$labelName not found")) |
| val direction = parseOption[String](labelGroup, "direction").getOrElse("out") |
| val limit = { |
| parseOption[Int](labelGroup, "limit") match { |
| case None => defaultLimit |
| case Some(l) if l < 0 => hardLimit |
| case Some(l) if l >= 0 => Math.min(l, hardLimit) |
| } |
| } |
| val offset = parseOption[Int](labelGroup, "offset").getOrElse(0) |
| val interval = extractInterval(label, labelGroup) |
| val duration = extractDuration(label, labelGroup) |
| val scoring = extractScoring(label, labelGroup).getOrElse(Nil).toList |
| val exclude = parseOption[Boolean](labelGroup, "exclude").getOrElse(false) |
| val include = parseOption[Boolean](labelGroup, "include").getOrElse(false) |
| val hasFilter = extractHas(label, labelGroup) |
| |
| val indexName = (labelGroup \ "index").asOpt[String].getOrElse(LabelIndex.DefaultName) |
| val whereClauseOpt = (labelGroup \ "where").asOpt[String] |
| val where = extractWhere(label, whereClauseOpt) |
| val includeDegree = (labelGroup \ "includeDegree").asOpt[Boolean].getOrElse(true) |
| val rpcTimeout = (labelGroup \ "rpcTimeout").asOpt[Int].getOrElse(DefaultRpcTimeout) |
| val maxAttempt = (labelGroup \ "maxAttempt").asOpt[Int].getOrElse(DefaultMaxAttempt) |
| |
| val tgtVertexInnerIdOpt = (labelGroup \ "_to").asOpt[JsValue].filterNot(_ == JsNull).flatMap(jsValueToAny) |
| |
| val cacheTTL = (labelGroup \ "cacheTTL").asOpt[Long].getOrElse(-1L) |
| val timeDecayFactor = (labelGroup \ "timeDecay").asOpt[JsObject].map { jsVal => |
| val propName = (jsVal \ "propName").asOpt[String].getOrElse(LabelMeta.timestamp.name) |
| val propNameSeq = label.metaPropsInvMap.get(propName).getOrElse(LabelMeta.timestamp) |
| val initial = (jsVal \ "initial").asOpt[Double].getOrElse(1.0) |
| val decayRate = (jsVal \ "decayRate").asOpt[Double].getOrElse(0.1) |
| if (decayRate >= 1.0 || decayRate <= 0.0) throw new BadQueryException("decay rate should be 0.0 ~ 1.0") |
| val timeUnit = (jsVal \ "timeUnit").asOpt[Double].getOrElse(60 * 60 * 24.0) |
| TimeDecay(initial, decayRate, timeUnit, propNameSeq) |
| } |
| val threshold = (labelGroup \ "threshold").asOpt[Double].getOrElse(QueryParam.DefaultThreshold) |
| // TODO: refactor this. dirty |
| val duplicate = parseOption[String](labelGroup, "duplicate").map(s => DuplicatePolicy(s)).getOrElse(DuplicatePolicy.First) |
| |
| val outputField = (labelGroup \ "outputField").asOpt[String].map(s => Json.arr(Json.arr(s))) |
| val transformer = (if (outputField.isDefined) outputField else (labelGroup \ "transform").asOpt[JsValue]) match { |
| case None => EdgeTransformer(EdgeTransformer.DefaultJson) |
| case Some(json) => EdgeTransformer(json) |
| } |
| val scorePropagateOp = (labelGroup \ "scorePropagateOp").asOpt[String].getOrElse("multiply") |
| val scorePropagateShrinkage = (labelGroup \ "scorePropagateShrinkage").asOpt[Long].getOrElse(500l) |
| val sample = (labelGroup \ "sample").asOpt[Int].getOrElse(-1) |
| val shouldNormalize = (labelGroup \ "normalize").asOpt[Boolean].getOrElse(false) |
| val cursorOpt = (labelGroup \ "cursor").asOpt[String] |
| // FIXME: Order of command matter |
| QueryParam(labelName = labelName, |
| direction = direction, |
| offset = offset, |
| limit = limit, |
| sample = sample, |
| maxAttempt = maxAttempt, |
| rpcTimeout = rpcTimeout, |
| cacheTTLInMillis = cacheTTL, |
| indexName = indexName, where = where, threshold = threshold, |
| rank = RankParam(scoring), intervalOpt = interval, durationOpt = duration, |
| exclude = exclude, include = include, has = hasFilter, duplicatePolicy = duplicate, |
| includeDegree = includeDegree, scorePropagateShrinkage = scorePropagateShrinkage, |
| scorePropagateOp = scorePropagateOp, shouldNormalize = shouldNormalize, |
| whereRawOpt = whereClauseOpt, cursorOpt = cursorOpt, |
| tgtVertexIdOpt = tgtVertexInnerIdOpt, |
| edgeTransformer = transformer, timeDecay = timeDecayFactor |
| ) |
| } |
| } |
| |
| private def parse[R](js: JsValue, key: String)(implicit read: Reads[R]): R = { |
| (js \ key).validate[R] match { |
| case JsError(errors) => |
| val msg = (JsError.toJson(errors) \ "obj").as[List[JsValue]].flatMap(x => (x \ "msg").toOption) |
| val e = Json.obj("args" -> key, "error" -> msg) |
| throw new GraphExceptions.JsonParseException(Json.obj("error" -> key).toString) |
| case JsSuccess(result, _) => result |
| } |
| } |
| |
| private def parseOption[R](js: JsValue, key: String)(implicit read: Reads[R]): Option[R] = { |
| (js \ key).validateOpt[R] match { |
| case JsError(errors) => |
| val msg = (JsError.toJson(errors) \ "obj").as[List[JsValue]].flatMap(x => (x \ "msg").toOption) |
| val e = Json.obj("args" -> key, "error" -> msg) |
| throw new GraphExceptions.JsonParseException(Json.obj("error" -> key).toString) |
| case JsSuccess(result, _) => result |
| } |
| } |
| |
| def jsToStr(js: JsValue): String = js match { |
| case JsString(s) => s |
| case _ => js.toString() |
| } |
| |
| def parseBulkFormat(str: String): Seq[(GraphElement, String)] = { |
| val edgeStrs = str.split("\\n").filterNot(_.isEmpty) |
| val elementsWithTsv = for { |
| edgeStr <- edgeStrs |
| str <- GraphUtil.parseString(edgeStr) |
| element <- graph.toGraphElement(str) |
| } yield (element, str) |
| |
| elementsWithTsv |
| } |
| |
| def parseJsonFormat(jsValue: JsValue, operation: String): Seq[(S2Edge, String)] = { |
| val jsValues = toJsValues(jsValue) |
| jsValues.flatMap(toEdgeWithTsv(_, operation)) |
| } |
| |
| private def toEdgeWithTsv(jsValue: JsValue, operation: String): Seq[(S2Edge, String)] = { |
| val srcIds = (jsValue \ "from").asOpt[JsValue].map(Seq(_)).getOrElse(Nil) ++ (jsValue \ "froms").asOpt[Seq[JsValue]].getOrElse(Nil) |
| val tgtIds = (jsValue \ "to").asOpt[JsValue].map(Seq(_)).getOrElse(Nil) ++ (jsValue \ "tos").asOpt[Seq[JsValue]].getOrElse(Nil) |
| |
| val label = parse[String](jsValue, "label") |
| val timestamp = parse[Long](jsValue, "timestamp") |
| val direction = parseOption[String](jsValue, "direction").getOrElse("out") |
| val propsJson = (jsValue \ "props").asOpt[JsObject].getOrElse(Json.obj()) |
| |
| for { |
| srcId <- srcIds.flatMap(jsValueToAny(_).toSeq) |
| tgtId <- tgtIds.flatMap(jsValueToAny(_).toSeq) |
| } yield { |
| // val edge = Management.toEdge(graph, timestamp, operation, srcId, tgtId, label, direction, fromJsonToProperties(propsJson)) |
| val edge = graph.toEdge(srcId, tgtId, label, direction, fromJsonToProperties(propsJson), ts = timestamp, operation = operation) |
| val tsv = (jsValue \ "direction").asOpt[String] match { |
| case None => Seq(timestamp, operation, "e", srcId, tgtId, label, propsJson.toString).mkString("\t") |
| case Some(dir) => Seq(timestamp, operation, "e", srcId, tgtId, label, propsJson.toString, dir).mkString("\t") |
| } |
| |
| (edge, tsv) |
| } |
| } |
| |
| def toVertices(jsValue: JsValue, operation: String, serviceName: Option[String] = None, columnName: Option[String] = None) = { |
| toJsValues(jsValue).map(toVertex(_, operation, serviceName, columnName)) |
| } |
| |
| def toVertex(jsValue: JsValue, operation: String, serviceName: Option[String] = None, columnName: Option[String] = None): S2Vertex = { |
| val id = parse[JsValue](jsValue, "id") |
| val ts = parseOption[Long](jsValue, "timestamp").getOrElse(System.currentTimeMillis()) |
| val sName = if (serviceName.isEmpty) parse[String](jsValue, "serviceName") else serviceName.get |
| val cName = if (columnName.isEmpty) parse[String](jsValue, "columnName") else columnName.get |
| val props = fromJsonToProperties((jsValue \ "props").asOpt[JsObject].getOrElse(Json.obj())) |
| graph.toVertex(sName, cName, id.toString, props, ts, operation) |
| } |
| |
| def toPropElements(jsObj: JsValue) = Try { |
| val propName = (jsObj \ "name").as[String] |
| val dataType = InnerVal.toInnerDataType((jsObj \ "dataType").as[String]) |
| val defaultValue = (jsObj \ "defaultValue").as[JsValue] match { |
| case JsString(s) => s |
| case _@js => js.toString |
| } |
| Prop(propName, defaultValue, dataType) |
| } |
| |
| def toPropsElements(jsValue: JsLookupResult): Seq[Prop] = for { |
| jsObj <- jsValue.asOpt[Seq[JsValue]].getOrElse(Nil) |
| } yield { |
| val propName = (jsObj \ "name").as[String] |
| val dataType = InnerVal.toInnerDataType((jsObj \ "dataType").as[String]) |
| val defaultValue = (jsObj \ "defaultValue").as[JsValue] match { |
| case JsString(s) => s |
| case _@js => js.toString |
| } |
| Prop(propName, defaultValue, dataType) |
| } |
| |
| def toIndicesElements(jsValue: JsLookupResult): Seq[Index] = { |
| val indices = for { |
| jsObj <- jsValue.as[Seq[JsValue]] |
| indexName = (jsObj \ "name").as[String] |
| propNames = (jsObj \ "propNames").as[Seq[String]] |
| direction = (jsObj \ "direction").asOpt[String].map(GraphUtil.toDirection) |
| options = (jsObj \ "options").asOpt[JsValue].map(_.toString) |
| } yield { |
| Index(indexName, propNames, direction, options) |
| } |
| |
| val (pk, others) = indices.partition(index => index.name == LabelIndex.DefaultName) |
| val (both, inOut) = others.partition(index => index.direction.isEmpty) |
| val (in, out) = inOut.partition(index => index.direction.get == GraphUtil.directions("in")) |
| |
| pk ++ both ++ in ++ out |
| } |
| |
| def toLabelElements(jsValue: JsValue): Try[Label] = { |
| val labelName = parse[String](jsValue, "label") |
| val srcServiceName = parse[String](jsValue, "srcServiceName") |
| val tgtServiceName = parse[String](jsValue, "tgtServiceName") |
| val srcColumnName = parse[String](jsValue, "srcColumnName") |
| val tgtColumnName = parse[String](jsValue, "tgtColumnName") |
| val srcColumnType = parse[String](jsValue, "srcColumnType") |
| val tgtColumnType = parse[String](jsValue, "tgtColumnType") |
| val serviceName = (jsValue \ "serviceName").asOpt[String].getOrElse(tgtServiceName) |
| val isDirected = (jsValue \ "isDirected").asOpt[Boolean].getOrElse(true) |
| |
| val allProps = toPropsElements(jsValue \ "props") |
| val indices = toIndicesElements(jsValue \ "indices") |
| |
| val consistencyLevel = (jsValue \ "consistencyLevel").asOpt[String].getOrElse("weak") |
| |
| // expect new label don`t provide hTableName |
| val hTableName = (jsValue \ "hTableName").asOpt[String] |
| val hTableTTL = (jsValue \ "hTableTTL").asOpt[Int] |
| val schemaVersion = (jsValue \ "schemaVersion").asOpt[String].getOrElse(HBaseType.DEFAULT_VERSION) |
| val isAsync = (jsValue \ "isAsync").asOpt[Boolean].getOrElse(false) |
| val compressionAlgorithm = (jsValue \ "compressionAlgorithm").asOpt[String].getOrElse(DefaultCompressionAlgorithm) |
| val options = (jsValue \ "options").asOpt[JsValue].map(_.toString()) |
| |
| graph.management.createLabel(labelName, srcServiceName, srcColumnName, srcColumnType, |
| tgtServiceName, tgtColumnName, tgtColumnType, isDirected, serviceName, |
| indices, allProps, consistencyLevel, hTableName, hTableTTL, schemaVersion, isAsync, compressionAlgorithm, options) |
| } |
| |
| def toIndexElements(jsValue: JsValue) = Try { |
| val labelName = parse[String](jsValue, "label") |
| val indices = toIndicesElements(jsValue \ "indices") |
| (labelName, indices) |
| } |
| |
| def toServiceElements(jsValue: JsValue) = { |
| val serviceName = parse[String](jsValue, "serviceName") |
| val cluster = (jsValue \ "cluster").asOpt[String].getOrElse(DefaultCluster) |
| val hTableName = (jsValue \ "hTableName").asOpt[String].getOrElse(s"${serviceName}-${DefaultPhase}") |
| val preSplitSize = (jsValue \ "preSplitSize").asOpt[Int].getOrElse(1) |
| val hTableTTL = (jsValue \ "hTableTTL").asOpt[Int] |
| val compressionAlgorithm = (jsValue \ "compressionAlgorithm").asOpt[String].getOrElse(DefaultCompressionAlgorithm) |
| (serviceName, cluster, hTableName, preSplitSize, hTableTTL, compressionAlgorithm) |
| } |
| |
| def toServiceColumnElements(jsValue: JsValue) = Try { |
| val serviceName = parse[String](jsValue, "serviceName") |
| val columnName = parse[String](jsValue, "columnName") |
| val columnType = parse[String](jsValue, "columnType") |
| val props = toPropsElements(jsValue \ "props") |
| (serviceName, columnName, columnType, props) |
| } |
| |
| def toCheckEdgeParam(jsValue: JsValue) = { |
| for { |
| json <- jsValue.asOpt[Seq[JsValue]].getOrElse(Nil) |
| from <- (json \ "from").asOpt[JsValue].flatMap(jsValueToAny(_)) |
| to <- (json \ "to").asOpt[JsValue].flatMap(jsValueToAny(_)) |
| labelName <- (json \ "label").asOpt[String] |
| direction = (json \ "direction").asOpt[String].getOrElse("out") |
| } yield { |
| graph.toEdge(from, to, labelName, direction, Map.empty) |
| } |
| } |
| |
| def toGraphElements(str: String): Seq[GraphElement] = { |
| val edgeStrs = str.split("\\n") |
| |
| for { |
| edgeStr <- edgeStrs |
| str <- GraphUtil.parseString(edgeStr) |
| element <- graph.toGraphElement(str) |
| } yield element |
| } |
| |
| def toDeleteParam(json: JsValue) = { |
| val labelName = (json \ "label").as[String] |
| val labels = Label.findByName(labelName).map { l => Seq(l) }.getOrElse(Nil) |
| val direction = (json \ "direction").asOpt[String].getOrElse("out") |
| |
| val ids = (json \ "ids").asOpt[List[JsValue]].getOrElse(Nil) |
| val ts = (json \ "timestamp").asOpt[Long].getOrElse(System.currentTimeMillis()) |
| val vertices = toVertices(labelName, direction, ids) |
| (labels, direction, ids, ts, vertices) |
| } |
| |
| def toFetchAndDeleteParam(json: JsValue) = { |
| val labelName = (json \ "label").as[String] |
| val fromOpt = (json \ "from").asOpt[JsValue] |
| val toOpt = (json \ "to").asOpt[JsValue] |
| val direction = (json \ "direction").asOpt[String].getOrElse("out") |
| val indexOpt = (json \ "index").asOpt[String] |
| val propsOpt = (json \ "props").asOpt[JsObject] |
| (labelName, fromOpt, toOpt, direction, indexOpt, propsOpt) |
| } |
| |
| def parseExperiment(jsQuery: JsValue): Seq[ExperimentParam] = jsQuery.as[Seq[JsObject]].map { obj => |
| def _require(field: String) = throw new RuntimeException(s"${field} not found") |
| |
| val accessToken = (obj \ "accessToken").asOpt[String].getOrElse(_require("accessToken")) |
| val experimentName = (obj \ "experiment").asOpt[String].getOrElse(_require("experiment")) |
| val uuid = (obj \ "#uuid").get match { |
| case JsString(s) => s |
| case JsNumber(n) => n.toString |
| case _ => _require("#uuid") |
| } |
| val body = (obj \ "params").asOpt[JsObject].getOrElse(Json.obj()) |
| val impKeyOpt = (obj \ Experiment.ImpressionKey).asOpt[String] |
| |
| (body, accessToken, experimentName, uuid, impKeyOpt) |
| } |
| } |
