This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 664e2a8b579 Fix asset event extra filter matching wrong events for
dotted keys on SQLite (#69675)
664e2a8b579 is described below
commit 664e2a8b579884e24e35f7400d788ad9983d7a02
Author: Steve Ahn <[email protected]>
AuthorDate: Fri Jul 31 12:35:08 2026 -0700
Fix asset event extra filter matching wrong events for dotted keys on
SQLite (#69675)
---
airflow-core/newsfragments/69675.bugfix.rst | 1 +
airflow-core/src/airflow/utils/sqlalchemy.py | 4 +-
.../core_api/routes/public/test_assets.py | 65 ++++++++++++++++++++++
3 files changed, 69 insertions(+), 1 deletion(-)
diff --git a/airflow-core/newsfragments/69675.bugfix.rst
b/airflow-core/newsfragments/69675.bugfix.rst
new file mode 100644
index 00000000000..38ca8fd6068
--- /dev/null
+++ b/airflow-core/newsfragments/69675.bugfix.rst
@@ -0,0 +1 @@
+Fix asset event ``extra`` filtering silently matching the wrong events on
SQLite when a filter key contains JSON-path metacharacters such as dots
diff --git a/airflow-core/src/airflow/utils/sqlalchemy.py
b/airflow-core/src/airflow/utils/sqlalchemy.py
index 0064c0ec3c8..61ac586bda6 100644
--- a/airflow-core/src/airflow/utils/sqlalchemy.py
+++ b/airflow-core/src/airflow/utils/sqlalchemy.py
@@ -229,7 +229,9 @@ def _default_json_contains(element, compiler, **kw):
clauses = []
for k, v in element.kv_dict.items():
- path = f"$.{k}"
+ # Quote the key (json.dumps also covers embedded quotes/backslashes)
so metacharacters
+ # like "." match literally, as PostgreSQL ``@>`` and MySQL
``JSON_CONTAINS`` do.
+ path = f"$.{json.dumps(k, ensure_ascii=False)}"
clauses.append(func.json_extract(element.column, literal(path)) ==
literal(v))
if len(clauses) == 1:
return compiler.process(clauses[0], **kw)
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index c8e6b65ccaf..f0e60462bf3 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
@@ -1324,6 +1324,71 @@ class TestGetAssetEventsExtraFilter(TestAssets):
assert response.json()["total_entries"] == expected_count
+class TestGetAssetEventsExtraFilterSpecialKeys(TestAssets):
+ """
+ Keys containing JSON-path metacharacters must be matched literally on
every backend.
+
+ PostgreSQL (``@>``) and MySQL (``JSON_CONTAINS``) compare keys literally
by containment.
+ The SQLite fallback builds a ``json_extract`` path from the key, where an
unquoted ``.``
+ or ``[`` is interpreted as path navigation instead — silently missing
literal dotted keys
+ and wrongly matching nested objects.
+ """
+
+ @pytest.fixture
+ def _setup(self, session):
+ self.create_assets(num=1, session=session)
+ events = [
+ AssetEvent(
+ asset_id=1,
+ extra={"spark.executor.memory": "4g"},
+ source_task_id="t1",
+ source_dag_id="d1",
+ source_run_id="r1",
+ timestamp=DEFAULT_DATE,
+ ),
+ AssetEvent(
+ asset_id=1,
+ extra={"spark": {"executor": {"memory": "4g"}}},
+ source_task_id="t1",
+ source_dag_id="d1",
+ source_run_id="r2",
+ timestamp=DEFAULT_DATE,
+ ),
+ AssetEvent(
+ asset_id=1,
+ extra={"partitions[0]": "2024-01-01"},
+ source_task_id="t1",
+ source_dag_id="d1",
+ source_run_id="r3",
+ timestamp=DEFAULT_DATE,
+ ),
+ ]
+ session.add_all(events)
+ session.commit()
+
+ @pytest.mark.usefixtures("_setup")
+ @pytest.mark.parametrize(
+ ("params", "expected_count"),
+ [
+ # Matches only the event whose extra has the literal dotted key,
+ # not the one nesting the same path as objects.
+ ({"extra": "spark.executor.memory=4g"}, 1),
+ ({"extra": "partitions[0]=2024-01-01"}, 1),
+ ({"extra": "spark.executor.memory=8g"}, 0),
+ ],
+ )
+ def test_extra_filter_metacharacter_keys_match_literally(self,
test_client, params, expected_count):
+ response = test_client.get("/assets/events", params=params)
+ assert response.status_code == 200
+ assert response.json()["total_entries"] == expected_count
+
+ @pytest.mark.usefixtures("_setup")
+ def test_extra_filter_dotted_key_matches_the_literal_key_event(self,
test_client):
+ response = test_client.get("/assets/events", params={"extra":
"spark.executor.memory=4g"})
+ assert response.status_code == 200
+ assert [e["source_run_id"] for e in response.json()["asset_events"]]
== ["r1"]
+
+
class TestGetAssetEndpoint(TestAssets):
@provide_session
def test_should_respond_200(self, test_client, *, session):
