Add extra field filtering for asset events

airflow-core/docs/authoring-and-scheduling/assets.rst

@@ -247,6 +247,26 @@ Inlet asset events can be read with the ``inlet_events`` accessor in the executi
 
 Each value in the ``inlet_events`` mapping is a sequence-like object that orders past events of a given asset by ``timestamp``, earliest to latest. It supports most of Python's list interface, so you can use ``[-1]`` to access the last event, ``[-2:]`` for the last two, etc. The accessor is lazy and only hits the database when you access items inside it.
 
+The accessor also supports chaining methods to filter events before fetching them. For example, to retrieve only events where specific ``extra`` keys match given values:
+
+.. code-block:: python
+
+    @task(inlets=[regional_sales])
+    def process_high_priority(*, inlet_events):
+        events = inlet_events[regional_sales].extra("region", "us").extra("env", "prod")
+        for event in events:
+            print(event.extra)
+
+The ``.extra(key, value)`` method can be chained multiple times; all conditions are combined with AND logic. Other chaining methods include ``.after(timestamp)``, ``.before(timestamp)``, ``.ascending()``, and ``.limit(n)``.
+
+The REST API also supports filtering asset events by extra key-value pairs using the ``extra`` query parameter (repeat for multiple conditions):
+
+.. code-block:: bash
+
+    curl "http://<airflow-host>/api/v2/assets/events?extra=region%3Dus&extra=env%3Dprod"
+
+Each ``extra`` value uses ``key=value`` format. Multiple entries are combined with AND logic.
+
 Dependency between ``@asset``, ``@task``, and classic operators
 ---------------------------------------------------------------
 

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``d2f4e1b3c5a7`` (head) | ``9ff64e1c35d3`` | ``3.3.0``         | Add partition_date to asset_partition_dag_run.               |
+| ``5a5d3253e946`` (head) | ``d2f4e1b3c5a7`` | ``3.4.0``         | Add GIN index on asset_event.extra for PostgreSQL.           |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``d2f4e1b3c5a7``        | ``9ff64e1c35d3`` | ``3.3.0``         | Add partition_date to asset_partition_dag_run.               |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``9ff64e1c35d3``        | ``dd5f3a8e2b91`` | ``3.3.0``         | Add indexes on dag_run.created_dag_version_id and            |
 |                         |                  |                   | task_instance.dag_version_id.                                |

airflow-core/src/airflow/api_fastapi/common/parameters.py

@@ -68,6 +68,7 @@
 from airflow.models.variable import Variable
 from airflow.models.xcom import XComModel
 from airflow.typing_compat import Self
+from airflow.utils.sqlalchemy import JsonContains
 from airflow.utils.state import DagRunState, TaskInstanceState
 from airflow.utils.types import DagRunType
 
