address review feedback for DuplicateRowCount (#729)

Co-authored-by: anniezc <anniezc@amazon.com>

Author: awang923

src/main/scala/com/amazon/deequ/analyzers/DuplicateRowCount.scala

@@ -44,7 +44,7 @@ case class DuplicateRowCount(columns: Seq[String], where: Option[String] = None,
 
   override def fromAggregationResult(result: Row, offset: Int, fullColumn: Option[Column]): DoubleMetric = {
     if (result.isNullAt(offset)) {
-      // WHERE clause matched zero rows — return 0 (no duplicates) instead of empty metric
+      // WHERE clause matched zero rows — return 0 (no duplicates)
       toSuccessMetric(0.0, fullColumn)
     } else {
       val conditionColumn = where.map { expression => expr(expression) }
@@ -78,7 +78,7 @@ case class DuplicateRowCount(columns: Seq[String], where: Option[String] = None,
     }
   }
 
-  /** For empty columns, resolve all DataFrame columns at calculation time */
+  /** For empty columns, resolve all DataFrame columns and re-wrap metric entity */
   override def calculate(
       data: DataFrame,
       aggregateWith: Option[StateLoader] = None,
@@ -104,6 +104,9 @@ case class DuplicateRowCount(columns: Seq[String], where: Option[String] = None,
   }
 
   override def filterCondition: Option[String] = where
+
+  override def columnsReferenced(): Option[Set[String]] =
+    if (columns.isEmpty || where.isDefined) None else Some(columns.toSet)
 }
 
 object DuplicateRowCount {

src/main/scala/com/amazon/deequ/analyzers/GroupingAnalyzers.scala

@@ -67,6 +67,12 @@ object FrequencyBasedAnalyzer {
       where: Option[String] = None)
     : FrequenciesAndNumRows = {
 
+    // Resolve empty groupingColumns to all DataFrame columns.
+    // Only DuplicateRowCount can reach here with empty columns (it overrides preconditions
+    // to skip atLeastOne check). All other analyzers enforce atLeastOne(columnsToGroupOn)
+    // which rejects empty columns before this method is called.
+    require(groupingColumns.nonEmpty || data.columns.nonEmpty,
+      "groupingColumns is empty and DataFrame has no columns")
     val resolvedColumns = if (groupingColumns.isEmpty) data.columns.toSeq else groupingColumns
     val columnsToGroupBy = resolvedColumns.map { name => col(name) }.toArray
     val projectionColumns = columnsToGroupBy :+ col(COUNT_COL)

src/main/scala/com/amazon/deequ/constraints/Constraint.scala

@@ -174,7 +174,13 @@ object Constraint {
     val constraint = AnalysisBasedConstraint[FrequenciesAndNumRows, Double, Long](
       duplicateRowCount, assertion, Some(_.toLong), hint)
 
-    new NamedConstraint(constraint, s"DuplicateRowCountConstraint($duplicateRowCount)")
+    if (columns.nonEmpty) {
+      new RowLevelGroupedConstraint(constraint,
+        s"DuplicateRowCountConstraint($duplicateRowCount)",
+        duplicateRowCount.columns)
+    } else {
+      new NamedConstraint(constraint, s"DuplicateRowCountConstraint($duplicateRowCount)")
+    }
   }
 
   /**

src/main/scala/com/amazon/deequ/dqdl/execution/executors/DeequRulesExecutor.scala

@@ -20,8 +20,10 @@ import com.amazon.deequ.{VerificationResult, VerificationSuite}
 import com.amazon.deequ.constraints.{RowLevelAssertedConstraint, RowLevelConstraint, RowLevelGroupedConstraint}
 import com.amazon.deequ.dqdl.execution.DQDLExecutor
 import com.amazon.deequ.dqdl.model.{DeequExecutableRule, DeequMetricMapping, Failed, NoColumn, RuleOutcome, SingularColumn}
+import com.amazon.deequ.dqdl.translation.rules.DuplicateRowCountRule
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.{col, concat, lit}
+import scala.collection.JavaConverters._
 import software.amazon.glue.dqdl.model.DQRule
 
 case class DeequExecutionResult(outcomes: Map[DQRule, RuleOutcome], rowLevelData: DataFrame)
@@ -40,9 +42,13 @@ object DeequRulesExecutor extends DQDLExecutor.RuleExecutor[DeequExecutableRule]
       return DeequExecutionResult(Map.empty, df)
     }
 
+    // Resolve empty DuplicateRowCount columns to all DataFrame columns
+    // so RowLevelGroupedConstraint gets a populated column list for row-level results
+    val resolvedRules = rules.map(rule => resolveDuplicateRowCountColumns(rule, df))
+
     val verificationResult = VerificationSuite()
       .onData(df.toDF())
-      .addChecks(rules.map(_.check))
+      .addChecks(resolvedRules.map(_.check))
       .run()
 
     val rowLevelData = VerificationResult.rowLevelResultsAsDataFrame(
@@ -68,7 +74,7 @@ object DeequRulesExecutor extends DQDLExecutor.RuleExecutor[DeequExecutableRule]
         if (hasRowLevel) Some(check.description) else None
     }.toSet
 
-    val outcomes = rules.map { r =>
+    val outcomes = resolvedRules.map { r =>
       val rowLevelOutcome = if (rowLevelCheckDescriptions.contains(r.check.description)) {
         SingularColumn(r.check.description)
       } else {
@@ -102,4 +108,41 @@ object DeequRulesExecutor extends DQDLExecutor.RuleExecutor[DeequExecutableRule]
       }
     }.toMap
   }
+
+  /**
+   * Resolves empty columns in DuplicateRowCount rules to all DataFrame columns.
+   * This ensures RowLevelGroupedConstraint gets a populated column list for row-level results.
+   */
+  private def resolveDuplicateRowCountColumns(rule: DeequExecutableRule, df: DataFrame): DeequExecutableRule = {
+    import com.amazon.deequ.checks.{Check, CheckLevel}
+    import com.amazon.deequ.dqdl.util.DQDLUtility.addWhereClause
+    import software.amazon.glue.dqdl.model.condition.number.NumberBasedCondition
+
+    // Only resolve for DuplicateRowCount with no explicit columns
+    val isDuplicateRowCountNoColumns = rule.dqRule.getRuleType == "DuplicateRowCount" &&
+      !rule.dqRule.getParameters.asScala.exists(_._1.startsWith("TargetColumn"))
+
+    if (!isDuplicateRowCountNoColumns) {
+      rule
+    } else {
+      // Re-derive the assertion from the DQRule condition
+      val condition = rule.dqRule.getCondition.asInstanceOf[NumberBasedCondition]
+      val converter = new DuplicateRowCountRule()
+      val doubleAssertion = converter.assertionAsScala(rule.dqRule, condition)
+      val longAssertion: Long => Boolean = (v: Long) => doubleAssertion(v.toDouble)
+
+      // Rebuild check with all DataFrame columns
+      val allColumns = df.columns.toSeq
+      val check = Check(CheckLevel.Error, rule.check.description)
+        .hasDuplicateRowCount(allColumns, longAssertion)
+
+      val resolvedCheck = if (rule.dqRule.getWhereClause != null && !rule.dqRule.getWhereClause.isEmpty) {
+        addWhereClause(rule.dqRule, check)
+      } else {
+        check
+      }
+
+      DeequExecutableRule(rule.dqRule, resolvedCheck, rule.deequMetricMappings)
+    }
+  }
 }

src/main/scala/com/amazon/deequ/repository/AnalysisResultSerde.scala

@@ -356,6 +356,7 @@ private[deequ] object AnalyzerSerializer
         result.add(COLUMNS_FIELD, context.serialize(duplicateRowCount.columns.asJava,
           new TypeToken[JList[String]]() {}.getType))
         result.addProperty(WHERE_FIELD, duplicateRowCount.where.orNull)
+        result.add(ANALYZER_OPTIONS_FIELD, context.serialize(duplicateRowCount.analyzerOptions.orNull))
 
       case histogram: Histogram if histogram.binningUdf.isEmpty =>
         result.addProperty(ANALYZER_NAME_FIELD, "Histogram")
@@ -598,7 +599,8 @@ private[deequ] object AnalyzerDeserializer
       case "DuplicateRowCount" =>
         DuplicateRowCount(
           getColumnsAsSeq(context, json),
-          getOptionalWhereParam(json))
+          getOptionalWhereParam(json),
+          analyzerOptions = getOptionalAnalyzerOptions(json))
 
       case "Histogram" =>
         Histogram(

src/test/scala/com/amazon/deequ/analyzers/AnalyzerTests.scala

@@ -63,6 +63,21 @@ class AnalyzerTests extends AnyWordSpec with Matchers with SparkContextSpec with
     }
   }
 
+  "DuplicateRowCount analyzer" should {
+    "compute correct metrics" in withSparkSession { sparkSession =>
+      import sparkSession.implicits._
+      val df = Seq(("a", 1), ("b", 2), ("a", 1), ("c", 3)).toDF("col1", "col2")
+      val result = DuplicateRowCount(Seq("col1", "col2")).calculate(df).value
+      result shouldBe Success(2.0)
+    }
+    "return 0 when no duplicates" in withSparkSession { sparkSession =>
+      import sparkSession.implicits._
+      val df = Seq(("a", 1), ("b", 2), ("c", 3)).toDF("col1", "col2")
+      val result = DuplicateRowCount(Seq("col1", "col2")).calculate(df).value
+      result shouldBe Success(0.0)
+    }
+  }
+
   "Completeness analyzer" should {
 
     "compute correct metrics" in withSparkSession { sparkSession =>

src/test/scala/com/amazon/deequ/analyzers/DuplicateRowCountTest.scala

@@ -150,5 +150,117 @@ class DuplicateRowCountTest extends AnyWordSpec with Matchers with SparkContextS
       val result = DuplicateRowCount(Seq("col1", "col2")).calculate(df)
       result.value shouldBe scala.util.Success(0.0)
     }
+
+    "correctly merge state across partitions" in withSparkSession { session =>
+      import session.implicits._
+      // Partition A: ("a",1) is unique
+      val dfA = Seq(("a", 1), ("b", 2)).toDF("col1", "col2")
+      // Partition B: ("a",1) appears again - now duplicate across A+B
+      val dfB = Seq(("a", 1), ("c", 3)).toDF("col1", "col2")
+
+      val analyzer = DuplicateRowCount(Seq("col1", "col2"))
+      val stateA = analyzer.computeStateFrom(dfA).get
+      val stateB = analyzer.computeStateFrom(dfB).get
+      val merged = stateA.sum(stateB)
+
+      val metric = analyzer.computeMetricFrom(Some(merged))
+      // ("a",1) has count 2 after merge -> 2 duplicate rows
+      metric.value shouldBe Success(2.0)
+    }
+
+    "correctly merge state with overlapping groups" in withSparkSession { session =>
+      import session.implicits._
+      // Partition A: ("a",1) appears twice
+      val dfA = Seq(("a", 1), ("a", 1), ("b", 2)).toDF("col1", "col2")
+      // Partition B: ("a",1) appears once more
+      val dfB = Seq(("a", 1), ("c", 3)).toDF("col1", "col2")
+
+      val analyzer = DuplicateRowCount(Seq("col1", "col2"))
+      val stateA = analyzer.computeStateFrom(dfA).get
+      val stateB = analyzer.computeStateFrom(dfB).get
+      val merged = stateA.sum(stateB)
+
+      val metric = analyzer.computeMetricFrom(Some(merged))
+      // ("a",1) has count 3 after merge -> 3 duplicate rows
+      metric.value shouldBe Success(3.0)
+    }
+
+    "produce correct row-level results" in withSparkSession { session =>
+      import session.implicits._
+      import com.amazon.deequ.{VerificationSuite, VerificationResult}
+      import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
+
+      val df = Seq(("a", 1), ("b", 2), ("a", 1), ("c", 3)).toDF("col1", "col2")
+
+      val result = VerificationSuite()
+        .onData(df)
+        .addCheck(Check(CheckLevel.Error, "dup-check")
+          .hasDuplicateRowCount(Seq("col1", "col2"), _ == 2))
+        .run()
+
+      // Verify the check passes
+      result.status shouldBe CheckStatus.Success
+
+      // Verify row-level results: true = duplicate, false = unique
+      val rowLevelDf = VerificationResult.rowLevelResultsAsDataFrame(session, result, df)
+      val flags = rowLevelDf.select("`dup-check`").collect().map(_.getBoolean(0))
+      // 2 rows are duplicates (true), 2 rows are unique (false)
+      flags.count(_ == true) shouldBe 2
+      flags.count(_ == false) shouldBe 2
+    }
+
+    "work with empty columns through VerificationSuite" in withSparkSession { session =>
+      import session.implicits._
+      import com.amazon.deequ.{VerificationSuite, VerificationResult}
+      import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
+
+      val df = Seq(("a", 1), ("b", 2), ("a", 1), ("c", 3)).toDF("col1", "col2")
+
+      val result = VerificationSuite()
+        .onData(df)
+        .addCheck(Check(CheckLevel.Error, "dup-empty-cols")
+          .hasDuplicateRowCount(Seq.empty, _ == 2))
+        .run()
+
+      // Empty columns resolves to all columns at runtime
+      result.status shouldBe CheckStatus.Success
+    }
+
+    "not crash with empty columns through constraint path" in withSparkSession { session =>
+      import session.implicits._
+      import com.amazon.deequ.constraints.Constraint
+
+      val df = Seq(("a", 1), ("b", 2), ("a", 1)).toDF("col1", "col2")
+      // Should not throw NoSuchElementException (NamedConstraint fallback for empty columns)
+      val constraint = Constraint.duplicateRowCountConstraint(Seq.empty, _ == 2)
+      constraint should not be null
+    }
+
+    "produce row-level results for empty columns through DeequRulesExecutor" in withSparkSession { session =>
+      import session.implicits._
+      import com.amazon.deequ.{VerificationSuite, VerificationResult}
+      import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
+
+      val df = Seq(("a", 1), ("b", 2), ("a", 1), ("c", 3)).toDF("col1", "col2")
+
+      // Simulate what DeequRulesExecutor does: resolve empty columns then run
+      val allColumns = df.columns.toSeq
+      val result = VerificationSuite()
+        .onData(df)
+        .addCheck(Check(CheckLevel.Error, "dup-resolved")
+          .hasDuplicateRowCount(allColumns, _ == 2))
+        .run()
+
+      result.status shouldBe CheckStatus.Success
+
+      // With resolved columns, RowLevelGroupedConstraint is used -> row-level results exist
+      val rowLevelDf = VerificationResult.rowLevelResultsAsDataFrame(session, result, df)
+      rowLevelDf.columns should contain ("dup-resolved")
+
+      // Verify flags: 2 duplicates (true), 2 unique (false)
+      val flags = rowLevelDf.select("`dup-resolved`").collect().map(_.getBoolean(0))
+      flags.count(_ == true) shouldBe 2
+      flags.count(_ == false) shouldBe 2
+    }
   }
 }

src/test/scala/com/amazon/deequ/analyzers/NullHandlingTests.scala

@@ -161,5 +161,29 @@ class NullHandlingTests extends AnyWordSpec
     assert(metric.value.failed.get.isInstanceOf[EmptyStateException])
   }
 
