Skip to content

feat: add is_valid_email row check function #1158

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mwojtyczka merged 12 commits into from
Jun 3, 2026
Merged

feat: add is_valid_email row check function #1158

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
5f57700
feat: add is_valid_email row check function
aarushisingh04 May 5, 2026
72ddac9
docs: add documentation for is_valid_email and remove redundant tests
aarushisingh04 May 6, 2026
9d25af0
Merge branch 'main' into feat/is-valid-email
aarushisingh04 May 6, 2026
6898c08
Update regex pattern
ghanse May 7, 2026
41ee18d
Fix test
ghanse May 7, 2026
28e63b5
Add tests
ghanse May 9, 2026
7ff99a8
Fix tests
ghanse May 11, 2026
47492b4
Fix tests
ghanse May 11, 2026
6213ef9
Merge branch 'main' into feat/is-valid-email
mwojtyczka Jun 2, 2026
acf06d8
Fix perf tests
ghanse Jun 2, 2026
67358f4
Merge remote-tracking branch 'origin/feat/is-valid-email' into feat/i…
ghanse Jun 2, 2026
0bb06c0
Merge branch 'main' into feat/is-valid-email
ghanse Jun 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions docs/dqx/docs/reference/quality_checks.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ You can also define your own custom checks in Python (see [Creating custom check
| `is_older_than_n_days` | Checks whether the values in one input column are at least N days older than the values in another column. | `column`: column to check (can be a string column name or a column expression); `days`: number of days; `curr_date`: current date, if not provided current_date() function is used; `negate`: if the condition should be negated |
| `is_older_than_col2_for_n_days` | Checks whether the values in one input column are at least N days older than the values in another column. | `column1`: first column to check (can be a string column name or a column expression); `column2`: second column to check (can be a string column name or a column expression); `days`: number of days; `negate`: if the condition should be negated |
| `regex_match` | Checks whether the values in the input column match a given regex. | `column`: column to check (can be a string column name or a column expression); regex: regex to check; `negate`: if the condition should be negated (true) or not |
| `is_valid_email` | Checks whether the values in the input column have valid email address format. | `column`: column to check (can be a string column name or a column expression) |
| `is_valid_ipv4_address` | Checks whether the values in the input column have valid IPv4 address format. | `column` to check (can be a string column name or a column expression) |
| `is_ipv4_address_in_cidr` | Checks whether the values in the input column have valid IPv4 address format and fall within the given CIDR block. | `column`: column to check (can be a string column name or a column expression); `cidr_block`: CIDR block string |
| `is_valid_ipv6_address` | Checks whether the values in the input column have valid IPv6 address format. | `column` to check (can be a string column name or a column expression) |
Expand Down Expand Up @@ -524,6 +525,13 @@ For brevity, the `name` field in the examples is omitted and it will be auto-gen
regex: '[0-9]+'
negate: false

# is_valid_email check
- criticality: error
check:
function: is_valid_email
arguments:
column: col1

# is_valid_ipv4_address check
- criticality: error
check:
Expand Down Expand Up @@ -1264,6 +1272,13 @@ checks = [
}
),

# is_valid_email check
DQRowRule(
criticality="error",
check_func=check_funcs.is_valid_email,
column="col1"
),

# is_valid_ipv4_address check
DQRowRule(
criticality="error",
Expand Down
56 changes: 56 additions & 0 deletions src/databricks/labs/dqx/check_funcs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,12 @@
IPV4_MAX_OCTET_COUNT = 4
IPV4_BIT_LENGTH = 32

# Email helpers (RFC 5322 §3.2.3, §3.2.4 + RFC 5321 §4.1.3, §4.5.3.1).
_EMAIL_ATEXT = r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]"
_EMAIL_DOMAIN_LABEL = r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?"
_EMAIL_QTEXT = r"[\x21\x23-\x5B\x5D-\x7E]" # printable ASCII except '"' (0x22) and '\' (0x5C)
_EMAIL_QPAIR = r"\\[\x09\x20-\x7E]" # quoted-pair: '\' + VCHAR or WSP
Comment thread
mwojtyczka marked this conversation as resolved.
Outdated

# Curated aggregate functions for data quality checks
# These are univariate (single-column) aggregate functions suitable for DQ monitoring
# Maps function names to human-readable display names for error messages
Expand Down Expand Up @@ -69,6 +75,24 @@ class DQPattern(Enum):

IPV4_ADDRESS = rf"^{_IPV4_OCTET}\.{_IPV4_OCTET}\.{_IPV4_OCTET}\.{_IPV4_OCTET}$"
IPV4_CIDR_BLOCK = rf"{IPV4_ADDRESS[:-1]}/{_IPV4_CIDR_SUFFIX}$"
# RFC 5322 pragmatic subset: dot-atom or quoted-string local part, dot-atom or
# IP-literal domain, with RFC 5321 length caps (local ≤ 64, total ≤ 254).
# Excludes CFWS, obsolete grammar, and SMTPUTF8/IDN. ReDoS-safe.
EMAIL_ADDRESS = (
rf"^(?=.{{1,254}}$)"
# Local part: dot-atom (with 64-char cap) or quoted-string
rf"(?:"
rf"(?=[^@]{{1,64}}@){_EMAIL_ATEXT}+(?:\.{_EMAIL_ATEXT}+)*"
rf'|"(?:{_EMAIL_QTEXT}|{_EMAIL_QPAIR})*"'
Comment thread
ghanse marked this conversation as resolved.
rf")"
rf"@"
# Domain: LDH hostname with alphabetic TLD, or IP-literal
rf"(?:"
rf"(?:{_EMAIL_DOMAIN_LABEL}\.)+[A-Za-z]{{2,63}}"
rf"|\[(?:{_IPV4_OCTET}(?:\.{_IPV4_OCTET}){{3}}|IPv6:[A-Fa-f0-9:]+)\]"
Comment thread
ghanse marked this conversation as resolved.

@mwojtyczka mwojtyczka May 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(Cosmetic) Captured-group count grows.

_IPV4_OCTET = r"(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)" is a capturing group. Each reuse here adds 4 more capture groups beyond what's needed (the existing IPV4_ADDRESS does the same). No correctness impact, but switching _IPV4_OCTET to non-capturing (?:…) once at its definition would clean this up across all callers. Optional; can be a follow-up.

rf")"
rf"$"
)


def make_condition(condition: Column, message: Column | str, alias: str) -> Column:
Expand Down Expand Up @@ -971,6 +995,38 @@ def is_valid_ipv4_address(column: str | Column) -> Column:
return _matches_pattern(column, DQPattern.IPV4_ADDRESS)


@register_rule("row")
def is_valid_email(column: str | Column) -> Column:
"""Checks whether the values in the input column are valid email addresses.

Validates against a pragmatic RFC 5322 subset:

* Local part: dot-atom (RFC 5322 §3.2.3) or quoted-string (§3.2.4).
* Domain: dot-atom hostname with LDH labels (RFC 1035 §2.3.4) and an
alphabetic TLD, or an IP-literal (RFC 5321 §4.1.3) — bracketed IPv4
with octet validation, or *[IPv6:...]* matched syntactically only
(callers requiring semantic IPv6 validation should re-check via
*ipaddress.IPv6Address*).
* Length caps per RFC 5321 §4.5.3.1: local-part max 64 octets, total
address max 254 octets.

Comments and folding whitespace (CFWS), obsolete grammar
(*obs-local-part*, *obs-domain*, *obs-qtext*, *obs-qp*), and
internationalized addresses (RFC 6531 / SMTPUTF8) are not supported;
Comment thread
ghanse marked this conversation as resolved.
a separate check would be needed for each. Numeric and single-character
TLDs are rejected (ICANN policy + practical interoperability).

Null values are treated as passing the check (no violation reported).

Args:
column: column to check; can be a string column name or a column expression

Returns:
Column object for condition
"""
return _matches_pattern(column, DQPattern.EMAIL_ADDRESS)


@register_rule("row")
def is_ipv4_address_in_cidr(column: str | Column, cidr_block: str) -> Column:
"""
Expand Down
103 changes: 103 additions & 0 deletions tests/integration/test_row_checks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
has_valid_json_schema,
is_valid_timestamp,
is_valid_ipv4_address,
is_valid_email,
is_ipv4_address_in_cidr,
is_valid_ipv6_address,
is_ipv6_address_in_cidr,
Expand Down Expand Up @@ -1842,6 +1843,108 @@ def test_col_is_valid_ipv4_address(spark):
assertDataFrameEqual(actual, expected)


def test_col_is_valid_email(spark):
schema_email = "a: string"
local_part_at_char_limit = "a" * 64 # 64 characters total
full_text_at_char_limit = "a" * 64 + "@" + "b" * 186 + ".co" # 254 characters total

test_df = spark.createDataFrame(
[
["user@example.com"],
["User.Name+tag@sub.domain.org"],
["a@b.cd"], # minimum valid
["1234567890@numbers.io"], # all-numeric local
["mixed.case@MixedCase.Org"], # case variations
[local_part_at_char_limit + "@b.com"], # exactly at 64-character limit for local parts
[full_text_at_char_limit], # exactly at 268-character limit for the full text
Comment thread
ghanse marked this conversation as resolved.
Outdated
['"a"@example.com'],
['"a@b"@example.com'], # "@" inside quoted local
["user@[192.0.2.1]"],
["user@[IPv6:2001:db8::1]"],
[".user@example.com"], # leading "."
["user.@example.com"], # trailing "."
["us..er@example.com"], # consecutive "."
["user@-example.com"], # leading hyphen on label
["user@example-.com"], # trailing hyphen on label
["user@.example.com"], # empty subdomain label
["user@example..com"], # double "." in domain
["user@example.c"], # single character TLD
["user@example.123"], # numeric TLD
["user@example"], # no TLD
["userexample.com"], # no "@"
["user@@example.com"], # double "@"
["user@"], # missing domain
["@example.com"], # missing local part
["plainaddress"], # no "@" or "."
["a" + local_part_at_char_limit + "@example.com"], # local part more than 64 characters
["a" + full_text_at_char_limit], # full text more than 268 characters
["user@[999.0.0.1]"], # invalid IPv4 octet
["user@[256.0.0.1]"], # IPv4 octet over 255
["user@[]"], # empty IP literal
["user@[192.0.2.1"], # missing closing bracket
["user@localhost"], # no TLD
["missing@tld"], # no domain "."
[None], # null - passes (no violation reported)
],
schema_email,
)

actual = test_df.select(is_valid_email("a"))

def violation(value: str) -> str:
return f"Value '{value}' in Column 'a' does not match pattern 'EMAIL_ADDRESS'"

checked_schema = "a_does_not_match_pattern_email_address: string"
checked_data = [
# Valid (no violation reported)
[None],
[None],
[None],
[None],
[None],
[None],
[None],
[None],
[None],
[None],
[None],
# Invalid - dot-atom violations
[violation(".user@example.com")],
[violation("user.@example.com")],
[violation("us..er@example.com")],
# Invalid - LDH domain violations
[violation("user@-example.com")],
[violation("user@example-.com")],
[violation("user@.example.com")],
[violation("user@example..com")],
# Invalid - TLD constraints
[violation("user@example.c")],
[violation("user@example.123")],
[violation("user@example")],
# Invalid - structural
[violation("userexample.com")],
[violation("user@@example.com")],
[violation("user@")],
[violation("@example.com")],
[violation("plainaddress")],
# Invalid - length caps
[violation("a" + local_part_at_char_limit + "@example.com")],
[violation("a" + full_text_at_char_limit)],
# Invalid - IP-literal edge cases
[violation("user@[999.0.0.1]")],
[violation("user@[256.0.0.1]")],
[violation("user@[]")],
[violation("user@[192.0.2.1")],
# Pre-existing coverage retained
[violation("user@localhost")],
[violation("missing@tld")],
[None],
]
expected = spark.createDataFrame(checked_data, checked_schema)

assertDataFrameEqual(actual, expected)


def test_is_ipv4_address_in_cidr(spark):
schema_ipv4 = "a: string, b: string"

Expand Down
19 changes: 19 additions & 0 deletions tests/unit/test_build_rules.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
is_aggr_not_equal,
foreign_key,
is_valid_ipv4_address,
is_valid_email,
is_ipv4_address_in_cidr,
is_not_less_than,
is_not_greater_than,
Expand Down Expand Up @@ -247,6 +248,7 @@ def test_build_rules():
DQRowRule(
criticality="warn", check_func=is_ipv4_address_in_cidr, column="g", check_func_args=["192.168.1.0/24"]
),
DQRowRule(criticality="warn", check_func=is_valid_email, column="g"),
DQDatasetRule(
criticality="error",
check_func=sql_query,
Expand Down Expand Up @@ -505,6 +507,12 @@ def test_build_rules():
column="g",
check_func_args=["192.168.1.0/24"],
),
DQRowRule(
name="g_does_not_match_pattern_email_address",
criticality="warn",
check_func=is_valid_email,
column="g",
),
DQDatasetRule(
criticality="error",
check_func=sql_query,
Expand Down Expand Up @@ -757,6 +765,11 @@ def test_build_rules_by_metadata():
"arguments": {"column": "a", "cidr_block": "192.168.1.0/24"},
},
},
{
"name": "a_is_valid_email",
"criticality": "error",
"check": {"function": "is_valid_email", "arguments": {"column": "a"}},
},
]

actual_rules = deserialize_checks(checks)
Expand Down Expand Up @@ -1023,6 +1036,12 @@ def test_build_rules_by_metadata():
column="a",
check_func_kwargs={"cidr_block": "192.168.1.0/24"},
),
DQRowRule(
name="a_is_valid_email",
criticality="error",
check_func=is_valid_email,
column="a",
),
]

assert pprint.pformat(actual_rules) == pprint.pformat(expected_rules)
Expand Down