@@ -553,6 +554,60 @@ def depends_prefix_search(
     return depends_prefix_search
 
 
+class _JsonKVFilter(BaseParam[dict[str, str]]):
+    """
+    Filter on a JSON column by multiple key-value pairs (AND logic).
+
+    Uses dialect-aware SQL: ``@>`` (JSONB containment, GIN-indexable) on
+    PostgreSQL, ``JSON_CONTAINS`` on MySQL, and ``JSON_EXTRACT`` on SQLite.
+    """
+
+    def __init__(
+        self,
+        attribute: ColumnElement,
+        value: dict[str, str] | None = None,
+        skip_none: bool = True,
+    ) -> None:
+        super().__init__(skip_none=skip_none)
+        self.attribute: ColumnElement = attribute
+        self.value = value
+
+    def to_orm(self, select: Select) -> Select:
+        if not self.value:
+            return select
+        return select.where(JsonContains(self.attribute, self.value))
+
+    @classmethod
+    def depends(cls, *args: Any, **kwargs: Any) -> Self:
+        raise NotImplementedError("Use json_kv_filter_factory instead.")
+
+
+def json_kv_filter_factory(
+    attribute: ColumnElement,
+    param_name: str = "extra",
+) -> Callable[[list[str]], _JsonKVFilter]:
+    DESCRIPTION = (
+        "Filter by JSON key-value pairs. Repeat for multiple conditions (AND logic). "
+        "Format: key=value (e.g. extra=region=us&extra=env=prod)."
+    )
+
+    def depends_json_kv(
+        values: list[str] = Query(alias=param_name, default_factory=list, description=DESCRIPTION),
+    ) -> _JsonKVFilter:
+        kv_dict: dict[str, str] = {}
+        for item in values:
+            if "=" not in item:
+                raise HTTPException(
+                    status_code=HTTP_422_UNPROCESSABLE_CONTENT,
+                    detail=f"Invalid {param_name} parameter format: {item!r}. Expected 'key=value'.",
+                )
+            k, v = item.split("=", 1)
+            kv_dict[k] = v
+        return _JsonKVFilter(attribute, kv_dict or None)
+
+    return depends_json_kv
+
+
 class SortParam(BaseParam[list[str]]):
     """Order result by the attribute."""
 
@@ -1587,6 +1642,7 @@ def _transform_ti_states(states: list[str] | None) -> list[TaskInstanceState | N
 QueryAssetDagIdPatternSearch = Annotated[
     _DagIdAssetReferenceFilter, Depends(_DagIdAssetReferenceFilter.depends)
 ]
+QueryAssetEventExtraFilter = Annotated[_JsonKVFilter, Depends(json_kv_filter_factory(AssetEvent.extra))]
 QueryPartitionedDagRunHasCreatedDagRunIdFilter = Annotated[
     FilterParam[bool | None],
     Depends(

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -478,6 +478,18 @@ paths:
           \ stays index-compatible under locale-aware collations \u2014 e.g. `test_`\
           \ effectively matches items starting with `test`, and `s3://` matches items\
           \ starting with `s3`."
+      - name: extra
+        in: query
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
+          description: 'Filter by JSON key-value pairs. Repeat for multiple conditions
+            (AND logic). Format: key=value (e.g. extra=region=us&extra=env=prod).'
+          title: Extra
+        description: 'Filter by JSON key-value pairs. Repeat for multiple conditions
+          (AND logic). Format: key=value (e.g. extra=region=us&extra=env=prod).'
       - name: timestamp_gte
         in: query
         required: false

airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

@@ -37,6 +37,7 @@
     QueryAssetAliasNamePatternSearch,
     QueryAssetAliasNamePrefixPatternSearch,
     QueryAssetDagIdPatternSearch,
+    QueryAssetEventExtraFilter,
     QueryAssetNamePatternSearch,
     QueryAssetNamePrefixPatternSearch,
     QueryLimit,
@@ -321,6 +322,7 @@ def get_asset_events(
     ],
     name_pattern: QueryAssetNamePatternSearch,
     name_prefix_pattern: QueryAssetNamePrefixPatternSearch,
+    extra_filter: QueryAssetEventExtraFilter,
     timestamp_range: Annotated[RangeFilter, Depends(datetime_range_filter_factory("timestamp", AssetEvent))],
     session: SessionDep,
 ) -> AssetEventCollectionResponse:
@@ -339,6 +341,7 @@ def get_asset_events(
             source_map_index,
             name_pattern,
             name_prefix_pattern,
+            extra_filter,
             timestamp_range,
         ],
         order_by=order_by,

airflow-core/src/airflow/api_fastapi/execution_api/routes/asset_events.py

@@ -30,6 +30,7 @@
     AssetEventsResponse,
 )
 from airflow.models.asset import AssetAliasModel, AssetEvent, AssetModel
+from airflow.utils.sqlalchemy import JsonContains
 
 router = APIRouter(
     responses={
@@ -44,7 +45,7 @@ def _get_asset_events_through_sql_clauses(
 ) -> AssetEventsResponse:
     order_by_clause = AssetEvent.timestamp.asc() if ascending else AssetEvent.timestamp.desc()
     asset_events_query = select(AssetEvent).join(join_clause).where(where_clause).order_by(order_by_clause)
-    if limit:
+    if limit is not None:
         asset_events_query = asset_events_query.limit(limit)
     asset_events = session.scalars(asset_events_query)
     return AssetEventsResponse.model_validate(
@@ -73,6 +74,23 @@ def _get_asset_events_through_sql_clauses(
     )
 
 
+def _parse_extra_params(extra: list[str] | None) -> dict[str, str]:
+    """Parse repeated ``key=value`` query params into a dict."""
+    result: dict[str, str] = {}
+    for item in extra or []:
+        if "=" not in item:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail={
+                    "reason": "Invalid parameter",
+                    "message": f"Invalid extra parameter format: {item!r}. Expected 'key=value'.",
+                },
+            )
+        k, v = item.split("=", 1)
+        result[k] = v
+    return result
+
+
 @router.get("/by-asset")
 def get_asset_event_by_asset_name_uri(
     name: Annotated[str | None, Query(description="The name of the Asset")],
@@ -82,6 +100,12 @@ def get_asset_event_by_asset_name_uri(
     before: Annotated[UtcDateTime | None, Query(description="The end of the time range")] = None,
     ascending: Annotated[bool, Query(description="Whether to sort results in ascending order")] = True,
     limit: Annotated[int | None, Query(description="The maximum number of results to return")] = None,
+    extra: Annotated[
+        list[str] | None,
+        Query(
+            description="Filter by extra JSON key-value pairs. Format: key=value. Repeat for AND logic.",
+        ),
+    ] = None,
 ) -> AssetEventsResponse:
     if name and uri:
         where_clause = and_(AssetModel.name == name, AssetModel.uri == uri)
@@ -102,6 +126,9 @@ def get_asset_event_by_asset_name_uri(
         where_clause = and_(where_clause, AssetEvent.timestamp >= after)
     if before:
         where_clause = and_(where_clause, AssetEvent.timestamp <= before)
+    extra_dict = _parse_extra_params(extra)
+    if extra_dict:
+        where_clause = and_(where_clause, JsonContains(AssetEvent.extra, extra_dict))
 
     return _get_asset_events_through_sql_clauses(
         join_clause=AssetEvent.asset,
@@ -120,12 +147,21 @@ def get_asset_event_by_asset_alias(
     before: Annotated[UtcDateTime | None, Query(description="The end of the time range")] = None,
     ascending: Annotated[bool, Query(description="Whether to sort results in ascending order")] = True,
     limit: Annotated[int | None, Query(description="The maximum number of results to return")] = None,
+    extra: Annotated[
+        list[str] | None,
+        Query(
+            description="Filter by extra JSON key-value pairs. Format: key=value. Repeat for AND logic.",
+        ),
+    ] = None,
 ) -> AssetEventsResponse:
     where_clause = AssetAliasModel.name == name
     if after:
         where_clause = and_(where_clause, AssetEvent.timestamp >= after)
     if before:
         where_clause = and_(where_clause, AssetEvent.timestamp <= before)
+    extra_dict = _parse_extra_params(extra)
+    if extra_dict:
+        where_clause = and_(where_clause, JsonContains(AssetEvent.extra, extra_dict))
 
     return _get_asset_events_through_sql_clauses(
         join_clause=AssetEvent.source_aliases,

airflow-core/src/airflow/migrations/env.py

@@ -27,6 +27,12 @@
 from airflow import models, settings
 from airflow.utils.db import compare_server_default, compare_type
 
+_POSTGRES_ONLY_INDEXES = frozenset(
+    {
+        "idx_asset_event_extra_gin",
+    }
+)
+
 
 def include_object(_, name, type_, *args):
     """Filter objects for autogenerating revisions."""
@@ -36,6 +42,10 @@ def include_object(_, name, type_, *args):
     # Only create migrations for objects that are in the target metadata
     if type_ == "table" and name not in target_metadata.tables:
         return False
+    # Indexes created by raw SQL in migrations (e.g. Postgres GIN) are not
+    # represented in the SQLAlchemy model; hide them from autogenerate.
+    if type_ == "index" and name in _POSTGRES_ONLY_INDEXES:
+        return False
     return True
 
 

airflow-core/src/airflow/migrations/versions/0124_3_4_0_add_gin_index_on_asset_event_extra.py

@@ -0,0 +1,57 @@
+#
+# 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.
+
+"""
+Add GIN index on asset_event.extra for PostgreSQL.
+
+Revision ID: 5a5d3253e946
+Revises: d2f4e1b3c5a7
+Create Date: 2026-04-01 23:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+from alembic import op
+from sqlalchemy import text
+
+# revision identifiers, used by Alembic.
+revision = "5a5d3253e946"
+down_revision = "d2f4e1b3c5a7"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+    """Add GIN index on asset_event.extra for PostgreSQL only."""
+    conn = op.get_bind()
+    if conn.dialect.name == "postgresql":
+        op.execute(
+            text(
+                "CREATE INDEX IF NOT EXISTS idx_asset_event_extra_gin "
+                "ON asset_event USING GIN ((extra::jsonb) jsonb_ops)"
+            )
+        )
+
+
+def downgrade():
+    """Remove GIN index on asset_event.extra."""
+    conn = op.get_bind()
+    if conn.dialect.name == "postgresql":
+        op.execute(text("DROP INDEX IF EXISTS idx_asset_event_extra_gin"))

airflow-core/src/airflow/ui/openapi-gen/queries/common.ts

@@ -36,8 +36,9 @@ export const UseAssetServiceGetAssetAliasKeyFn = ({ assetAliasId }: {
 export type AssetServiceGetAssetEventsDefaultResponse = Awaited<ReturnType<typeof AssetService.getAssetEvents>>;
 export type AssetServiceGetAssetEventsQueryResult<TData = AssetServiceGetAssetEventsDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useAssetServiceGetAssetEventsKey = "AssetServiceGetAssetEvents";
-export const UseAssetServiceGetAssetEventsKeyFn = ({ assetId, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }: {
+export const UseAssetServiceGetAssetEventsKeyFn = ({ assetId, extra, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }: {
   assetId?: number;
+  extra?: string[];
   limit?: number;
   namePattern?: string;
   namePrefixPattern?: string;
@@ -51,7 +52,7 @@ export const UseAssetServiceGetAssetEventsKeyFn = ({ assetId, limit, namePattern
   timestampGte?: string;
   timestampLt?: string;
   timestampLte?: string;
-} = {}, queryKey?: Array<unknown>) => [useAssetServiceGetAssetEventsKey, ...(queryKey ?? [{ assetId, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }])];
+} = {}, queryKey?: Array<unknown>) => [useAssetServiceGetAssetEventsKey, ...(queryKey ?? [{ assetId, extra, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }])];
 export type AssetServiceGetAssetQueuedEventsDefaultResponse = Awaited<ReturnType<typeof AssetService.getAssetQueuedEvents>>;
 export type AssetServiceGetAssetQueuedEventsQueryResult<TData = AssetServiceGetAssetQueuedEventsDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useAssetServiceGetAssetQueuedEventsKey = "AssetServiceGetAssetQueuedEvents";

airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts

@@ -83,15 +83,17 @@ export const ensureUseAssetServiceGetAssetAliasData = (queryClient: QueryClient,
 *
 * **Performance note:** this full-match pattern is evaluated as ``ILIKE '%term%'`` and most of the time prevents the database from using B-tree indexes, which can be very slow on large tables. Prefer the equivalent ``name_prefix_pattern`` parameter when possible.
 * @param data.namePrefixPattern Prefix match — returns items whose value starts with the given string (case-sensitive, index-friendly). Use the pipe `|` operator for OR logic (e.g. `dag1|dag2`). Use `~` to match all. Wildcard characters (`%`, `_`) are treated as literal characters. Trailing non-alphanumeric characters in the prefix are stripped before matching so the range scan stays index-compatible under locale-aware collations — e.g. `test_` effectively matches items starting with `test`, and `s3://` matches items starting with `s3`.
+* @param data.extra Filter by JSON key-value pairs. Repeat for multiple conditions (AND logic). Format: key=value (e.g. extra=region=us&extra=env=prod).
 * @param data.timestampGte
 * @param data.timestampGt
 * @param data.timestampLte
 * @param data.timestampLt
 * @returns AssetEventCollectionResponse Successful Response
 * @throws ApiError
 */
-export const ensureUseAssetServiceGetAssetEventsData = (queryClient: QueryClient, { assetId, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }: {
+export const ensureUseAssetServiceGetAssetEventsData = (queryClient: QueryClient, { assetId, extra, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }: {
   assetId?: number;
+  extra?: string[];
   limit?: number;
   namePattern?: string;
   namePrefixPattern?: string;
@@ -105,7 +107,7 @@ export const ensureUseAssetServiceGetAssetEventsData = (queryClient: QueryClient
   timestampGte?: string;
   timestampLt?: string;
   timestampLte?: string;
-} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseAssetServiceGetAssetEventsKeyFn({ assetId, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }), queryFn: () => AssetService.getAssetEvents({ assetId, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }) });
+} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseAssetServiceGetAssetEventsKeyFn({ assetId, extra, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }), queryFn: () => AssetService.getAssetEvents({ assetId, extra, limit, namePattern, namePrefixPattern, offset, orderBy, sourceDagId, sourceMapIndex, sourceRunId, sourceTaskId, timestampGt, timestampGte, timestampLt, timestampLte }) });
 /**
 * Get Asset Queued Events
 * Get queued asset events for an asset.