diff --git a/griffin-doc/measure/measure-configuration-guide/accuracy.md b/griffin-doc/measure/measure-configuration-guide/accuracy.md
index 95fe54b..f0ad0e5 100644
--- a/griffin-doc/measure/measure-configuration-guide/accuracy.md
+++ b/griffin-doc/measure/measure-configuration-guide/accuracy.md
@@ -242,4 +242,4 @@
 A new column `__status` has been added to the original data set on which this measure was executed. The value of this
 column can be either `bad` or `good` which can be used to calculate the metrics/ separate data based on quality etc.
 
-_Note:_ This output is for `ConsoleSink`. 
\ No newline at end of file
+_Note:_ These outputs are for `ConsoleSink`. 
\ No newline at end of file
diff --git a/griffin-doc/measure/measure-configuration-guide/completeness.md b/griffin-doc/measure/measure-configuration-guide/completeness.md
index accd5a5..e932156 100644
--- a/griffin-doc/measure/measure-configuration-guide/completeness.md
+++ b/griffin-doc/measure/measure-configuration-guide/completeness.md
@@ -31,9 +31,9 @@
 users to define SQL-like expressions which describe their definition of completeness. For a tabular data set with
 columns `name`, `email` and `age`, some examples of such completeness definitions are mentioned below,
 
-- `name is NULL`
-- `name is NULL and age is NULL`
-- `email NOT RLIKE '^[a-zA-Z0-9+_.-]+@[a-zA-Z0-9.-]+$'`
+- `name is NOT NULL`
+- `name is NOT NULL and age is NOT NULL`
+- `email RLIKE '^[a-zA-Z0-9+_.-]+@[a-zA-Z0-9.-]+$'`
 
 ### Configuration
 
