Merge branch 'master' into add-columndatatype-rule

Author: sudsali

README.md

@@ -158,6 +158,9 @@ Deequ also supports [DQDL](https://docs.aws.amazon.com/glue/latest/dg/dqdl.html)
 - **ColumnExists**: `ColumnExists "column"`
 - **RowCountMatch**: `RowCountMatch "referenceDataset" >= 0.9`
 - **DataFreshness**: `DataFreshness "Order_Date" <= 24 hours`
+- **Composite Rules**: Combine multiple rules with `and` / `or` operators
+  - Simple: `(RowCount > 0) and (IsComplete "column")`
+  - Nested: `(Rule1) or ((Rule2) and (Rule3))`
 
 ### Scala Example
 
@@ -223,6 +226,57 @@ Dataset<Row> results = EvaluateDataQuality.process(df, ruleset);
 results.show();
 ```
 
+### Composite Rules Example
+
+Composite rules allow you to combine multiple data quality checks using logical operators (`and`, `or`). This enables complex validation scenarios:
+
+```scala
+import com.amazon.deequ.dqdl.EvaluateDataQuality
+import org.apache.spark.sql.SparkSession
+
+val spark = SparkSession.builder()
+  .appName("Composite Rules Example")
+  .master("local[*]")
+  .getOrCreate()
+
+import spark.implicits._
+
+val df = Seq(
+  (1, "Alice", 25, "alice@example.com"),
+  (2, "Bob", 30, "bob@example.com"),
+  (3, "Charlie", 35, "charlie@example.com")
+).toDF("id", "name", "age", "email")
+
+// Simple AND: Both conditions must be true
+val andRule = """Rules=[(RowCount > 0) and (IsComplete "email")]"""
+val andResults = EvaluateDataQuality.process(df, andRule)
+andResults.show()
+
+// Simple OR: At least one condition must be true
+val orRule = """Rules=[(RowCount > 100) or (IsUnique "id")]"""
+val orResults = EvaluateDataQuality.process(df, orRule)
+orResults.show()
+
+// Nested composition: Complex logic with multiple levels
+val nestedRule = """Rules=[
+  ((IsComplete "name") and (IsComplete "email")) or 
+  ((RowCount > 0) and (IsUnique "id"))
+]"""
+val nestedResults = EvaluateDataQuality.process(df, nestedRule)
+nestedResults.show()
+
+// Multiple composite rules in one ruleset
+val multipleRules = """Rules=[
+  (RowCount > 0) and (IsComplete "id"),
+  (IsUnique "id") or (IsUnique "email"),
+  ((Mean "age" > 20) and (Mean "age" < 50)) or (RowCount < 10)
+]"""
+val multipleResults = EvaluateDataQuality.process(df, multipleRules)
+multipleResults.show()
+```
+
+**Note:** Composite rules currently support dataset-level evaluation only. Row-level evaluation (identifying which specific rows pass/fail) is not yet implemented.
+
 ## Citation
 
 If you would like to reference this package in a research paper, please cite:

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

@@ -1,5 +1,5 @@
 /**
- * Copyright 2025 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * 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
@@ -16,8 +16,8 @@
 
 package com.amazon.deequ.dqdl.execution
 
-import com.amazon.deequ.dqdl.execution.executors.{ColumnDataTypeExecutor, DataFreshnessExecutor, DeequRulesExecutor, ReferentialIntegrityExecutor, RowCountMatchExecutor, UnsupportedRulesExecutor}
-import com.amazon.deequ.dqdl.model.{ColumnDataTypeExecutableRule, DataFreshnessExecutableRule, DeequExecutableRule, ExecutableRule, Failed, ReferentialIntegrityExecutableRule, RowCountMatchExecutableRule, RuleOutcome, UnsupportedExecutableRule}
+import com.amazon.deequ.dqdl.execution.executors.{AggregateMatchExecutor, ColumnNamesMatchPatternExecutor, CompositeRulesExecutor, ColumnDataTypeExecutor, DataFreshnessExecutor, DatasetMatchExecutor, DeequRulesExecutor, ReferentialIntegrityExecutor, RowCountMatchExecutor, UnsupportedRulesExecutor}
+import com.amazon.deequ.dqdl.model.{AggregateMatchExecutableRule, ColumnNamesMatchPatternExecutableRule, CompositeExecutableRule, ColumnDataTypeExecutableRule, DataFreshnessExecutableRule, DatasetMatchExecutableRule, DeequExecutableRule, ExecutableRule, Failed, ReferentialIntegrityExecutableRule, RowCountMatchExecutableRule, RuleOutcome, UnsupportedExecutableRule}
 import org.apache.spark.sql.DataFrame
 import software.amazon.glue.dqdl.model.DQRule
 
@@ -35,11 +35,15 @@ object DQDLExecutor {
   // Map from rule class to its executor
   private val executors = Map[Class[_ <: ExecutableRule], RuleExecutor[_ <: ExecutableRule]](
     classOf[DeequExecutableRule] -> DeequRulesExecutor,
+    classOf[CompositeExecutableRule] -> CompositeRulesExecutor,
     classOf[UnsupportedExecutableRule] -> UnsupportedRulesExecutor,
     classOf[RowCountMatchExecutableRule] -> RowCountMatchExecutor,
     classOf[ReferentialIntegrityExecutableRule] -> ReferentialIntegrityExecutor,
     classOf[DataFreshnessExecutableRule] -> DataFreshnessExecutor,
     classOf[ColumnDataTypeExecutableRule] -> ColumnDataTypeExecutor
+    classOf[ColumnNamesMatchPatternExecutableRule] -> ColumnNamesMatchPatternExecutor,
+    classOf[DatasetMatchExecutableRule] -> DatasetMatchExecutor,
+    classOf[AggregateMatchExecutableRule] -> AggregateMatchExecutor
   )
 
   def executeRules(rules: Seq[ExecutableRule], df: DataFrame,

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

@@ -0,0 +1,82 @@
+/**
+ * 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.dqdl.execution.DQDLExecutor
+import com.amazon.deequ.dqdl.model.{AggregateMatchExecutableRule, AggregateOperation, Avg, Failed, Passed, RuleOutcome, Sum}
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.functions.{avg, sum}
+import org.apache.spark.sql.types.DoubleType
+import software.amazon.glue.dqdl.model.DQRule
+
+import scala.util.{Failure, Success, Try}
+
+object AggregateMatchExecutor extends DQDLExecutor.RuleExecutor[AggregateMatchExecutableRule] {
+
+  private val PrimaryAlias = "primary"
+
+  override def executeRules(rules: Seq[AggregateMatchExecutableRule], df: DataFrame,
+                            additionalDataSources: Map[String, DataFrame] = Map.empty): Map[DQRule, RuleOutcome] = {
+    val dataSources = additionalDataSources + (PrimaryAlias -> df)
+    rules.map { rule =>
+      rule.dqRule -> evaluateRule(rule, dataSources)
+    }.toMap
+  }
+
+  private def evaluateRule(rule: AggregateMatchExecutableRule,
+                           dataSources: Map[String, DataFrame]): RuleOutcome = {
+    val result = for {
+      first <- evaluateAggregate(rule.firstAggregateOperation, dataSources)
+      second <- evaluateAggregate(rule.secondAggregateOperation, dataSources)
+    } yield computeRatio(first, second)
+
+    result match {
+      case Right(ratio) =>
+        val metricName = rule.evaluatedMetricName.get
+        val metrics = Map(metricName -> 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 Left(errorMsg) =>
+        RuleOutcome(rule.dqRule, Failed, Some(errorMsg))
+    }
+  }
+
+  private def evaluateAggregate(op: AggregateOperation,
+                                dataSources: Map[String, DataFrame]): Either[String, Double] = {
+    dataSources.get(op.dataSourceAlias) match {
+      case Some(ds) =>
+        val colOp = op match {
+          case Avg(_, _) => avg(op.column).cast(DoubleType)
+          case Sum(_, _) => sum(op.column).cast(DoubleType)
+        }
+        Try(ds.select(colOp).collect().head.getAs[Double](0)) match {
+          case Success(v) => Right(v)
+          case Failure(ex) => Left(s"Exception: ${ex.getClass.getName}")
+        }
+      case None =>
+        Left(s"${op.dataSourceAlias} not found in additional sources")
+    }
+  }
+
+  private def computeRatio(first: Double, second: Double): Double = {
+    if (first == 0 && second == 0) 1.0 else if (second == 0) 0.0 else first / second
+  }
+}

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

@@ -0,0 +1,53 @@
+/**
+ * Copyright 2025 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.dqdl.execution.DQDLExecutor
+import com.amazon.deequ.dqdl.model.{ColumnNamesMatchPatternExecutableRule, Failed, Passed, RuleOutcome}
+import org.apache.spark.sql.DataFrame
+import software.amazon.glue.dqdl.model.DQRule
+
+import scala.util.matching.Regex
+import scala.util.Try
+
+object ColumnNamesMatchPatternExecutor extends DQDLExecutor.RuleExecutor[ColumnNamesMatchPatternExecutableRule] {
+
+  override def executeRules(rules: Seq[ColumnNamesMatchPatternExecutableRule], df: DataFrame,
+                            additionalDataSources: Map[String, DataFrame] = Map.empty): Map[DQRule, RuleOutcome] = {
+    rules.map { rule =>
+      val pattern = Try(new Regex(rule.pattern)).getOrElse {
+        throw new IllegalArgumentException(
+          s"Invalid regex pattern '${rule.pattern}' for ColumnNamesMatchPattern rule")
+      }
+      val columns = df.columns
+      val columnCount = columns.length
+      val unmatchedColumns = columns.filter { col => pattern.findAllMatchIn(col).isEmpty }
+      val metricValue = (columnCount - unmatchedColumns.length).toDouble / columnCount.toDouble
+      val evaluatedMetric = Map("Dataset.*.ColumnNamesPatternMatchRatio" -> metricValue)
+
+      val outcome = if (unmatchedColumns.isEmpty) {
+        RuleOutcome(rule.dqRule, Passed, None, evaluatedMetric)
+      } else {
+        RuleOutcome(rule.dqRule, Failed,
+          Some(s"Columns that do not match the pattern: ${unmatchedColumns.mkString(", ")}"),
+          evaluatedMetric)
+      }
+
+      rule.dqRule -> outcome
+    }.toMap
+  }
+}

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

@@ -0,0 +1,96 @@
+/**
+ * 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.dqdl.execution.DQDLExecutor
+import com.amazon.deequ.dqdl.model.{CompositeExecutableRule, ExecutableRule, RuleOutcome}
+import com.amazon.deequ.dqdl.translation.RuleOutcomeTranslator
+import org.apache.spark.sql.DataFrame
+import software.amazon.glue.dqdl.model.DQRule
+
+/**
+ * Executor for composite rules that combine multiple rules with AND/OR operators.
+ *
+ * This executor handles the evaluation of composite rules by:
+ * 1. Flattening the rule tree to extract all leaf (non-composite) rules
+ * 2. Executing all leaf rules once (avoiding duplicate execution)
+ * 3. Composing the outcomes using the specified logical operators
+ *
+ * Example composite rules:
+ * - `(RowCount > 0) and (IsComplete "column")`
+ * - `(ColumnValues "col" > 10) or (IsUnique "col")`
+ * - `((Rule1) and (Rule2)) or (Rule3)` (nested composition)
+ *
+ * Note: Currently only supports dataset-level evaluation.
+ */
+object CompositeRulesExecutor extends DQDLExecutor.RuleExecutor[CompositeExecutableRule] {
+
+  /**
+   * Executes composite rules by evaluating all nested rules and composing their outcomes.
+   *
+   * The execution strategy:
+   * 1. Flattens all composite rules to extract unique leaf rules
+   * 2. Executes each leaf rule once (shared across multiple composites if needed)
+   * 3. Uses RuleOutcomeTranslator to compose outcomes with AND/OR logic
+   *
+   * @param rules The composite rules to execute
+   * @param df The DataFrame to evaluate rules against
+   * @param additionalDataSources Additional DataFrames for dataset comparison rules
+   * @return Map of rules to their composed outcomes
+   */
+  override def executeRules(
+    rules: Seq[CompositeExecutableRule],
+    df: DataFrame,
+    additionalDataSources: Map[String, DataFrame] = Map.empty
+  ): Map[DQRule, RuleOutcome] = {
+
+    // Flatten all nested rules to get leaf rules
+    val allNestedRules = rules.flatMap(flattenRules).distinct
+
+    // Execute all nested rules
+    val nestedOutcomes = DQDLExecutor.executeRules(
+      allNestedRules,
+      df,
+      additionalDataSources
+    )
+
+    // Compose outcomes for each composite rule
+    rules.map { compositeRule =>
+      val outcome = RuleOutcomeTranslator.collectOutcome(
+        compositeRule.dqRule,
+        nestedOutcomes
+      )
+      compositeRule.dqRule -> outcome
+    }.toMap
+  }
+
+  /**
+   * Recursively flattens composite rules to extract all leaf (non-composite) rules.
+   *
+   * This ensures that each leaf rule is executed only once, even if it appears
+   * in multiple composite rules or at different nesting levels.
+   *
+   * @param rule The composite rule to flatten
+   * @return Sequence of all leaf rules contained within the composite
+   */
+  private def flattenRules(rule: CompositeExecutableRule): Seq[ExecutableRule] = {
+    rule.nestedRules.flatMap {
+      case composite: CompositeExecutableRule => flattenRules(composite)
+      case other => Seq(other)
+    }
+  }
+}