Add DatasetMatch DQDL Rule Support

Author: Joshua Zexter

src/main/scala/com/amazon/deequ/dqdl/execution/DQDLExecutor.scala

@@ -16,8 +16,8 @@
 
 package com.amazon.deequ.dqdl.execution
 
-import com.amazon.deequ.dqdl.execution.executors.{DeequRulesExecutor, ReferentialIntegrityExecutor, RowCountMatchExecutor, UnsupportedRulesExecutor}
-import com.amazon.deequ.dqdl.model.{DeequExecutableRule, ExecutableRule, Failed, ReferentialIntegrityExecutableRule, RowCountMatchExecutableRule, RuleOutcome, UnsupportedExecutableRule}
+import com.amazon.deequ.dqdl.execution.executors.{DatasetMatchExecutor, DeequRulesExecutor, ReferentialIntegrityExecutor, RowCountMatchExecutor, UnsupportedRulesExecutor}
+import com.amazon.deequ.dqdl.model.{DatasetMatchExecutableRule, DeequExecutableRule, ExecutableRule, Failed, ReferentialIntegrityExecutableRule, RowCountMatchExecutableRule, RuleOutcome, UnsupportedExecutableRule}
 import org.apache.spark.sql.DataFrame
 import software.amazon.glue.dqdl.model.DQRule
 
