merging https://github.com/daumkakao/s2graph/pull/14 into master
diff --git a/app/com/daumkakao/s2graph/rest/config/Config.scala b/app/com/daumkakao/s2graph/rest/config/Config.scala
index 3be7b28..4bef00a 100644
--- a/app/com/daumkakao/s2graph/rest/config/Config.scala
+++ b/app/com/daumkakao/s2graph/rest/config/Config.scala
@@ -1,7 +1,8 @@
package com.daumkakao.s2graph.rest.config
import java.util.concurrent.TimeUnit
-import com.codahale.metrics.{Metric, ConsoleReporter, MetricRegistry}
+import com.codahale.metrics.{Slf4jReporter, Metric, ConsoleReporter, MetricRegistry}
+import org.slf4j.LoggerFactory
import play.api.{Play, Logger}
object Config {
@@ -32,10 +33,15 @@
trait Instrumented extends nl.grons.metrics.scala.InstrumentedBuilder {
val metricRegistry: MetricRegistry = Config.metricRegistry
- val consoleReporter = ConsoleReporter.forRegistry(metricRegistry)
+ val reporter = Slf4jReporter.forRegistry(metricRegistry)
+ .outputTo(LoggerFactory.getLogger(classOf[Instrumented]))
.convertRatesTo(TimeUnit.SECONDS)
.convertDurationsTo(TimeUnit.MILLISECONDS)
.build()
+// val consoleReporter = ConsoleReporter.forRegistry(metricRegistry)
+// .convertRatesTo(TimeUnit.SECONDS)
+// .convertDurationsTo(TimeUnit.MILLISECONDS)
+// .build()
val stats = new collection.mutable.HashMap[String, Metric]
diff --git a/app/controllers/ApplicationController.scala b/app/controllers/ApplicationController.scala
index 8384fd9..ab08044 100644
--- a/app/controllers/ApplicationController.scala
+++ b/app/controllers/ApplicationController.scala
@@ -17,7 +17,7 @@
var connectionOption = CONNECTION -> "Keep-Alive"
var keepAliveOption = "Keep-Alive" -> "timeout=5, max=100"
- consoleReporter.start(10, TimeUnit.SECONDS)
+ reporter.start(10, TimeUnit.SECONDS)
def updateHealthCheck(isHealthy: Boolean) = Action { request =>
this.isHealthy = isHealthy
diff --git a/app/controllers/RequestParser.scala b/app/controllers/RequestParser.scala
index 945368b..c5af045 100644
--- a/app/controllers/RequestParser.scala
+++ b/app/controllers/RequestParser.scala
@@ -84,59 +84,6 @@
WhereParser(label).parse(where)
}
}
- case class WhereParser(label: Label) extends JavaTokenParsers with JSONParser {
-
- val metaProps = label.metaPropsInvMap ++ Map(LabelMeta.from.name -> LabelMeta.from, LabelMeta.to.name -> LabelMeta.to)
-
- def where: Parser[Where] = rep(clause) ^^ (Where(_))
-
- def clause: Parser[Clause] = (predicate | parens) * (
- "and" ^^^ { (a: Clause, b: Clause) => And(a, b) } |
- "or" ^^^ { (a: Clause, b: Clause) => Or(a, b) })
-
- def parens: Parser[Clause] = "(" ~> clause <~ ")"
-
- def boolean = ("true" ^^^ (true) | "false" ^^^ (false))
-
- /** floating point is not supported yet **/
- def predicate = (
- (ident ~ "=" ~ ident | ident ~ "=" ~ decimalNumber | ident ~ "=" ~ stringLiteral) ^^ {
- case f ~ "=" ~ s =>
- metaProps.get(f) match {
- case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
- case Some(metaProp) =>
- Equal(metaProp.seq, toInnerVal(s, metaProp.dataType))
- }
-
- }
- | (ident ~ "between" ~ ident ~ "and" ~ ident | ident ~ "between" ~ decimalNumber ~ "and" ~ decimalNumber
- | ident ~ "between" ~ stringLiteral ~ "and" ~ stringLiteral) ^^ {
- case f ~ "between" ~ minV ~ "and" ~ maxV =>
- metaProps.get(f) match {
- case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
- case Some(metaProp) =>
- Between(metaProp.seq, toInnerVal(minV, metaProp.dataType), toInnerVal(maxV, metaProp.dataType))
- }
- }
- | (ident ~ "in" ~ "(" ~ rep(ident | decimalNumber | stringLiteral | "true" | "false" | ",") ~ ")") ^^ {
- case f ~ "in" ~ "(" ~ vals ~ ")" =>
- metaProps.get(f) match {
- case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
- case Some(metaProp) =>
- val values = vals.filter(v => v != ",").map { v =>
- toInnerVal(v, metaProp.dataType)
- }
- IN(metaProp.seq, values.toSet)
- }
- })
-
- def parse(sql: String): Option[Where] = {
- parseAll(where, sql) match {
- case Success(r, q) => Some(r)
- case x => println(x); None
- }
- }
- }
def toQuery(jsValue: JsValue, isEdgeQuery: Boolean = true): Query = {
try {
diff --git a/loader/src/main/scala/subscriber/GraphSubscriber.scala b/loader/src/main/scala/subscriber/GraphSubscriber.scala
index b1d6051..aec5c4f 100644
--- a/loader/src/main/scala/subscriber/GraphSubscriber.scala
+++ b/loader/src/main/scala/subscriber/GraphSubscriber.scala
@@ -4,11 +4,11 @@
import com.daumkakao.s2graph.core._
import com.daumkakao.s2graph.core.Graph
-import com.typesafe.config.ConfigFactory
+import com.typesafe.config.{Config, ConfigFactory}
import kafka.javaapi.producer.Producer
import kafka.producer.KeyedMessage
import org.apache.hadoop.hbase.HBaseConfiguration
-import org.apache.hadoop.hbase.client.{HConnection, HConnectionManager}
+import org.apache.hadoop.hbase.client.{HTableInterface, HConnection, HConnectionManager}
import org.apache.spark.{Accumulable, SparkContext}
import s2.spark.{HashMapParam, SparkApp, WithKafka}
import subscriber.parser.{GraphParser, GraphParsers}
@@ -26,67 +26,39 @@
zkQuorum = zkAddr.getOrElse("localhost")
kafkaBrokers = kafkaBrokerList.getOrElse("localhost:9092")
val s = s"""
-db.default.driver=com.mysql.jdbc.Driver
-db.default.url="$database"
-db.default.user=graph
-db.default.password=graph
+ |logger.root=ERROR
+ |
+ |# Logger used by the framework:
+ |logger.play=INFO
+ |
+ |# Logger provided to your application:
+ |logger.application=DEBUG
+ |
+ |# APP PHASE
+ |phase=dev
+ |
+ |# DB
+ |db.default.driver=com.mysql.jdbc.Driver
+ |db.default.url="$database"
+ |db.default.user=graph
+ |db.default.password=graph
+ |
+ |# Query server
+ |is.query.server=true
+ |is.write.server=true
+ |
+ |# Local Cache
+ |cache.ttl.seconds=60
+ |cache.max.size=100000
+ |
+ |# HBASE
+ |hbase.client.operation.timeout=60000
+ |
+ |# Kafka
+ |kafka.metadata.broker.list="$kafkaBrokers"
+ |kafka.producer.pool.size=0
+ | """.stripMargin
-is.query.server=true
-is.analyzer=false
-is.test.query.server=false
-test.sample.prob=0.1
-
-cache.ttl.seconds=60000
-cache.max.size=100000
-
-hbase.connection.pool.size=1
-hbase.table.pool.size=10
-hbase.client.ipc.pool.size=1
-zookeeper.recovery.retry=10
-zookeeper.session.timeout=180000
-hbase.zookeeper.quorum="$zkQuorum"
-hbase.table.name="s2graph-alpha"
-hbase.client.operation.timeout=10000
-hbase.client.retries.number=10
-hbase.client.write.operation.timeout=10000
-hbase.client.write.retries.number=10
-
-kafka.metadata.broker.list="$kafkaBrokers"
-kafka.request.required.acks=1
-kafka.producer.type="sync"
-kafka.producer.buffer.flush.time=1000
-kafka.producer.buffer.size=1000
-kafka.producer.pool.size=1
-kafka.aggregate.flush.timeout=1000
-
-# Aggregator
-client.aggregate.buffer.size=100
-client.aggregate.buffer.flush.time=10000
-client.aggregate.pool.size=1
-
-
-# blocking execution context
-contexts {
- query {
- fork-join-executor {
- parallelism-min = 1
- parallelism-max = 1
- }
- }
- blocking {
- fork-join-executor {
- parallelism-min = 1
- parallelism-max = 1
- }
- }
- scheduler {
- fork-join-executor {
- parallelism-min = 1
- parallelism-max = 1
- }
- }
-}
- """
println(s)
ConfigFactory.parseString(s)
}
@@ -98,17 +70,11 @@
lazy val producer = new Producer[String, String](kafkaConf(GraphConfig.kafkaBrokers))
-
+ var config: Config = _
private val writeBufferSize = 1024 * 1024 * 8
private val sleepPeriod = 10000
private val maxTryNum = 10
-// lazy val graph = {
-// println(System.getProperty("phase"))
-// Graph.apply(GraphConfig.apply(System.getProperty("phase"), None, None))(ExecutionContext.Implicits.global)
-// println(Graph.config)
-// println(Graph.hbaseConfig)
-// Graph
-// }
+
def toOption(s: String) = {
s match {
case "" | "none" => None
@@ -116,16 +82,14 @@
}
}
def apply(phase: String, dbUrl: String, zkQuorum: String, kafkaBrokerList: String) : Unit = {
- apply(phase, toOption(dbUrl), toOption(zkQuorum), toOption(kafkaBrokerList))
+ config = GraphConfig(phase, toOption(dbUrl), toOption(zkQuorum), toOption(kafkaBrokerList))
+ Graph.apply(config)(ExecutionContext.Implicits.global)
}
- def apply(phase: String, dbUrl: Option[String], zkQuorum: Option[String], kafkaBrokerList: Option[String]): Unit = {
- Graph.apply(GraphConfig(phase, dbUrl, zkQuorum, kafkaBrokerList))(ExecutionContext.Implicits.global)
- }
+// def apply(phase: String, dbUrl: Option[String], zkQuorum: Option[String], kafkaBrokerList: Option[String]): Unit = {
+// Graph.apply(GraphConfig(phase, dbUrl, zkQuorum, kafkaBrokerList))(ExecutionContext.Implicits.global)
+// }
def report(key: String, value: Option[String], topic: String = "report") = {
- // val ts = System.currentTimeMillis().toString
val msg = Seq(Some(key), value).flatten.mkString("\t")
-
- // val producer = new Producer[String, String](kafkaConf(Config.KAFKA_METADATA_BROKER_LIST))
val kafkaMsg = new KeyedMessage[String, String](topic, msg)
producer.send(kafkaMsg)
}
@@ -133,15 +97,15 @@
/**
* bulkMutates read connection and table info from database.
*/
- def store(msgs: Seq[String])(mapAccOpt: Option[HashMapAccumulable]): Iterable[(String, Long)] = {
- // assert(msgs.size >= maxSize)
+ def store(msgs: Seq[String], labelToReplace: Option[String] = None)(mapAccOpt: Option[HashMapAccumulable]): Iterable[(String, Long)] = {
val counts = HashMap[String, Long]()
val statFunc = storeStat(counts)(mapAccOpt)_
val elements = (for (msg <- msgs) yield {
statFunc("total", 1)
- val element = Graph.toGraphElement(msg)
+ val element = Graph.toGraphElement(msg, labelToReplace)
+ println(element)
element match {
case Some(e) =>
statFunc("parseOk", 1)
@@ -167,6 +131,64 @@
}
counts
}
+ def storeBulk(conn: HConnection, tableName: String)(msgs: Seq[String], labelToReplace: Option[String] = None)(mapAccOpt: Option[HashMapAccumulable]): Iterable[(String, Long)] = {
+ val counts = HashMap[String, Long]()
+ val statFunc = storeStat(counts)(mapAccOpt) _
+ val edges = (for (msg <- msgs) yield {
+ statFunc("total", 1)
+ try {
+ Graph.toGraphElement(msg, labelToReplace) match {
+ case Some(e) if e.isInstanceOf[Edge] =>
+ statFunc("parseOk", 1)
+ Some(e.asInstanceOf[Edge])
+ case None =>
+ statFunc("errorParsing", 1)
+ None
+ }
+ } catch {
+ case e: Throwable =>
+ System.err.println(s"$msg $e", e)
+ None
+ }
+ }).flatten.toList
+
+ println(edges.take(10))
+
+ /**
+ * don't use database for connection and table.
+ */
+ // throw all exception to caller.
+
+ def storeRec(edges: List[Edge], tryNum: Int = maxTryNum): Unit = {
+ if (tryNum <= 0) {
+ statFunc("errorStore", edges.size)
+ throw new RuntimeException(s"retry failed after $maxTryNum")
+ }
+ val table = conn.getTable(tableName)
+ table.setAutoFlush(false, false)
+ table.setWriteBufferSize(writeBufferSize)
+ try {
+ // we only care about insertBulk
+ val puts = edges.flatMap { edge =>
+ edge.relatedEdges.flatMap { relatedEdge =>
+ relatedEdge.insertBulk() ++ relatedEdge.buildVertexPuts()
+ }
+ }
+ println(puts.take(10))
+ table.put(puts)
+ statFunc("storeOk", msgs.size)
+ } catch {
+ case e: Throwable =>
+ e.printStackTrace()
+ Thread.sleep(sleepPeriod)
+ storeRec(edges, tryNum - 1)
+ } finally {
+ table.close()
+ }
+ }
+ storeRec(edges)
+ counts
+ }
def storeStat(counts: HashMap[String, Long])(mapAccOpt: Option[HashMapAccumulable])(key: String, value: Int) = {
counts.put(key, counts.getOrElse(key, 0L) + value)
mapAccOpt match {
@@ -174,108 +196,62 @@
case Some(mapAcc) => mapAcc += (key -> value)
}
}
-//
-// /**
-// * caller of this method should split msgs into reasonable size.
-// */
-// def storeBulk(conn: HConnection, msgs: Seq[String])(mapAccOpt: Option[HashMapAccumulable], zkQuorum: String, tableName: String): Iterable[(String, Long)] = {
-// val counts = HashMap[String, Long]()
-// val statFunc = storeStat(counts)(mapAccOpt)_
-// val edges = (for (msg <- msgs) yield {
-// statFunc("total", 1)
-//
-// val edge = Graph.toEdge(msg)
-// edge match {
-// case Some(e) =>
-// statFunc("parseOk", 1)
-// edge
-// case None =>
-// statFunc("errorParsing", 1)
-// None
-// }
-// }).flatten.toList
-//
-// storeRec(edges)
-// /**
-// * don't use database for connection and table.
-// */
-// // throw all exception to caller.
-//
-// def storeRec(edges: List[Edge], tryNum: Int = maxTryNum): Unit = {
-// if (tryNum <= 0) {
-// statFunc("errorStore", edges.size)
-// throw new RuntimeException(s"retry failed after $maxTryNum")
-// }
-// try {
-//
-// }
-// try {
-//
-// Graph.bulkMutates(edges)
-// // on bulk mode, we don`t actually care about WAL log on hbase. so skip this process to make things faster.
-// val puts = edges.flatMap(e => e.buildPutsAll ++ e.buildVertexPuts).map { p =>
-// // p.setDurability(Durability.SKIP_WAL)
-// p
-// }
-// table.put(puts)
-// statFunc("storeOk", msgs.size)
-// } catch {
-// case e: Throwable =>
-// e.printStackTrace()
-// Thread.sleep(sleepPeriod)
-// storeRec(edges, tryNum - 1)
-// // statFunc("errorStore", msgs.size)
-// // throw e
-// } finally {
-// table.close()
-// }
-// }
-//
-// counts
-// }
+
}
-/**
- * do not use Graph.bulkMutates since it automatically read zkQuorum and hbase TableName from database.
- * batch insert should reference database only for parsing not getting connection and table!
- */
object GraphSubscriber extends SparkApp with WithKafka {
val sleepPeriod = 5000
-
+ val usages =
+ s"""
+ |/**
+ | * this job read edge format(TSV) from HDFS file system then bulk load edges into s2graph. assumes that newLabelName is already created by API.
+ | * params:
+ | * 1. hdfsPath: where is your data in hdfs. require full path with hdfs:// predix
+ | * 2. dbUrl: jdbc database connection string to specify database for meta.
+ | * 3. originalLabelName: label to copy with.
+ | * 4. newLabelName: label field will be replaced to this value.
+ | * 5. zkQuorum: target hbase zkQuorum where this job will publish data to.
+ | * 6. hTableName: target hbase physical table name where this job will publish data to.
+ | * 7. batchSize: how many edges will be batched for Put request to target hbase.
+ | * 8. kafkaBrokerList: using kafka as fallback queue. when something goes wrong during batch, data needs to be replay will be stored in kafka.
+ | * 9. kafkaTopic: fallback queue topic.
+ | * after this job finished, s2graph will have data with sequence corresponding newLabelName.
+ | * change this newLabelName to ogirinalName if you want to online replace of label.
+ | *
+ | */
+ """.stripMargin
override def run() = {
/**
* Main function
*/
- if (args.length < 3) {
- System.err.println("Usage: GraphSubscriber <hdfsPath> <batchSize> <dbUrl> <zkQuorum> <tableName> <kafkaBrokerList> <kafkaTopic> <preSplitSize>")
+ println(args.toList)
+ if (args.length != 9) {
+ System.err.println(usages)
System.exit(1)
}
val hdfsPath = args(0)
- val batchSize = args(1).toInt
- val dbUrl = args(2)
- val zkQuorum = args(3)
- val tableName = args(4)
- val kafkaBrokerList = args(5)
- val kafkaTopic = args(6)
- val preSplitSize = if (args.length > 7) args(7).toInt else 20
+ val dbUrl = args(1)
+ val oldLabelName = args(2)
+ val newLabelName = args(3)
+ val zkQuorum = args(4)
+ val hTableName = args(5)
+ val batchSize = args(6).toInt
+ val kafkaBrokerList = args(7)
+ val kafkaTopic = args(8)
val conf = sparkConf(s"$hdfsPath: GraphSubscriber")
val sc = new SparkContext(conf)
-
val mapAcc = sc.accumulable(HashMap.empty[String, Long], "counter")(HashMapParam[String, Long](_ + _))
- val fallbackTopic = s"${tableName}_batch_failed"
+
+
try {
- tableName match {
- case "s2graph" | "s2graph-alpha" | "s2graph-sandbox" => System.err.println("try to create master table!")
- case _ => Management.createTable(zkQuorum, tableName, List("e", "v"), preSplitSize, None)
- }
- // not sure how fast htable get table can recognize new table so sleep.
- Thread.sleep(sleepPeriod)
-
+ import GraphSubscriberHelper._
// set local driver setting.
val phase = System.getProperty("phase")
GraphSubscriberHelper.apply(phase, dbUrl, zkQuorum, kafkaBrokerList)
+ /** copy when oldLabel exist and newLabel done exist. otherwise ignore. */
+ Management.copyLabel(oldLabelName, newLabelName, toOption(hTableName))
val msgs = sc.textFile(hdfsPath)
msgs.foreachPartition(partition => {
@@ -290,21 +266,22 @@
partition.grouped(batchSize).foreach { msgs =>
try {
val start = System.currentTimeMillis()
+// val counts =
+// GraphSubscriberHelper.store(msgs, GraphSubscriberHelper.toOption(newLabelName))(Some(mapAcc))
val counts =
-// if (isBulk) GraphSubscriberHelper.storeBulk(conn, msgs)(Some(mapAcc), zkQuorum, tableName)
- GraphSubscriberHelper.store(msgs)(Some(mapAcc))
+ GraphSubscriberHelper.storeBulk(conn, hTableName)(msgs, GraphSubscriberHelper.toOption(newLabelName))(Some(mapAcc))
for ((k, v) <- counts) {
mapAcc += (k, v)
}
val duration = System.currentTimeMillis() - start
- println(s"[Success]: store, $mapAcc, $duration, $zkQuorum, $tableName")
+ println(s"[Success]: store, $mapAcc, $duration, $zkQuorum, $hTableName")
} catch {
case e: Throwable =>
println(s"[Failed]: store $e")
msgs.foreach { msg =>
- GraphSubscriberHelper.report(msg, Some(e.getMessage()), topic = fallbackTopic)
+ GraphSubscriberHelper.report(msg, Some(e.getMessage()), topic = kafkaTopic)
}
}
}
@@ -314,9 +291,8 @@
logInfo(s"counter: $mapAcc")
println(s"Stats: ${mapAcc}")
- // if (shouldUpdate) Label.prependHBaseTableName(labelName, tableName)
} catch {
- case e: IOException =>
+ case e: Throwable =>
println(s"job failed with exception: $e")
throw e
}
diff --git a/loader/src/test/scala/subscriber/GraphSubscriberTest.scala b/loader/src/test/scala/subscriber/GraphSubscriberTest.scala
index 13a3084..4e5b60a 100644
--- a/loader/src/test/scala/subscriber/GraphSubscriberTest.scala
+++ b/loader/src/test/scala/subscriber/GraphSubscriberTest.scala
@@ -1,20 +1,44 @@
package subscriber
+import com.daumkakao.s2graph.core.{Label, Service, Management}
import org.scalatest.{ FunSuite, Matchers }
+import play.api.libs.json.{JsBoolean, JsNumber}
import s2.spark.WithKafka
-import kafka.javaapi.producer.Producer
-import config.Config
class GraphSubscriberTest extends FunSuite with Matchers with WithKafka {
-
- test("kafka producer") {
- val brokerList = Config.KAFKA_METADATA_BROKER_LIST
- println(brokerList)
- val producer = new Producer[String, String](kafkaConf(brokerList))
- println(producer)
- }
-
+ val phase = "dev"
+ val dbUrl = "jdbc:mysql://localhost:3306/graph_dev"
+ val zkQuorum = "localhost"
+ val kafkaBrokerList = "localhost:9099"
+ val currentTs = System.currentTimeMillis()
+ val op = "insertBulk"
+ val testLabelName = "s2graph_label_test"
+ val labelToReplace = "s2graph_label_test_new"
+ val serviceName = "s2graph"
+ val columnName = "user_id"
+ val columnType = "long"
+ val indexProps = Seq("time" -> JsNumber(0), "weight" -> JsNumber(0))
+ val props = Seq("is_hidden" -> JsBoolean(false), "is_blocked" -> JsBoolean(false))
+ val hTableName = "s2graph-dev_new"
+ val ttl = 86000
+ val testStrings = List("1431788400000\tinsertBulk\te\t147229417\t99240432\ts2graph_label_test\t{\"is_hidden\": true}")
+
+ GraphSubscriberHelper.apply(phase, dbUrl, zkQuorum, kafkaBrokerList)
+
test("GraphSubscriberHelper.store") {
-
+ // actually we need to delete labelToReplace first for each test.
+ Management.copyLabel(testLabelName, labelToReplace, Some(hTableName))
+
+//
+// val msgs = (for {
+// i <- (1 until 10)
+// j <- (100 until 110)
+// } yield {
+// s"$currentTs\t$op\tedge\t$i\t$j\t$testLabelName"
+// }).toSeq
+ val msgs = testStrings
+
+ val stat = GraphSubscriberHelper.store(msgs, Some(labelToReplace))(None)
+ println(stat)
}
}
\ No newline at end of file
diff --git a/s2core/build.sbt b/s2core/build.sbt
index c27a05c..716fbdf 100644
--- a/s2core/build.sbt
+++ b/s2core/build.sbt
@@ -27,6 +27,7 @@
"commons-pool" % "commons-pool" % "1.6",
"org.scalikejdbc" %% "scalikejdbc" % "2.1.+",
"com.twitter" %% "util-collection" % "6.12.1",
- "org.hbase" % "asynchbase" % "1.7.0-SNAPSHOT"
+ "org.hbase" % "asynchbase" % "1.7.0-SNAPSHOT",
+ "org.scalatest" %% "scalatest" % "2.2.1" % "test"
)
diff --git a/s2core/src/main/scala/com/daumkakao/s2graph/core/Edge.scala b/s2core/src/main/scala/com/daumkakao/s2graph/core/Edge.scala
index 6bb9081..dd73fd0 100644
--- a/s2core/src/main/scala/com/daumkakao/s2graph/core/Edge.scala
+++ b/s2core/src/main/scala/com/daumkakao/s2graph/core/Edge.scala
@@ -261,7 +261,9 @@
}
edgePuts
}
-
+ def insertBulk() = {
+ edgesWithInvertedIndex.buildPut :: edgesWithIndex.flatMap(e => e.buildPuts())
+ }
def insert() = {
edgesWithInvertedIndex.buildPutAsync :: edgesWithIndex.flatMap(e => e.buildPutsAsync)
}
diff --git a/s2core/src/main/scala/com/daumkakao/s2graph/core/Graph.scala b/s2core/src/main/scala/com/daumkakao/s2graph/core/Graph.scala
index 8a8240e..b528834 100644
--- a/s2core/src/main/scala/com/daumkakao/s2graph/core/Graph.scala
+++ b/s2core/src/main/scala/com/daumkakao/s2graph/core/Graph.scala
@@ -765,11 +765,17 @@
*
*/
- def toGraphElement(s: String): Option[GraphElement] = {
+ def toGraphElement(s: String, labelNameToReplace: Option[String] = None): Option[GraphElement] = {
val parts = GraphUtil.split(s)
try {
val logType = parts(2)
val element = if (logType == "edge" | logType == "e") {
+ /** current only edge is considered to be bulk loaded */
+ labelNameToReplace match {
+ case None =>
+ case Some(toReplace) =>
+ parts(5) = toReplace
+ }
toEdge(parts)
} else if (logType == "vertex" | logType == "v") {
toVertex(parts)
@@ -779,7 +785,6 @@
element
} catch {
case e: Throwable =>
- logger.error(s"toGraphElement: $s => $e", e)
None
}
}
diff --git a/s2core/src/main/scala/com/daumkakao/s2graph/core/JSONParser.scala b/s2core/src/main/scala/com/daumkakao/s2graph/core/JSONParser.scala
index 8b1b085..4c81e35 100644
--- a/s2core/src/main/scala/com/daumkakao/s2graph/core/JSONParser.scala
+++ b/s2core/src/main/scala/com/daumkakao/s2graph/core/JSONParser.scala
@@ -3,6 +3,8 @@
import play.api.libs.json._
import HBaseElement.InnerVal
+import scala.util.parsing.combinator.JavaTokenParsers
+
trait JSONParser {
def innerValToJsValue(innerVal: InnerVal): JsValue = {
@@ -76,4 +78,76 @@
case _ => value.toString
}
}
+ case class WhereParser(label: Label) extends JavaTokenParsers with JSONParser {
+
+ val metaProps = label.metaPropsInvMap ++ Map(LabelMeta.from.name -> LabelMeta.from, LabelMeta.to.name -> LabelMeta.to)
+
+ def where: Parser[Where] = rep(clause) ^^ (Where(_))
+
+ def clause: Parser[Clause] = (predicate | parens) * (
+ "and" ^^^ { (a: Clause, b: Clause) => And(a, b) } |
+ "or" ^^^ { (a: Clause, b: Clause) => Or(a, b) })
+
+ def parens: Parser[Clause] = "(" ~> clause <~ ")"
+
+ def boolean = ("true" ^^^ (true) | "false" ^^^ (false))
+
+ /** floating point is not supported yet **/
+ def predicate = (
+ (ident ~ "=" ~ ident | ident ~ "=" ~ decimalNumber | ident ~ "=" ~ stringLiteral) ^^ {
+ case f ~ "=" ~ s =>
+ metaProps.get(f) match {
+ case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
+ case Some(metaProp) =>
+ Equal(metaProp.seq, toInnerVal(s, metaProp.dataType))
+ }
+ }
+ | (ident ~ "between" ~ ident ~ "and" ~ ident | ident ~ "between" ~ decimalNumber ~ "and" ~ decimalNumber
+ | ident ~ "between" ~ stringLiteral ~ "and" ~ stringLiteral) ^^ {
+ case f ~ "between" ~ minV ~ "and" ~ maxV =>
+ metaProps.get(f) match {
+ case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
+ case Some(metaProp) =>
+ Between(metaProp.seq, toInnerVal(minV, metaProp.dataType), toInnerVal(maxV, metaProp.dataType))
+ }
+ }
+ | (ident ~ "in" ~ "(" ~ rep(ident | decimalNumber | stringLiteral | "true" | "false" | ",") ~ ")") ^^ {
+ case f ~ "in" ~ "(" ~ vals ~ ")" =>
+ metaProps.get(f) match {
+ case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
+ case Some(metaProp) =>
+ val values = vals.filter(v => v != ",").map { v =>
+ toInnerVal(v, metaProp.dataType)
+ }
+ IN(metaProp.seq, values.toSet)
+ }
+ }
+ | (ident ~ "!=" ~ ident | ident ~ "!=" ~ decimalNumber | ident ~ "!=" ~ stringLiteral) ^^ {
+ case f ~ "!=" ~ s =>
+ metaProps.get(f) match {
+ case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
+ case Some(metaProp) =>
+ Not(Equal(metaProp.seq, toInnerVal(s, metaProp.dataType)))
+ }
+ }
+ | (ident ~ "not in" ~ "(" ~ rep(ident | decimalNumber | stringLiteral | "true" | "false" | ",") ~ ")") ^^ {
+ case f ~ "not in" ~ "(" ~ vals ~ ")" =>
+ metaProps.get(f) match {
+ case None => throw new RuntimeException(s"where clause contains not existing property name: $f")
+ case Some(metaProp) =>
+ val values = vals.filter(v => v != ",").map { v =>
+ toInnerVal(v, metaProp.dataType)
+ }
+ Not(IN(metaProp.seq, values.toSet))
+ }
+ }
+ )
+
+ def parse(sql: String): Option[Where] = {
+ parseAll(where, sql) match {
+ case Success(r, q) => Some(r)
+ case x => println(x); None
+ }
+ }
+ }
}
\ No newline at end of file
diff --git a/s2core/src/main/scala/com/daumkakao/s2graph/core/Labels.scala b/s2core/src/main/scala/com/daumkakao/s2graph/core/Labels.scala
index 39f293d..ad326c6 100644
--- a/s2core/src/main/scala/com/daumkakao/s2graph/core/Labels.scala
+++ b/s2core/src/main/scala/com/daumkakao/s2graph/core/Labels.scala
@@ -230,6 +230,8 @@
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)
/**
diff --git a/s2core/src/main/scala/com/daumkakao/s2graph/core/Management.scala b/s2core/src/main/scala/com/daumkakao/s2graph/core/Management.scala
index bc249ec..1f87a1a 100644
--- a/s2core/src/main/scala/com/daumkakao/s2graph/core/Management.scala
+++ b/s2core/src/main/scala/com/daumkakao/s2graph/core/Management.scala
@@ -29,7 +29,24 @@
*
* talk_friend 10 -> story:friend 10 ->
*/
+ def copyLabel(oldLabelName: String, newLabelName: String, hTableName: Option[String]) = {
+ for {
+ old <- Label.findByName(oldLabelName)
+ } {
+ Label.findByName(newLabelName) match {
+ case None =>
+ val (indexProps, metaProps) = old.metaPropsInvMap.partition(nameMeta => nameMeta._2.usedInIndex)
+ createLabel(newLabelName, old.srcService.serviceName, old.srcColumnName, old.srcColumnType,
+ old.tgtService.serviceName, old.tgtColumnName, old.tgtColumnType,
+ old.isDirected, old.serviceName,
+ indexProps.map(t => t._1 -> innerValToJsValue(t._2.defaultInnerVal)).toSeq,
+ metaProps.map(t => t._1 -> innerValToJsValue(t._2.defaultInnerVal)).toSeq,
+ old.consistencyLevel, hTableName, old.hTableTTL)
+ case Some(_) =>
+ }
+ }
+ }
/**
* label
*/
diff --git a/s2core/src/main/scala/com/daumkakao/s2graph/core/QueryParam.scala b/s2core/src/main/scala/com/daumkakao/s2graph/core/QueryParam.scala
index 33d1a39..92489ee 100644
--- a/s2core/src/main/scala/com/daumkakao/s2graph/core/QueryParam.scala
+++ b/s2core/src/main/scala/com/daumkakao/s2graph/core/QueryParam.scala
@@ -127,6 +127,11 @@
}
}
+case class Not(val self: Clause) extends Clause {
+ override def filter(edge: Edge): Boolean = {
+ !self.filter(edge)
+ }
+}
case class And(val left: Clause, val right: Clause) extends Clause {
override def filter(edge: Edge): Boolean = {
left.filter(edge) && right.filter(edge)
diff --git a/s2core/src/test/scala/com/daumkakao/s2graph/core/ManagementTest.scala b/s2core/src/test/scala/com/daumkakao/s2graph/core/ManagementTest.scala
new file mode 100644
index 0000000..26a33dc
--- /dev/null
+++ b/s2core/src/test/scala/com/daumkakao/s2graph/core/ManagementTest.scala
@@ -0,0 +1,29 @@
+package com.daumkakao.s2graph.core
+
+import com.typesafe.config.ConfigFactory
+import org.scalatest.{Matchers, FunSuite}
+
+import scala.concurrent.ExecutionContext
+
+/**
+ * Created by shon on 5/18/15.
+ */
+class ManagementTest extends FunSuite with Matchers {
+ val zkQuorum = "tokyo062.kr2.iwilab.com"
+ val database = "jdbc:mysql://nuk151.kr2.iwilab.com:13306/graph_alpha"
+ val config = ConfigFactory.parseString(
+ s"""
+ | db.default.url="$database"
+ | cache.ttl.second=60
+ | cache.max.size=1000
+ | hbase.client.operation.timeout=1000
+ | hbase.zookeeper.quorum="$zkQuorum"
+ """.stripMargin)
+ val labelToCopy = "graph_test"
+ val newLabelName = "graph_test_tc"
+ val hTableName = "graph_test_tc"
+ Graph.apply(config)(ExecutionContext.Implicits.global)
+ test("test copy label") {
+ Management.copyLabel(labelToCopy, newLabelName, Some(hTableName))
+ }
+}
diff --git a/test/controllers/FutureHelpersSpec.scala b/test/controllers/FutureHelpersSpec.scala
index 227257f..03c49e0 100644
--- a/test/controllers/FutureHelpersSpec.scala
+++ b/test/controllers/FutureHelpersSpec.scala
@@ -1,4 +1,6 @@
-package contollers;
+package test.controllers
+
+
//package kgraph
//import scala.concurrent.Await
//import scala.concurrent._
diff --git a/test/controllers/GraphSpec.scala b/test/controllers/GraphSpec.scala
deleted file mode 100644
index 30d7b5b..0000000
--- a/test/controllers/GraphSpec.scala
+++ /dev/null
@@ -1,101 +0,0 @@
-package controllers
-
-import com.daumkakao.s2graph.core.Management._
-import com.daumkakao.s2graph.core._
-import org.specs2.mutable.Specification
-import play.api.libs.json.Json
-import scala.Option.option2Iterable
-/**
- * test-only kgraph.GraphSpec
- */
-class GraphSpec extends Specification {
- sequential
-
- private val phase = "alpha"
- // Graph.tableName = tableName
- System.setProperty("phase", phase)
-// Config.phase = phase
- private val serviceName = "s2graph"
- private val labelName = "graph_test"
-
- val msgs = """
- ["1417079166891\tu\te\t156598730\t158524147\tgraph_test\t{\"is_blocked\":true}","1417079166000\td\te\t4079848\t54739565\tgraph_test","1417079166000\ti\te\t102735518\t49629673\tgraph_test","1417079166000\ti\te\t15120422\t160902634\tgraph_test","1417079166000\ti\te\t160251573\t7960026\tgraph_test","1417079167000\td\te\t816808\t47823955\tgraph_test","1417079166000\ti\te\t160251573\t43063061\tgraph_test","1417079167000\ti\te\t141391508\t151985238\tgraph_test","1417079167360\tu\te\t43237903\t49005293\tgraph_test\t{\"is_blocked\":true}","1417079167000\ti\te\t9788808\t167206573\tgraph_test","1417079167000\ti\te\t106040854\t166946583\tgraph_test","1417079166000\ti\te\t15120422\t35674614\tgraph_test","1417079167519\tu\te\t29460631\t28640554\tgraph_test\t{\"is_blocked\":false}","1417079166000\ti\te\t15120422\t6366975\tgraph_test","1417079167000\tu\te\t4610363\t146176454\tgraph_test\t{\"is_hidden\":true}","1417079167000\td\te\t4935659\t1430321\tgraph_test","1417079166000\ti\te\t15120422\t136919643\tgraph_test","1417079167000\td\te\t84547561\t17923231\tgraph_test","1417079167000\td\te\t108876864\t19695266\tgraph_test","1417079166000\td\te\t14644128\t11853672\tgraph_test","1417079167000\ti\te\t142522237\t167208738\tgraph_test"]"""
-
- val createService = """{"serviceName": "s2graph"}"""
-
- val createLabel = """
- {"label": "graph_test", "srcServiceName": "s2graph", "srcColumnName": "account_id", "srcColumnType": "long", "tgtServiceName": "s2graph", "tgtColumnName": "account_id", "tgtColumnType": "long", "indexProps": {"time": 0, "weight":0}}
- """
-
- // 0.
- AdminController.createServiceInner(Json.parse(createService))
- // 1.
- AdminController.deleteLabel(labelName)
- // 2.
- AdminController.createLabelInner(Json.parse(createLabel))
-
- val label = Label.findByName("graph_test").get
-
- val insertEdges = """
-[
- {"from":1,"to":101,"label":"graph_test","props":{"time":-1, "weight":10},"timestamp":193829192},
- {"from":1,"to":102,"label":"graph_test","props":{"time":0, "weight":11},"timestamp":193829193},
- {"from":1,"to":103,"label":"graph_test","props":{"time":1, "weight":12},"timestamp":193829194},
- {"from":1,"to":104,"label":"graph_test","props":{"time":-2, "weight":1},"timestamp":193829195}
-]
- """
-// val insertResults = List(
-// Edge(Vertex(CompositeId(label.srcColumn.id.get, InnerVal.withLong(1), true)), Vertex(CompositeId(label.tgtColumn.id.get, InnerVal.withLong(103), true)), LabelWithDirection(label.id.get, 0), 0, 193829192, )
-// )
- val insertEdgesTsv = """["193829192\ti\te\t1\t101\tgraph_test\t{\"time\":1, \"weight\": 10, \"is_hidden\": false}","193829192\ti\te\t1\t102\tgraph_test\t{\"time\":1, \"weight\": 19, \"is_hidden\": false}","193829192\ti\te\t1\t103\tgraph_test\t{\"time\":1, \"weight\": 10, \"is_hidden\": false}"]"""
-
- val updateEdges = """
-[
- {"from":1,"to":102,"label":"graph_test","timestamp":193829199, "props": {"is_hidden":true}},
- {"from":1,"to":102,"label":"graph_test","timestamp":193829299, "props": {"weight":100}}
-]
- """
- val incrementEdges = """
-[
- {"from":1,"to":102,"label":"graph_test","timestamp":193829399, "props": {"weight":100}},
- {"from":1,"to":102,"label":"graph_test","timestamp":193829499, "props": {"weight":100}},
- {"from":1,"to":103,"label":"graph_test","timestamp":193829599, "props": {"weight":1000}},
- {"from":1,"to":102,"label":"graph_test","timestamp":193829699, "props": {"weight":-200}},
- {"from":1,"to":102,"label":"graph_test","timestamp":193829699, "props": {"weight":400}}
-]
- """
-
- val queryEdges = """
-{
- "srcVertices": [{"columnName": "account_id", "id":1}],
- "steps": [
- [{"label": "graph_test", "direction": "out", "limit": 100, "scoring":{"time": 0, "weight": 1}}]
- ]
-}
- """
- val query = toQuery(Json.parse(queryEdges))
-
- "RequestParsor" should {
- "Json parsed edge should match Graph.toEdge" in {
- val jsons = Json.parse(insertEdges)
- val jsonEdges = toEdges(jsons, "insert")
- val tsvs = GraphUtil.parseString(insertEdgesTsv)
- val tsvEdges = tsvs.map(tsv => Graph.toEdge(tsv)).flatten
- jsonEdges must containTheSameElementsAs(tsvEdges)
- }
- }
-// "Graph" should {
-// "test insert" in {
-// EdgeController.tryMutates(Json.parse(insertEdges), "insert")
-// val edgesWithRank = Graph.getEdgesSync(query)
-// edgesWithRank must have size(1)
-// val head = edgesWithRank.head.toArray
-// head must have size(4)
-// val (firstEdge, firstRAnk) = head(0)
-// firstEdge.tgtVertex.innerId must beEqualTo(InnerVal.withLong(103))
-// firstEdge.ts must beEqualTo (193829194)
-//
-// true
-// }
-// }
-}
\ No newline at end of file
diff --git a/test/controllers/IntegritySpec.scala b/test/controllers/IntegritySpec.scala
index f4f916b..9a7b8ae 100644
--- a/test/controllers/IntegritySpec.scala
+++ b/test/controllers/IntegritySpec.scala
@@ -1,8 +1,10 @@
-package controllers
+package test.controllers
+
import com.daumkakao.s2graph.core._
import com.daumkakao.s2graph.rest.config.Config
import com.daumkakao.s2graph.rest.actors._
import com.twitter.io.exp.VarSource.Ok
+import controllers.AdminController
import org.omg.CosNaming.NamingContextPackage.NotFound
import scala.concurrent.{ Await, Future }
import scala.concurrent.duration._
@@ -450,9 +452,8 @@
def initialize = {
running(FakeApplication()) {
- GraphAggregatorActor.init()
- KafkaAggregatorActor.init()
- Graph(Config.conf)(ExecutionContext.Implicits.global)
+// KafkaAggregatorActor.init()
+ Graph(Config.conf.underlying)(ExecutionContext.Implicits.global)
// 1. createService
var result = AdminController.createServiceInner(Json.parse(createService))
@@ -479,8 +480,7 @@
// println(s">> Label deleted : $testLabelName, $result")
// 2. delete service ( currently NOT supported )
- GraphAggregatorActor.shutdown()
- KafkaAggregatorActor.shutdown()
+// KafkaAggregatorActor.shutdown()
}
}
diff --git a/test/controllers/QuerySpec.scala b/test/controllers/QuerySpec.scala
index 8bc4975..3f0a01a 100644
--- a/test/controllers/QuerySpec.scala
+++ b/test/controllers/QuerySpec.scala
@@ -1,15 +1,14 @@
-package controllers
+package test.controllers
-import com.daumkakao.s2graph.core.Management._
import com.daumkakao.s2graph.core._
import com.daumkakao.s2graph.rest.actors._
import com.daumkakao.s2graph.rest.config.Config
+import controllers.{AdminController, RequestParser}
import org.specs2.matcher.Matchers
import org.specs2.mutable.Specification
import play.api.libs.json.{ JsArray, JsObject, Json }
import play.api.test.Helpers._
import play.api.test._
-import com.wordnik.swagger.annotations.Api
import scala.Array.canBuildFrom
import scala.concurrent.ExecutionContext
@@ -53,7 +52,7 @@
// logger.debug(s">> Social Controller test Acts result : ${contentAsJson(acts)}")
status(acts) must equalTo(OK)
- contentType(acts) must beSome.which(_ == "text/plain")
+ contentType(acts) must beSome.which(_ == "application/json")
val jsRslt = contentAsJson(acts)
println("======")
println(jsRslt)
@@ -101,7 +100,7 @@
// logger.debug(s">> Social Controller test Acts result : ${contentAsJson(acts)}")
status(acts) must equalTo(OK)
- contentType(acts) must beSome.which(_ == "text/plain")
+ contentType(acts) must beSome.which(_ == "application/json")
val jsRslt = contentAsJson(acts)
println("======")
println(jsRslt)
@@ -143,7 +142,7 @@
// logger.debug(s">> Social Controller test Acts result : ${contentAsJson(acts)}")
status(acts) must equalTo(OK)
- contentType(acts) must beSome.which(_ == "text/plain")
+ contentType(acts) must beSome.which(_ == "application/json")
val jsRslt = contentAsJson(acts)
println("======")
println(jsRslt)
@@ -161,10 +160,11 @@
}
-abstract class QuerySpecificationBase extends Specification {
- protected val testServiceName = "s2graph_test"
+abstract class QuerySpecificationBase extends Specification with RequestParser {
+ protected val testServiceName = "s2graph"
protected val testLabelName = "s2graph_label_test"
protected val testColumnName = "user_id"
+ val asyncFlushInterval = 1500 // in mill
val createService =
s"""
@@ -248,28 +248,33 @@
def initialize = {
running(FakeApplication()) {
- GraphAggregatorActor.init()
- KafkaAggregatorActor.init()
- Graph(Config.conf)(ExecutionContext.Implicits.global)
+ Graph(Config.conf.underlying)(ExecutionContext.Implicits.global)
// 1. createService
var result = AdminController.createServiceInner(Json.parse(createService))
println(s">> Service created : $createService, $result")
// 2. createLabel
- result = AdminController.createLabelInner(Json.parse(createLabel))
- println(s">> Label created : $createLabel, $result")
-
+ try {
+ AdminController.createLabelInner(Json.parse(createLabel))
+ println(s">> Label created : $createLabel, $result")
+ } catch {
+ case e: Throwable =>
+ }
+
// 3. insert edges
val jsArrStr = s"[${edgesInsertInitial.mkString(",")}]"
println(s">> 1st step Inserts : $jsArrStr")
val inserts = toEdges(Json.parse(jsArrStr), "insert")
Graph.mutateEdges(inserts)
+ Thread.sleep(asyncFlushInterval)
+
println(s"<< 1st step Inserted : $jsArrStr")
val jsArrStr2nd = s"[${edgesInsert2ndDepth.mkString(",")}]"
println(s">> 2nd step Inserts : $jsArrStr2nd")
val inserts2nd = toEdges(Json.parse(jsArrStr2nd), "insert")
Graph.mutateEdges(inserts2nd)
+ Thread.sleep(asyncFlushInterval)
println(s"<< 2nd step Inserted : $inserts2nd")
}
}
@@ -277,26 +282,26 @@
def cleanup = {
running(FakeApplication()) {
- Graph(Config.conf)(ExecutionContext.Implicits.global)
+ Graph(Config.conf.underlying)(ExecutionContext.Implicits.global)
// 1. delete edges
val jsArrStr = s"[${edgesInsertInitial.mkString(",")}]"
println(s"Deletes : $jsArrStr")
val deletes = toEdges(Json.parse(jsArrStr), "delete")
Graph.mutateEdges(deletes)
+ Thread.sleep(asyncFlushInterval)
val jsArrStr2nd = s"[${edgesInsert2ndDepth.mkString(",")}]"
println(s">> 2nd step Deletes : $jsArrStr2nd")
val deletes2nd = toEdges(Json.parse(jsArrStr2nd), "delete")
Graph.mutateEdges(deletes2nd)
+ Thread.sleep(asyncFlushInterval)
println(s"<< 2nd step Deleted : $deletes2nd")
// 2. delete label ( currently NOT supported )
// var result = AdminController.deleteLabelInner(testLabelName)
// println(s">> Label deleted : $testLabelName, $result")
// 3. delete service ( currently NOT supported )
- GraphAggregatorActor.shutdown()
- KafkaAggregatorActor.shutdown()
}
}
diff --git a/test/controllers/RequestParserSpec.scala b/test/controllers/RequestParserSpec.scala
index 9e51414..acf0f82 100644
--- a/test/controllers/RequestParserSpec.scala
+++ b/test/controllers/RequestParserSpec.scala
@@ -1,4 +1,4 @@
-package controllers
+package test.controllers
import models._
import com.daumkakao.s2graph.core._
@@ -14,6 +14,7 @@
class RequestParserSpec extends Specification {
// dummy data for dummy edge
+ val testLabelName = "s2graph_label_test"
val ts = System.currentTimeMillis()
val srcId = CompositeId(0, InnerVal.withLong(1), isEdge = true, useHash = true)
val tgtId = CompositeId(0, InnerVal.withLong(2), isEdge = true, useHash = true)
@@ -30,10 +31,9 @@
val checkedOpt = for (label <- Label.findByName(labelName)) yield {
val labelMetas = LabelMeta.findAllByLabelId(label.id.get, useCache = false)
val metaMap = labelMetas.map { m => m.name -> m.seq } toMap
-
val whereOpt = WhereParser(label).parse(sql)
whereOpt must beSome
- play.api.Logger.error(whereOpt.toString)
+ play.api.Logger.debug(whereOpt.toString)
// val props = Json.obj("is_hidden" -> true, "is_blocked" -> false)
// val propsInner = Management.toProps(label, props).map { case (k, v) => k -> InnerValWithTs.withInnerVal(v, ts) }.toMap
@@ -49,18 +49,20 @@
"RequestParser WhereParser" should {
"check where clause not nested" in {
running(FakeApplication()) {
- val labelOpt = Label.findByName("graph_test")
+ val labelOpt = Label.findByName(testLabelName)
labelOpt must beSome[Label]
val label = labelOpt.get
val labelWithDir = LabelWithDirection(label.id.get, 0)
+
val js = Json.obj("is_hidden" -> true, "is_blocked" -> false, "weight" -> 10, "time" -> 3, "name" -> "abc")
val propsInner = Management.toProps(label, js).map { case (k, v) => k -> InnerValWithTs.withInnerVal(v, ts) }.toMap
val edge = Edge(srcVertex, tgtVertex, labelWithDir, 0.toByte, ts, 0, propsInner)
play.api.Logger.debug(s"$edge")
- val f = validate("graph_test")(edge)_
+ val f = validate(testLabelName)(edge)_
f("is_hidden = false")(false)
+ f("is_hidden != false")(true)
f("is_hidden = true and is_blocked = true")(false)
f("is_hidden = true and is_blocked = false")(true)
f("time in (1, 2, 3) and is_blocked = true")(false)
@@ -68,15 +70,14 @@
f("time in (1, 2, 3) and is_blocked = false")(true)
f("time in (1, 2, 4) and is_blocked = false")(false)
f("time in (1, 2, 4) or is_blocked = false")(true)
+ f("time not in (1, 2, 4)")(true)
f("time in (1, 2, 3) and weight between 10 and 20 and is_blocked = false")(true)
f("time in (1, 2, 4) or weight between 10 and 20 or is_blocked = true")(true)
- f("name = abc")(true)
- f("name = xxx")(false)
}
}
"check where clause nested" in {
running(FakeApplication()) {
- val labelOpt = Label.findByName("graph_test")
+ val labelOpt = Label.findByName(testLabelName)
labelOpt must beSome[Label]
val label = labelOpt.get
val labelWithDir = LabelWithDirection(label.id.get, 0)
@@ -84,7 +85,7 @@
val propsInner = Management.toProps(label, js).map { case (k, v) => k -> InnerValWithTs.withInnerVal(v, ts) }.toMap
val edge = Edge(srcVertex, tgtVertex, labelWithDir, 0.toByte, ts, 0, propsInner)
- val f = validate("graph_test")(edge)_
+ val f = validate(testLabelName)(edge)_
f("(time in (1, 2, 3) and is_blocked = true) or is_hidden = false")(false)
f("(time in (1, 2, 3) or is_blocked = true) or is_hidden = false")(true)
@@ -96,15 +97,15 @@
f("(time in (1, 2, 4) or weight between 1 and 9) or is_hidden = true")(true)
f("(time in (1, 2, 3) or weight between 1 and 10) and is_hidden = false")(false)
- f("(name in (a, abc, c) and weight between 1 and 10) or is_hidden = false")(true)
- f("name between a and b or is_hidden = false")(true)
- f("name = abc and is_hidden = true")(true)
+// f("(name in (a, abc, c) and weight between 1 and 10) or is_hidden = false")(true)
+// f("name between a and b or is_hidden = false")(true)
+// f("name = abc and is_hidden = true")(true)
}
}
"check where clause with from/to long" in {
running(FakeApplication()) {
- val labelOpt = Label.findByName("graph_test")
+ val labelOpt = Label.findByName(testLabelName)
labelOpt must beSome[Label]
val label = labelOpt.get
val labelWithDir = LabelWithDirection(label.id.get, 0)
@@ -112,7 +113,7 @@
val propsInner = Management.toProps(label, js).map { case (k, v) => k -> InnerValWithTs.withInnerVal(v, ts) }.toMap
val edge = Edge(srcVertex, tgtVertex, labelWithDir, 0.toByte, ts, 0, propsInner)
- val f = validate("graph_test")(edge)_
+ val f = validate(testLabelName)(edge)_
// f("from = abc")(false)
f("_from = 2 or _to = 2")(true)
