Data quality tooling usually arrives as a platform: a service to deploy next to the cluster that is already holding the data. And its output stops at a verdict: "3,412 rows failed" says the table is bad, not which rule rejected which row, so the quarantine it writes cannot actually be worked.

One class, Cola, with a registry of rule types. A rule is a plain dict (a type plus that check’s parameters), so a validation block can be written by hand, generated, or stored next to the pipeline that runs it. Thresholds are one small DSL shared by every metric: > 0, < 5%, between 10 and 20, <= 2h.

from sparquet_cola import Cola

cola = Cola()

for r in cola.run(df, [
    {"type": "row_count", "min": 1},
    {"type": "not_null", "columns": ["id", "email"]},
    {"type": "missing_percent", "column": "cpf", "must_be": "< 5%", "warn": "= 0"},
    {"type": "sql", "failed_rows": "SELECT * FROM _validation_df WHERE amount < 0"},
]):
    print(r)   # [FAIL] check: missing_percent(cpf) = 8 violates must_be (< 5%)

Every row-level rule carries a code: declared, or rendered from the rule itself, always the same string for the same rule, because that string lands in your data. annotate writes it beside each rejected row, built from predicates the split already computes, so it costs no extra pass.

split = cola.split(df, [
    {"type": "not_null", "columns": ["id", "email"]},
    {"type": "range", "column": "age", "min": 1, "max": 99, "code": "AGE_RANGE"},
], annotate="dq_codes")

split.valid.write.format("delta").save(".../silver_ok")
split.invalid.write.format("delta").save(".../silver_quarantine")

split.invalid.select("id", "dq_codes").show(truncate=False)
# +----+-------------------+
# | id | dq_codes          |
# +----+-------------------+
# | 7  | [AGE_RANGE]       |
# | 8  | [not_null(email)] |
# +----+-------------------+

Each target expands into an independent rule with its own result, its own code and its own share of the quarantine: a single verdict over two columns would not say which of them broke. Ambiguous forms (an empty target list, a nested targets, a code on the parent) are refused at parse time instead of being silently degraded.

{"type": "regex", "targets": [
    {"column": "document",  "pattern": "^[0-9]{11}$"},
    {"column": "document2", "pattern": "^[0-9]{12}$"}]}

# regex(document,^[0-9]{11}$)
# regex(document2,^[0-9]{12}$)

Until v0.4.0 each check fired its own action, often more than one, so a block of a dozen rules scanned the data a dozen times. Now run collects the columns of every check that answered, issues a single df.agg(...), and hands each check back its slice. Checks registered from outside, which cannot know the mechanism exists, keep their own action and keep working unchanged.

tests/test_run_batch_spark.py runs each rule alone and again inside a block, comparing verdict, message, failed_count, severity and metric_value field by field, so the two paths cannot drift.

class NoFutureDateCheck(BaseCheck):
    def aggregations(self, df):
        return [F.count(F.when(F.col(self.params["column"]) > F.current_date(), F.lit(1)))]

    def from_aggregations(self, values):
        failed = int(values[0])
        if failed:
            return CheckResult("no_future_date", False, f"{failed} future dates", failed)
        return CheckResult("no_future_date", True)

• PyPI: pip install sparquet-cola, imported as sparquet_cola
• Drops into a Spark job, a notebook or an Airflow task
• Documented in English, Portuguese and Spanish
• The engine behind the validations block of Sparquet