Native declarative data validation / expectations API (df.validate(...))

[Liusixuuu]

Summary

I'd like to propose a first-class, declarative data validation / expectations API for Daft — something like df.validate([...]) — that compiles checks into native Daft expressions so they execute lazily and distributed in the engine, instead of requiring an external framework.

Motivation

When building AI and multimodal pipelines on Daft, there's currently no built-in way to assert data-quality constraints (non-null, value ranges, allowed categories, schema/dtype, row-count bounds, uniqueness) as part of the query plan. Today users either:

• write ad-hoc filter + count checks (verbose, not reusable, no standardized report), or
• pull in Pandera / Great Expectations, which forces materialization to pandas/Arrow and breaks Daft's lazy, distributed, out-of-core execution.

For pipelines feeding LLM prompts or embedding generation, silently malformed rows (empty strings, out-of-range values, unexpected nulls in multimodal columns) are a common and costly failure mode — they waste inference spend and corrupt downstream outputs.

Proposed API

import daft
from daft.validation import Expect

df = daft.read_parquet("s3://bucket/products/")

report = df.validate([
    Expect.column("price").to_be_between(0, 1e6),
    Expect.column("category").to_be_in(["a", "b", "c"]),
    Expect.column("image_url").to_not_be_null(),
    Expect.column("id").to_be_unique(),
    Expect.row_count().to_be_at_least(1),
    Expect.schema({"price": daft.DataType.float64()}),
])

report.summary()           # pass/fail per expectation + failed row counts
report.raise_on_failure()  # opt-in hard failure for CI / pipeline gating

# Or keep only valid rows, lazily
clean_df = df.where_valid(expectations)

Key properties

• Native execution: expectations lower to Daft expressions — no forced materialization, so validation stays lazy, distributed, and out-of-core.
• Structured report: serializable, per-expectation pass/fail with failed-row counts and an optional sample of failing rows.
• Soft and strict modes: report-only for exploration, raise_on_failure for CI/ETL gating.
• Multimodal-aware (stretch): expectations like "image decodable" or "embedding dimensionality matches," which external tabular validators can't express.

Alternatives considered

• Pandera / Great Expectations: break Daft's execution model; don't scale to distributed/out-of-core.
• Manual filter/count: works but verbose, error-prone, no standardized/serializable report.

Open questions

• Should validation be a separate df.validate(...) entry point, or expressible as regular expressions plus a thin reporting layer?

Happy to help scope this down to a minimal v1 (e.g. a handful of tabular expectations + report) and contribute if there's interest.

[everettVT]

This is Fascinating! I love this proposal.

We support automatic type coercion for Pydantic Types so I'd love to derive/compile validation udfs from pydantic models.

[Liusixuuu]

Thanks @everettVT, I get your point. A single Pydantic model could drive both validation and the struct coercion you already support, instead of users learning a separate API.

Happy to scope a minimal v1: df.validate(model) for dtype/null/range/enum via native-expression compilation, a serializable report, and raise_on_failure, with the per-row UDF fallback and multimodal checks (image-decodable, embedding dim) as follow-ups.

Here's a first cut of the Pydantic-field to Daft-expression mapping. I've tried to use current expression methods but flagged the ones I'm not certain map cleanly:

Pydantic field constraint	Daft expression	Notes
field type (int/float/str/...)	schema dtype check via DataType.is_*() introspection	strict — report mismatch, don't coerce
non-Optional[T]	col(x).not_null()	Optional ⇒ nulls allowed
ge/le/gt/lt (conint/confloat)	col(x) >= lo, col(x) <= hi (combine with &)	comparison operators confirmed; unsure if there's a between helper
Enum / Literal	col(x).is_in([...])
min_length/max_length (constr)	col(x).str.length() + comparisons
pattern=... (constr)	regex match — str.match?	str.contains is substring; need to confirm the regex method name in current API
field marked unique	distinct-count vs count	couldn't confirm an exact (non-approx) distinct-count method. Is there one, or should this lower to a groupby().count() check?
custom @field_validator	per-row model_validate UDF	fallback path, flagged non-vectorized

Would you want v1 scoped to just the vectorizable per-field constraints (dtype/null/range/enum/length), leaving uniqueness, regex, and custom validators for a second pass? If that split sounds right, I can start on the field-introspection to expression compiler.

[everettVT]

I'm definitely curious about it as a prototype.

I agree with @srilman that the api design will probably end up needing iteration.

One thought I had was to add validation to col().

If you could perform validation directly in columns like

import daft 

df = daft.from_pydict({
    "name":["Bob","bill","Ben"],
    "value":[1,2,3]
})

df = df.expect(
    col("name").is_capitalized(on_err="log")
)

df.validate() # materialize

[srilman]

Hi @Liusixuuu, I agree with @everettVT's point about Pydantic, and I think you mapping from makes sense. Before starting to implement though, I wanted to ask what your thoughts on the DataFrame API was.

I think having a core expression like is_valid that can take in multiple expressions and return a boolean expression would be the most flexible, as you can use it wherever you want expressions to go. On top of that, we can have a helper at the DataFrame row level. But I'm not sure about the report = df.validate(...) API you presented above. Do you mind explaining your thoughts?

[Liusixuuu]

Hi @srilman, my initial intention forreport = df.validate(...) was for answering "which checks failed, how many rows each, and a sample of bad rows". Like a aggregate report.

I agree that is_valid should be the right primitive. It's a plain expression and more flexible.

Here's the design combining @everettVT's advice:

# 1. expression primitive, usable anywhere (where, with_column, etc.)
is_valid(
    col("price").between(0, 1e6),
    col("category").is_in(["a", "b", "c"]),
)

# 2. collect named checks, then materialize the report
df.expect(
    col("price").between(0, 1e6).alias("price_range"),
    col("category").is_in(["a", "b", "c"]).alias("category_valid"),
)
df.validate(on_err="log") (# if keep this, maybe move the on_err here to keep the expression pure)

Does this looks cleaner? I'm curious about your thoughts.