feat: add event filtering framework with pre-deserialization MCL optimization

[sgomezvillamor]

Overview

This PR introduces a comprehensive event filtering framework for DataHub Actions pipelines, with a special optimization for MetadataChangeLog (MCL) events that enables pre-deserialization filtering to reduce processing overhead.

Key Changes

1. New Filter Framework

• Filter base class (datahub_actions/filter/filter.py): Abstract base for all pipeline filters
• EventTypeFilter (datahub_actions/plugin/filter/event_type_filter.py): Pluggable filter implementation that matches events by type and body predicates with OR/AND semantics
• Filter registry and utilities: filter_registry.py, match_util.py, filter_stats.py for filter discovery and matching logic
• Pipeline integration: Filters are now evaluated before transformers with AND semantics across multiple filters

2. MCL Pre-Deserialization Optimization

• Added enable_mcl_pre_deserialization_filter config flag to KafkaEventSourceConfig
• Implemented set_filters() method in KafkaEventSource to extract filter criteria for MCL events
• MCL events can be dropped before expensive avrogen deserialization by checking top-level Avro fields (entityType, aspectName, entityUrn, changeType)
• Added metrics tracking (MCL_EARLY_FILTER_METRIC) for pre-deserialization filter results

3. Configuration Updates

• New filters: section in pipeline YAML (replaces deprecated filter: section)
• Supports multiple filters with pluggable types
• Updated example configs in docker/datahub-actions/config/ to use new filter syntax
• Added deprecation warning for old filter: section

4. Documentation

• Updated docs/actions/README.md with new filter configuration examples
• Updated docs/actions/concepts.md to document the Filter component and its role in pipelines
• Clarified filter execution order (before transformers)

5. Testing

• Comprehensive unit tests for EventTypeFilter matching logic (tests/unit/plugin/filter/test_event_type_filter.py)
• Integration tests for filter pipeline behavior (tests/unit/filter/test_filter_pipeline_integration.py)
• Extended Kafka source tests with pre-deserialization filter scenarios (tests/unit/plugin/source/kafka/test_kafka_source.py)

Design Rationale

Why pre-deserialization filtering for MCL only?

• MCL events have scalar fields (entityType, aspectName, etc.) accessible at the Avro message level without deserialization
• EntityChangeEvent (ECE) events arrive JSON-encoded inside a PlatformEvent envelope, requiring full deserialization to inspect
• This optimization provides significant performance gains for MCL-heavy pipelines with selective filtering

Filter semantics:

• Multiple filters: AND (event must pass all filters)
• Multiple event types in a filter: OR (event type must match any listed type)
• Multiple body predicates: OR (event body must satisfy any predicate)
• Within a predicate: AND (all key/value pairs must match)

Breaking Changes

The deprecated filter: section in pipeline YAML is replaced by filters: (list-based). Old configs will emit a deprecation warning but continue to work via the FilterTransformer.

Test Coverage

• Unit tests for EventTypeFilter covering type matching, body predicates, and predicate combinations
• Integration tests verifying filter execution order and interaction with transformers
• Kafka source tests validating pre-deserialization filter behavior and metrics
• All existing tests pass with new filter framework integrated

https://claude.ai/code/session_018Pohzav3u7c3dLpZG2yeKV

[github-actions]

Linear: ING-2765

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ All tests successful. No failed tests found.

📢 Thoughts on this report? Let us know!

[sgomezvillamor]

✅ Local Testing & Metrics Validation

I've tested this PR locally with Prometheus metrics enabled and can confirm the new MCL pre-deserialization filter feature is working as expected.

Test Setup

• Generated MCL traffic with datahub docker ingest-sample-data
• Triggered executions with available system sources
• Enabled Prometheus metrics on datahub-actions (port 8000)

Metrics Results

Accumulated totals across running pipelines:

Pipeline	Passed	Rejected	Entirely Rejected	Total MCL Filtered	Total Messages*
ingestion_executor	8	471	0	479	728
datahub_doc_propagation_action	0	473	0	473	722
TOTAL	8	944	0	952	1,450

* Total Messages includes both MCL and PlatformEvent messages

Filter Efficiency

Pipeline	Efficiency (Rejected/Total)	Performance Benefit
ingestion_executor	471/479 = 98.3%	98.3% of MCL events avoided avrogen deserialization
datahub_doc_propagation_action	473/473 = 100%	All MCL events avoided avrogen deserialization
Overall	944/952 = 99.2%	🎯 99.2% optimization rate

Key Observations

1. New metric is working correctly: kafka_mcl_early_filter_total is being incremented with proper labels (pipeline_name, result)
2. Filter results are accurate:
   • passed: MCL events matching filter criteria (8 execution requests)
   • rejected: MCL events filtered before avrogen (JSON) deserialization (944 events)
   • entirely_rejected: Not applicable (both pipelines accept some MCL event types)
3. Significant performance benefit: 99.2% of MCL events were rejected before expensive avrogen (JSON) deserialization, demonstrating the optimization's effectiveness

Configuration Verified

Both test pipelines have enable_mcl_pre_deserialization_filter: true configured and are using the new filters: section with event_type filters as documented.