+  "DuplicateRowCount" should {
+    "exclude all-null rows and return 0 duplicates" in withSparkSession { session =>
+      import session.implicits._
+      val df = Seq(
+        (None: Option[String], None: Option[String]),
+        (None: Option[String], None: Option[String]),
+        (Some("a"), Some("b"))
+      ).toDF("col1", "col2")
+      val result = DuplicateRowCount(Seq("col1", "col2")).calculate(df)
+      assert(result.value == Success(0.0))
+    }
+
+    "treat partial nulls as equal for grouping" in withSparkSession { session =>
+      import session.implicits._
+      val df = Seq(
+        (Some("a"), None: Option[String]),
+        (Some("a"), None: Option[String]),
+        (Some("b"), Some("c"))
+      ).toDF("col1", "col2")
+      val result = DuplicateRowCount(Seq("col1", "col2")).calculate(df)
+      assert(result.value == Success(2.0))
+    }
+  }
+
 
 }

src/test/scala/com/amazon/deequ/analyzers/StateProviderTest.scala

@@ -74,6 +74,8 @@ class StateProviderTest extends AnyWordSpec
       assertCorrectlyRestoresFrequencyBasedState(provider, provider,
         Uniqueness(Seq("att1", "count")), data)
       assertCorrectlyRestoresFrequencyBasedState(provider, provider, Entropy("att1"), data)
