http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/persist/LoggerPersist.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/LoggerPersist.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/persist/LoggerPersist.scala
deleted file mode 100644
index f0ea639..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/LoggerPersist.scala
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
-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.griffin.measure.batch.persist
-
-import java.util.Date
-
-import org.apache.griffin.measure.batch.result._
-import org.apache.griffin.measure.batch.utils.HdfsUtil
-import org.apache.spark.rdd.RDD
-
-// persist result and data to hdfs
-case class LoggerPersist(config: Map[String, Any], metricName: String, 
timeStamp: Long) extends Persist {
-
-  val MaxLogLines = "max.log.lines"
-
-  val maxLogLines = try { config.getOrElse(MaxLogLines, 100).toString.toInt } 
catch { case _ => 100 }
-
-  def available(): Boolean = true
-
-  def start(msg: String): Unit = {
-    info(s"${metricName} start")
-  }
-  def finish(): Unit = {
-    info(s"${metricName} finish")
-  }
-
-  def result(rt: Long, result: Result): Unit = {
-    try {
-      val resStr = result match {
-        case ar: AccuracyResult => {
-          s"match percentage: ${ar.matchPercentage}\ntotal count: 
${ar.getTotal}\nmiss count: ${ar.getMiss}, match count: ${ar.getMatch}"
-        }
-        case pr: ProfileResult => {
-          s"match percentage: ${pr.matchPercentage}\ntotal count: 
${pr.getTotal}\nmiss count: ${pr.getMiss}, match count: ${pr.getMatch}"
-        }
-        case _ => {
-          s"result: ${result}"
-        }
-      }
-      info(s"${metricName} result: \n${resStr}")
-    } catch {
-      case e: Throwable => error(e.getMessage)
-    }
-  }
-
-  // need to avoid string too long
-  private def rddRecords(records: RDD[String]): Unit = {
-    try {
-      val recordCount = records.count.toInt
-      val count = if (maxLogLines < 0) recordCount else 
scala.math.min(maxLogLines, recordCount)
-      if (count > 0) {
-        val recordsArray = records.take(count)
-        recordsArray.foreach(println)
-      }
-    } catch {
-      case e: Throwable => error(e.getMessage)
-    }
-  }
-
-  def missRecords(records: RDD[String]): Unit = {
-    info(s"${metricName} miss records: ")
-    rddRecords(records)
-  }
-  def matchRecords(records: RDD[String]): Unit = {
-    info(s"${metricName} match records: ")
-    rddRecords(records)
-  }
-
-  def log(rt: Long, msg: String): Unit = {
-    info(s"${rt}: ${msg}")
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/persist/MultiPersists.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/MultiPersists.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/persist/MultiPersists.scala
deleted file mode 100644
index 71b7ea7..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/MultiPersists.scala
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
-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.griffin.measure.batch.persist
-
-import org.apache.griffin.measure.batch.result._
-import org.apache.griffin.measure.batch.utils.{HttpUtil, JsonUtil}
-import org.apache.spark.rdd.RDD
-
-import scala.util.Try
-
-// persist result and data by multiple persists
-case class MultiPersists(persists: Iterable[Persist]) extends Persist {
-
-  val timeStamp: Long = persists match {
-    case Nil => 0
-    case _ => persists.head.timeStamp
-  }
-
-  val config: Map[String, Any] = Map[String, Any]()
-
-  def available(): Boolean = { persists.exists(_.available()) }
-
-  def start(msg: String): Unit = { persists.foreach(_.start(msg)) }
-  def finish(): Unit = { persists.foreach(_.finish()) }
-
-  def result(rt: Long, result: Result): Unit = { persists.foreach(_.result(rt, 
result)) }
-
-  def missRecords(records: RDD[String]): Unit = { 
persists.foreach(_.missRecords(records)) }
-  def matchRecords(records: RDD[String]): Unit = { 
persists.foreach(_.matchRecords(records)) }
-
-  def log(rt: Long, msg: String): Unit = { persists.foreach(_.log(rt, msg)) }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/persist/Persist.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/Persist.scala 
b/measure/src/main/scala/org/apache/griffin/measure/batch/persist/Persist.scala
deleted file mode 100644
index b17beea..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/Persist.scala
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
-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.griffin.measure.batch.persist
-
-import org.apache.griffin.measure.batch.log.Loggable
-import org.apache.griffin.measure.batch.result._
-import org.apache.spark.rdd.RDD
-
-import scala.util.Try
-
-
-trait Persist extends Loggable with Serializable {
-  val timeStamp: Long
-
-  val config: Map[String, Any]
-
-  def available(): Boolean
-
-  def start(msg: String): Unit
-  def finish(): Unit
-
-  def result(rt: Long, result: Result): Unit
-
-  def missRecords(records: RDD[String]): Unit
-  def matchRecords(records: RDD[String]): Unit
-
-  def log(rt: Long, msg: String): Unit
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/persist/PersistFactory.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/PersistFactory.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/persist/PersistFactory.scala
deleted file mode 100644
index 44752fc..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/persist/PersistFactory.scala
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
-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.griffin.measure.batch.persist
-
-import org.apache.griffin.measure.batch.config.params.env._
-
-import scala.util.{Success, Try}
-
-
-case class PersistFactory(persistParams: Iterable[PersistParam], metricName: 
String) extends Serializable {
-
-  val HDFS_REGEX = """^(?i)hdfs$""".r
-  val HTTP_REGEX = """^(?i)http$""".r
-  val LOG_REGEX = """^(?i)log$""".r
-
-  def getPersists(timeStamp: Long): MultiPersists = {
-    MultiPersists(persistParams.flatMap(param => getPersist(timeStamp, param)))
-  }
-
-  // get the persists configured
-  private def getPersist(timeStamp: Long, persistParam: PersistParam): 
Option[Persist] = {
-    val persistConfig = persistParam.config
-    val persistTry = persistParam.persistType match {
-      case HDFS_REGEX() => Try(HdfsPersist(persistConfig, metricName, 
timeStamp))
-      case HTTP_REGEX() => Try(HttpPersist(persistConfig, metricName, 
timeStamp))
-      case LOG_REGEX() => Try(LoggerPersist(persistConfig, metricName, 
timeStamp))
-      case _ => throw new Exception("not supported persist type")
-    }
-    persistTry match {
-      case Success(persist) if (persist.available) => Some(persist)
-      case _ => None
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/result/AccuracyResult.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/AccuracyResult.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/result/AccuracyResult.scala
deleted file mode 100644
index 4ea35a0..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/AccuracyResult.scala
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
-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.griffin.measure.batch.result
-
-// result for accuracy: miss count, total count
-case class AccuracyResult(miss: Long, total: Long) extends Result {
-
-  type T = AccuracyResult
-
-  def update(delta: T): T = {
-    AccuracyResult(delta.miss, total)
-  }
-
-  def eventual(): Boolean = {
-    this.miss <= 0
-  }
-
-  def differsFrom(other: T): Boolean = {
-    (this.miss != other.miss) || (this.total != other.total)
-  }
-
-  def getMiss = miss
-  def getTotal = total
-  def getMatch = total - miss
-
-  def matchPercentage: Double = if (getTotal <= 0) 0 else getMatch.toDouble / 
getTotal * 100
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/result/ProfileResult.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/ProfileResult.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/result/ProfileResult.scala
deleted file mode 100644
index 3d5cc69..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/ProfileResult.scala
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
-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.griffin.measure.batch.result
-
-// result for profile: match count, total count
-case class ProfileResult(matchCount: Long, totalCount: Long) extends Result {
-
-  type T = ProfileResult
-
-  def update(delta: T): T = {
-    ProfileResult(matchCount + delta.matchCount, totalCount)
-  }
-
-  def eventual(): Boolean = {
-    this.matchCount >= totalCount
-  }
-
-  def differsFrom(other: T): Boolean = {
-    (this.matchCount != other.matchCount) || (this.totalCount != 
other.totalCount)
-  }
-
-  def getMiss = totalCount - matchCount
-  def getTotal = totalCount
-  def getMatch = matchCount
-
-  def matchPercentage: Double = if (getTotal <= 0) 0 else getMatch.toDouble / 
getTotal * 100
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/result/Result.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/Result.scala 
b/measure/src/main/scala/org/apache/griffin/measure/batch/result/Result.scala
deleted file mode 100644
index 67b4f74..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/Result.scala
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
-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.griffin.measure.batch.result
-
-
-trait Result extends Serializable {
-
-  type T <: Result
-
-  def update(delta: T): T
-
-  def eventual(): Boolean
-
-  def differsFrom(other: T): Boolean
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/result/ResultInfo.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/ResultInfo.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/result/ResultInfo.scala
deleted file mode 100644
index 03bb4b4..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/result/ResultInfo.scala
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
-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.griffin.measure.batch.result
-
-
-sealed trait ResultInfo {
-  type T
-  val key: String
-  val tp: String
-  def wrap(value: T) = (key -> value)
-}
-
-final case object TimeGroupInfo extends ResultInfo {
-  type T = Long
-  val key = "__time__"
-  val tp = "bigint"
-}
-
-final case object NextFireTimeInfo extends ResultInfo {
-  type T = Long
-  val key = "__next_fire_time__"
-  val tp = "bigint"
-}
-
-final case object MismatchInfo extends ResultInfo {
-  type T = String
-  val key = "__mismatch__"
-  val tp = "string"
-}
-
-final case object TargetInfo extends ResultInfo {
-  type T = Map[String, Any]
-  val key = "__target__"
-  val tp = "map"
-}
-
-final case object ErrorInfo extends ResultInfo {
-  type T = String
-  val key = "__error__"
-  val tp = "string"
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/CalculationUtil.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/CalculationUtil.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/CalculationUtil.scala
deleted file mode 100644
index 7faf031..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/CalculationUtil.scala
+++ /dev/null
@@ -1,315 +0,0 @@
-/*
-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.griffin.measure.batch.rule
-
-import scala.util.{Success, Try}
-
-
-object CalculationUtil {
-
-  implicit def option2CalculationValue(v: Option[_]): CalculationValue = 
CalculationValue(v)
-
-  // redefine the calculation method of operators in DSL
-  case class CalculationValue(value: Option[_]) extends Serializable {
-
-    def + (other: Option[_]): Option[_] = {
-      Try {
-        (value, other) match {
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: String), Some(v2)) => Some(v1 + v2.toString)
-          case (Some(v1: Byte), Some(v2)) => Some(v1 + v2.toString.toByte)
-          case (Some(v1: Short), Some(v2)) => Some(v1 + v2.toString.toShort)
-          case (Some(v1: Int), Some(v2)) => Some(v1 + v2.toString.toInt)
-          case (Some(v1: Long), Some(v2)) => Some(v1 + v2.toString.toLong)
-          case (Some(v1: Float), Some(v2)) => Some(v1 + v2.toString.toFloat)
-          case (Some(v1: Double), Some(v2)) => Some(v1 + v2.toString.toDouble)
-          case _ => value
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def - (other: Option[_]): Option[_] = {
-      Try {
-        (value, other) match {
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: Byte), Some(v2)) => Some(v1 - v2.toString.toByte)
-          case (Some(v1: Short), Some(v2)) => Some(v1 - v2.toString.toShort)
-          case (Some(v1: Int), Some(v2)) => Some(v1 - v2.toString.toInt)
-          case (Some(v1: Long), Some(v2)) => Some(v1 - v2.toString.toLong)
-          case (Some(v1: Float), Some(v2)) => Some(v1 - v2.toString.toFloat)
-          case (Some(v1: Double), Some(v2)) => Some(v1 - v2.toString.toDouble)
-          case _ => value
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def * (other: Option[_]): Option[_] = {
-      Try {
-        (value, other) match {
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(s1: String), Some(n2: Int)) => Some(s1 * n2)
-          case (Some(s1: String), Some(n2: Long)) => Some(s1 * n2.toInt)
-          case (Some(v1: Byte), Some(v2)) => Some(v1 * v2.toString.toByte)
-          case (Some(v1: Short), Some(v2)) => Some(v1 * v2.toString.toShort)
-          case (Some(v1: Int), Some(v2)) => Some(v1 * v2.toString.toInt)
-          case (Some(v1: Long), Some(v2)) => Some(v1 * v2.toString.toLong)
-          case (Some(v1: Float), Some(v2)) => Some(v1 * v2.toString.toFloat)
-          case (Some(v1: Double), Some(v2)) => Some(v1 * v2.toString.toDouble)
-          case _ => value
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def / (other: Option[_]): Option[_] = {
-      Try {
-        (value, other) match {
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: Byte), Some(v2)) => Some(v1 / v2.toString.toByte)
-          case (Some(v1: Short), Some(v2)) => Some(v1 / v2.toString.toShort)
-          case (Some(v1: Int), Some(v2)) => Some(v1 / v2.toString.toInt)
-          case (Some(v1: Long), Some(v2)) => Some(v1 / v2.toString.toLong)
-          case (Some(v1: Float), Some(v2)) => Some(v1 / v2.toString.toFloat)
-          case (Some(v1: Double), Some(v2)) => Some(v1 / v2.toString.toDouble)
-          case _ => value
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def % (other: Option[_]): Option[_] = {
-      Try {
-        (value, other) match {
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: Byte), Some(v2)) => Some(v1 % v2.toString.toByte)
-          case (Some(v1: Short), Some(v2)) => Some(v1 % v2.toString.toShort)
-          case (Some(v1: Int), Some(v2)) => Some(v1 % v2.toString.toInt)
-          case (Some(v1: Long), Some(v2)) => Some(v1 % v2.toString.toLong)
-          case (Some(v1: Float), Some(v2)) => Some(v1 % v2.toString.toFloat)
-          case (Some(v1: Double), Some(v2)) => Some(v1 % v2.toString.toDouble)
-          case _ => value
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def unary_- (): Option[_] = {
-      value match {
-        case None => None
-        case Some(null) => None
-        case Some(v: String) => Some(v.reverse.toString)
-        case Some(v: Boolean) => Some(!v)
-        case Some(v: Byte) => Some(-v)
-        case Some(v: Short) => Some(-v)
-        case Some(v: Int) => Some(-v)
-        case Some(v: Long) => Some(-v)
-        case Some(v: Float) => Some(-v)
-        case Some(v: Double) => Some(-v)
-        case Some(v) => Some(v)
-        case _ => None
-      }
-    }
-
-
-    def === (other: Option[_]): Option[Boolean] = {
-      (value, other) match {
-        case (None, None) => Some(true)
-        case (Some(v1), Some(v2)) => Some(v1 == v2)
-        case _ => Some(false)
-      }
-    }
-
-    def =!= (other: Option[_]): Option[Boolean] = {
-      (value, other) match {
-        case (None, None) => Some(false)
-        case (Some(v1), Some(v2)) => Some(v1 != v2)
-        case _ => Some(true)
-      }
-    }
-
-    def > (other: Option[_]): Option[Boolean] = {
-      Try {
-        (value, other) match {
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: String), Some(v2: String)) => Some(v1 > v2)
-          case (Some(v1: Byte), Some(v2)) => Some(v1 > v2.toString.toDouble)
-          case (Some(v1: Short), Some(v2)) => Some(v1 > v2.toString.toDouble)
-          case (Some(v1: Int), Some(v2)) => Some(v1 > v2.toString.toDouble)
-          case (Some(v1: Long), Some(v2)) => Some(v1 > v2.toString.toDouble)
-          case (Some(v1: Float), Some(v2)) => Some(v1 > v2.toString.toDouble)
-          case (Some(v1: Double), Some(v2)) => Some(v1 > v2.toString.toDouble)
-          case _ => None
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def >= (other: Option[_]): Option[Boolean] = {
-      Try {
-        (value, other) match {
-          case (None, None) | (Some(null), Some(null)) => Some(true)
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: String), Some(v2: String)) => Some(v1 >= v2)
-          case (Some(v1: Byte), Some(v2)) => Some(v1 >= v2.toString.toDouble)
-          case (Some(v1: Short), Some(v2)) => Some(v1 >= v2.toString.toDouble)
-          case (Some(v1: Int), Some(v2)) => Some(v1 >= v2.toString.toDouble)
-          case (Some(v1: Long), Some(v2)) => Some(v1 >= v2.toString.toDouble)
-          case (Some(v1: Float), Some(v2)) => Some(v1 >= v2.toString.toDouble)
-          case (Some(v1: Double), Some(v2)) => Some(v1 >= v2.toString.toDouble)
-          case _ => None
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def < (other: Option[_]): Option[Boolean] = {
-      Try {
-        (value, other) match {
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: String), Some(v2: String)) => Some(v1 < v2)
-          case (Some(v1: Byte), Some(v2)) => Some(v1 < v2.toString.toDouble)
-          case (Some(v1: Short), Some(v2)) => Some(v1 < v2.toString.toDouble)
-          case (Some(v1: Int), Some(v2)) => Some(v1 < v2.toString.toDouble)
-          case (Some(v1: Long), Some(v2)) => Some(v1 < v2.toString.toDouble)
-          case (Some(v1: Float), Some(v2)) => Some(v1 < v2.toString.toDouble)
-          case (Some(v1: Double), Some(v2)) => Some(v1 < v2.toString.toDouble)
-          case _ => None
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-    def <= (other: Option[_]): Option[Boolean] = {
-      Try {
-        (value, other) match {
-          case (None, None) | (Some(null), Some(null)) => Some(true)
-          case (None, _) | (_, None) => None
-          case (Some(null), _) | (_, Some(null)) => None
-          case (Some(v1: String), Some(v2: String)) => Some(v1 <= v2)
-          case (Some(v1: Byte), Some(v2)) => Some(v1 <= v2.toString.toDouble)
-          case (Some(v1: Short), Some(v2)) => Some(v1 <= v2.toString.toDouble)
-          case (Some(v1: Int), Some(v2)) => Some(v1 <= v2.toString.toDouble)
-          case (Some(v1: Long), Some(v2)) => Some(v1 <= v2.toString.toDouble)
-          case (Some(v1: Float), Some(v2)) => Some(v1 <= v2.toString.toDouble)
-          case (Some(v1: Double), Some(v2)) => Some(v1 <= v2.toString.toDouble)
-          case _ => None
-        }
-      } match {
-        case Success(opt) => opt
-        case _ => None
-      }
-    }
-
-
-    def in (other: Iterable[Option[_]]): Option[Boolean] = {
-      other.foldLeft(Some(false): Option[Boolean]) { (res, next) =>
-        optOr(res, ===(next))
-      }
-    }
-
-    def not_in (other: Iterable[Option[_]]): Option[Boolean] = {
-      other.foldLeft(Some(true): Option[Boolean]) { (res, next) =>
-        optAnd(res, =!=(next))
-      }
-    }
-
-    def between (other: Iterable[Option[_]]): Option[Boolean] = {
-      if (other.size < 2) None else {
-        val (begin, end) = (other.head, other.tail.head)
-        if (begin.isEmpty && end.isEmpty) Some(value.isEmpty)
-        else optAnd(>=(begin), <=(end))
-      }
-    }
-
-    def not_between (other: Iterable[Option[_]]): Option[Boolean] = {
-      if (other.size < 2) None else {
-        val (begin, end) = (other.head, other.tail.head)
-        if (begin.isEmpty && end.isEmpty) Some(value.nonEmpty)
-        else optOr(<(begin), >(end))
-      }
-    }
-
-    def unary_! (): Option[Boolean] = {
-      optNot(value)
-    }
-
-    def && (other: Option[_]): Option[Boolean] = {
-      optAnd(value, other)
-    }
-
-    def || (other: Option[_]): Option[Boolean] = {
-      optOr(value, other)
-    }
-
-
-    private def optNot(a: Option[_]): Option[Boolean] = {
-      a match {
-        case None => None
-        case Some(null) => None
-        case Some(v: Boolean) => Some(!v)
-        case _ => None
-      }
-    }
-    private def optAnd(a: Option[_], b: Option[_]): Option[Boolean] = {
-      (a, b) match {
-        case (None, _) | (_, None) => None
-        case (Some(null), _) | (_, Some(null)) => None
-        case (Some(false), _) | (_, Some(false)) => Some(false)
-        case (Some(b1: Boolean), Some(b2: Boolean)) => Some(b1 && b2)
-        case _ => None
-      }
-    }
-    private def optOr(a: Option[_], b: Option[_]): Option[Boolean] = {
-      (a, b) match {
-        case (None, _) | (_, None) => None
-        case (Some(null), _) | (_, Some(null)) => None
-        case (Some(true), _) | (_, Some(true)) => Some(true)
-        case (Some(b1: Boolean), Some(b2: Boolean)) => Some(b1 || b2)
-        case _ => None
-      }
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/ExprValueUtil.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/ExprValueUtil.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/ExprValueUtil.scala
deleted file mode 100644
index e4e4055..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/ExprValueUtil.scala
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
-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.griffin.measure.batch.rule
-
-import org.apache.griffin.measure.batch.rule.expr._
-import org.apache.spark.sql.Row
-
-import scala.util.{Success, Try}
-
-object ExprValueUtil {
-
-  // from origin data such as a Row of DataFrame, with existed expr value map, 
calculate related expression, get the expression value
-  // for now, one expr only get one value, not supporting one expr get 
multiple values
-  // params:
-  // - originData: the origin data such as a Row of DataFrame
-  // - expr: the expression to be calculated
-  // - existExprValueMap: existed expression value map, which might be used to 
get some existed expression value during calculation
-  // output: the calculated expression value
-  private def calcExprValue(originData: Option[Any], expr: Expr, 
existExprValueMap: Map[String, Any]): Option[Any] = {
-    Try {
-      expr match {
-        case selection: SelectionExpr => {
-          selection.selectors.foldLeft(originData) { (dt, selector) =>
-            calcExprValue(dt, selector, existExprValueMap)
-          }
-        }
-        case selector: IndexFieldRangeSelectExpr => {
-          originData match {
-            case Some(row: Row) => {
-              if (selector.fields.size == 1) {
-                selector.fields.head match {
-                  case i: IndexDesc => Some(row.getAs[Any](i.index))
-                  case f: FieldDesc => Some(row.getAs[Any](f.field))
-                  case _ => None
-                }
-              } else None
-            }
-            case _ => None
-          }
-        }
-        case _ => expr.calculate(existExprValueMap)
-      }
-    } match {
-      case Success(v) => v
-      case _ => None
-    }
-  }
-
-  // try to calculate expr from data and initExprValueMap, generate a new 
expression value map
-  // depends on origin data and existed expr value map
-  def genExprValueMap(data: Option[Any], expr: Expr, initExprValueMap: 
Map[String, Any]): Map[String, Any] = {
-    val valueOpt = calcExprValue(data, expr, initExprValueMap)
-    if (valueOpt.nonEmpty) {
-      initExprValueMap + (expr._id -> valueOpt.get)
-    } else initExprValueMap
-  }
-
-  // try to calculate some exprs from data and initExprValueMap, generate a 
new expression value map
-  // depends on origin data and existed expr value map
-  def genExprValueMap(data: Option[Any], exprs: Iterable[Expr], 
initExprValueMap: Map[String, Any]): Map[String, Any] = {
-    exprs.foldLeft(initExprValueMap) { (evMap, expr) =>
-      ExprValueUtil.genExprValueMap(None, expr, evMap)
-    }
-  }
-
-  // with exprValueMap, calculate expressions, update the expression value map
-  // only depends on existed expr value map, only calculation, not need origin 
data
-  def updateExprValueMap(exprs: Iterable[Expr], exprValueMap: Map[String, 
Any]): Map[String, Any] = {
-    exprs.foldLeft(Map[String, Any]()) { (evMap, expr) =>
-      val valueOpt = expr.calculate(exprValueMap)
-      if (valueOpt.nonEmpty) {
-        evMap + (expr._id -> valueOpt.get)
-      } else evMap
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleAnalyzer.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleAnalyzer.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleAnalyzer.scala
deleted file mode 100644
index 124df04..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleAnalyzer.scala
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
-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.griffin.measure.batch.rule
-
-import org.apache.griffin.measure.batch.rule.expr._
-
-case class RuleAnalyzer(rule: StatementExpr) extends Serializable {
-
-  val constData = ""
-  private val SourceData = "source"
-  private val TargetData = "target"
-
-  val constCacheExprs: Iterable[Expr] = rule.getCacheExprs(constData)
-  private val sourceCacheExprs: Iterable[Expr] = rule.getCacheExprs(SourceData)
-  private val targetCacheExprs: Iterable[Expr] = rule.getCacheExprs(TargetData)
-
-  private val sourcePersistExprs: Iterable[Expr] = 
rule.getPersistExprs(SourceData)
-  private val targetPersistExprs: Iterable[Expr] = 
rule.getPersistExprs(TargetData)
-
-  val constFinalCacheExprs: Iterable[Expr] = 
rule.getFinalCacheExprs(constData).toSet
-  private val sourceFinalCacheExprs: Iterable[Expr] = 
rule.getFinalCacheExprs(SourceData).toSet ++ sourcePersistExprs.toSet
-  private val targetFinalCacheExprs: Iterable[Expr] = 
rule.getFinalCacheExprs(TargetData).toSet ++ targetPersistExprs.toSet
-
-  private val groupbyExprPairs: Seq[(Expr, Expr)] = 
rule.getGroupbyExprPairs((SourceData, TargetData))
-  private val sourceGroupbyExprs: Seq[Expr] = groupbyExprPairs.map(_._1)
-  private val targetGroupbyExprs: Seq[Expr] = groupbyExprPairs.map(_._2)
-
-  private val whenClauseExprOpt: Option[LogicalExpr] = rule.getWhenClauseExpr
-
-  val sourceRuleExprs: RuleExprs = RuleExprs(sourceGroupbyExprs, 
sourceCacheExprs,
-    sourceFinalCacheExprs, sourcePersistExprs, whenClauseExprOpt)
-  val targetRuleExprs: RuleExprs = RuleExprs(targetGroupbyExprs, 
targetCacheExprs,
-    targetFinalCacheExprs, targetPersistExprs, whenClauseExprOpt)
-
-}
-
-
-// for a single data source
-// groupbyExprs: in accuracy case, these exprs could be groupby exprs
-//                  Data keys for accuracy case, generated by the equal 
statements, to improve the calculation efficiency
-// cacheExprs: the exprs value could be caculated independently, and cached 
for later use
-//                  Cached for the finalCacheExprs calculation, it has some 
redundant values, saving it wastes a lot
-// finalCacheExprs: the root of cachedExprs, cached for later use, plus with 
persistExprs
-//                  Cached for the calculation usage, and can be saved for the 
re-calculation in streaming mode
-// persistExprs: the expr values should be persisted, only the direct 
selection exprs are persistable
-//                  Persisted for record usage, to record the missing data, 
need be readable as raw data
-// whenClauseExprOpt: when clause of rule, to determine if the row of data 
source is filtered
-//                  Can be prep-calculated to filter some data in data 
connector
-case class RuleExprs(groupbyExprs: Seq[Expr],
-                     cacheExprs: Iterable[Expr],
-                     finalCacheExprs: Iterable[Expr],
-                     persistExprs: Iterable[Expr],
-                     whenClauseExprOpt: Option[LogicalExpr]
-                    ) {
-  // for example: for a rule "$source.name = $target.name AND $source.age < 
$target.age + (3 * 4)"
-  // in this rule, for the target data source, the targetRuleExprs looks like 
below
-  // groupbyExprs: $target.name
-  // cacheExprs: $target.name, $target.age, $target.age + (3 * 4)
-  // finalCacheExprs: $target.name, $target.age + (3 * 4), $target.age
-  // persistExprs: $target.name, $target.age
-  // whenClauseExprOpt: None
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleFactory.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleFactory.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleFactory.scala
deleted file mode 100644
index b1a4b29..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleFactory.scala
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
-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.griffin.measure.batch.rule
-
-import org.apache.griffin.measure.batch.config.params.user._
-
-import scala.util.Failure
-//import org.apache.griffin.measure.batch.rule.expr_old._
-import org.apache.griffin.measure.batch.rule.expr._
-
-import scala.util.{Success, Try}
-
-
-case class RuleFactory(evaluateRuleParam: EvaluateRuleParam) {
-
-  val ruleParser: RuleParser = RuleParser()
-
-  def generateRule(): StatementExpr = {
-    val rules = evaluateRuleParam.rules
-    val statement = parseExpr(rules) match {
-      case Success(se) => se
-      case Failure(ex) => throw ex
-    }
-    statement
-  }
-
-  private def parseExpr(rules: String): Try[StatementExpr] = {
-    Try {
-      val result = ruleParser.parseAll(ruleParser.rule, rules)
-      if (result.successful) result.get
-      else throw new Exception("parse rule error!")
-//      throw new Exception("parse rule error!")
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleParser.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleParser.scala 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleParser.scala
deleted file mode 100644
index 7e030ed..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/RuleParser.scala
+++ /dev/null
@@ -1,240 +0,0 @@
-/*
-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.griffin.measure.batch.rule
-
-import org.apache.griffin.measure.batch.rule.expr._
-
-import scala.util.parsing.combinator._
-
-case class RuleParser() extends JavaTokenParsers with Serializable {
-
-  /**
-    * BNF representation for grammar as below:
-    *
-    * <rule> ::= <logical-statement> [WHEN <logical-statement>]
-    * rule: mapping-rule [WHEN when-rule]
-    * - mapping-rule: the first level opr should better not be OR | NOT, 
otherwise it can't automatically find the groupby column
-    * - when-rule: only contain the general info of data source, not the 
special info of each data row
-    *
-    * <logical-statement> ::= [NOT] <logical-expression> [(AND | OR) 
<logical-expression>]+ | "(" <logical-statement> ")"
-    * logical-statement: return boolean value
-    * logical-operator: "AND" | "&&", "OR" | "||", "NOT" | "!"
-    *
-    * <logical-expression> ::= <math-expr> (<compare-opr> <math-expr> | 
<range-opr> <range-expr>)
-    * logical-expression example: $source.id = $target.id, $source.page_id IN 
('3214', '4312', '60821')
-    *
-    * <compare-opr> ::= "=" | "!=" | "<" | ">" | "<=" | ">="
-    * <range-opr> ::= ["NOT"] "IN" | "BETWEEN"
-    * <range-expr> ::= "(" [<math-expr>] [, <math-expr>]+ ")"
-    * range-expr example: ('3214', '4312', '60821'), (10, 15), ()
-    *
-    * <math-expr> ::= [<unary-opr>] <math-factor> [<binary-opr> <math-factor>]+
-    * math-expr example: $source.price * $target.count, "hello" + " " + 
"world" + 123
-    *
-    * <binary-opr> ::= "+" | "-" | "*" | "/" | "%"
-    * <unary-opr> ::= "+" | "-"
-    *
-    * <math-factor> ::= <literal> | <selection> | "(" <math-expr> ")"
-    *
-    * <selection> ::= <selection-head> [ <field-sel> | <function-operation> | 
<index-field-range-sel> | <filter-sel> ]+
-    * selection example: $source.price, $source.json(), $source['state'], 
$source.numList[3], $target.json().mails['org' = 'apache'].names[*]
-    *
-    * <selection-head> ::= $source | $target
-    *
-    * <field-sel> ::= "." <field-string>
-    *
-    * <function-operation> ::= "." <function-name> "(" <arg> [, <arg>]+ ")"
-    * <function-name> ::= <name-string>
-    * <arg> ::= <math-expr>
-    *
-    * <index-field-range-sel> ::= "[" <index-field-range> [, 
<index-field-range>]+ "]"
-    * <index-field-range> ::= <index-field> | (<index-field>, <index-field>) | 
"*"
-    * index-field-range: 2 means the 3rd item, (0, 3) means first 4 items, * 
means all items, 'age' means item 'age'
-    * <index-field> ::= <index> | <field-quote> | <all-selection>
-    * index: 0 ~ n means position from start, -1 ~ -n means position from end
-    * <field-quote> ::= ' <field-string> ' | " <field-string> "
-    *
-    * <filter-sel> ::= "[" <field-quote> <filter-compare-opr> <math-expr> "]"
-    * <filter-compare-opr> ::= "=" | "!=" | "<" | ">" | "<=" | ">="
-    * filter-sel example: ['name' = 'URL'], $source.man['age' > 
$source.graduate_age + 5 ]
-    *
-    * When <math-expr> in the selection, it mustn't contain the different 
<selection-head>, for example:
-    * $source.tags[1+2]             valid
-    * $source.tags[$source.first]   valid
-    * $source.tags[$target.first]   invalid
-    * -- Such job is for validation, not for parser
-    *
-    *
-    * <literal> ::= <literal-string> | <literal-number> | <literal-time> | 
<literal-boolean> | <literal-null> | <literal-none>
-    * <literal-string> ::= <any-string>
-    * <literal-number> ::= <integer> | <double>
-    * <literal-time> ::= <integer> ("d"|"h"|"m"|"s"|"ms")
-    * <literal-boolean> ::= true | false
-    * <literal-null> ::= null | undefined
-    * <literal-none> ::= none
-    *
-    */
-
-  object Keyword {
-    def WhenKeywords: Parser[String] = """(?i)when""".r
-    def UnaryLogicalKeywords: Parser[String] = """(?i)not""".r
-    def BinaryLogicalKeywords: Parser[String] = """(?i)and|or""".r
-    def RangeKeywords: Parser[String] = """(?i)(not\s+)?(in|between)""".r
-    def DataSourceKeywords: Parser[String] = """(?i)\$(source|target)""".r
-    def Keywords: Parser[String] = WhenKeywords | UnaryLogicalKeywords | 
BinaryLogicalKeywords | RangeKeywords | DataSourceKeywords
-  }
-  import Keyword._
-
-  object Operator {
-    def NotLogicalOpr: Parser[String] = """(?i)not""".r | "!"
-    def AndLogicalOpr: Parser[String] = """(?i)and""".r | "&&"
-    def OrLogicalOpr: Parser[String] = """(?i)or""".r | "||"
-    def CompareOpr: Parser[String] = """!?==?""".r | """<=?""".r | """>=?""".r
-    def RangeOpr: Parser[String] = RangeKeywords
-
-    def UnaryMathOpr: Parser[String] = "+" | "-"
-    def BinaryMathOpr1: Parser[String] = "*" | "/" | "%"
-    def BinaryMathOpr2: Parser[String] = "+" | "-"
-
-    def FilterCompareOpr: Parser[String] = """!?==?""".r | """<=?""".r | 
""">=?""".r
-
-    def SqBracketPair: (Parser[String], Parser[String]) = ("[", "]")
-    def BracketPair: (Parser[String], Parser[String]) = ("(", ")")
-    def Dot: Parser[String] = "."
-    def AllSelection: Parser[String] = "*"
-    def SQuote: Parser[String] = "'"
-    def DQuote: Parser[String] = "\""
-    def Comma: Parser[String] = ","
-  }
-  import Operator._
-
-  object SomeString {
-    def AnyString: Parser[String] = """[^'\"{}\[\]()=<>.$@,;+\-*/\\]*""".r
-    def SimpleFieldString: Parser[String] = """\w+""".r
-    def FieldString: Parser[String] = """[\w\s]+""".r
-    def NameString: Parser[String] = """[a-zA-Z_]\w*""".r
-  }
-  import SomeString._
-
-  object SomeNumber {
-    def IntegerNumber: Parser[String] = """[+\-]?\d+""".r
-    def DoubleNumber: Parser[String] = """[+\-]?(\.\d+|\d+\.\d*)""".r
-    def IndexNumber: Parser[String] = IntegerNumber
-  }
-  import SomeNumber._
-
-  // -- literal --
-  def literal: Parser[LiteralExpr] = literialString | literialTime | 
literialNumber | literialBoolean | literialNull | literialNone
-  def literialString: Parser[LiteralStringExpr] = (SQuote ~> AnyString <~ 
SQuote | DQuote ~> AnyString <~ DQuote) ^^ { LiteralStringExpr(_) }
-  def literialNumber: Parser[LiteralNumberExpr] = (DoubleNumber | 
IntegerNumber) ^^ { LiteralNumberExpr(_) }
-  def literialTime: Parser[LiteralTimeExpr] = """(\d+(d|h|m|s|ms))+""".r ^^ { 
LiteralTimeExpr(_) }
-  def literialBoolean: Parser[LiteralBooleanExpr] = ("""(?i)true""".r | 
"""(?i)false""".r) ^^ { LiteralBooleanExpr(_) }
-  def literialNull: Parser[LiteralNullExpr] = ("""(?i)null""".r | 
"""(?i)undefined""".r) ^^ { LiteralNullExpr(_) }
-  def literialNone: Parser[LiteralNoneExpr] = """(?i)none""".r ^^ { 
LiteralNoneExpr(_) }
-
-  // -- selection --
-  // <selection> ::= <selection-head> [ <field-sel> | <function-operation> | 
<index-field-range-sel> | <filter-sel> ]+
-  def selection: Parser[SelectionExpr] = selectionHead ~ rep(selector) ^^ {
-    case head ~ selectors => SelectionExpr(head, selectors)
-  }
-  def selector: Parser[SelectExpr] = (functionOperation | fieldSelect | 
indexFieldRangeSelect | filterSelect)
-
-  def selectionHead: Parser[SelectionHead] = DataSourceKeywords ^^ { 
SelectionHead(_) }
-  // <field-sel> ::= "." <field-string>
-  def fieldSelect: Parser[IndexFieldRangeSelectExpr] = Dot ~> 
SimpleFieldString ^^ {
-    case field => IndexFieldRangeSelectExpr(FieldDesc(field) :: Nil)
-  }
-  // <function-operation> ::= "." <function-name> "(" <arg> [, <arg>]+ ")"
-  def functionOperation: Parser[FunctionOperationExpr] = Dot ~ NameString ~ 
BracketPair._1 ~ repsep(argument, Comma) ~ BracketPair._2 ^^ {
-    case _ ~ func ~ _ ~ args ~ _ => FunctionOperationExpr(func, args)
-  }
-  def argument: Parser[MathExpr] = mathExpr
-  // <index-field-range-sel> ::= "[" <index-field-range> [, 
<index-field-range>]+ "]"
-  def indexFieldRangeSelect: Parser[IndexFieldRangeSelectExpr] = 
SqBracketPair._1 ~> rep1sep(indexFieldRange, Comma) <~ SqBracketPair._2 ^^ {
-    case ifrs => IndexFieldRangeSelectExpr(ifrs)
-  }
-  // <index-field-range> ::= <index-field> | (<index-field>, <index-field>) | 
"*"
-  def indexFieldRange: Parser[FieldDescOnly] = indexField | BracketPair._1 ~ 
indexField ~ Comma ~ indexField ~ BracketPair._2 ^^ {
-    case _ ~ if1 ~ _ ~ if2 ~ _ => FieldRangeDesc(if1, if2)
-  }
-  // <index-field> ::= <index> | <field-quote> | <all-selection>
-  // *here it can parse <math-expr>, but for simple situation, not supported 
now*
-  def indexField: Parser[FieldDescOnly] = IndexNumber ^^ { IndexDesc(_) } | 
fieldQuote | AllSelection ^^ { AllFieldsDesc(_) }
-  // <field-quote> ::= ' <field-string> ' | " <field-string> "
-  def fieldQuote: Parser[FieldDesc] = (SQuote ~> FieldString <~ SQuote | 
DQuote ~> FieldString <~ DQuote) ^^ { FieldDesc(_) }
-  // <filter-sel> ::= "[" <field-quote> <filter-compare-opr> <math-expr> "]"
-  def filterSelect: Parser[FilterSelectExpr] = SqBracketPair._1 ~> fieldQuote 
~ FilterCompareOpr ~ mathExpr <~ SqBracketPair._2 ^^ {
-    case field ~ compare ~ value => FilterSelectExpr(field, compare, value)
-  }
-
-  // -- math --
-  // <math-factor> ::= <literal> | <selection> | "(" <math-expr> ")"
-  def mathFactor: Parser[MathExpr] = (literal | selection | BracketPair._1 ~> 
mathExpr <~ BracketPair._2) ^^ { MathFactorExpr(_) }
-  // <math-expr> ::= [<unary-opr>] <math-factor> [<binary-opr> <math-factor>]+
-  // <unary-opr> ::= "+" | "-"
-  def unaryMathExpr: Parser[MathExpr] = rep(UnaryMathOpr) ~ mathFactor ^^ {
-    case Nil ~ a => a
-    case list ~ a => UnaryMathExpr(list, a)
-  }
-  // <binary-opr> ::= "+" | "-" | "*" | "/" | "%"
-  def binaryMathExpr1: Parser[MathExpr] = unaryMathExpr ~ rep(BinaryMathOpr1 ~ 
unaryMathExpr) ^^ {
-    case a ~ Nil => a
-    case a ~ list => BinaryMathExpr(a, list.map(c => (c._1, c._2)))
-  }
-  def binaryMathExpr2: Parser[MathExpr] = binaryMathExpr1 ~ rep(BinaryMathOpr2 
~ binaryMathExpr1) ^^ {
-    case a ~ Nil => a
-    case a ~ list => BinaryMathExpr(a, list.map(c => (c._1, c._2)))
-  }
-  def mathExpr: Parser[MathExpr] = binaryMathExpr2
-
-  // -- logical expression --
-  // <range-expr> ::= "(" [<math-expr>] [, <math-expr>]+ ")"
-  def rangeExpr: Parser[RangeDesc] = BracketPair._1 ~> repsep(mathExpr, Comma) 
<~ BracketPair._2 ^^ { RangeDesc(_) }
-  // <logical-expression> ::= <math-expr> (<compare-opr> <math-expr> | 
<range-opr> <range-expr>)
-  def logicalExpr: Parser[LogicalExpr] = mathExpr ~ CompareOpr ~ mathExpr ^^ {
-    case left ~ opr ~ right => LogicalCompareExpr(left, opr, right)
-  } | mathExpr ~ RangeOpr ~ rangeExpr ^^ {
-    case left ~ opr ~ range => LogicalRangeExpr(left, opr, range)
-  } | mathExpr ^^ { LogicalSimpleExpr(_) }
-
-  // -- logical statement --
-  def logicalFactor: Parser[LogicalExpr] = logicalExpr | BracketPair._1 ~> 
logicalStatement <~ BracketPair._2
-  def notLogicalStatement: Parser[LogicalExpr] = rep(NotLogicalOpr) ~ 
logicalFactor ^^ {
-    case Nil ~ a => a
-    case list ~ a => UnaryLogicalExpr(list, a)
-  }
-  def andLogicalStatement: Parser[LogicalExpr] = notLogicalStatement ~ 
rep(AndLogicalOpr ~ notLogicalStatement) ^^ {
-    case a ~ Nil => a
-    case a ~ list => BinaryLogicalExpr(a, list.map(c => (c._1, c._2)))
-  }
-  def orLogicalStatement: Parser[LogicalExpr] = andLogicalStatement ~ 
rep(OrLogicalOpr ~ andLogicalStatement) ^^ {
-    case a ~ Nil => a
-    case a ~ list => BinaryLogicalExpr(a, list.map(c => (c._1, c._2)))
-  }
-  // <logical-statement> ::= [NOT] <logical-expression> [(AND | OR) 
<logical-expression>]+ | "(" <logical-statement> ")"
-  def logicalStatement: Parser[LogicalExpr] = orLogicalStatement
-
-  // -- rule --
-  // <rule> ::= <logical-statement> [WHEN <logical-statement>]
-  def rule: Parser[StatementExpr] = logicalStatement ~ opt(WhenKeywords ~> 
logicalStatement) ^^ {
-    case ls ~ Some(ws) => WhenClauseStatementExpr(ls, ws)
-    case ls ~ _ => SimpleStatementExpr(ls)
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/AnalyzableExpr.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/AnalyzableExpr.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/AnalyzableExpr.scala
deleted file mode 100644
index dd68d0e..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/AnalyzableExpr.scala
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-
-trait AnalyzableExpr extends Serializable {
-  def getGroupbyExprPairs(dsPair: (String, String)): Seq[(Expr, Expr)] = Nil
-  def getWhenClauseExpr(): Option[LogicalExpr] = None
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Cacheable.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Cacheable.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Cacheable.scala
deleted file mode 100644
index f4779e4..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Cacheable.scala
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-trait Cacheable extends DataSourceable {
-  protected def cacheUnit: Boolean = false
-  def cacheable(ds: String): Boolean = {
-    cacheUnit && !conflict() && ((ds.isEmpty && dataSources.isEmpty) || 
(ds.nonEmpty && contains(ds)))
-  }
-  protected def getCacheExprs(ds: String): Iterable[Cacheable]
-
-  protected def persistUnit: Boolean = false
-  def persistable(ds: String): Boolean = {
-    persistUnit && ((ds.isEmpty && dataSources.isEmpty) || (ds.nonEmpty && 
contains(ds)))
-  }
-  protected def getPersistExprs(ds: String): Iterable[Cacheable]
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Calculatable.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Calculatable.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Calculatable.scala
deleted file mode 100644
index 25fbae4..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Calculatable.scala
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-trait Calculatable extends Serializable {
-
-  def calculate(values: Map[String, Any]): Option[Any]
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/DataSourceable.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/DataSourceable.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/DataSourceable.scala
deleted file mode 100644
index a5fbd28..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/DataSourceable.scala
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-trait DataSourceable extends Serializable {
-  val dataSources: Set[String]
-  protected def conflict(): Boolean = dataSources.size > 1
-  def contains(ds: String): Boolean = dataSources.contains(ds)
-  def dataSourceOpt: Option[String] = {
-    if (dataSources.size == 1) Some(dataSources.head) else None
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Describable.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Describable.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Describable.scala
deleted file mode 100644
index 7518eea..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Describable.scala
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-trait Describable extends Serializable {
-
-  val desc: String
-
-  protected def describe(v: Any): String = {
-    v match {
-      case s: Describable => s"${s.desc}"
-      case s: String => s"'${s}'"
-      case a => s"${a}"
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Expr.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Expr.scala 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Expr.scala
deleted file mode 100644
index f0b9c29..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/Expr.scala
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-trait Expr extends Serializable with Describable with Cacheable with 
Calculatable {
-
-  protected val _defaultId: String = ExprIdCounter.emptyId
-
-  val _id = ExprIdCounter.genId(_defaultId)
-
-  protected def getSubCacheExprs(ds: String): Iterable[Expr] = Nil
-  final def getCacheExprs(ds: String): Iterable[Expr] = {
-    if (cacheable(ds)) getSubCacheExprs(ds).toList :+ this else 
getSubCacheExprs(ds)
-  }
-
-  protected def getSubFinalCacheExprs(ds: String): Iterable[Expr] = Nil
-  final def getFinalCacheExprs(ds: String): Iterable[Expr] = {
-    if (cacheable(ds)) Nil :+ this else getSubFinalCacheExprs(ds)
-  }
-
-  protected def getSubPersistExprs(ds: String): Iterable[Expr] = Nil
-  final def getPersistExprs(ds: String): Iterable[Expr] = {
-    if (persistable(ds)) getSubPersistExprs(ds).toList :+ this else 
getSubPersistExprs(ds)
-  }
-
-  final def calculate(values: Map[String, Any]): Option[Any] = {
-    values.get(_id) match {
-      case Some(v) => Some(v)
-      case _ => calculateOnly(values)
-    }
-  }
-  protected def calculateOnly(values: Map[String, Any]): Option[Any]
-
-}
-

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprDescOnly.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprDescOnly.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprDescOnly.scala
deleted file mode 100644
index 4043cc6..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprDescOnly.scala
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-trait ExprDescOnly extends Describable {
-
-}
-
-
-case class SelectionHead(expr: String) extends ExprDescOnly {
-  private val headRegex = """\$(\w+)""".r
-  val head: String = expr match {
-    case headRegex(v) => v.toLowerCase
-    case _ => expr
-  }
-  val desc: String = "$" + head
-}
-
-case class RangeDesc(elements: Iterable[MathExpr]) extends ExprDescOnly {
-  val desc: String = {
-    val rangeDesc = elements.map(_.desc).mkString(", ")
-    s"(${rangeDesc})"
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprIdCounter.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprIdCounter.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprIdCounter.scala
deleted file mode 100644
index 0df52d8..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/ExprIdCounter.scala
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-import java.util.concurrent.atomic.AtomicLong
-
-import scala.collection.mutable.{Set => MutableSet}
-
-object ExprIdCounter {
-
-  private val idCounter: AtomicLong = new AtomicLong(0L)
-
-  private val existIdSet: MutableSet[String] = MutableSet.empty[String]
-
-  private val invalidIdRegex = """^\d+$""".r
-
-  val emptyId: String = ""
-
-  def genId(defaultId: String): String = {
-    defaultId match {
-      case emptyId => increment.toString
-      case invalidIdRegex() => increment.toString
-//      case defId if (exist(defId)) => s"${increment}#${defId}"
-      case defId if (exist(defId)) => s"${defId}"
-      case _ => {
-        insertUserId(defaultId)
-        defaultId
-      }
-    }
-  }
-
-  private def exist(id: String): Boolean = {
-    existIdSet.contains(id)
-  }
-
-  private def insertUserId(id: String): Unit = {
-    existIdSet += id
-  }
-
-  private def increment(): Long = {
-    idCounter.incrementAndGet()
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/FieldDescOnly.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/FieldDescOnly.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/FieldDescOnly.scala
deleted file mode 100644
index e2951f4..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/FieldDescOnly.scala
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-import scala.util.{Success, Try}
-
-trait FieldDescOnly extends Describable with DataSourceable {
-
-}
-
-case class IndexDesc(expr: String) extends FieldDescOnly {
-  val index: Int = {
-    Try(expr.toInt) match {
-      case Success(v) => v
-      case _ => throw new Exception(s"${expr} is invalid index")
-    }
-  }
-  val desc: String = describe(index)
-  val dataSources: Set[String] = Set.empty[String]
-}
-
-case class FieldDesc(expr: String) extends FieldDescOnly {
-  val field: String = expr
-  val desc: String = describe(field)
-  val dataSources: Set[String] = Set.empty[String]
-}
-
-case class AllFieldsDesc(expr: String) extends FieldDescOnly {
-  val allFields: String = expr
-  val desc: String = allFields
-  val dataSources: Set[String] = Set.empty[String]
-}
-
-case class FieldRangeDesc(startField: FieldDescOnly, endField: FieldDescOnly) 
extends FieldDescOnly {
-  val desc: String = {
-    (startField, endField) match {
-      case (f1: IndexDesc, f2: IndexDesc) => s"(${f1.desc}, ${f2.desc})"
-      case _ => throw new Exception("invalid field range description")
-    }
-  }
-  val dataSources: Set[String] = Set.empty[String]
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LiteralExpr.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LiteralExpr.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LiteralExpr.scala
deleted file mode 100644
index de80346..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LiteralExpr.scala
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-import scala.util.{Failure, Success, Try}
-
-trait LiteralExpr extends Expr {
-  val value: Option[Any]
-  def calculateOnly(values: Map[String, Any]): Option[Any] = value
-  val dataSources: Set[String] = Set.empty[String]
-}
-
-case class LiteralStringExpr(expr: String) extends LiteralExpr {
-  val value: Option[String] = Some(expr)
-  val desc: String = s"'${value.getOrElse("")}'"
-}
-
-case class LiteralNumberExpr(expr: String) extends LiteralExpr {
-  val value: Option[Any] = {
-    if (expr.contains(".")) {
-      Try (expr.toDouble) match {
-        case Success(v) => Some(v)
-        case _ => throw new Exception(s"${expr} is invalid number")
-      }
-    } else {
-      Try (expr.toLong) match {
-        case Success(v) => Some(v)
-        case _ => throw new Exception(s"${expr} is invalid number")
-      }
-    }
-  }
-  val desc: String = value.getOrElse("").toString
-}
-
-case class LiteralTimeExpr(expr: String) extends LiteralExpr {
-  final val TimeRegex = """(\d+)(d|h|m|s|ms)""".r
-  val value: Option[Long] = {
-    Try {
-      expr match {
-        case TimeRegex(time, unit) => {
-          val t = time.toLong
-          unit match {
-            case "d" => t * 24 * 60 * 60 * 1000
-            case "h" => t * 60 * 60 * 1000
-            case "m" => t * 60 * 1000
-            case "s" => t * 1000
-            case "ms" => t
-            case _ => throw new Exception(s"${expr} is invalid time format")
-          }
-        }
-        case _ => throw new Exception(s"${expr} is invalid time format")
-      }
-    } match {
-      case Success(v) => Some(v)
-      case Failure(ex) => throw ex
-    }
-  }
-  val desc: String = expr
-}
-
-case class LiteralBooleanExpr(expr: String) extends LiteralExpr {
-  final val TrueRegex = """(?i)true""".r
-  final val FalseRegex = """(?i)false""".r
-  val value: Option[Boolean] = expr match {
-    case TrueRegex() => Some(true)
-    case FalseRegex() => Some(false)
-    case _ => throw new Exception(s"${expr} is invalid boolean")
-  }
-  val desc: String = value.getOrElse("").toString
-}
-
-case class LiteralNullExpr(expr: String) extends LiteralExpr {
-  val value: Option[Any] = Some(null)
-  val desc: String = "null"
-}
-
-case class LiteralNoneExpr(expr: String) extends LiteralExpr {
-  val value: Option[Any] = None
-  val desc: String = "none"
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LogicalExpr.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LogicalExpr.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LogicalExpr.scala
deleted file mode 100644
index fdb665d..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/LogicalExpr.scala
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-import org.apache.griffin.measure.batch.rule.CalculationUtil._
-
-trait LogicalExpr extends Expr with AnalyzableExpr {
-  override def cacheUnit: Boolean = true
-}
-
-case class LogicalSimpleExpr(expr: MathExpr) extends LogicalExpr {
-  def calculateOnly(values: Map[String, Any]): Option[Any] = 
expr.calculate(values)
-  val desc: String = expr.desc
-  val dataSources: Set[String] = expr.dataSources
-  override def cacheUnit: Boolean = false
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = 
expr.getCacheExprs(ds)
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = 
expr.getFinalCacheExprs(ds)
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = 
expr.getPersistExprs(ds)
-}
-
-case class LogicalCompareExpr(left: MathExpr, compare: String, right: 
MathExpr) extends LogicalExpr {
-  private val (eqOpr, neqOpr, btOpr, bteOpr, ltOpr, lteOpr) = ("""==?""".r, 
"""!==?""".r, ">", ">=", "<", "<=")
-  def calculateOnly(values: Map[String, Any]): Option[Any] = {
-    val (lv, rv) = (left.calculate(values), right.calculate(values))
-    compare match {
-      case this.eqOpr() => lv === rv
-      case this.neqOpr() => lv =!= rv
-      case this.btOpr => lv > rv
-      case this.bteOpr => lv >= rv
-      case this.ltOpr => lv < rv
-      case this.lteOpr => lv <= rv
-      case _ => None
-    }
-  }
-  val desc: String = s"${left.desc} ${compare} ${right.desc}"
-  val dataSources: Set[String] = left.dataSources ++ right.dataSources
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = {
-    left.getCacheExprs(ds) ++ right.getCacheExprs(ds)
-  }
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = {
-    left.getFinalCacheExprs(ds) ++ right.getFinalCacheExprs(ds)
-  }
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = {
-    left.getPersistExprs(ds) ++ right.getPersistExprs(ds)
-  }
-
-  override def getGroupbyExprPairs(dsPair: (String, String)): Seq[(Expr, 
Expr)] = {
-    if (compare == "=" || compare == "==") {
-      (left.dataSourceOpt, right.dataSourceOpt) match {
-        case (Some(dsPair._1), Some(dsPair._2)) => (left, right) :: Nil
-        case (Some(dsPair._2), Some(dsPair._1)) => (right, left) :: Nil
-        case _ => Nil
-      }
-    } else Nil
-  }
-}
-
-case class LogicalRangeExpr(left: MathExpr, rangeOpr: String, range: 
RangeDesc) extends LogicalExpr {
-  private val (inOpr, ninOpr, btwnOpr, nbtwnOpr) = ("""(?i)in""".r, 
"""(?i)not\s+in""".r, """(?i)between""".r, """(?i)not\s+between""".r)
-  def calculateOnly(values: Map[String, Any]): Option[Any] = {
-    val (lv, rvs) = (left.calculate(values), 
range.elements.map(_.calculate(values)))
-    rangeOpr match {
-      case this.inOpr() => lv in rvs
-      case this.ninOpr() => lv not_in rvs
-      case this.btwnOpr() => lv between rvs
-      case this.nbtwnOpr() => lv not_between rvs
-      case _ => None
-    }
-  }
-  val desc: String = s"${left.desc} ${rangeOpr} ${range.desc}"
-  val dataSources: Set[String] = left.dataSources ++ 
range.elements.flatMap(_.dataSources).toSet
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = {
-    left.getCacheExprs(ds) ++ range.elements.flatMap(_.getCacheExprs(ds))
-  }
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = {
-    left.getFinalCacheExprs(ds) ++ 
range.elements.flatMap(_.getFinalCacheExprs(ds))
-  }
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = {
-    left.getPersistExprs(ds) ++ range.elements.flatMap(_.getPersistExprs(ds))
-  }
-}
-
-// -- logical statement --
-//case class LogicalFactorExpr(self: LogicalExpr) extends LogicalExpr {
-//  def calculate(values: Map[String, Any]): Option[Any] = 
self.calculate(values)
-//  val desc: String = self.desc
-//}
-
-case class UnaryLogicalExpr(oprList: Iterable[String], factor: LogicalExpr) 
extends LogicalExpr {
-  private val notOpr = """(?i)not|!""".r
-  def calculateOnly(values: Map[String, Any]): Option[Any] = {
-    val fv = factor.calculate(values)
-    oprList.foldRight(fv) { (opr, v) =>
-      opr match {
-        case this.notOpr() => !v
-        case _ => None
-      }
-    }
-  }
-  val desc: String = oprList.foldRight(factor.desc) { (prev, ex) => s"${prev} 
${ex}" }
-  val dataSources: Set[String] = factor.dataSources
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = {
-    factor.getCacheExprs(ds)
-  }
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = {
-    factor.getFinalCacheExprs(ds)
-  }
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = {
-    factor.getPersistExprs(ds)
-  }
-
-  override def getGroupbyExprPairs(dsPair: (String, String)): Seq[(Expr, 
Expr)] = {
-    val notOprList = oprList.filter { opr =>
-      opr match {
-        case this.notOpr() => true
-        case _ => false
-      }
-    }
-    if (notOprList.size % 2 == 0) factor.getGroupbyExprPairs(dsPair) else Nil
-  }
-}
-
-case class BinaryLogicalExpr(first: LogicalExpr, others: Iterable[(String, 
LogicalExpr)]) extends LogicalExpr {
-  private val (andOpr, orOpr) = ("""(?i)and|&&""".r, """(?i)or|\|\|""".r)
-  def calculateOnly(values: Map[String, Any]): Option[Any] = {
-    val fv = first.calculate(values)
-    others.foldLeft(fv) { (v, pair) =>
-      val (opr, next) = pair
-      val nv = next.calculate(values)
-      opr match {
-        case this.andOpr() => v && nv
-        case this.orOpr() => v || nv
-        case _ => None
-      }
-    }
-  }
-  val desc: String = others.foldLeft(first.desc) { (ex, next) => s"${ex} 
${next._1} ${next._2.desc}" }
-  val dataSources: Set[String] = first.dataSources ++ 
others.flatMap(_._2.dataSources).toSet
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = {
-    first.getCacheExprs(ds) ++ others.flatMap(_._2.getCacheExprs(ds))
-  }
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = {
-    first.getFinalCacheExprs(ds) ++ others.flatMap(_._2.getFinalCacheExprs(ds))
-  }
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = {
-    first.getPersistExprs(ds) ++ others.flatMap(_._2.getPersistExprs(ds))
-  }
-
-  override def getGroupbyExprPairs(dsPair: (String, String)): Seq[(Expr, 
Expr)] = {
-    if (others.isEmpty) first.getGroupbyExprPairs(dsPair)
-    else {
-      val isAnd = others.exists(_._1 match {
-        case this.andOpr() => true
-        case _ => false
-      })
-      if (isAnd) {
-        first.getGroupbyExprPairs(dsPair) ++ 
others.flatMap(_._2.getGroupbyExprPairs(dsPair))
-      } else Nil
-    }
-  }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/6fd22ae7/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/MathExpr.scala
----------------------------------------------------------------------
diff --git 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/MathExpr.scala
 
b/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/MathExpr.scala
deleted file mode 100644
index f02d894..0000000
--- 
a/measure/src/main/scala/org/apache/griffin/measure/batch/rule/expr/MathExpr.scala
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
-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.griffin.measure.batch.rule.expr
-
-import org.apache.griffin.measure.batch.rule.CalculationUtil._
-
-trait MathExpr extends Expr {
-
-}
-
-case class MathFactorExpr(self: Expr) extends MathExpr {
-  def calculateOnly(values: Map[String, Any]): Option[Any] = 
self.calculate(values)
-  val desc: String = self.desc
-  val dataSources: Set[String] = self.dataSources
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = {
-    self.getCacheExprs(ds)
-  }
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = {
-    self.getFinalCacheExprs(ds)
-  }
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = {
-    self.getPersistExprs(ds)
-  }
-}
-
-case class UnaryMathExpr(oprList: Iterable[String], factor: Expr) extends 
MathExpr {
-  private val (posOpr, negOpr) = ("+", "-")
-  def calculateOnly(values: Map[String, Any]): Option[Any] = {
-    val fv = factor.calculate(values)
-    oprList.foldRight(fv) { (opr, v) =>
-      opr match {
-        case this.posOpr => v
-        case this.negOpr => -v
-        case _ => None
-      }
-    }
-  }
-  val desc: String = oprList.foldRight(factor.desc) { (prev, ex) => 
s"${prev}${ex}" }
-  val dataSources: Set[String] = factor.dataSources
-  override def cacheUnit: Boolean = true
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = {
-    factor.getCacheExprs(ds)
-  }
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = {
-    factor.getFinalCacheExprs(ds)
-  }
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = {
-    factor.getPersistExprs(ds)
-  }
-}
-
-case class BinaryMathExpr(first: MathExpr, others: Iterable[(String, 
MathExpr)]) extends MathExpr {
-  private val (addOpr, subOpr, mulOpr, divOpr, modOpr) = ("+", "-", "*", "/", 
"%")
-  def calculateOnly(values: Map[String, Any]): Option[Any] = {
-    val fv = first.calculate(values)
-    others.foldLeft(fv) { (v, pair) =>
-      val (opr, next) = pair
-      val nv = next.calculate(values)
-      opr match {
-        case this.addOpr => v + nv
-        case this.subOpr => v - nv
-        case this.mulOpr => v * nv
-        case this.divOpr => v / nv
-        case this.modOpr => v % nv
-        case _ => None
-      }
-    }
-  }
-  val desc: String = others.foldLeft(first.desc) { (ex, next) => s"${ex} 
${next._1} ${next._2.desc}" }
-  val dataSources: Set[String] = first.dataSources ++ 
others.flatMap(_._2.dataSources).toSet
-  override def cacheUnit: Boolean = true
-  override def getSubCacheExprs(ds: String): Iterable[Expr] = {
-    first.getCacheExprs(ds) ++ others.flatMap(_._2.getCacheExprs(ds))
-  }
-  override def getSubFinalCacheExprs(ds: String): Iterable[Expr] = {
-    first.getFinalCacheExprs(ds) ++ others.flatMap(_._2.getFinalCacheExprs(ds))
-  }
-  override def getSubPersistExprs(ds: String): Iterable[Expr] = {
-    first.getPersistExprs(ds) ++ others.flatMap(_._2.getPersistExprs(ds))
-  }
-}
\ No newline at end of file


Reply via email to