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 9 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
58 changes: 58 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; valid only inside a quoted part

# 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,25 @@ 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 or quoted-string limited to 64-characters
rf"(?=[^@]{{1,64}}@)"
rf"(?:"
rf"{_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 +996,39 @@ 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 subset of RFC 5322:

* 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
addresses will use octet validation while *[IPv6:...] addresses* are
only matched syntactically; Callers requiring semantic IPv6 domain
validation should implement a custom Python check that uses a Pandas
UDF to call *ipaddress.IPv6Address* methods.
* Length caps (per RFC 5321 §4.5.3.1): 64 octets for local-parts, 254
octets for the full address.

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 will pass the check with 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
18 changes: 16 additions & 2 deletions tests/integration/test_apply_checks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5903,7 +5903,8 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak
schema = (
"col1: string, col2: int, col3: int, col4 array<int>, col5: date, col6: timestamp, "
"col7: map<string, int>, col8: struct<field1: int>, col10: int, col11: string, "
"col_ipv4: string, col_ipv6: string, col_json_str: string, col_json_str2: string"
"col_ipv4: string, col_ipv6: string, col_json_str: string, col_json_str2: string, "
"col_email: string"
)
test_df = spark.createDataFrame(
[
Expand All @@ -5922,6 +5923,7 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak
"2001:0db8:85a3:08d3:1319:8a2e:0370:7344",
'{"key1": "1"}',
'{"a" : 1, "b": 2}',
"user@example.com",
],
[
"val2",
Expand All @@ -5938,6 +5940,7 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak
"2001:0db8:85a3:08d3:ffff:ffff:ffff:ffff",
'{"key1": "1", "key2": "2"}',
'{ "a" : 1, "b": 1000, "c": {"1": 8}}',
'"quoted_user"@example.co.uk',
],
[
"val3",
Expand All @@ -5954,6 +5957,7 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak
"2001:db8:85a3:8d3:1319:8a2e:3.112.115.68",
'{"key1": "[1, 2, 3]"}',
'{ "a" : 1, "b": 1023455, "c": null }',
"user@[12.96.144.202]",
],
],
schema,
Expand Down Expand Up @@ -5994,6 +5998,7 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak
"2001:0db8:85a3:08d3:1319:8a2e:0370:7344",
'{"key1": "1"}',
'{"a" : 1, "b": 2}',
"user@example.com",
None,
None,
],
Expand All @@ -6012,6 +6017,7 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak
"2001:0db8:85a3:08d3:ffff:ffff:ffff:ffff",
'{"key1": "1", "key2": "2"}',
'{ "a" : 1, "b": 1000, "c": {"1": 8}}',
'"quoted_user"@example.co.uk',
None,
None,
],
Expand All @@ -6030,6 +6036,7 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak
"2001:db8:85a3:8d3:1319:8a2e:3.112.115.68",
'{"key1": "[1, 2, 3]"}',
'{ "a" : 1, "b": 1023455, "c": null }',
"user@[12.96.144.202]",
None,
None,
],
Expand Down Expand Up @@ -6183,7 +6190,8 @@ def test_apply_checks_all_checks_as_yaml(ws, spark):
schema = (
"col1: string, col2: int, col3: int, col4 array<int>, col5: date, col6: timestamp, "
"col7: map<string, int>, col8: struct<field1: int>, col10: int, col11: string, "
"col_ipv4: string, col_ipv6: string, col_json_str: string, col_json_str2: string"
"col_ipv4: string, col_ipv6: string, col_json_str: string, col_json_str2: string, "
"col_email: string"
)
test_df = spark.createDataFrame(
[
Expand All @@ -6202,6 +6210,7 @@ def test_apply_checks_all_checks_as_yaml(ws, spark):
"2001:0db8:85a3:08d3:0000:0000:0000:0001",
'{"key1": "1"}',
'{"a" : 1, "b": 2}',
"user@example.com",
],
[
"val2",
Expand All @@ -6218,6 +6227,7 @@ def test_apply_checks_all_checks_as_yaml(ws, spark):
"2001:0db8:85a3:08d3:0000:0000:0000:1",
'{"key1": "1", "key2": "2"}',
'{ "a" : 1, "b": 1000, "c": {"1": 8}}',
'"quoted_user"@example.co.uk',
],
[
"val3",
Expand All @@ -6234,6 +6244,7 @@ def test_apply_checks_all_checks_as_yaml(ws, spark):
"2001:0db8:85a3:08d3:0000::2",
'{"key1": "[1, 2, 3]"}',
'{ "a" : 1, "b": 1023455, "c": null }',
"user@[12.96.144.202]",
],
],
schema,
Expand Down Expand Up @@ -6262,6 +6273,7 @@ def test_apply_checks_all_checks_as_yaml(ws, spark):
"2001:0db8:85a3:08d3:0000:0000:0000:0001",
'{"key1": "1"}',
'{"a" : 1, "b": 2}',
"user@example.com",
None,
None,
],
Expand All @@ -6280,6 +6292,7 @@ def test_apply_checks_all_checks_as_yaml(ws, spark):
"2001:0db8:85a3:08d3:0000:0000:0000:1",
'{"key1": "1", "key2": "2"}',
'{ "a" : 1, "b": 1000, "c": {"1": 8}}',
'"quoted_user"@example.co.uk',
None,
None,
],
Expand All @@ -6298,6 +6311,7 @@ def test_apply_checks_all_checks_as_yaml(ws, spark):
"2001:0db8:85a3:08d3:0000::2",
'{"key1": "[1, 2, 3]"}',
'{ "a" : 1, "b": 1023455, "c": null }',
"user@[12.96.144.202]",
None,
None,
],
Expand Down
Loading