The feature is production-ready from a metrics and functionality perspective! 🚀

[skrydal]

It seems that performance improvement is functionally equal to the old code, why would we want to disable it? Moreover - disable it by default. I think this would contribute to the over-complication of flags. I propose we enable the feature universally and not introduce the flag at all.

[sgomezvillamor]

Agreed - enabled by default in commit 39b0c7c

Having the flag for optional optimizations is still good practice though - it allows operators to:

• Disable if they encounter edge cases we didn't anticipate
• A/B test performance impact in production
• Gradually roll out optimizations with confidence

The flag provides a safety valve without requiring code changes or deployments.

[skrydal]

Two filter execution paths run in parallel: the deprecated filter: becomes a FilterTransformer here, while filters: runs through the new framework. Since match_util
already extracted the shared matching logic, the deprecated path could be auto-translated into a FilterSpec(type="event_type", ...) and pushed onto filters — removing FilterTransformer
and collapsing to a single code path. Backward-compatible, ~100 fewer lines, and the legacy config would even benefit from MCL pre-deserialization for free.

[sgomezvillamor]

For now, we warn it as deprecated and let users migrate at their own pace. An automatic forward migration can be considered when we actually remove FilterTransform.

Why not do it now? IMO FilterTransform is fundamentally flawed — it only covers one event type, while actions can consume multiple. I'd rather users migrate in a controlled way, which also nudges them to rethink their filters — many of which probably belong in the action itself, not formalized as a filter transform at all.

Is that a blocking concern?

[skrydal]

Why can't we list the predicates directly, without key event?

[sgomezvillamor]

Initially this was shaped by the legacy filter transformer, which had:

event_type: str
event: Dict[str, Any]

and we just promoted event_type as a top-level key.

At some point I considered removing the event key as you suggest now, but kept it as a good practice, event it's the key for pattern matching constraints, leaving room to extend the spec with other constraint types in the future.

[skrydal]

I am confused about new filters structure. What would be the use-case for filters have more than one item?

[sgomezvillamor]

Filters with a clear, single responsibility are more reusable and composable.
It's the same pattern than transforms.

Currently, an event must match all filters: AND behavior.
In the future, we could add an OR mode at the pipeline level, but that's not needed yet, so not included.

[skrydal]

I understand, but considering an event is always of one type (is it?), then why would we want to have several filters? We still can list explicitly the field combinations via event list, right?
It would make sense if filters were OR, unless I do not understand something.

[sgomezvillamor]

A list of filters doesn't seem necessary now because we only have one filter implementation (EventTypeFilter), so concatenating multiple instances would be unnecessary. Indeed.

However, users can already register and use custom filters via the plugin registry, and we may add new filter implementations in the future. In that case, having a collection of filters becomes essential.

Also, as said before, the AND semantics across filters (event must pass ALL filters) enables powerful composition patterns - like combining EventTypeFilter with eg future RateLimitFilter or WhateverFancyFilter implementations.

[skrydal]

I haven't confirmed it practically, but Claude insists that the optimization path does not handle correct below case:

The minimal failing case is one EventTypeFilter, two OR predicates, where the predicates don't have the same set of keys:

  filters:
    - type: event_type
      config:
        filter:
          MetadataChangeLogEvent_v1:
            event:
              - entityType: dataset                          # predicate 1 — no aspectName constraint
              - entityType: chart
                aspectName: documentation                    # predicate 2 — has aspectName

Pipeline filter (per EventTypeFilter.matches): pass if (entityType=dataset) OR (entityType=chart AND aspectName=documentation). So an MCL with entityType=dataset, aspectName=schemaMetadata
passes.

Now trace set_filters merge on mcl_predicates = [{entityType: dataset}, {entityType: chart, aspectName: documentation}]:

Step	early_criteria after
predicate 1: entityType=dataset	{entityType: dataset}
predicate 2: entityType=chart → conflicts, pop	{}
predicate 2: aspectName=documentation → new key, add	{aspectName: documentation}

So the early filter ends up requiring aspectName=documentation. The MCL above (entityType=dataset, aspectName=schemaMetadata) gets early-rejected before deserialization — even though the
pipeline filter would have passed it. The Action never sees the event.

[sgomezvillamor]

@skrydal Excellent catch!

You're absolutely right. The conservative approach of the early filtering was not properly implemented. The bug was in the merge logic attempting to combine OR predicates using AND semantics.

The Issue:
The code tried to merge [{entityType: dataset}, {entityType: chart, aspectName: documentation}] into a single criteria dict, which resulted in {aspectName: documentation} after conflict resolution. This incorrectly rejected events that matched the first predicate.

The Fix:
We now maintain a list of criteria (one per predicate) and implement correct OR semantics:

• Pass if ANY criteria matches (OR across predicates)
• Each criteria requires ALL its fields to match (AND within predicate)
• Conservative by design: We only filter out events we can definitively reject. If resolving predicates requires deserialization (nested fields like aspect.value.executorId), we let the event pass through to be properly filtered after deserialization.
• Optimization is disabled if ANY predicate has no extractable fields (since we can't safely reject anything due to OR semantics).

Fix and comprehensive tests added in commit 191dc0c