Add extra field filtering for asset events

[hussein-awala]

Summary

Add the ability to filter asset events by their extra JSON field using key-value pairs across the full stack (Core REST API, Execution API, Task SDK).

Changes

REST API (Core)

• New extra repeated query parameter on the /assets/events endpoint (e.g. ?extra=region=us&extra=env=prod).
• Multiple extra params are combined with AND logic.
• Uses a _JsonKVFilter parameter class backed by a dialect-aware JsonContains SQL construct.

Execution API

• Both /by-asset and /by-asset-alias endpoints accept extra as a repeated key=value query parameter.

Task SDK

• InletEventsAccessor.extra(key, value) is chainable — each call accumulates into a dict for multi-condition filtering:

  inlet_events[my_asset].extra("region", "us").extra("env", "prod")

• Updated comms models (GetAssetEventByAsset, GetAssetEventByAssetAlias) with extra: dict[str, str] | None.
• Updated SDK client to serialize the dict as repeated extra=key=value query params.
• Updated supervisor to forward the extra dict to the client.

Dialect-aware JSON containment (JsonContains)

Compiles to database-native JSON containment on each backend:

• PostgreSQL — @> JSONB containment (GIN-indexable).
• MySQL — JSON_CONTAINS.
• SQLite — per-key json_extract comparisons.

All dialects use bound parameters to avoid SQL injection.

Database

• New PostgreSQL-only GIN index migration on asset_event.extra (extra::jsonb) for efficient containment queries. The index is created via raw SQL and excluded from non-PostgreSQL backends (handled in env.py and the schema-sync test).

Documentation

• Updated assets.rst with examples for both Task SDK chaining and REST API usage.

Test plan

• Core API: parametrized tests for various key-value combinations and combined filters.
• Execution API: tests for both by-asset and by-alias endpoints with extra filtering.
• SDK client: mock transport tests for dict serialization and alias routing.
• Supervisor: updated request test cases and added new extra-filter cases.

[Copilot]

Pull request overview

Adds support for filtering asset events by extra JSON key/value pairs across the Airflow stack (Core REST API, Execution API, Task SDK), including client serialization, docs, and test coverage, plus a Postgres index intended to speed up containment queries.

Changes:

• Introduces repeated extra=key=value query filtering for asset events in Core REST API and Execution API endpoints.
• Extends Task SDK inlet event accessors and client/supervisor plumbing to forward extra filters.
• Adds a new Postgres GIN index migration for asset_event.extra, updates OpenAPI artifacts and documentation, and expands unit tests.

Reviewed changes

Copilot reviewed 30 out of 30 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
task-sdk/tests/task_sdk/execution_time/test_supervisor.py	Updates supervisor request expectations to include extra, adds new request cases.
task-sdk/tests/task_sdk/execution_time/test_context.py	Adds accessor chaining tests for .extra(key, value) filtering.
task-sdk/tests/task_sdk/api/test_client.py	Adds tests verifying dict-to-repeated-query-param serialization for extra.
task-sdk/src/airflow/sdk/execution_time/supervisor.py	Forwards extra from supervisor messages to SDK client calls.
task-sdk/src/airflow/sdk/execution_time/context.py	Adds .extra() chaining on InletEventsAccessor and forwards extra to supervisor message.
task-sdk/src/airflow/sdk/execution_time/comms.py	Adds `extra: dict[str, str]
task-sdk/src/airflow/sdk/api/client.py	Serializes extra dict to repeated extra=key=value query parameters.
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_events.py	Adds Execution API tests for extra filtering for both by-asset and by-alias endpoints.
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py	Adds Core REST API tests for extra filtering and combinations.
airflow-core/src/airflow/utils/sqlalchemy.py	Adds JsonContains custom SQL construct for dialect-aware JSON containment.
airflow-core/src/airflow/utils/db.py	Updates migration head mapping for Airflow 3.3.0.
airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts	Updates generated UI request types to include extra and adjusts ValidationError shape.
airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts	Wires extra through generated UI service call parameters.
airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts	Updates generated UI schemas (including ValidationError changes).
airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts	Adds extra to generated suspense query hook signature/key.
airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts	Adds extra to generated query hook signature/key.
airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts	Adds extra to generated prefetch helper signature/key.
airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts	Adds extra to generated ensureQueryData helper signature/key.
airflow-core/src/airflow/ui/openapi-gen/queries/common.ts	Adds extra into the generated query-key function inputs.
airflow-core/src/airflow/migrations/versions/0111_3_3_0_add_gin_index_on_asset_event_extra.py	Adds a Postgres-only GIN index migration for asset_event.extra.
airflow-core/src/airflow/api_fastapi/execution_api/routes/asset_events.py	Adds extra filtering to Execution API asset-event endpoints.
airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py	Adds extra filter dependency to Core REST /assets/events.
airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml	Adds extra query parameter to OpenAPI and updates ValidationError schema.
airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml	Updates private UI OpenAPI components (including ValidationError changes).
airflow-core/src/airflow/api_fastapi/common/parameters.py	Adds reusable _JsonKVFilter and json_kv_filter_factory for JSON KV filtering.
airflow-core/src/airflow/api_fastapi/auth/managers/simple/ui/openapi-gen/requests/types.gen.ts	Updates generated auth-manager UI request types (ValidationError changes).
airflow-core/src/airflow/api_fastapi/auth/managers/simple/ui/openapi-gen/requests/schemas.gen.ts	Updates generated auth-manager UI schemas (ValidationError changes).
airflow-core/src/airflow/api_fastapi/auth/managers/simple/openapi/v2-simple-auth-manager-generated.yaml	Updates generated auth-manager OpenAPI components (ValidationError changes).
airflow-core/docs/migrations-ref.rst	Documents the new migration as the head revision.
airflow-core/docs/authoring-and-scheduling/assets.rst	Documents Task SDK chaining and REST API usage for extra filtering.

Comments suppressed due to low confidence (1)

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

• The /by-asset-alias endpoint applies extra filtering via AssetEvent.extra[k].as_string() == v, while /by-asset uses the new dialect-aware JsonContains. This creates inconsistent behavior across DB backends and duplicates parsing logic. Recommend reusing _parse_extra_params() here and applying a single JsonContains(AssetEvent.extra, extra_dict) condition (like /by-asset) for consistent cross-dialect SQL and index usage on Postgres.

    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)
    for item in extra:
        if "=" in item:
            k, v = item.split("=", 1)
            where_clause = and_(where_clause, AssetEvent.extra[k].as_string() == v)

    return _get_asset_events_through_sql_clauses(
        join_clause=AssetEvent.source_aliases,
        where_clause=where_clause,
        session=session,
        ascending=ascending,
        limit=limit,
    )

[Copilot]

Pull request overview

Copilot reviewed 30 out of 30 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 30 out of 30 changed files in this pull request and generated 3 comments.

[pierrejeambrun]

Looking good overall.

[choo121600]

overall looks good. It just seems to need a rebase.

[pierrejeambrun]

Thanks! LGTM.

We might one another pair of eyes on this since the change set is big.