Copilot commented on code in PR #64610:
URL: https://github.com/apache/airflow/pull/64610#discussion_r3147217105
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -442,6 +443,66 @@ def depends_search(
return depends_search
+class _RegexParam(BaseParam[str]):
+ """
+ Filter using database-level regex matching (regexp_match).
+
+ SQLAlchemy's ``regexp_match`` is supported on PostgreSQL (``~`` operator),
+ MySQL/MariaDB (``REGEXP``), and SQLite (via Python ``re.match``).
+ Note: SQLite uses ``re.match`` semantics (anchored at the start of the
+ string), while PostgreSQL uses ``re.search`` (matches anywhere).
+ """
+
+ def __init__(self, attribute: ColumnElement, skip_none: bool = True) ->
None:
+ super().__init__(skip_none=skip_none)
+ self.attribute: ColumnElement = attribute
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+ return select.where(self.attribute.regexp_match(self.value))
+
+ @classmethod
+ def depends(cls, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use regex_param_factory instead, depends is
not implemented.")
+
+
+def _validate_regex_pattern(value: str | None) -> str | None:
+ """Validate that the regex pattern is syntactically correct."""
+ if value is None:
+ return value
+ try:
+ re.compile(value)
+ except re.error as e:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"Invalid regular expression: {e}",
+ )
+ return value
+
+
+_DEFAULT_REGEX_DESCRIPTION = (
+ "Regex filter. Uses database-native regex "
+ "(PostgreSQL ~ operator, MySQL REGEXP, SQLite re.match). "
Review Comment:
Regex validation is done with `re.compile()`, but the actual matching is
performed by the database regex engine (PostgreSQL POSIX regex, MySQL regex,
SQLite `re.match`). This can incorrectly reject patterns that are valid on some
backends (or accept patterns that later error on the DB). Either document that
`partition_key_pattern` must be compatible with Python `re` (a cross-backend
subset), or move validation to a backend-aware check / let the database raise
on invalid patterns.
```suggestion
"""
Return the regex pattern unchanged.
Regex matching is performed by the database backend via ``regexp_match``,
so Python ``re`` compilation is not a reliable cross-backend validator:
PostgreSQL and MySQL/MariaDB have different regex engines from Python,
while SQLite uses ``re.match`` semantics. Invalid patterns should be
rejected by the backend that actually evaluates them.
"""
return value
_DEFAULT_REGEX_DESCRIPTION = (
"Regex filter. Uses database-native regex "
"(PostgreSQL ~ operator, MySQL REGEXP, SQLite re.match). "
"Pattern syntax must be compatible with the active database backend; "
"invalid patterns may be rejected when the database evaluates the query.
"
```
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/asset_events.py:
##########
@@ -73,6 +74,31 @@ def _get_asset_events_through_sql_clauses(
)
+def _validate_partition_key_params(partition_key: str | None,
partition_key_pattern: str | None) -> None:
+ """Validate partition_key and partition_key_pattern are not both set, and
that the pattern is valid regex."""
+ if partition_key is not None and partition_key_pattern is not None:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail={
+ "reason": "Mutually exclusive parameters",
+ "message": "partition_key and partition_key_pattern cannot
both be provided. "
+ "Use partition_key for exact match or partition_key_pattern
for regex filtering.",
+ },
+ )
+ if partition_key_pattern is None:
+ return
+ try:
+ re.compile(partition_key_pattern)
+ except re.error as e:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
Review Comment:
`partition_key_pattern` is validated with Python `re.compile()`, but the
filter is executed via the database regex operator. This can lead to false
negatives/positives across backends (e.g. DB-specific regex tokens rejected by
Python, or Python-valid patterns rejected by the DB). Consider aligning this
with the core API helper (and documenting the supported regex flavor), or
handling validation in a backend-aware way.
##########
airflow-core/docs/authoring-and-scheduling/assets.rst:
##########
@@ -547,5 +561,33 @@ including ``partition_key`` in the request body):
"partition_key": "us|2026-03-10T09:00:00"
}'
+You can also query asset events filtered by partition key using the REST API.
+Two parameters are available:
+
+- ``partition_key`` for **exact match** — uses the B-tree index for fast
lookups:
+
+.. code-block:: bash
+
+ curl
"http://<airflow-host>/api/v2/assets/events?partition_key=us|2026-03-10"
+
+- ``partition_key_pattern`` for **regex filtering** — uses database-native
regex
+ (PostgreSQL ``~`` operator, MySQL ``REGEXP``, SQLite ``re.match``):
+
+.. code-block:: bash
+
+ curl "http://<airflow-host>/api/v2/assets/events?partition_key_pattern=^us"
Review Comment:
The curl example includes an unescaped `|` character in the query string.
`|` is not a valid URL character per RFC 3986 and may be rejected by
proxies/clients unless percent-encoded. Consider updating the example to
URL-encode reserved characters (e.g. `%7C`, `%5E`) or use `curl
--data-urlencode` to avoid copy/paste failures.
```suggestion
curl
"http://<airflow-host>/api/v2/assets/events?partition_key=us%7C2026-03-10"
- ``partition_key_pattern`` for **regex filtering** — uses database-native
regex
(PostgreSQL ``~`` operator, MySQL ``REGEXP``, SQLite ``re.match``):
.. code-block:: bash
curl
"http://<airflow-host>/api/v2/assets/events?partition_key_pattern=%5Eus"
```
##########
asset-event-query-proposal.md:
##########
@@ -0,0 +1,250 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ -->
+
+# Proposal: Enhanced Asset Event Querying in Airflow — Towards Deprecating
Census
+
+**Author:** Hussein Awala
+**Date:** April 2026
+**Status:** In progress — community discussion pending
+**Target Release:** Apache Airflow 3.3
+
+---
+
+## TL;DR
+
+I'm contributing a set of features to Apache Airflow that will make **Asset
Events** a first-class mechanism for tracking dataset watermarks, metadata, and
cross-pipeline coordination — covering the core use cases currently served by
Census. Because Airflow is the **standard scheduler at Datadog**, building
these capabilities natively means every team already using Airflow gets them
for free, with better performance, simpler integrations, and no extra
infrastructure.
+
+This document describes my four-step plan, the current status of each step,
and how it maps to the Census deprecation effort.
+
+---
+
+## Context
+
+### Census today
+
+Census stores watermarks, metadata, and scope markers in **JSONB columns**
where every field — dataset name, partition key, extras — lives inside JSON
blobs. Even though GIN indexes exist, the dataset name itself is a JSONB field,
so queries cannot efficiently narrow down rows before scanning JSON content.
This design:
+
+- **Limits query performance** — filtering by dataset name still requires a
GIN index lookup into JSONB rather than a simple B-tree index scan on a
dedicated column, and combining it with partition or extra filters compounds
the cost.
+- **Requires a separate service** to run and operate alongside Airflow.
+- **Duplicates scheduling metadata** that Airflow already manages through
Asset Events.
+
+Airflow does support **XCom** for inter-task communication, and XCom is
already replacing some Census usage. However, reading an XCom value requires
knowing the exact `dag_id`, `run_id`, and `task_id` of the producer, which is
not always possible — and more importantly, it cannot serve as a **data
contract between teams**, since each team needs the flexibility to restructure
their DAGs and internal tasks without breaking consumers. Asset Events solve
this by decoupling the contract (asset name + partition key + extra metadata)
from the internal DAG structure.
+
+For background on how Census is used today and the roadmap discussion, see:
+
+- [Luigi — Using Census to track datasets execution
tokens](https://datadoghq.atlassian.net/wiki/spaces/adp/pages/2867364337/Luigi+-+Using+Census+to+track+datasets+execution+tokens)
+- [Future of Census — As of
2026.01](https://datadoghq.atlassian.net/wiki/spaces/adp/pages/5366547407/Future+of+Census+-+As+of+2026.01)
Review Comment:
This proposal doc contains company-specific statements and references to
private/internal URLs (e.g. datadoghq.atlassian.net). These links won't be
accessible to most contributors/users and are generally not appropriate to keep
in the Apache Airflow repo. Suggest removing the internal references/links,
moving this document to an external/public location, or rewriting it as an
Airflow-focused design doc with public references only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
