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 fc6e10f..b1eaad3 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
@@ -25,7 +25,6 @@
 
 import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
 import org.apache.griffin.measure.execution.Measure
-import org.apache.griffin.measure.step.builder.ConstantColumns
 
 /**
  * Duplication Measure
@@ -40,14 +39,9 @@
  */
 case class DuplicationMeasure(measureParam: MeasureParam) extends Measure {
 
+  import DuplicationMeasure._
   import Measure._
 
-  private final val Duplicate: String = "duplicate"
-  private final val Unique: String = "unique"
-  private final val NonUnique: String = "non_unique"
-  private final val Distinct: String = "distinct"
-  private final val __Temp: String = "__temp"
-
   private final val duplicationMeasures = Seq(Duplicate, Unique, NonUnique, Distinct)
 
   val exprs: String = getFromConfig[String](Expression, null)
@@ -68,18 +62,11 @@
     val nonUniqueCol =
       when(col(Unique) =!= 1 and (col(__Temp) - col(NonUnique) === 0), 1).otherwise(0)
 
-    val keyCols = {
-      if (StringUtil.isNullOrEmpty(exprs)) input.columns
-      else exprs.split(",").map(_.trim)
-    }.distinct
-
-    keyCols.foreach(c =>
-      assert(input.columns.contains(c), s"Provided column '$c' does not exist in the dataset."))
-
-    val window = Window.partitionBy(keyCols.map(col): _*).orderBy(ConstantColumns.tmst)
+    val cols = keyCols(input).map(col)
+    val window = Window.partitionBy(cols: _*).orderBy(cols: _*)
 
     val aggDf = input
-      .select(col("*"), row_number().over(window).as(__Temp))
+      .select(col(AllColumns), row_number().over(window).as(__Temp))
       .withColumn(Duplicate, duplicateCol)
       .withColumn(Unique, count(lit(1)).over(window))
       .withColumn(Unique, uniqueCol)
@@ -90,7 +77,7 @@
       .drop(__Temp)
 
     val metricAggCols = duplicationMeasures.map(m => sum(m).as(m))
-    val selectCols = duplicationMeasures.flatMap(e => Seq(lit(e), col(e)))
+    val selectCols = duplicationMeasures.flatMap(e => Seq(lit(e), col(e).cast("string")))
     val metricColumn = map(selectCols: _*).as(valueColumn)
     val metricDf = aggDf
       .agg(metricAggCols.head, metricAggCols.tail: _*)
@@ -102,6 +89,13 @@
   }
 
   private def validate(): Unit = {
+    val input = SparkSession.getDefaultSession.get.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(
       !StringUtil.isNullOrEmpty(badnessExpr),
       s"Invalid value '$badnessExpr' provided for $BadRecordDefinition")
@@ -111,4 +105,18 @@
       case _ => false
     }, s"Invalid value '$badnessExpr' was provided for $BadRecordDefinition")
   }
+
+  private def keyCols(input: DataFrame): Array[String] = {
+    if (StringUtil.isNullOrEmpty(exprs)) input.columns
+    else exprs.split(",").map(_.trim)
+  }.distinct
+
+}
+
+object DuplicationMeasure {
+  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"
 }
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
new file mode 100644
index 0000000..7fed3d9
--- /dev/null
+++ b/measure/src/test/scala/org/apache/griffin/measure/execution/impl/DuplicationMeasureTest.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.execution.impl
+
+import org.apache.commons.lang3.StringUtils
+
+import org.apache.griffin.measure.configuration.dqdefinition.MeasureParam
+import org.apache.griffin.measure.execution.Measure._
+import org.apache.griffin.measure.execution.impl.DuplicationMeasure._
+
+class DuplicationMeasureTest extends MeasureTest {
+  var param: MeasureParam = _
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    param = MeasureParam(
+      "param",
+      "Duplication",
+      "source",
+      Map(Expression -> "name", BadRecordDefinition -> "duplicate"))
+  }
+
+  "DuplicationMeasure" should "validate expression config" in {
+    // Validations for Duplication Expr
+
+    // Empty
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map.empty[String, String]))
+    }
+
+    // Empty
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map(Expression -> StringUtils.EMPTY)))
+    }
+
+    // Null
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map(Expression -> null)))
+    }
+
+    // Incorrect Type
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map(Expression -> 1234)))
+    }
+
+    // Validations for BadRecordDefinition
+
+    // Empty
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map(Expression -> "name")))
+    }
+
+    // Empty
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map(Expression -> "name", BadRecordDefinition -> "")))
+    }
+
+    // Null
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map(Expression -> "name", BadRecordDefinition -> null)))
+    }
+
+    // Incorrect Type
+    assertThrows[AssertionError] {
+      DuplicationMeasure(param.copy(config = Map(Expression -> "name", BadRecordDefinition -> 435)))
+    }
+
+    // Incorrect Value
+    assertThrows[AssertionError] {
+      DuplicationMeasure(
+        param.copy(config = Map(Expression -> "name", BadRecordDefinition -> "xyz")))
+    }
+  }
+
+  it should "support metric writing" in {
+    val measure = DuplicationMeasure(param)
+    assertResult(true)(measure.supportsMetricWrite)
+  }
+
+  it should "support record writing" in {
+    val measure = DuplicationMeasure(param)
+    assertResult(true)(measure.supportsRecordWrite)
+  }
+
+  it should "execute defined measure expr" in {
+    val measure = DuplicationMeasure(param.copy(config = Map(BadRecordDefinition -> "duplicate")))
+    val (recordsDf, metricsDf) = measure.execute(context, None)
+
+    assertResult(recordsDf.schema)(recordDfSchema)
+    assertResult(metricsDf.schema)(metricDfSchema)
+
+    assertResult(recordsDf.count())(source.count())
+    assertResult(metricsDf.count())(1L)
+
+    val row = metricsDf.head()
+    assertResult(param.getDataSource)(row.getAs[String](DataSource))
+    assertResult(param.getName)(row.getAs[String](MeasureName))
+    assertResult(param.getType.toString)(row.getAs[String](MeasureType))
+
+    val metricMap = row.getAs[Map[String, String]](Metrics)
+    println(metricMap)
+    assertResult(metricMap(Duplicate))("0")
+    assertResult(metricMap(Unique))("5")
+    assertResult(metricMap(NonUnique))("0")
+    assertResult(metricMap(Distinct))("5")
+  }
+
+  it should "supported complex measure expr" in {
+    val measure = DuplicationMeasure(
+      param.copy(config = Map(Expression -> "name", BadRecordDefinition -> "duplicate")))
+    val (recordsDf, metricsDf) = measure.execute(context, None)
+
+    assertResult(recordsDf.schema)(recordDfSchema)
+    assertResult(metricsDf.schema)(metricDfSchema)
+
+    assertResult(recordsDf.count())(source.count())
+    assertResult(metricsDf.count())(1L)
+
+    val row = metricsDf.head()
+    assertResult(param.getDataSource)(row.getAs[String](DataSource))
+    assertResult(param.getName)(row.getAs[String](MeasureName))
+    assertResult(param.getType.toString)(row.getAs[String](MeasureType))
+
+    val metricMap = row.getAs[Map[String, String]](Metrics)
+    assertResult(metricMap(Duplicate))("1")
+    assertResult(metricMap(Unique))("3")
+    assertResult(metricMap(NonUnique))("1")
+    assertResult(metricMap(Distinct))("4")
+  }
+
+}