Skip to content

Commit b64afe0

Browse files
shriyavanvarishriyavanvari
authored
Add composite rule evaluation with AND/OR operators (#665)
* Add composite rule evaluation with AND/OR operators Authored-by: Shriya Vanvari <svanvari@amazon.com>
1 parent 2d45b81 commit b64afe0

8 files changed

Lines changed: 488 additions & 6 deletions

File tree

‎README.md‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,9 @@ Deequ also supports [DQDL](https://docs.aws.amazon.com/glue/latest/dg/dqdl.html)
158158
- **ColumnExists**: `ColumnExists "column"`
159159
- **RowCountMatch**: `RowCountMatch "referenceDataset" >= 0.9`
160160
- **DataFreshness**: `DataFreshness "Order_Date" <= 24 hours`
161+
- **Composite Rules**: Combine multiple rules with `and` / `or` operators
162+
- Simple: `(RowCount > 0) and (IsComplete "column")`
163+
- Nested: `(Rule1) or ((Rule2) and (Rule3))`
161164

162165
### Scala Example
163166

@@ -223,6 +226,57 @@ Dataset<Row> results = EvaluateDataQuality.process(df, ruleset);
223226
results.show();
224227
```
225228

229+
### Composite Rules Example
230+
231+
Composite rules allow you to combine multiple data quality checks using logical operators (`and`, `or`). This enables complex validation scenarios:
232+
233+
```scala
234+
import com.amazon.deequ.dqdl.EvaluateDataQuality
235+
import org.apache.spark.sql.SparkSession
236+
237+
val spark = SparkSession.builder()
238+
.appName("Composite Rules Example")
239+
.master("local[*]")
240+
.getOrCreate()
241+
242+
import spark.implicits._
243+
244+
val df = Seq(
245+
(1, "Alice", 25, "alice@example.com"),
246+
(2, "Bob", 30, "bob@example.com"),
247+
(3, "Charlie", 35, "charlie@example.com")
248+
).toDF("id", "name", "age", "email")
249+
250+
// Simple AND: Both conditions must be true
251+
val andRule = """Rules=[(RowCount > 0) and (IsComplete "email")]"""
252+
val andResults = EvaluateDataQuality.process(df, andRule)
253+
andResults.show()
254+
255+
// Simple OR: At least one condition must be true
256+
val orRule = """Rules=[(RowCount > 100) or (IsUnique "id")]"""
257+
val orResults = EvaluateDataQuality.process(df, orRule)
258+
orResults.show()
259+
260+
// Nested composition: Complex logic with multiple levels
261+
val nestedRule = """Rules=[
262+
((IsComplete "name") and (IsComplete "email")) or
263+
((RowCount > 0) and (IsUnique "id"))
264+
]"""
265+
val nestedResults = EvaluateDataQuality.process(df, nestedRule)
266+
nestedResults.show()
267+
268+
// Multiple composite rules in one ruleset
269+
val multipleRules = """Rules=[
270+
(RowCount > 0) and (IsComplete "id"),
271+
(IsUnique "id") or (IsUnique "email"),
272+
((Mean "age" > 20) and (Mean "age" < 50)) or (RowCount < 10)
273+
]"""
274+
val multipleResults = EvaluateDataQuality.process(df, multipleRules)
275+
multipleResults.show()
276+
```
277+
278+
**Note:** Composite rules currently support dataset-level evaluation only. Row-level evaluation (identifying which specific rows pass/fail) is not yet implemented.
279+
226280
## Citation
227281

228282
If you would like to reference this package in a research paper, please cite:

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,8 @@
1616

1717
package com.amazon.deequ.dqdl.execution
1818

19-
import com.amazon.deequ.dqdl.execution.executors.{ColumnNamesMatchPatternExecutor, DataFreshnessExecutor, DeequRulesExecutor, ReferentialIntegrityExecutor, RowCountMatchExecutor, UnsupportedRulesExecutor}
20-
import com.amazon.deequ.dqdl.model.{ColumnNamesMatchPatternExecutableRule, DataFreshnessExecutableRule, DeequExecutableRule, ExecutableRule, Failed, ReferentialIntegrityExecutableRule, RowCountMatchExecutableRule, RuleOutcome, UnsupportedExecutableRule}
19+
import com.amazon.deequ.dqdl.execution.executors.{ColumnNamesMatchPatternExecutor, CompositeRulesExecutor, DataFreshnessExecutor, DeequRulesExecutor, ReferentialIntegrityExecutor, RowCountMatchExecutor, UnsupportedRulesExecutor}
20+
import com.amazon.deequ.dqdl.model.{ColumnNamesMatchPatternExecutableRule, CompositeExecutableRule, DataFreshnessExecutableRule, DeequExecutableRule, ExecutableRule, Failed, ReferentialIntegrityExecutableRule, RowCountMatchExecutableRule, RuleOutcome, UnsupportedExecutableRule}
2121
import org.apache.spark.sql.DataFrame
2222
import software.amazon.glue.dqdl.model.DQRule
2323

@@ -35,6 +35,7 @@ object DQDLExecutor {
3535
// Map from rule class to its executor
3636
private val executors = Map[Class[_ <: ExecutableRule], RuleExecutor[_ <: ExecutableRule]](
3737
classOf[DeequExecutableRule] -> DeequRulesExecutor,
38+
classOf[CompositeExecutableRule] -> CompositeRulesExecutor,
3839
classOf[UnsupportedExecutableRule] -> UnsupportedRulesExecutor,
3940
classOf[RowCountMatchExecutableRule] -> RowCountMatchExecutor,
4041
classOf[ReferentialIntegrityExecutableRule] -> ReferentialIntegrityExecutor,

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

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/**
2+
* Copyright 2026 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"). You may not
5+
* use this file except in compliance with the License. A copy of the License
6+
* is located at
7+
*
8+
* http://aws.amazon.com/apache2.0/
9+
*
10+
* or in the "license" file accompanying this file. This file is distributed on
11+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+
* express or implied. See the License for the specific language governing
13+
* permissions and limitations under the License.
14+
*
15+
*/
16+
17+
package com.amazon.deequ.dqdl.execution.executors
18+
19+
import com.amazon.deequ.dqdl.execution.DQDLExecutor
20+
import com.amazon.deequ.dqdl.model.{CompositeExecutableRule, ExecutableRule, RuleOutcome}
21+
import com.amazon.deequ.dqdl.translation.RuleOutcomeTranslator
22+
import org.apache.spark.sql.DataFrame
23+
import software.amazon.glue.dqdl.model.DQRule
24+
25+
/**
26+
* Executor for composite rules that combine multiple rules with AND/OR operators.
27+
*
28+
* This executor handles the evaluation of composite rules by:
29+
* 1. Flattening the rule tree to extract all leaf (non-composite) rules
30+
* 2. Executing all leaf rules once (avoiding duplicate execution)
31+
* 3. Composing the outcomes using the specified logical operators
32+
*
33+
* Example composite rules:
34+
* - `(RowCount > 0) and (IsComplete "column")`
35+
* - `(ColumnValues "col" > 10) or (IsUnique "col")`
36+
* - `((Rule1) and (Rule2)) or (Rule3)` (nested composition)
37+
*
38+
* Note: Currently only supports dataset-level evaluation.
39+
*/
40+
object CompositeRulesExecutor extends DQDLExecutor.RuleExecutor[CompositeExecutableRule] {
41+
42+
/**
43+
* Executes composite rules by evaluating all nested rules and composing their outcomes.
44+
*
45+
* The execution strategy:
46+
* 1. Flattens all composite rules to extract unique leaf rules
47+
* 2. Executes each leaf rule once (shared across multiple composites if needed)
48+
* 3. Uses RuleOutcomeTranslator to compose outcomes with AND/OR logic
49+
*
50+
* @param rules The composite rules to execute
51+
* @param df The DataFrame to evaluate rules against
52+
* @param additionalDataSources Additional DataFrames for dataset comparison rules
53+
* @return Map of rules to their composed outcomes
54+
*/
55+
override def executeRules(
56+
rules: Seq[CompositeExecutableRule],
57+
df: DataFrame,
58+
additionalDataSources: Map[String, DataFrame] = Map.empty
59+
): Map[DQRule, RuleOutcome] = {
60+
61+
// Flatten all nested rules to get leaf rules
62+
val allNestedRules = rules.flatMap(flattenRules).distinct
63+
64+
// Execute all nested rules
65+
val nestedOutcomes = DQDLExecutor.executeRules(
66+
allNestedRules,
67+
df,
68+
additionalDataSources
69+
)
70+
71+
// Compose outcomes for each composite rule
72+
rules.map { compositeRule =>
73+
val outcome = RuleOutcomeTranslator.collectOutcome(
74+
compositeRule.dqRule,
75+
nestedOutcomes
76+
)
77+
compositeRule.dqRule -> outcome
78+
}.toMap
79+
}
80+
81+
/**
82+
* Recursively flattens composite rules to extract all leaf (non-composite) rules.
83+
*
84+
* This ensures that each leaf rule is executed only once, even if it appears
85+
* in multiple composite rules or at different nesting levels.
86+
*
87+
* @param rule The composite rule to flatten
88+
* @return Sequence of all leaf rules contained within the composite
89+
*/
90+
private def flattenRules(rule: CompositeExecutableRule): Seq[ExecutableRule] = {
91+
rule.nestedRules.flatMap {
92+
case composite: CompositeExecutableRule => flattenRules(composite)
93+
case other => Seq(other)
94+
}
95+
}
96+
}

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

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ import com.amazon.deequ.analyzers.FilteredRowOutcome.FilteredRowOutcome
2020
import com.amazon.deequ.checks.Check
2121
import com.amazon.deequ.dqdl.util.DQDLUtility.convertWhereClauseForMetric
2222
import org.apache.spark.sql.Column
23-
import software.amazon.glue.dqdl.model.DQRule
23+
import software.amazon.glue.dqdl.model.{DQRule, DQRuleLogicalOperator}
2424

2525
trait ExecutableRule {
2626
val dqRule: DQRule
@@ -73,6 +73,24 @@ case class ColumnNamesMatchPatternExecutableRule(dqRule: DQRule,
7373
Some("Dataset.*.ColumnNamesPatternMatchRatio")
7474
}
7575

76+
/**
77+
* Represents a composite rule that combines multiple nested rules using logical operators (AND/OR).
78+
* Composite rules allow complex data quality checks by composing simpler rules.
79+
*
80+
* Example: `(RowCount > 0) and (IsComplete "column")`
81+
*
82+
* Note: Currently only supports dataset-level evaluation. Row-level evaluation is not yet implemented.
83+
*
84+
* @param dqRule The DQDL rule definition
85+
* @param nestedRules The child rules to be evaluated and combined
86+
* @param operator The logical operator (AND/OR) used to combine nested rule outcomes
87+
*/
88+
case class CompositeExecutableRule(dqRule: DQRule,
89+
nestedRules: Seq[ExecutableRule],
90+
operator: DQRuleLogicalOperator) extends ExecutableRule {
91+
override val evaluatedMetricName: Option[String] = None
92+
}
93+
7694
case class DeequMetricMapping(entity: String,
7795
instance: String,
7896
name: String,

‎src/main/scala/com/amazon/deequ/dqdl/model/RuleOutcome.scala‎

Lines changed: 65 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,46 @@
1717
package com.amazon.deequ.dqdl.model
1818

1919
import com.amazon.deequ.checks.{CheckResult, CheckStatus}
20-
import software.amazon.glue.dqdl.model.DQRule
20+
import software.amazon.glue.dqdl.model.{DQRule, DQRuleLogicalOperator}
21+
22+
/**
23+
* Represents row-level evaluation outcomes for data quality rules.
24+
* This is used to track which specific rows pass or fail a rule.
25+
*
26+
* Note: Row-level evaluation is not yet implemented in DQDL. This data model
27+
* is provided for future extensibility and compatibility with AwsGlueMlDataQualityETL.
28+
*
29+
* A row-level outcome can be either:
30+
* - NoColumn: No row-level results available (default for dataset-level rules)
31+
* - SingularColumn: A single column contains the row-level results
32+
* - CompositeOutcome: A composite of multiple row-level outcomes combined with an operator
33+
*/
34+
sealed trait RowLevelOutcome
35+
36+
/**
37+
* Indicates that a rule does not produce row-level results.
38+
* This is the default for most DQDL rules which only evaluate at the dataset level.
39+
*/
40+
case class NoColumn() extends RowLevelOutcome
41+
42+
/**
43+
* Indicates that a rule produces row-level results in a single column.
44+
*
45+
* @param columnName The name of the column containing row-level boolean results
46+
*/
47+
case class SingularColumn(columnName: String) extends RowLevelOutcome
48+
49+
/**
50+
* Represents the composition of multiple row-level outcomes using logical operators.
51+
* Used by composite rules to combine row-level results from nested rules.
52+
*
53+
* @param columnName The name of the column that will contain the composed results
54+
* @param components The row-level outcomes from nested rules to be combined
55+
* @param operator The logical operator (AND/OR) used to combine the components
56+
*/
57+
case class CompositeOutcome(columnName: String,
58+
components: Seq[RowLevelOutcome],
59+
operator: DQRuleLogicalOperator) extends RowLevelOutcome
2160

2261
sealed trait OutcomeStatus {
2362
def asString: String
@@ -43,11 +82,35 @@ case object Failed extends OutcomeStatus {
4382
def asString: String = "Failed"
4483
}
4584

85+
/**
86+
* Represents the outcome of evaluating a data quality rule.
87+
*
88+
* @param rule The DQDL rule that was evaluated
89+
* @param outcome The status of the evaluation (Passed or Failed)
90+
* @param failureReason Optional message explaining why the rule failed
91+
* @param evaluatedMetrics Map of metric names to their computed values
92+
* @param evaluatedRule Optional rule with evaluated metric values filled in
93+
* @param rowLevelOutcome Information about row-level results (currently unused in DQDL)
94+
*/
4695
case class RuleOutcome(rule: DQRule,
4796
outcome: OutcomeStatus,
4897
failureReason: Option[String],
4998
evaluatedMetrics: Map[String, Double] = Map.empty,
50-
evaluatedRule: Option[DQRule] = None) {
99+
evaluatedRule: Option[DQRule] = None,
100+
rowLevelOutcome: RowLevelOutcome = NoColumn()) {
101+
102+
/**
103+
* Returns a copy of this outcome with the specified row-level outcome.
104+
* Note: Row-level evaluation is not yet implemented in DQDL.
105+
*/
106+
def withRowLevelOutcome(outcome: RowLevelOutcome): RuleOutcome =
107+
copy(rowLevelOutcome = outcome)
108+
109+
/**
110+
* Returns a copy of this outcome with the specified evaluated rule.
111+
*/
112+
def withEvaluatedRule(rule: DQRule): RuleOutcome =
113+
copy(evaluatedRule = Some(rule))
51114
}
52115

53116
object RuleOutcome {

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

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717
package com.amazon.deequ.dqdl.translation
1818

1919
import com.amazon.deequ.analyzers.FilteredRowOutcome
20-
import com.amazon.deequ.dqdl.model.{DeequExecutableRule, ExecutableRule, UnsupportedExecutableRule}
20+
import com.amazon.deequ.dqdl.model.{CompositeExecutableRule, DeequExecutableRule, ExecutableRule, UnsupportedExecutableRule}
2121
import com.amazon.deequ.dqdl.translation.rules.ColumnCorrelationRule
2222
import com.amazon.deequ.dqdl.translation.rules.CompletenessRule
2323
import com.amazon.deequ.dqdl.translation.rules.CustomSqlRule
@@ -84,6 +84,16 @@ object DQDLRuleTranslator {
8484

8585
private[dqdl] def toExecutableRule(rule: DQRule): ExecutableRule = {
8686
rule.getRuleType match {
87+
case "Composite" =>
88+
// Validate nested rules exist
89+
if (rule.getNestedRules == null || rule.getNestedRules.isEmpty) {
90+
UnsupportedExecutableRule(rule, Some("Composite rule must have at least one nested rule"))
91+
} else {
92+
// Recursively translate nested rules
93+
val nestedExecutableRules = rule.getNestedRules.asScala.map(toExecutableRule).toSeq
94+
CompositeExecutableRule(rule, nestedExecutableRules, rule.getOperator)
95+
}
96+
8797
case "DataFreshness" =>
8898
DataFreshnessRule.toExecutableRule(rule, FilteredRowOutcome.TRUE) match {
8999
case Right(executableRule) => executableRule

0 commit comments

Comments
 (0)