+      assertCorrectlyRestoresFrequencyBasedState(provider, provider,
+        DuplicateRowCount(Seq("att1", "count")), data)
 
       assertCorrectlyApproxQuantileState(provider, provider, ApproxQuantile("price", 0.5), data)
     }

src/test/scala/com/amazon/deequ/checks/CheckTest.scala

@@ -282,6 +282,19 @@ class CheckTest extends AnyWordSpec with Matchers with SparkContextSpec with Fix
       assert(constraintStatuses(9) == ConstraintStatus.Success)
     }
 
+    "return the correct check status for hasDuplicateRowCount" in withSparkSession { sparkSession =>
+      import sparkSession.implicits._
+      val df = Seq(("a", 1), ("b", 2), ("a", 1), ("c", 3)).toDF("col1", "col2")
+
+      val check = Check(CheckLevel.Error, "duplicate-row-count-check")
+        .hasDuplicateRowCount(Seq("col1", "col2"), _ == 2)
+
+      val context = runChecks(df, check)
+      val result = check.evaluate(context)
+
+      assert(result.status == CheckStatus.Success)
+    }
+
     "return the correct check status for hasUniqueValueRatio" in withSparkSession { sparkSession =>
 
       val check = Check(CheckLevel.Error, "unique-value-ratio-check")