@@ -37,7 +37,8 @@ object DQDLExecutor {
     classOf[DeequExecutableRule] -> DeequRulesExecutor,
     classOf[UnsupportedExecutableRule] -> UnsupportedRulesExecutor,
     classOf[RowCountMatchExecutableRule] -> RowCountMatchExecutor,
-    classOf[ReferentialIntegrityExecutableRule] -> ReferentialIntegrityExecutor
+    classOf[ReferentialIntegrityExecutableRule] -> ReferentialIntegrityExecutor,
+    classOf[DatasetMatchExecutableRule] -> DatasetMatchExecutor
   )
 
   def executeRules(rules: Seq[ExecutableRule], df: DataFrame,

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

@@ -0,0 +1,63 @@
+/**
+ * Copyright 2026 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ * use this file except in compliance with the License. A copy of the License
+ * is located at
+ *
+ * http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file 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 com.amazon.deequ.dqdl.execution.executors
+
+import com.amazon.deequ.comparison.{DataSynchronization, DatasetMatchFailed}
+import com.amazon.deequ.dqdl.execution.DQDLExecutor
+import com.amazon.deequ.dqdl.model.{DatasetMatchExecutableRule, Failed, Passed, RuleOutcome}
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.functions.{col, lit}
+import software.amazon.glue.dqdl.model.DQRule
+
+import scala.util.{Failure, Success, Try}
+
+object DatasetMatchExecutor extends DQDLExecutor.RuleExecutor[DatasetMatchExecutableRule] {
+
+  override def executeRules(rules: Seq[DatasetMatchExecutableRule], df: DataFrame,
+                            additionalDataSources: Map[String, DataFrame]): Map[DQRule, RuleOutcome] = {
+    rules.map { rule =>
+      rule.dqRule -> (additionalDataSources.get(rule.referenceDatasetAlias) match {
+        case Some(referenceDF) =>
+          val matchCols = rule.matchColumnMappings.orElse {
+            Some(df.columns.filterNot(rule.keyColumnMappings.contains).map(c => c -> c).toMap)
+          }
+          val outcomeCol = java.util.UUID.randomUUID().toString
+
+          Try(DataSynchronization.columnMatchRowLevel(
+            df, referenceDF, rule.keyColumnMappings, matchCols, Some(outcomeCol))) match {
+            case Success(Left(DatasetMatchFailed(errorMessage, _, _))) =>
+              RuleOutcome(rule.dqRule, Failed, Some(errorMessage))
+            case Success(Right(augmentedDF)) =>
+              val filteredCount = augmentedDF.filter(col(outcomeCol) === lit(true)).count()
+              val totalCount = df.count()
+              val ratio = filteredCount.toDouble / totalCount
+              val metrics = Map(s"Dataset.${rule.referenceDatasetAlias}.DatasetMatch" -> ratio)
+
+              if (rule.assertion(ratio)) RuleOutcome(rule.dqRule, Passed, None, metrics)
+              else RuleOutcome(rule.dqRule, Failed,
+                Some(s"Value: $ratio does not meet the constraint requirement."), metrics)
+            case Failure(ex) =>
+              RuleOutcome(rule.dqRule, Failed,
+                Some(s"Exception thrown while evaluating rule: ${ex.getClass.getName}"))
+          }
+        case None =>
+          RuleOutcome(rule.dqRule, Failed,
+            Some(s"${rule.referenceDatasetAlias} not found in additional sources"))
+      })
+    }.toMap
+  }
+}

src/main/scala/com/amazon/deequ/dqdl/model/ExecutableRule.scala

@@ -57,6 +57,15 @@ case class ReferentialIntegrityExecutableRule(dqRule: DQRule,
     Some(s"Column.$referenceDatasetAlias.ReferentialIntegrity")
 }
 
+case class DatasetMatchExecutableRule(dqRule: DQRule,
+                                      referenceDatasetAlias: String,
+                                      keyColumnMappings: Map[String, String],
+                                      matchColumnMappings: Option[Map[String, String]],
+                                      assertion: Double => Boolean) extends ExecutableRule {
+  override val evaluatedMetricName: Option[String] =
+    Some(s"Dataset.$referenceDatasetAlias.DatasetMatch")
+}
+
 case class DeequMetricMapping(entity: String,
                               instance: String,
                               name: String,

src/main/scala/com/amazon/deequ/dqdl/translation/DQDLRuleTranslator.scala

@@ -35,6 +35,7 @@ import com.amazon.deequ.dqdl.translation.rules.ColumnLengthRule
 import com.amazon.deequ.dqdl.translation.rules.ColumnExistsRule
 import com.amazon.deequ.dqdl.translation.rules.RowCountMatchRule
 import com.amazon.deequ.dqdl.translation.rules.ReferentialIntegrityRule
+import com.amazon.deequ.dqdl.translation.rules.DatasetMatchRule
 import software.amazon.glue.dqdl.model.DQRule
 import software.amazon.glue.dqdl.model.DQRuleset
 
@@ -87,6 +88,11 @@ object DQDLRuleTranslator {
           case Right(executableRule) => executableRule
           case Left(message) => UnsupportedExecutableRule(rule, Some(message))
         }
+      case "DatasetMatch" =>
+        DatasetMatchRule.toExecutableRule(rule) match {
+          case Right(executableRule) => executableRule
+          case Left(message) => UnsupportedExecutableRule(rule, Some(message))
+        }
       case _ =>
         translateRule(rule) match {
           case Right(deequExecutableRule) => deequExecutableRule

src/main/scala/com/amazon/deequ/dqdl/translation/rules/DatasetMatchRule.scala

@@ -0,0 +1,52 @@
+/**
+ * Copyright 2026 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ * use this file except in compliance with the License. A copy of the License
+ * is located at
+ *
+ * http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file 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 com.amazon.deequ.dqdl.translation.rules
+
+import com.amazon.deequ.dqdl.execution.DefaultOperandEvaluator
+import com.amazon.deequ.dqdl.model.DatasetMatchExecutableRule
+import software.amazon.glue.dqdl.model.DQRule
+import software.amazon.glue.dqdl.model.condition.number.NumberBasedCondition
+
+import scala.collection.JavaConverters._
+
+object DatasetMatchRule {
+
+  def toExecutableRule(rule: DQRule): Either[String, DatasetMatchExecutableRule] = {
+    val params = rule.getParameters.asScala.mapValues(_.replaceAll("\"", ""))
+
+    val referenceAlias = params("ReferenceDatasetAlias")
+    val keyMappings = parseColumnMappings(params("KeyColumnMappings"))
+
+    keyMappings match {
+      case None => Left("Could not parse KeyColumnMappings")
+      case Some(keyMap) =>
+        val matchMappings = params.get("MatchColumnMappings").flatMap(parseColumnMappings)
+        val condition = rule.getCondition.asInstanceOf[NumberBasedCondition]
+        val assertion: Double => Boolean = d => condition.evaluate(d, rule, DefaultOperandEvaluator)
+        Right(DatasetMatchExecutableRule(rule, referenceAlias, keyMap, matchMappings, assertion))
+    }
+  }
+
+  private def parseColumnMappings(mappings: String): Option[Map[String, String]] = {
+    if (mappings.isEmpty) return None
+    Some(mappings.split(",").map { mapping =>
+      val parts = mapping.trim.split("->")
+      if (parts.length == 2) parts(0).trim -> parts(1).trim
+      else parts(0).trim -> parts(0).trim
+    }.toMap)
+  }
+}