@@ -49,7 +49,7 @@
       "type": "completeness",
       "data.source": "crime_report_source",
       "config": {
-        "expr": "zipcode is null OR city is null"
+        "expr": "zipcode is not null and city is not null"
       },
       "out": [
         {
@@ -84,8 +84,8 @@
 `config` object for completeness measure contains only one key `expr`. The value for `expr` is a SQL-like expression
 string which definition this completeness. For more complex definitions, expressions can be clubbed with `AND` and `OR`.
 
-_Note:_ This expression describes the bad or incomplete records. This means that for `"expr": "zipcode is NULL"` the
-records which contain `null` in zipcode column are considered as incomplete.
+_Note:_ This expression describes the good or complete records. This means that for `"expr": "zipcode is NOT NULL"` the
+records which contain `null` in zipcode column are incomplete.
 
 It can be defined as mentioned below,
 
@@ -94,7 +94,7 @@
   ...
 
   "config": {
-    "expr": "zipcode is null OR city is null"
+    "expr": "zipcode is NOT null AND city is NOT null"
   }
 
   ...
@@ -136,7 +136,6 @@
 {
   "applicationId": "local-1623452412322",
   "job_name": "Batch-All-Measures-Example",
-  "tmst": 1623452423891,
   "measure_name": "completeness_measure",
   "metrics": [
     {
@@ -209,4 +208,4 @@
 A new column `__status` has been added to the original data set on which this measure was executed. The value of this
 column can be either `bad` or `good` which can be used to calculate the metrics/ separate data based on quality etc.
 
-_Note:_ This output is for `ConsoleSink`. 
\ No newline at end of file
+_Note:_ These outputs are for `ConsoleSink`. 
\ No newline at end of file
diff --git a/griffin-doc/measure/measure-configuration-guide/duplication.md b/griffin-doc/measure/measure-configuration-guide/duplication.md
index 15e7408..71a79df 100644
--- a/griffin-doc/measure/measure-configuration-guide/duplication.md
+++ b/griffin-doc/measure/measure-configuration-guide/duplication.md
@@ -254,4 +254,4 @@
 column can be either `bad` or `good` which can be used to calculate the metrics/ separate data based on quality etc.
 These values for `__status` column are based on the value of user-defined key `bad.record.definition`.
 
-_Note:_ This output is for `ConsoleSink`. 
\ No newline at end of file
+_Note:_ These outputs are for `ConsoleSink`. 
\ No newline at end of file
diff --git a/griffin-doc/measure/measure-configuration-guide/profiling.md b/griffin-doc/measure/measure-configuration-guide/profiling.md
index b6ee6fa..5c7873b 100644
--- a/griffin-doc/measure/measure-configuration-guide/profiling.md
+++ b/griffin-doc/measure/measure-configuration-guide/profiling.md
@@ -207,4 +207,6 @@
 ```
 
 _Note:_ Some mathematical metrics are bound to the type of attribute under consideration, for example standard deviation
-cannot be calculated for a column name of string type, thus, the value for these metrics are null for such columns.
\ No newline at end of file
+cannot be calculated for a column name of string type, thus, the value for these metrics are null for such columns.
+
+_Note:_ This output is for `ConsoleSink`. 
\ No newline at end of file
diff --git a/griffin-doc/measure/measure-configuration-guide/schema_conformance.md b/griffin-doc/measure/measure-configuration-guide/schema_conformance.md
index 768c0ca..2f47dde 100644
--- a/griffin-doc/measure/measure-configuration-guide/schema_conformance.md
+++ b/griffin-doc/measure/measure-configuration-guide/schema_conformance.md
@@ -215,4 +215,4 @@
 A new column `__status` has been added to the source data set that acted as input to this measure. The value of this
 column can be either `bad` or `good` which can be used to calculate the metrics/ separate data based on quality etc.
 
-_Note:_ This output is for `ConsoleSink`. 
\ No newline at end of file
+_Note:_ These outputs are for `ConsoleSink`. 
diff --git a/griffin-doc/measure/measure-configuration-guide/sparksql.md b/griffin-doc/measure/measure-configuration-guide/sparksql.md
index d8ca39e..e82c743 100644
--- a/griffin-doc/measure/measure-configuration-guide/sparksql.md
+++ b/griffin-doc/measure/measure-configuration-guide/sparksql.md
@@ -211,7 +211,7 @@
 column can be either `bad` or `good` which can be used to calculate the metrics/ separate data based on quality etc.
 These values for `__status` column are based on the value of user-defined key `bad.record.definition`.
 
-_Note:_ This output is for `ConsoleSink`.
+_Note:_ These outputs are for `ConsoleSink`.
 
 **Further Reading**
 
@@ -224,4 +224,4 @@
 `bad.record.definition: "__measure_{measure_name}"` in the measure config.
 
 For example if the user-defined name of the measure is `spark_sql_measure`, the alias column name
-becomes `__measure_spark_sql_measure`.
\ No newline at end of file
+becomes `__measure_spark_sql_measure`.
diff --git a/measure/pom.xml b/measure/pom.xml
index 3cd8bc6..6d97107 100644
--- a/measure/pom.xml
+++ b/measure/pom.xml
@@ -58,6 +58,10 @@
         <scoverage.plugin.version>1.4.0</scoverage.plugin.version>
         <mockito.version>3.2.2.0</mockito.version>
         <scalatest.version>3.2.3</scalatest.version>
+        <maven.surefire.plugin.version>2.22.2</maven.surefire.plugin.version>
+        <scoverage.plugin.version>1.4.0</scoverage.plugin.version>
+        <scalatest.maven.plugin.version>2.0.2</scalatest.maven.plugin.version>
+        <scapegoat.version>1.4.8</scapegoat.version>
     </properties>
 
     <dependencies>
@@ -68,6 +72,13 @@
             <version>${scala.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.sksamuel.scapegoat</groupId>
+            <artifactId>scalac-scapegoat-plugin_${scala.version}</artifactId>
+            <version>${scapegoat.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- Spark Dependencies -->
         <dependency>
             <groupId>org.apache.spark</groupId>
@@ -249,9 +260,13 @@
                         <arg>-deprecation</arg>
                         <arg>-feature</arg>
                         <arg>-explaintypes</arg>
+                        <arg>-P:scapegoat:dataDir:${project.build.directory}</arg>
+                        <arg>-P:scapegoat:reports:scalastyle</arg>
+                        <arg>
+                            -P:scapegoat:overrideLevels:UnsafeTraversableMethods=Warning:OptionGet=Warning:TryGet=Warning
+                        </arg>
                     </args>
                     <jvmArgs>
-                        <jvmArg>-Xms64m</jvmArg>
                         <jvmArg>-Xmx1024m</jvmArg>
                     </jvmArgs>
                     <javacArgs>
@@ -260,19 +275,41 @@
                         <javacArg>${java.version}</javacArg>
                         <javacArg>-Xlint:all,-serial,-path,-try</javacArg>
                     </javacArgs>
+                    <compilerPlugins>
+                        <compilerPlugin>
+                            <groupId>com.sksamuel.scapegoat</groupId>
+                            <artifactId>scalac-scapegoat-plugin_${scala.version}</artifactId>
+                            <version>${scapegoat.version}</version>
+                        </compilerPlugin>
+                    </compilerPlugins>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>${maven.surefire.plugin.version}</version>
+                <configuration>
+                    <skipTests>true</skipTests>
                 </configuration>
             </plugin>
 
             <plugin>
                 <groupId>org.scalatest</groupId>
                 <artifactId>scalatest-maven-plugin</artifactId>
-                <version>2.0.2</version>
+                <version>${scalatest.maven.plugin.version}</version>
                 <configuration>
                     <testFailureIgnore>false</testFailureIgnore>
                     <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
                     <junitxml>.</junitxml>
                     <filereports>TestSuiteReport.txt</filereports>
+                    <parallel>false</parallel>
+                    <forkMode>once</forkMode>
                     <stderr/>
+                    <argLine>
+                        -ea -Xmx4g -Xss4m -XX:ReservedCodeCacheSize=1g -Dio.netty.tryReflectionSetAccessible=true
+                        -Dfile.encoding=${project.build.sourceEncoding}
+                    </argLine>
                     <systemProperties>
                         <java.io.tmpdir>${project.build.directory}/tmp</java.io.tmpdir>
                     </systemProperties>
@@ -297,18 +334,29 @@
                         <goals>
                             <goal>shade</goal>
                         </goals>
+                        <configuration>
+                            <relocations>
+                                <relocation>
+                                    <pattern>org.apache.http</pattern>
+                                    <shadedPattern>griffin.org.apache.http</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>com.zaxxer</pattern>
+                                    <shadedPattern>griffin.com.zaxxer</shadedPattern>
+                                </relocation>
+                            </relocations>
+                            <finalName>${project.build.finalName}-with-dependencies</finalName>
+                            <transformers>
+                                <transformer
+                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+                                <transformer
+                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                                    <mainClass>org.apache.griffin.measure.Application</mainClass>
+                                </transformer>
+                            </transformers>
+                        </configuration>
                     </execution>
                 </executions>
-                <configuration>
-                    <createDependencyReducedPom>false</createDependencyReducedPom>
-                    <minimizeJar>true</minimizeJar>
-                    <relocations>
-                        <relocation>
-                            <pattern>org.apache.http</pattern>
-                            <shadedPattern>griffin.org.apache.http</shadedPattern>
-                        </relocation>
-                    </relocations>
-                </configuration>
             </plugin>
 
             <plugin>
@@ -324,7 +372,7 @@
                 </configuration>
                 <executions>
                     <execution>
-                        <phase>compile</phase>
+                        <phase>validate</phase>
                         <goals>
                             <goal>format</goal>
                         </goals>
@@ -350,7 +398,7 @@
                 </configuration>
                 <executions>
                     <execution>
-                        <phase>compile</phase>
+                        <phase>validate</phase>
                         <goals>
                             <goal>check</goal>
                         </goals>
@@ -359,86 +407,18 @@
             </plugin>
 
             <plugin>
-                <groupId>org.jacoco</groupId>
-                <artifactId>jacoco-maven-plugin</artifactId>
-                <version>0.8.2</version>
-                <executions>
-                    <execution>
-                        <id>default-prepare-agent</id>
-                        <goals>
-                            <goal>prepare-agent</goal>
-                        </goals>
-                    </execution>
-                    <execution>
-                        <id>report</id>
-                        <phase>test</phase>
-                        <goals>
-                            <goal>report</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-
-            <plugin>
                 <groupId>org.scoverage</groupId>
                 <artifactId>scoverage-maven-plugin</artifactId>
                 <version>${scoverage.plugin.version}</version>
                 <configuration>
                     <scalaVersion>${scala.version}</scalaVersion>
-                    <aggregate>true</aggregate>
+                    <aggregateOnly>true</aggregateOnly>
                     <highlighting>true</highlighting>
                 </configuration>
             </plugin>
 
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-jar-plugin</artifactId>
-                <version>3.2.0</version>
-                <executions>
-                    <execution>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>jar</goal>
-                        </goals>
-                        <configuration>
-                            <finalName>${project.artifactId}-${project.version}</finalName>
-                            <classifier>spark-${spark.major.version}_scala-${scala.binary.version}</classifier>
-                        </configuration>
-                    </execution>
-                </executions>
-            </plugin>
-
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
-                <version>3.3.0</version>
-                <configuration>
-                    <tarLongFileMode>posix</tarLongFileMode>
-                    <finalName>
-                        ${project.artifactId}-${project.version}-spark-${spark.major.version}_scala-${scala.binary.version}
-                    </finalName>
-                    <archive>
-                        <manifest>
-                            <mainClass>org.apache.griffin.measure.Application</mainClass>
-                        </manifest>
-                    </archive>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                </configuration>
-                <executions>
-                    <execution>
-                        <id>make-assembly</id>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>single</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-enforcer-plugin</artifactId>
                 <version>3.0.0-M2</version>
                 <executions>
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala
index 1765f4d..7c695c6 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/DQConfig.scala
@@ -89,7 +89,7 @@
 
       assert(
         repeatedMeasures.isEmpty,
-        s"Measure names must be unique. " +
+        "Measure names must be unique. " +
           s"Duplicate Measures names ['${repeatedMeasures.mkString("', '")}'] were found.")
 
       val invalidMeasureSources = measures
@@ -100,7 +100,7 @@
 
       assert(
         invalidMeasureSources.isEmpty,
-        s"Measure source(s) undefined." +
+        "Measure source(s) undefined." +
           s" Unknown source(s) ['${invalidMeasureSources.mkString("', '")}'] were found.")
     } else if (evaluateRule != null) {
       evaluateRule.validate()
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala
index ab35fb6..6698726 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala
@@ -75,7 +75,7 @@
   }
 
   def dataBySql(ms: Long): (Option[DataFrame], TimeRange) = {
-    val path: String = s"/_sql?format=csv"
+    val path: String = "/_sql?format=csv"
     info(s"ElasticSearchGriffinDataConnector data : sql: $sql")
     val dfOpt =
       try {
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala
index 0caac17..cedcfda 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/Measure.scala
@@ -98,10 +98,10 @@
    *
    * @return tuple of records dataframe and metric dataframe
    */
-  def impl(): (DataFrame, DataFrame)
+  def impl(dataSource: DataFrame): (DataFrame, DataFrame)
 
   /**
-   * Implementation should define validtion checks in this method (if required).
+   * Implementation should define validation checks in this method (if required).
    * This method needs to be called explicitly call this method (preferably during measure creation).
    *
    * Defaults to no-op.
@@ -111,35 +111,18 @@
   /**
    * Executes this measure specific transformation on input data source.
    *
-   * @param batchId batch id to append in case of streaming source.
    * @return enriched tuple of records dataframe and metric dataframe
    */
-  def execute(batchId: Option[Long] = None): (DataFrame, DataFrame) = {
-    val (recordsDf, metricDf) = impl()
+  def execute(dataSource: DataFrame): (DataFrame, DataFrame) = {
+    val (recordsDf, metricDf) = impl(dataSource)
 
     val processedRecordDf = preProcessRecords(recordsDf)
     val processedMetricDf = preProcessMetrics(metricDf)
 
-    val res = batchId match {
-      case Some(batchId) =>
-        implicit val bId: Long = batchId
-        (appendBatchIdIfAvailable(processedRecordDf), appendBatchIdIfAvailable(processedMetricDf))
-      case None => (processedRecordDf, processedMetricDf)
-    }
-
-    res
+    (processedRecordDf, processedMetricDf)
   }
 
-  /**
-   * Appends batch id to metrics in case of streaming sources.
-   *
-   * @param input metric dataframe
-   * @param batchId batch id to append
-   * @return updated metric dataframe
-   */
-  private def appendBatchIdIfAvailable(input: DataFrame)(implicit batchId: Long): DataFrame = {
-    input.withColumn(BatchId, typedLit[Long](batchId))
-  }
+  protected def nullToZero(column: Column): Column = when(column.isNull, 0).otherwise(column)
 
 }
 
@@ -152,7 +135,8 @@
   final val Expression = "expr"
   final val MeasureColPrefix = "__measure"
   final val Status = "__status"
-  final val BatchId = "__batch_id"
+  final val BatchId = "batch_id"
+
   final val MeasureName = "measure_name"
   final val MeasureType = "measure_type"
   final val MetricName = "metric_name"
@@ -164,6 +148,7 @@
   final val Total: String = "total"
   final val BadRecordDefinition = "bad.record.definition"
   final val AllColumns: String = "*"
+  final val RowNumber: String = "__row_number"
 
   final val emptyCol: Column = lit(StringUtils.EMPTY)
 
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala
index a84dff0..71a2ef3 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/MeasureExecutor.scala
@@ -18,13 +18,11 @@
 package org.apache.griffin.measure.execution
 
 import java.util.Date
-import java.util.concurrent.Executors
 
-import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService, Future}
+import scala.concurrent.{ExecutionContextExecutorService, Future}
 import scala.util._
 
-import org.apache.commons.lang3.StringUtils
-import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
 
 import org.apache.griffin.measure.Loggable
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
@@ -55,7 +53,8 @@
  *
  * @param context Instance of `DQContext`
  */
-case class MeasureExecutor(context: DQContext) extends Loggable {
+case class MeasureExecutor(context: DQContext, ec: ExecutionContextExecutorService)
+    extends Loggable {
 
   /**
    * SparkSession for this Griffin Application.
@@ -69,19 +68,6 @@
     .getBoolean("spark.griffin.measure.cacheDataSources", defaultValue = true)
 
   /**
-   * Size of thread pool for parallel measure execution.
-   * Defaults to number of processors available to the spark driver JVM.
-   */
-  private val numThreads: Int = sparkSession.sparkContext.getConf
-    .getInt("spark.griffin.measure.parallelism", Runtime.getRuntime.availableProcessors())
-
-  /**
-   * Service to handle threaded execution of tasks (measures).
-   */
-  private implicit val ec: ExecutionContextExecutorService =
-    ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(numThreads))
-
-  /**
    * Starting point of measure execution.
    *
    * @param measureParams Object representation(s) of user defined measure(s).
@@ -98,20 +84,18 @@
         val dataSourceName = measuresForSource._1
         val measureParams = measuresForSource._2
 
-        withCacheIfNecessary(dataSourceName, {
-          val dataSource = sparkSession.read.table(dataSourceName)
+        val dataSource = sparkSession.read.table(dataSourceName)
 
-          if (dataSource.isStreaming) {
-            // TODO this is a no op as streaming queries need to be registered.
+        if (dataSource.isStreaming) {
+          // TODO this is a no op as streaming queries need to be registered.
 
-            dataSource.writeStream
-              .foreachBatch((_, batchId) => {
-                executeMeasures(measureParams, Some(batchId))
-              })
-          } else {
-            executeMeasures(measureParams)
-          }
-        })
+          dataSource.writeStream
+            .foreachBatch((batchDf: Dataset[Row], batchId: Long) => {
+              executeMeasures(batchDf, dataSourceName, measureParams, batchId)
+            })
+        } else {
+          executeMeasures(dataSource, dataSourceName, measureParams)
+        }
       })
   }
 
@@ -121,25 +105,31 @@
    * After the function is complete, the data source is uncached.
    *
    * @param dataSourceName name of data source
+   * @param numMeasures number of measures for each data source
    * @param f function to perform
-   * @param measureCountByDataSource number of measures for each data source
    * @return
    */
-  private def withCacheIfNecessary(dataSourceName: String, f: => Unit)(
-      implicit measureCountByDataSource: Map[String, Int]): Unit = {
-    val numMeasures = measureCountByDataSource(dataSourceName)
+  private def withCacheIfNecessary(
+      dataSourceName: String,
+      numMeasures: Int,
+      dataSource: DataFrame,
+      f: => Unit): Unit = {
     var isCached = false
     if (cacheDataSources && numMeasures > 1) {
-      info(
-        s"Caching data source with name '$dataSourceName' as $numMeasures measures are applied on it.")
-      sparkSession.catalog.cacheTable(dataSourceName)
-      isCached = true
+      if (!dataSource.isStreaming) {
+        info(
+          s"Caching data source with name '$dataSourceName' as $numMeasures measures are applied on it.")
+        dataSource.persist()
+        isCached = true
+      }
     }
 
     f
 
     if (isCached) {
-      sparkSession.catalog.uncacheTable(dataSourceName)
+      info(
+        s"Un-Caching data source with name '$dataSourceName' as measure execution is complete for it.")
+      dataSource.unpersist(true)
     }
   }
 
@@ -155,49 +145,74 @@
    * @param batchId Option batch Id in case of streaming sources to identify micro batches.
    */
   private def executeMeasures(
+      input: DataFrame,
+      dataSourceName: String,
       measureParams: Seq[MeasureParam],
-      batchId: Option[Long] = None): Unit = {
-    val batchDetailsOpt = batchId.map(bId => s"for batch id $bId").getOrElse(StringUtils.EMPTY)
+      batchId: Long = -1L): Unit = {
+    val numMeasures: Int = measureParams.length
 
-    // define the tasks
-    val tasks: Map[String, Future[_]] = (for (i <- measureParams.indices)
-      yield {
-        val measureParam = measureParams(i)
-        val measureName = measureParam.getName
+    withCacheIfNecessary(
+      dataSourceName,
+      numMeasures,
+      input, {
+        import java.util.concurrent.TimeUnit
 
-        (measureName, Future {
-          val currentContext = context.cloneDQContext(ContextId(new Date().getTime))
-          info(s"Started execution of measure with name '$measureName'")
+        import scala.concurrent.duration.Duration
 
-          val measure = createMeasure(measureParam)
-          val (recordsDf, metricsDf) = measure.execute(batchId)
+        // define the tasks
+        val tasks = (for (i <- measureParams.indices)
+          yield {
+            val measureParam = measureParams(i)
+            val measure = createMeasure(measureParam)
+            val measureName = measureParam.getName
 
-          persistMetrics(currentContext, measure, metricsDf)
-          persistRecords(currentContext, measure, recordsDf)
+            (measure, Future {
+              info(s"Started execution of measure with name '$measureName'")
 
-          MetricFlushStep(Some(measureParam)).execute(currentContext)
+              val (recordsDf, metricsDf) = measure.execute(input)
+              val currentContext = context.cloneDQContext(ContextId(new Date().getTime))
+
+              persistMetrics(currentContext, measure, metricsDf)
+              persistRecords(currentContext, measure, recordsDf)
+
+              MetricFlushStep(Some(measureParam)).execute(currentContext)
+            }(ec))
+          }).toMap
+
+        tasks.foreach(task => {
+          val measureName = task._1.measureParam.getName
+
+          task._2.onComplete {
+            case Success(_) =>
+              info(
+                s"Successfully executed measure with name '$measureName' on data source with name " +
+                  s"'$dataSourceName")
+            case Failure(exception) =>
+              error(
+                s"Error occurred while executing measure with name '$measureName' on data source with name " +
+                  s"'$dataSourceName'",
+                exception)
+          }(ec)
         })
-      }).toMap
 
-    tasks.foreach(task =>
-      task._2.onComplete {
-        case Success(_) =>
-          info(s"Successfully executed measure with name '${task._1}' $batchDetailsOpt")
-        case Failure(e) =>
-          error(s"Error executing measure with name '${task._1}' $batchDetailsOpt", e)
+        var deadline = Duration(10, TimeUnit.SECONDS).fromNow
+
+        while (!tasks.forall(_._2.isCompleted)) {
+          if (deadline.isOverdue()) {
+            val unfinishedMeasureNames = tasks
+              .filterNot(_._2.isCompleted)
+              .map(_._1.measureParam.getName)
+              .mkString("['", "', '", "']")
+
+            info(s"Measures with name $unfinishedMeasureNames are still executing.")
+            deadline = Duration(10, TimeUnit.SECONDS).fromNow
+          }
+        }
+
+        info(
+          "Completed execution of all measures for data source with " +
+            s"name '${measureParams.head.getDataSource}'.")
       })
-
-    Thread.sleep(1000)
-
-    while (!tasks.forall(_._2.isCompleted)) {
-      info(
-        s"Measures with name ${tasks.filterNot(_._2.isCompleted).keys.mkString("['", "', '", "']")} " +
-          s"are still executing.")
-      Thread.sleep(1000)
-    }
-
-    info(
-      s"Completed execution of all measures for data source with name '${measureParams.head.getDataSource}'.")
   }
 
   /**
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasure.scala
index d33efc8..d9192f7 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasure.scala
@@ -17,16 +17,15 @@
 
 package org.apache.griffin.measure.execution.impl
 
-import java.util.Locale
-
 import io.netty.util.internal.StringUtil
 import org.apache.spark.sql.{Column, DataFrame, SparkSession}
+import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.StringType
 
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure
-import org.apache.griffin.measure.step.builder.ConstantColumns
+import org.apache.griffin.measure.utils.CommonUtils.safeReduce
 
 /**
  * Accuracy Measure.
@@ -60,7 +59,7 @@
    * @param sourceCol name of source column
    * @param refCol name of reference column
    */
-  case class AccuracyExpr(sourceCol: String, refCol: String)
+  final case class AccuracyExpr(sourceCol: String, refCol: String)
 
   import AccuracyMeasure._
   import Measure._
@@ -100,42 +99,47 @@
    *
    *  @return tuple of records dataframe and metric dataframe
    */
-  override def impl(): (DataFrame, DataFrame) = {
-    val originalSource = sparkSession.read.table(measureParam.getDataSource)
-    val originalCols = originalSource.columns
+  override def impl(input: DataFrame): (DataFrame, DataFrame) = {
+    val originalCols = input.columns
 
-    val dataSource = addColumnPrefix(originalSource, SourcePrefixStr)
+    val dataSource = addColumnPrefix(input, SourcePrefixStr)
 
     val refDataSource =
-      addColumnPrefix(sparkSession.read.table(refSource).drop(ConstantColumns.tmst), refPrefixStr)
+      addColumnPrefix(sparkSession.read.table(refSource), refPrefixStr)
 
-    val accuracyExprs = exprOpt.get
+    val expr = exprOpt.getOrElse(throw new AssertionError(s"'$Expression' must be defined."))
+    val accuracyExprs = expr
       .map(toAccuracyExpr)
       .distinct
       .map(x => AccuracyExpr(s"$SourcePrefixStr${x.sourceCol}", s"$refPrefixStr${x.refCol}"))
 
-    val joinExpr =
+    val joinExpr = safeReduce(
       accuracyExprs
-        .map(e => col(e.sourceCol) === col(e.refCol))
-        .reduce(_ and _)
+        .map(e => col(e.sourceCol) === col(e.refCol)))(_ and _)
 
     val indicatorExpr =
-      accuracyExprs
-        .map(e => coalesce(col(e.sourceCol), emptyCol) notEqual coalesce(col(e.refCol), emptyCol))
-        .reduce(_ or _)
+      safeReduce(
+        accuracyExprs
+          .map(e =>
+            coalesce(col(e.sourceCol), emptyCol) notEqual coalesce(col(e.refCol), emptyCol)))(
+        _ or _)
 
-    val nullExpr = accuracyExprs.map(e => col(e.sourceCol).isNull).reduce(_ or _)
+    val nullExpr = safeReduce(accuracyExprs.map(e => col(e.sourceCol).isNull))(_ or _)
+
+    val cols = accuracyExprs.map(_.refCol).map(col)
+    val window = Window.partitionBy(cols: _*).orderBy(cols: _*)
 
     val recordsDf = removeColumnPrefix(
       dataSource
-        .join(refDataSource, joinExpr, "left")
+        .join(refDataSource.withColumn(RowNumber, row_number().over(window)), joinExpr, "left")
+        .where(col(RowNumber) === 1 or col(RowNumber).isNull)
         .withColumn(valueColumn, when(indicatorExpr or nullExpr, 1).otherwise(0)),
       SourcePrefixStr)
       .select((originalCols :+ valueColumn).map(col): _*)
 
     val selectCols =
       Seq(Total, AccurateStr, InAccurateStr).map(e =>
-        map(lit(MetricName), lit(e), lit(MetricValue), col(e).cast(StringType)))
+        map(lit(MetricName), lit(e), lit(MetricValue), nullToZero(col(e).cast(StringType))))
     val metricColumn: Column = array(selectCols: _*).as(valueColumn)
 
     val metricDf = recordsDf
@@ -165,8 +169,8 @@
    * Validates if the expression is not null and non empty along with some dataset specific validations.
    */
   override def validate(): Unit = {
-    assert(exprOpt.isDefined, s"'$Expression' must be defined.")
-    assert(exprOpt.get.flatten.nonEmpty, s"'$Expression' must not be empty or of invalid type.")
+    val expr = exprOpt.getOrElse(throw new AssertionError(s"'$Expression' must be defined."))
+    assert(expr.flatten.nonEmpty, s"'$Expression' must not be empty or of invalid type.")
 
     assert(
       !StringUtil.isNullOrEmpty(refSource),
@@ -178,12 +182,10 @@
 
     val datasourceName = measureParam.getDataSource
 
-    val dataSourceCols =
-      sparkSession.read.table(datasourceName).columns.map(_.toLowerCase(Locale.ROOT)).toSet
-    val refDataSourceCols =
-      sparkSession.read.table(refSource).columns.map(_.toLowerCase(Locale.ROOT)).toSet
+    val dataSourceCols = sparkSession.read.table(datasourceName).columns.toSet
+    val refDataSourceCols = sparkSession.read.table(refSource).columns.toSet
 
-    val accuracyExpr = exprOpt.get.map(toAccuracyExpr).distinct
+    val accuracyExpr = expr.map(toAccuracyExpr).distinct
     val (forDataSource, forRefDataSource) =
       accuracyExpr
         .map(
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasure.scala
index fbbadc6..fabe166 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasure.scala
@@ -39,15 +39,10 @@
 case class CompletenessMeasure(sparkSession: SparkSession, measureParam: MeasureParam)
     extends Measure {
 
+  import CompletenessMeasure._
   import Measure._
 
   /**
-   * Completeness Constants
-   */
-  final val Complete: String = "complete"
-  final val InComplete: String = "incomplete"
-
-  /**
    * Completeness measure supports record and metric write
    */
   override val supportsRecordWrite: Boolean = true
@@ -80,16 +75,15 @@
    *
    *  @return tuple of records dataframe and metric dataframe
    */
-  override def impl(): (DataFrame, DataFrame) = {
-    val exprStr = exprOpt.get
+  override def impl(input: DataFrame): (DataFrame, DataFrame) = {
+    val exprStr = exprOpt.getOrElse(throw new AssertionError(s"'$Expression' must be defined."))
 
     val selectCols =
       Seq(Total, Complete, InComplete).map(e =>
-        map(lit(MetricName), lit(e), lit(MetricValue), col(e).cast(StringType)))
+        map(lit(MetricName), lit(e), lit(MetricValue), nullToZero(col(e).cast(StringType))))
     val metricColumn: Column = array(selectCols: _*).as(valueColumn)
 
-    val input = sparkSession.read.table(measureParam.getDataSource)
-    val badRecordsDf = input.withColumn(valueColumn, when(expr(exprStr), 1).otherwise(0))
+    val badRecordsDf = input.withColumn(valueColumn, when(expr(exprStr), 0).otherwise(1))
 
     val metricDf = badRecordsDf
       .withColumn(Total, lit(1))
@@ -104,9 +98,16 @@
    * Validates if expression is defined and is non empty.
    */
   override def validate(): Unit = {
-    assert(exprOpt.isDefined, s"'$Expression' must be defined.")
-    assert(exprOpt.nonEmpty, s"'$Expression' must not be empty.")
-
-    assert(!StringUtil.isNullOrEmpty(exprOpt.get), s"'$Expression' must not be null or empty.")
+    val expr = exprOpt.getOrElse(throw new AssertionError(s"'$Expression' must be defined."))
+    assert(!StringUtil.isNullOrEmpty(expr), s"'$Expression' must not be null or empty.")
   }
 }
+
+object CompletenessMeasure {
+
+  /**
+   * Completeness Constants
+   */
+  final val Complete: String = "complete"
+  final val InComplete: String = "incomplete"
+}
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasure.scala
index 359a80e..12588cd 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasure.scala
@@ -26,6 +26,7 @@
 
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure
+import org.apache.griffin.measure.utils.CommonUtils.safeReduce
 
 /**
  * Duplication Measure.
@@ -95,38 +96,33 @@
    *
    *  @return tuple of records dataframe and metric dataframe
    */
-  override def impl(): (DataFrame, DataFrame) = {
-    val input = sparkSession.read.table(measureParam.getDataSource)
+  override def impl(input: DataFrame): (DataFrame, DataFrame) = {
     val cols = keyCols(input).map(col)
 
-    val isNullCol = cols.map(x => x.isNull).reduce(_ and _)
-    val duplicateCol = when(col(__Temp) > 1, 1).otherwise(0)
-    val uniqueCol = when(not(isNullCol) and col(Unique) === 1, 1).otherwise(0)
-    val distinctCol =
-      when(not(isNullCol) and (col(Unique) === 1 or col(NonUnique) === 1), 1).otherwise(0)
-    val nonUniqueCol =
-      when(not(isNullCol) and col(Unique) =!= 1 and (col(__Temp) - col(NonUnique) === 0), 1)
-        .otherwise(0)
+    val isNullCol = safeReduce(cols.map(x => x.isNull))(_ and _)
+    val uniqueCol = condition(col(Count) === 1)
+    val nonUniqueCol = condition(col(Count) > 1 and col(RowNumber) === 1)
+    val duplicateCol = condition(col(Count) > 1 and col(RowNumber) > 1)
+    val distinctCol = condition(col(Unique) === 1 or col(NonUnique) === 1)
 
     val window = Window.partitionBy(cols: _*).orderBy(cols: _*)
 
     val aggDf = input
-      .select(col(AllColumns), row_number().over(window).as(__Temp))
-      .withColumn(IsNull, isNullCol)
-      .withColumn(Duplicate, duplicateCol)
-      .withColumn(Unique, count(lit(1)).over(window))
+      .withColumn(IsNotNull, not(isNullCol))
+      .withColumn(RowNumber, row_number().over(window))
+      .withColumn(Count, count(lit(1)).over(window))
       .withColumn(Unique, uniqueCol)
-      .withColumn(NonUnique, min(__Temp).over(window))
       .withColumn(NonUnique, nonUniqueCol)
+      .withColumn(Duplicate, duplicateCol)
       .withColumn(Distinct, distinctCol)
       .withColumn(Total, lit(1))
       .withColumn(valueColumn, col(badnessExpr))
-      .drop(__Temp, IsNull)
+      .drop(IsNotNull, RowNumber, Count)
 
     val metricAggCols = duplicationMeasures.map(m => sum(m).as(m))
 
     val selectCols = duplicationMeasures.map(e =>
-      map(lit(MetricName), lit(e), lit(MetricValue), col(e).cast(StringType)))
+      map(lit(MetricName), lit(e), lit(MetricValue), nullToZero(col(e).cast(StringType))))
     val metricColumn: Column = array(selectCols: _*).as(valueColumn)
 
     val metricDf = aggDf
@@ -160,6 +156,11 @@
     }, s"Invalid value '$badnessExpr' was provided for $BadRecordDefinition")
   }
 
+  private def condition(c: Column, checkNotNull: Boolean = true): Column = {
+    val notNullExpr = if (checkNotNull) col(IsNotNull) else lit(true)
+    when(notNullExpr and c, 1).otherwise(0)
+  }
+
   private def keyCols(input: DataFrame): Array[String] = {
     if (StringUtil.isNullOrEmpty(exprs)) input.columns
     else exprs.split(",").map(_.trim)
@@ -171,10 +172,11 @@
  * Duplication measure constants
  */
 object DuplicationMeasure {
-  final val IsNull: String = "is_null"
   final val Duplicate: String = "duplicate"
   final val Unique: String = "unique"
   final val NonUnique: String = "non_unique"
   final val Distinct: String = "distinct"
-  final val __Temp: String = "__temp"
+
+  final val IsNotNull: String = "is_not_null"
+  final val Count: String = "count"
 }
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasure.scala
index 52d8bff..fedfe37 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasure.scala
@@ -17,8 +17,7 @@
 
 package org.apache.griffin.measure.execution.impl
 
-import java.util.Locale
-
+import io.netty.util.internal.StringUtil
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
@@ -26,7 +25,6 @@
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure
 import org.apache.griffin.measure.execution.Measure._
-import org.apache.griffin.measure.step.builder.ConstantColumns
 
 /**
  * Profiling measure.
@@ -71,12 +69,12 @@
    * profiling measure is to be executed. `expr` is an optional key for Profiling measure,
    * i.e., if it is not defined, all columns in the data set will be profiled.
    */
-  val exprOpt: Option[String] = Option(getFromConfig[String](Expression, null))
+  val exprs: String = getFromConfig[String](Expression, null)
 
   /**
-   * The value for this key is boolean. If this is `true`, the distinct counts will be approximated
-   * to allow up to 5% error. Approximate counts are usually faster by are less accurate. If this is set
-   * to `false`, then the counts will be 100% accurate.
+   * Several resultant metrics of profiling measure are floating-point numbers. This key controls to extent
+   * to which these floating-point numbers are rounded. For example, if `round.scale = 2` then all
+   * floating-point metric values will be rounded to 2 decimal places.
    */
   val roundScale: Int = getFromConfig[java.lang.Integer](RoundScaleStr, 3)
 
@@ -91,13 +89,15 @@
   val dataSetSample: Double = getFromConfig[java.lang.Double](DataSetSampleStr, 1.0)
 
   /**
-   * Several resultant metrics of profiling measure are floating-point numbers. This key controls to extent
-   * to which these floating-point numbers are rounded. For example, if `round.scale = 2` then all
-   * floating-point metric values will be rounded to 2 decimal places.
+   * The value for this key is boolean. If this is `true`, the distinct counts will be approximated
+   * to allow up to 5% error. Approximate counts are usually faster by are less accurate. If this is set
+   * to `false`, then the counts will be 100% accurate.
    */
   val approxDistinctCount: Boolean =
     getFromConfig[java.lang.Boolean](ApproxDistinctCountStr, true)
 
+  validate()
+
   /**
    * Various metrics are calculated for columns of the data set. If expr is correctly defined,
    * then metrics are generated for only the given subset of columns else, its generated for all columns.
@@ -118,26 +118,17 @@
    *
    *  @return tuple of records dataframe and metric dataframe
    */
-  override def impl(): (DataFrame, DataFrame) = {
+  override def impl(dataSource: DataFrame): (DataFrame, DataFrame) = {
     info(s"Selecting random ${dataSetSample * 100}% of the rows for profiling.")
-    val input = sparkSession.read.table(measureParam.getDataSource).sample(dataSetSample)
-    val profilingColNames = exprOpt
-      .getOrElse(input.columns.mkString(","))
-      .split(",")
-      .map(_.trim.toLowerCase(Locale.ROOT))
-      .toSet
+    val input = dataSource.sample(dataSetSample)
+    val profilingColNames = keyCols(input)
 
-    val profilingCols =
-      input.schema.fields.filter(f =>
-        profilingColNames.contains(f.name) && !f.name.equalsIgnoreCase(ConstantColumns.tmst))
-
-    assert(
-      profilingCols.nonEmpty,
-      s"Invalid columns [${profilingCols.map(_.name).mkString(", ")}] were provided for profiling.")
+    val profilingCols = input.schema.fields.filter(f => profilingColNames.contains(f.name))
 
     val profilingExprs = profilingCols.foldLeft(Array.empty[Column])((exprList, field) => {
       val colName = field.name
-      val profilingExprs = getProfilingExprs(field, roundScale, approxDistinctCount)
+      val profilingExprs =
+        getProfilingExprs(field, roundScale, approxDistinctCount, dataSetSample).map(nullToZero)
 
       exprList.:+(map(profilingExprs: _*).as(s"$DetailsPrefix$colName"))
     })
@@ -165,6 +156,24 @@
     (sparkSession.emptyDataFrame, metricDf)
   }
 
+  override def validate(): Unit = {
+    val input = sparkSession.read.table(measureParam.getDataSource)
+    val kc = keyCols(input)
+
+    assert(kc.nonEmpty, s"Columns defined in '$Expression' is empty.")
+    kc.foreach(c =>
+      assert(input.columns.contains(c), s"Provided column '$c' does not exist in the dataset."))
+
+    assert(
+      dataSetSample > 0.0d && dataSetSample <= 1.0d,
+      "Sample fraction of rows must be in range [0.0, 1.0].")
+  }
+
+  private def keyCols(input: DataFrame): Array[String] = {
+    if (StringUtil.isNullOrEmpty(exprs)) input.columns
+    else exprs.split(",").map(_.trim)
+  }.distinct
+
 }
 
 /**
@@ -227,7 +236,8 @@
   private def getProfilingExprs(
       field: StructField,
       roundScale: Int,
-      approxDistinctCount: Boolean): Seq[Column] = {
+      approxDistinctCount: Boolean,
+      dataSetSample: Double): Seq[Column] = {
     val colName = field.name
     val colType = field.dataType
 
@@ -243,12 +253,16 @@
         (lit(DistinctCount), countDistinct(column).as(DistinctCount))
       }
 
+    val distinctExpr = if (dataSetSample == 1) {
+      Seq(lit(distinctCountName), distinctCountExpr)
+    } else Nil
+
     Seq(
       Seq(lit(DataTypeStr), lit(colType.catalogString).as(DataTypeStr)),
       Seq(lit(Total), sum(lit(1)).as(Total)),
       Seq(lit(MinColLength), min(lengthColExpr).as(MinColLength)),
       Seq(lit(MaxColLength), max(lengthColExpr).as(MaxColLength)),
-      Seq(lit(AvgColLength), forNumericFn(colType, avg(lengthColExpr), AvgColLength)),
+      Seq(lit(AvgColLength), avg(lengthColExpr).as(AvgColLength)),
       Seq(lit(Min), forNumericFn(colType, min(column), Min)),
       Seq(lit(Max), forNumericFn(colType, max(column), Max)),
       Seq(lit(Avg), forNumericFn(colType, bround(avg(column), roundScale), Avg)),
@@ -257,7 +271,7 @@
         forNumericFn(colType, bround(stddev(column), roundScale), StdDeviation)),
       Seq(lit(Variance), forNumericFn(colType, bround(variance(column), roundScale), Variance)),
       Seq(lit(Kurtosis), forNumericFn(colType, bround(kurtosis(column), roundScale), Kurtosis)),
-      Seq(lit(distinctCountName), distinctCountExpr),
+      distinctExpr,
       Seq(lit(NullCount), sum(nullColExpr).as(NullCount))).flatten
   }
 }
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasure.scala
index 2111d6a..3247b9b 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasure.scala
@@ -17,14 +17,13 @@
 
 package org.apache.griffin.measure.execution.impl
 
-import java.util.Locale
-
 import org.apache.spark.sql.{Column, DataFrame, SparkSession}
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.{DataType, StringType}
 
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure
+import org.apache.griffin.measure.utils.CommonUtils.safeReduce
 
 /**
  * SchemaConformance Measure.
@@ -43,6 +42,7 @@
 case class SchemaConformanceMeasure(sparkSession: SparkSession, measureParam: MeasureParam)
     extends Measure {
 
+  import CompletenessMeasure._
   import Measure._
 
   /**
@@ -51,15 +51,13 @@
    * @param sourceCol name of source column
    * @param dataType name of reference column
    */
-  case class SchemaConformanceExpr(sourceCol: String, dataType: DataType)
+  final case class SchemaConformanceExpr(sourceCol: String, dataType: DataType)
 
   /**
    * SchemaConformance Constants
    */
   final val SourceColStr: String = "source.col"
   final val DataTypeStr: String = "type"
-  final val Complete: String = "complete"
-  final val InComplete: String = "incomplete"
 
   /**
    * SchemaConformance measure supports record and metric write
@@ -90,19 +88,21 @@
    *
    *  @return tuple of records dataframe and metric dataframe
    */
-  override def impl(): (DataFrame, DataFrame) = {
-    val givenExprs = exprOpt.get.map(toSchemaConformanceExpr).distinct
+  override def impl(input: DataFrame): (DataFrame, DataFrame) = {
+    val expr = exprOpt.getOrElse(throw new AssertionError(s"'$Expression' must be defined."))
+    val givenExprs = expr.map(toSchemaConformanceExpr).distinct
 
-    val incompleteExpr = givenExprs
-      .map(e => when(col(e.sourceCol).cast(e.dataType).isNull, true).otherwise(false))
-      .reduce(_ or _)
+    val incompleteExpr = safeReduce(
+      givenExprs
+        .map(e =>
+          when(col(e.sourceCol).cast(StringType).cast(e.dataType).isNull, true)
+            .otherwise(false)))(_ or _)
 
     val selectCols =
       Seq(Total, Complete, InComplete).map(e =>
-        map(lit(MetricName), lit(e), lit(MetricValue), col(e).cast(StringType)))
+        map(lit(MetricName), lit(e), lit(MetricValue), nullToZero(col(e).cast(StringType))))
     val metricColumn: Column = array(selectCols: _*).as(valueColumn)
 
-    val input = sparkSession.read.table(measureParam.getDataSource)
     val badRecordsDf = input.withColumn(valueColumn, when(incompleteExpr, 1).otherwise(0))
 
     val metricDf = badRecordsDf
@@ -132,14 +132,13 @@
    * Validates if the expression is not null and non empty along with some dataset specific validations.
    */
   override def validate(): Unit = {
-    assert(exprOpt.isDefined, s"'$Expression' must be defined.")
-    assert(exprOpt.get.flatten.nonEmpty, s"'$Expression' must not be empty or of invalid type.")
+    val expr = exprOpt.getOrElse(throw new AssertionError(s"'$Expression' must be defined."))
+    assert(expr.flatten.nonEmpty, s"'$Expression' must not be empty or of invalid type.")
 
     val datasourceName = measureParam.getDataSource
 
-    val dataSourceCols =
-      sparkSession.read.table(datasourceName).columns.map(_.toLowerCase(Locale.ROOT)).toSet
-    val schemaConformanceExprExpr = exprOpt.get.map(toSchemaConformanceExpr).distinct
+    val dataSourceCols = sparkSession.read.table(datasourceName).columns.toSet
+    val schemaConformanceExprExpr = expr.map(toSchemaConformanceExpr).distinct
 
     val forDataSource =
       schemaConformanceExprExpr.map(e => (e.sourceCol, dataSourceCols.contains(e.sourceCol)))
diff --git a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SparkSQLMeasure.scala b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SparkSQLMeasure.scala
index 0cdc090..2e79257 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SparkSQLMeasure.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/execution/impl/SparkSQLMeasure.scala
@@ -43,15 +43,10 @@
 case class SparkSQLMeasure(sparkSession: SparkSession, measureParam: MeasureParam)
     extends Measure {
 
+  import CompletenessMeasure._
   import Measure._
 
   /**
-   * SparkSQL measure constants
-   */
-  final val Complete: String = "complete"
-  final val InComplete: String = "incomplete"
-
-  /**
    * SparkSQL measure supports record and metric write
    */
   override val supportsRecordWrite: Boolean = true
@@ -87,8 +82,8 @@
    *
    *  @return tuple of records dataframe and metric dataframe
    */
-  override def impl(): (DataFrame, DataFrame) = {
-    val df = sparkSession.sql(expr).withColumn(valueColumn, sparkExpr(badnessExpr))
+  override def impl(dataSource: DataFrame): (DataFrame, DataFrame) = {
+    val df = dataSource.sparkSession.sql(expr).withColumn(valueColumn, sparkExpr(badnessExpr))
 
     assert(
       df.schema.exists(f => f.name.matches(valueColumn) && f.dataType.isInstanceOf[BooleanType]),
@@ -96,7 +91,7 @@
 
     val selectCols =
       Seq(Total, Complete, InComplete).map(e =>
-        map(lit(MetricName), lit(e), lit(MetricValue), col(e).cast(StringType)))
+        map(lit(MetricName), lit(e), lit(MetricValue), nullToZero(col(e).cast(StringType))))
     val metricColumn: Column = array(selectCols: _*).as(valueColumn)
 
     val badRecordsDf = df.withColumn(valueColumn, when(col(valueColumn), 1).otherwise(0))
diff --git a/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala b/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala
index 48231ac..6d6200a 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/launch/batch/BatchDQApp.scala
@@ -17,8 +17,9 @@
 
 package org.apache.griffin.measure.launch.batch
 
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{Executors, TimeUnit}
 
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService}
 import scala.util.Try
 
 import org.apache.spark.SparkConf
@@ -85,7 +86,18 @@
 
           if (dqParam.getMeasures != null && dqParam.getMeasures.nonEmpty) {
             Try {
-              MeasureExecutor(dqContext).execute(dqParam.getMeasures)
+              // Size of thread pool for parallel measure execution.
+              // Defaults to number of processors available to the spark driver JVM.
+              val numThreads: Int = sparkSession.sparkContext.getConf
+                .getInt(
+                  "spark.griffin.measure.parallelism",
+                  Runtime.getRuntime.availableProcessors())
+
+              // Service to handle threaded execution of tasks (measures).
+              val ec: ExecutionContextExecutorService =
+                ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(numThreads))
+
+              MeasureExecutor(dqContext, ec).execute(dqParam.getMeasures)
               true
             }
           } else {
diff --git a/measure/src/main/scala/org/apache/griffin/measure/utils/CommonUtils.scala b/measure/src/main/scala/org/apache/griffin/measure/utils/CommonUtils.scala
index 268b3a0..08bfb9b 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/utils/CommonUtils.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/utils/CommonUtils.scala
@@ -41,4 +41,17 @@
 
     result
   }
+
+  /**
+   *
+   * @param t `Traversable`
+   * @param op operation
+   * @tparam T1 any type
+   * @return instance of `T1`
+   */
+  def safeReduce[T1](t: Iterable[T1])(op: (T1, T1) => T1): T1 = {
+    val head :: tail = t.toList
+    tail.fold(head)(op)
+  }
+
 }
diff --git a/measure/src/test/resources/_completeness-batch-griffindsl.json b/measure/src/test/resources/_completeness-batch-griffindsl.json
index 2a7f6a4..9f0f748 100644
--- a/measure/src/test/resources/_completeness-batch-griffindsl.json
+++ b/measure/src/test/resources/_completeness-batch-griffindsl.json
@@ -21,7 +21,7 @@
       "type": "completeness",
       "data.source": "source",
       "config": {
-        "expr": "email is null or post_code is null or first_name is null"
+        "expr": "email is not null and post_code is not null and first_name is not null"
       },
       "out": [
         {
diff --git a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasureTest.scala b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasureTest.scala
index 2f81d1c..206761f 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasureTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/AccuracyMeasureTest.scala
@@ -136,7 +136,7 @@
 
   it should "execute defined measure expr" in {
     val measure = AccuracyMeasure(spark, param)
-    val (recordsDf, metricsDf) = measure.execute(None)
+    val (recordsDf, metricsDf) = measure.execute(source)
 
     assertResult(recordDfSchema)(recordsDf.schema)
     assertResult(metricDfSchema)(metricsDf.schema)
@@ -155,4 +155,27 @@
     assertResult(metricMap(InAccurateStr))("3")
   }
 
+  it should "support space in column name" in {
+    val newSource = source.withColumnRenamed("name", "first name")
+    newSource.createOrReplaceTempView("newSource")
+    val measure = AccuracyMeasure(spark, param)
+    val (recordsDf, metricsDf) = measure.execute(newSource)
+
+    assertResult(newSource.schema.add(Status, "string", nullable = false))(recordsDf.schema)
+    assertResult(metricDfSchema)(metricsDf.schema)
+
+    assertResult(newSource.count())(recordsDf.count())
+    assertResult(1L)(metricsDf.count())
+
+    val metricMap = metricsDf
+      .head()
+      .getAs[Seq[Map[String, String]]](Metrics)
+      .map(x => x(MetricName) -> x(MetricValue))
+      .toMap
+
+    assertResult(metricMap(Total))("5")
+    assertResult(metricMap(AccurateStr))("2")
+    assertResult(metricMap(InAccurateStr))("3")
+  }
+
 }
diff --git a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasureTest.scala b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasureTest.scala
index d6e8b74..db0127c 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasureTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/CompletenessMeasureTest.scala
@@ -23,13 +23,14 @@
 
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure._
+import org.apache.griffin.measure.execution.impl.CompletenessMeasure._
 
 class CompletenessMeasureTest extends MeasureTest {
   var param: MeasureParam = _
 
   override def beforeAll(): Unit = {
     super.beforeAll()
-    param = MeasureParam("param", "Completeness", "source", Map(Expression -> "name is null"))
+    param = MeasureParam("param", "Completeness", "source", Map(Expression -> "name is not null"))
   }
 
   "CompletenessMeasure" should "validate expression config" in {
@@ -61,8 +62,8 @@
   }
 
   it should "execute defined measure expr" in {
-    val measure = CompletenessMeasure(spark, param)
-    val (recordsDf, metricsDf) = measure.execute(None)
+    val measure = new CompletenessMeasure(spark, param)
+    val (recordsDf, metricsDf) = measure.execute(source)
 
     assertResult(recordDfSchema)(recordsDf.schema)
     assertResult(metricDfSchema)(metricsDf.schema)
@@ -77,15 +78,15 @@
       .toMap
 
     assertResult(metricMap(Total))("5")
-    assertResult(metricMap(measure.Complete))("4")
-    assertResult(metricMap(measure.InComplete))("1")
+    assertResult(metricMap(Complete))("4")
+    assertResult(metricMap(InComplete))("1")
   }
 
   it should "supported complex measure expr" in {
-    val measure = CompletenessMeasure(
+    val measure = new CompletenessMeasure(
       spark,
-      param.copy(config = Map(Expression -> "name is null or gender is null")))
-    val (recordsDf, metricsDf) = measure.execute(None)
+      param.copy(config = Map(Expression -> "name is not null and gender is not null")))
+    val (recordsDf, metricsDf) = measure.execute(source)
 
     assertResult(recordDfSchema)(recordsDf.schema)
     assertResult(metricDfSchema)(metricsDf.schema)
@@ -100,19 +101,19 @@
       .toMap
 
     assertResult(metricMap(Total))("5")
-    assertResult(metricMap(measure.Complete))("3")
-    assertResult(metricMap(measure.InComplete))("2")
+    assertResult(metricMap(Complete))("3")
+    assertResult(metricMap(InComplete))("2")
   }
 
   it should "throw runtime error for invalid expr" in {
     assertThrows[AnalysisException] {
-      CompletenessMeasure(spark, param.copy(config = Map(Expression -> "xyz is null")))
-        .execute()
+      new CompletenessMeasure(spark, param.copy(config = Map(Expression -> "xyz is null")))
+        .execute(source)
     }
 
     assertThrows[ParseException] {
-      CompletenessMeasure(spark, param.copy(config = Map(Expression -> "select 1")))
-        .execute()
+      new CompletenessMeasure(spark, param.copy(config = Map(Expression -> "select 1")))
+        .execute(source)
     }
   }
 
diff --git a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasureTest.scala b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasureTest.scala
index 7e22617..f807434 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasureTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasureTest.scala
@@ -107,7 +107,7 @@
   it should "execute defined measure expr" in {
     val measure =
       DuplicationMeasure(spark, param.copy(config = Map(BadRecordDefinition -> "duplicate")))
-    val (recordsDf, metricsDf) = measure.execute(None)
+    val (recordsDf, metricsDf) = measure.execute(source)
 
     assertResult(recordDfSchema)(recordsDf.schema)
     assertResult(metricDfSchema)(metricsDf.schema)
@@ -128,11 +128,11 @@
     assertResult(metricMap(Total))("5")
   }
 
-  it should "supported complex measure expr" in {
+  it should "support complex measure expr" in {
     val measure = DuplicationMeasure(
       spark,
       param.copy(config = Map(Expression -> "name", BadRecordDefinition -> "duplicate")))
-    val (recordsDf, metricsDf) = measure.execute(None)
+    val (recordsDf, metricsDf) = measure.execute(source)
 
     assertResult(recordDfSchema)(recordsDf.schema)
     assertResult(metricDfSchema)(metricsDf.schema)
@@ -153,4 +153,34 @@
     assertResult(metricMap(Total))("5")
   }
 
+  it should "support columns with space in their names" in {
+    val newSource = source.withColumnRenamed("name", "full name")
+    newSource.createOrReplaceTempView("newSource")
+    val measure = DuplicationMeasure(
+      spark,
+      param.copy(
+        dataSource = "newSource",
+        config = Map(Expression -> "full name", BadRecordDefinition -> "duplicate")))
+
+    val (recordsDf, metricsDf) = measure.execute(newSource)
+
+    assertResult(newSource.schema.add(Status, "string", nullable = false))(recordsDf.schema)
+    assertResult(metricDfSchema)(metricsDf.schema)
+
+    assertResult(newSource.count())(recordsDf.count())
+    assertResult(1L)(metricsDf.count())
+
+    val metricMap = metricsDf
+      .head()
+      .getAs[Seq[Map[String, String]]](Metrics)
+      .map(x => x(MetricName) -> x(MetricValue))
+      .toMap
+
+    assertResult(metricMap(Duplicate))("1")
+    assertResult(metricMap(Unique))("2")
+    assertResult(metricMap(NonUnique))("1")
+    assertResult(metricMap(Distinct))("3")
+    assertResult(metricMap(Total))("5")
+  }
+
 }
diff --git a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasureTest.scala b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasureTest.scala
index e4f1b21..d32730f 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasureTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/ProfilingMeasureTest.scala
@@ -70,7 +70,7 @@
     assertResult(3)(measure.roundScale)
     assertResult(true)(measure.approxDistinctCount)
 
-    val (_, metricsDf) = measure.execute(None)
+    val (_, metricsDf) = measure.execute(source)
 
     assertResult(1L)(metricsDf.count())
 
@@ -86,7 +86,7 @@
     assertResult(3)(measure.roundScale)
     assertResult(false)(measure.approxDistinctCount)
 
-    val (_, metricsDf) = measure.execute(None)
+    val (_, metricsDf) = measure.execute(source)
 
     assertResult(1L)(metricsDf.count())
 
@@ -97,7 +97,7 @@
   it should "throw runtime error for invalid expr" in {
     assertThrows[AssertionError] {
       ProfilingMeasure(spark, param.copy(config = Map(Expression -> "xyz")))
-        .execute()
+        .execute(source)
     }
   }
 
diff --git a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasureTest.scala b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasureTest.scala
index e705675..6c37f34 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasureTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SchemaConformanceMeasureTest.scala
@@ -18,9 +18,11 @@
 package org.apache.griffin.measure.execution.impl
 
 import org.apache.commons.lang3.StringUtils
+import org.apache.spark.sql.functions.lit
 
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure._
+import org.apache.griffin.measure.execution.impl.CompletenessMeasure._
 
 class SchemaConformanceMeasureTest extends MeasureTest {
   var param: MeasureParam = _
@@ -102,7 +104,7 @@
 
   it should "execute defined measure expr" in {
     val measure = SchemaConformanceMeasure(spark, param)
-    val (recordsDf, metricsDf) = measure.execute(None)
+    val (recordsDf, metricsDf) = measure.execute(source)
 
     assertResult(recordDfSchema)(recordsDf.schema)
     assertResult(metricDfSchema)(metricsDf.schema)
@@ -117,8 +119,65 @@
       .toMap
 
     assertResult(metricMap(Total))("5")
-    assertResult(metricMap(measure.Complete))("5")
-    assertResult(metricMap(measure.InComplete))("0")
+    assertResult(metricMap(Complete))("5")
+    assertResult(metricMap(InComplete))("0")
+  }
+
+  it should "be able to check for date column" in {
+    val newSource = source.withColumn("Test_date", lit("2009-07-30"))
+    newSource.createOrReplaceTempView("newSource")
+    val measure = SchemaConformanceMeasure(
+      spark,
+      param.copy(
+        dataSource = "newSource",
+        config = Map(Expression -> Seq(Map(SourceColStr -> "Test_date", DataTypeStr -> "date")))))
+    val (recordsDf, metricsDf) = measure.execute(newSource)
+
+    val newRecordDfSchema = newSource.schema.add(Status, "string", nullable = false)
+    assertResult(newRecordDfSchema)(recordsDf.schema)
+    assertResult(metricDfSchema)(metricsDf.schema)
+
+    assertResult(newSource.count())(recordsDf.count())
+    assertResult(1L)(metricsDf.count())
+
+    val metricMap = metricsDf
+      .head()
+      .getAs[Seq[Map[String, String]]](Metrics)
+      .map(x => x(MetricName) -> x(MetricValue))
+      .toMap
+
+    assertResult(metricMap(Total))("5")
+    assertResult(metricMap(Complete))("5")
+    assertResult(metricMap(InComplete))("0")
+  }
+
+  it should "be able to check for timestamp column and support space in column name" in {
+    val newSource = source.withColumn("Test date", lit("2009-07-30"))
+    newSource.createOrReplaceTempView("newSource")
+    val measure = SchemaConformanceMeasure(
+      spark,
+      param.copy(
+        dataSource = "newSource",
+        config =
+          Map(Expression -> Seq(Map(SourceColStr -> "Test date", DataTypeStr -> "timestamp")))))
+    val (recordsDf, metricsDf) = measure.execute(newSource)
+
+    val newRecordDfSchema = newSource.schema.add(Status, "string", nullable = false)
+    assertResult(newRecordDfSchema)(recordsDf.schema)
+    assertResult(metricDfSchema)(metricsDf.schema)
+
+    assertResult(newSource.count())(recordsDf.count())
+    assertResult(1L)(metricsDf.count())
+
+    val metricMap = metricsDf
+      .head()
+      .getAs[Seq[Map[String, String]]](Metrics)
+      .map(x => x(MetricName) -> x(MetricValue))
+      .toMap
+
+    assertResult(metricMap(Total))("5")
+    assertResult(metricMap(Complete))("5")
+    assertResult(metricMap(InComplete))("0")
   }
 
 }
diff --git a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SparkSqlMeasureTest.scala b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SparkSqlMeasureTest.scala
index 405e4a6..a2986e0 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SparkSqlMeasureTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/SparkSqlMeasureTest.scala
@@ -22,6 +22,7 @@
 
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure._
+import org.apache.griffin.measure.execution.impl.CompletenessMeasure._
 
 class SparkSqlMeasureTest extends MeasureTest {
   var param: MeasureParam = _
@@ -99,7 +100,7 @@
 
   it should "execute defined measure expr" in {
     val measure = SparkSQLMeasure(spark, param)
-    val (recordsDf, metricsDf) = measure.execute(None)
+    val (recordsDf, metricsDf) = measure.execute(source)
 
     assertResult(metricDfSchema)(metricsDf.schema)
     assertResult(source.count())(recordsDf.count())
@@ -112,8 +113,8 @@
       .toMap
 
     assertResult(metricMap(Total))("5")
-    assertResult(metricMap(measure.Complete))("3")
-    assertResult(metricMap(measure.InComplete))("2")
+    assertResult(metricMap(Complete))("3")
+    assertResult(metricMap(InComplete))("2")
   }
 
   it should "throw runtime error for invalid expr" in {
@@ -122,7 +123,7 @@
         spark,
         param.copy(config =
           Map(Expression -> "select * from source", BadRecordDefinition -> "name")))
-        .execute()
+        .execute(source)
     }
 
     assertThrows[AnalysisException] {
@@ -130,7 +131,7 @@
         spark,
         param.copy(config =
           Map(Expression -> "select 1 as my_value", BadRecordDefinition -> "name is null")))
-        .execute()
+        .execute(source)
     }
   }
 
diff --git a/measure/src/test/scala/org/apache/griffin/measure/job/BatchDQAppTest.scala b/measure/src/test/scala/org/apache/griffin/measure/job/BatchDQAppTest.scala
index b4f11cd..c7d002b 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/job/BatchDQAppTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/job/BatchDQAppTest.scala
@@ -147,18 +147,18 @@
         "max" -> "10013",
         "min_col_len" -> "5"),
       "first_name" -> Map(
-        "avg_col_len" -> null,
+        "avg_col_len" -> "6.0",
         "max_col_len" -> "6",
-        "variance" -> null,
-        "kurtosis" -> null,
-        "avg" -> null,
-        "min" -> null,
+        "variance" -> "0",
+        "kurtosis" -> "0",
+        "avg" -> "0",
+        "min" -> "0",
         "null_count" -> "0",
         "approx_distinct_count" -> "13",
         "total" -> "13",
-        "std_dev" -> null,
+        "std_dev" -> "0",
         "data_type" -> "string",
-        "max" -> null,
+        "max" -> "0",
         "min_col_len" -> "6"))
 
     runAndCheckResult(Map("profiling_measure" -> expectedMetrics))
diff --git a/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala b/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala
index e24bcd5..5e374c9 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala
@@ -17,7 +17,7 @@
 
 package org.apache.griffin.measure.job
 
-import scala.util.{Failure, Success}
+import scala.util._
 
 import org.scalatest.matchers.should._
 
@@ -38,11 +38,9 @@
   var dqApp: DQApp = _
 
   def getConfigFilePath(fileName: String): String = {
-    try {
-      getClass.getResource(fileName).getFile
-    } catch {
-      case _: NullPointerException => throw new Exception(s"resource [$fileName] not found")
-      case ex: Throwable => throw ex
+    Try(getClass.getResource(fileName).getFile).toOption match {
+      case Some(n) => n
+      case None => throw new Exception(s"resource [$fileName] not found")
     }
   }
 
diff --git a/pom.xml b/pom.xml
index e51c94e..d2d4ffe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -47,7 +47,7 @@
         <java.version>1.8</java.version>
         <scala.binary.version>2.11</scala.binary.version>
         <scala211.binary.version>2.11</scala211.binary.version>
-        <scala.version>${scala.binary.version}.0</scala.version>
+        <scala.version>${scala.binary.version}.12</scala.version>
 
         <maven.compiler.source>${java.version}</maven.compiler.source>
         <maven.compiler.target>${java.version}</maven.compiler.target>