refactor(ingestion): workunit processors

[sgomezvillamor]

Summary

This PR standardizes workunit processing across all DataHub ingestion sources by introducing a common processor interface with automatic discovery and pluggable observability.

Main Contributions

1. Standardized Processor Interface

All processors now inherit from a common WorkunitProcessor base class with:

• Consistent lifecycle (create(), should_enable(), process())
• Dependency injection via WorkunitProcessorContext
• Automatic report registration

2. Pluggable Sub-Reports for Observability

Each processor can define its own typed report class for metrics:

@dataclass
class ValidateInputFieldsProcessorReport(WorkunitProcessorReport):
    num_input_fields_filtered: int = 0
    num_workunits_with_invalid_fields: int = 0

class ValidateInputFieldsProcessor(WorkunitProcessor[ValidateInputFieldsProcessorReport]):
    def process(self, stream):
        self.report.num_input_fields_filtered += 1  # Properly typed!

3. Automatic Processor Discovery

Eliminated ~72 source overrides of get_workunit_processors(). Sources no longer need to manually instantiate processors - they're auto-discovered and enabled based on:

• should_enable() conditions (e.g., config flags)
• Explicit allow/exclude lists when needed

Before:

def get_workunit_processors(self):
    return [
        functools.partial(auto_status_aspect, self.ctx, self.report),
        functools.partial(auto_browse_path_v2, self.ctx, self.report),
        # ... manual wiring for each processor
    ]

After:

# Nothing needed - processors auto-discovered!
# Or opt-out when needed:
def get_excluded_workunit_processors(self):
    return [AutoIncrementalLineageProcessor]  # Type-safe

Observability

Startup logs show which processors are enabled:

[2026-06-11 17:42:50,191] INFO - Workunit processor 'AutoStatusAspectProcessor' enabled
[2026-06-11 17:42:50,191] INFO - Workunit processor 'ValidateInputFieldsProcessor' enabled
[2026-06-11 17:42:50,191] INFO - Workunit processor 'EnsureAspectSizeProcessor' enabled
[2026-06-11 17:42:50,191] INFO - Workunit processor 'AutoIncrementalLineageProcessor' disabled by should_enable()

Report structure with processor-specific metrics:

 'workunit_processor_reports': {'AutoStatusAspectProcessor': {'num_errors': 0, 'num_status_aspects_emitted': 48},
                                'AutoMaterializeReferencedTagsTermsProcessor': {'num_exceptions': 0},
                                'ValidateDuplicateSchemaFieldPathsProcessor': {'total_schema_aspects': 25,
                                                                               'schemas_with_duplicates': 0,
                                                                               'duplicated_field_paths': 0},
                                'ValidateEmptySchemaFieldPathsProcessor': {'total_schema_aspects': 25,
                                                                           'schemas_with_empty_fields': 0,
                                                                           'empty_field_paths': 0},
                                'AutoBrowsePathV2Processor': {'num_out_of_batch': 20,
                                                              'num_out_of_order': 6,
                                                              'num_browse_path_v2_emitted': 28,
                                                              'num_container_or_legacy_emitted': 0,
                                                              'num_fallback_emitted': 0},
                                'AutoWorkunitsReporterProcessor': {},
                                'AutoPatchLastModifiedProcessor': {'num_patches_emitted': 0},
                                'ValidateInputFieldsProcessor': {'num_input_fields_filtered': 0,
                                                                 'num_workunits_with_invalid_fields': 0,
                                                                 'num_workunits_skipped_entirely': 0},
                                'EnsureAspectSizeProcessor': {'num_truncations_by_aspect': {}},
                                'AutoStaleEntityRemovalProcessor': {}},

Architecture

New module structure:

• datahub.ingestion.api.workunit_processor - Base class and context
• datahub.ingestion.workunit_processors/ - All processor implementations:
  • auto_* - Enrichment processors (add metadata)
  • validate_* - Validation processors (remove invalid data)
  • ensure_* - Enforcement processors (enforce constraints)

Processor lifecycle:

1. Auto-discover all processors from registry
2. Filter by should_enable() conditions
3. Apply source allow/exclude lists
4. Instantiate via create() with dependency injection
5. Register reports with source report
6. Stream workunits through process() pipeline

Implementation Details

• Generic typing for type-safe report access (WorkunitProcessor[ReportT])
• Runtime type extraction via existing get_class_from_annotation() utility
• Naming convention documented in base class docstring
• Error handling: log + raise to halt ingestion on processor failures
• Breaking changes documented in docs/how/updating-datahub.md

Testing

• All existing tests updated for new processor API
• Fixed timezone-related test flakiness (8 test files)
• All processors now have report classes ready for metrics

https://claude.ai/code/session_0171jAojWBsvNMDz7havHgDy

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
14761	1	14760	118

View the top 2 failed test(s) by shortest run time

tests.integration.mode.test_mode_threading::test_threading_speedup

Stack Traces | 2.89s run time

tmp_path = PosixPath('.../pytest-of-runner/pytest-0/test_threading_speedup0')

    def test_threading_speedup(tmp_path):
        """Verify that max_threads > 1 provides wall-clock speedup with simulated latency.
    
        Uses 10 reports with 2 queries each. Each HTTP call sleeps 50ms.
        With 4 threads, expect ~2-4x wall-clock speedup.
    
        Note: No @time_machine.travel here -- time_machine patches time.monotonic()
        which would make our wall-clock measurements return 0.
        """
        num_reports = 10
        num_queries_per_report = 2
        latency = 0.05  # 50ms per request
    
        responses = _build_perf_response_map(num_reports, num_queries_per_report)
    
        # Sequential run (max_threads=1)
        with patch(
            "datahub.ingestion.source.mode.requests.Session",
            side_effect=lambda *a, **kw: ThreadSafeMockSession(responses, latency=latency),
        ):
            pipeline_seq = Pipeline.create(
                {
                    "run_id": "mode-perf-sequential",
                    "source": {
                        "type": "mode",
                        "config": {
                            "token": "xxxx",
                            "password": "xxxx",
                            "connect_uri": "https://app.mode.com/",
                            "workspace": "acryl",
                            "max_threads": 1,
                        },
                    },
                    "sink": {
                        "type": "file",
                        "config": {
                            "filename": f"{tmp_path}/perf_seq.json",
                        },
                    },
                }
            )
            t0 = time.perf_counter()
            pipeline_seq.run()
            sequential_time = time.perf_counter() - t0
    
        # Threaded run (max_threads=4)
        with patch(
            "datahub.ingestion.source.mode.requests.Session",
            side_effect=lambda *a, **kw: ThreadSafeMockSession(responses, latency=latency),
        ):
            pipeline_par = Pipeline.create(
                {
                    "run_id": "mode-perf-parallel",
                    "source": {
                        "type": "mode",
                        "config": {
                            "token": "xxxx",
                            "password": "xxxx",
                            "connect_uri": "https://app.mode.com/",
                            "workspace": "acryl",
                            "max_threads": 4,
                        },
                    },
                    "sink": {
                        "type": "file",
                        "config": {
                            "filename": f"{tmp_path}/perf_par.json",
                        },
                    },
                }
            )
            t0 = time.perf_counter()
            pipeline_par.run()
            parallel_time = time.perf_counter() - t0
    
        speedup = sequential_time / parallel_time if parallel_time > 0 else float("inf")
    
        print(
            f"\nPerf results: sequential={sequential_time:.2f}s, "
            f"parallel={parallel_time:.2f}s, speedup={speedup:.1f}x"
        )
    
        # With 4 threads and 50ms latency, we should see at least 1.5x speedup.
        # Using a conservative threshold to avoid flaky CI.
>       assert speedup > 1.5, (
            f"Expected >1.5x speedup but got {speedup:.2f}x "
            f"(seq={sequential_time:.2f}s, par={parallel_time:.2f}s)"
        )
E       AssertionError: Expected >1.5x speedup but got 0.59x (seq=0.96s, par=1.62s)
E       assert 0.5936169040388863 > 1.5

.../integration/mode/test_mode_threading.py:511: AssertionError

tests.integration.snowplow.test_snowplow_performance::test_parallel_fetching_performance

Stack Traces | 3.61s run time

pytestconfig = <_pytest.config.Config object at 0x7ff8caffa410>
tmp_path = PosixPath('.../pytest-of-runner/pytest-0/test_parallel_fetching_perform0')

    @pytest.mark.integration
    def test_parallel_fetching_performance(pytestconfig, tmp_path):
        """
        Test that parallel deployment fetching is significantly faster than sequential.
    
        This test compares performance with parallel fetching enabled vs disabled.
        """
        # Generate dataset with 100 schemas (enough to see performance difference)
        mock_data_structures = generate_mock_data_structures(100)
    
        # Simulate API delay (10ms per call to make difference measurable)
        def mock_get_deployments(schema_hash: str) -> List[DataStructureDeployment]:
            """Simulate API delay."""
            time.sleep(0.01)  # 10ms delay
            return [
                DataStructureDeployment(
                    version="1-0-0",
                    ts="2024-01-01T00:00:00Z",
                    initiator="Test User",
                    initiator_id="user123",
                )
            ]
    
        # Test 1: Sequential fetching (parallel disabled)
        config_sequential = {
            "bdp_connection": {
                "organization_id": "test-org",
                "api_key_id": "test-key",
                "api_key": "test-secret",
            },
            "field_tagging": {"track_field_versions": True},
            "performance": {
                "enable_parallel_fetching": False,
                "max_concurrent_api_calls": 10,
            },
        }
    
        with patch(
            "datahub.ingestion.source.snowplow.snowplow.SnowplowBDPClient"
        ) as mock_client_class:
            mock_client = mock_client_class.return_value
            mock_client._authenticate = lambda: None
            mock_client._jwt_token = "mock_token"
    
            from datahub.ingestion.source.snowplow.models.snowplow_models import (
                DataStructure,
            )
    
            mock_client.get_data_structures.return_value = [
                DataStructure.model_validate(ds) for ds in mock_data_structures
            ]
            mock_client.get_data_structure_deployments.side_effect = mock_get_deployments
    
            config = SnowplowSourceConfig.model_validate(config_sequential)
            source = SnowplowSource(config, create_mock_context())
            source.bdp_client = mock_client
    
            start_time = time.time()
            list(source.schema_processor._get_data_structures_filtered())
            sequential_time = time.time() - start_time
    
        # Test 2: Parallel fetching (parallel enabled)
        config_parallel = {
            "bdp_connection": {
                "organization_id": "test-org",
                "api_key_id": "test-key",
                "api_key": "test-secret",
            },
            "field_tagging": {"track_field_versions": True},
            "performance": {
                "enable_parallel_fetching": True,
                "max_concurrent_api_calls": 10,
            },
        }
    
        with patch(
            "datahub.ingestion.source.snowplow.snowplow.SnowplowBDPClient"
        ) as mock_client_class:
            mock_client = mock_client_class.return_value
            mock_client._authenticate = lambda: None
            mock_client._jwt_token = "mock_token"
            mock_client.get_data_structures.return_value = [
                DataStructure.model_validate(ds) for ds in mock_data_structures
            ]
            mock_client.get_data_structure_deployments.side_effect = mock_get_deployments
    
            config = SnowplowSourceConfig.model_validate(config_parallel)
            source = SnowplowSource(config, create_mock_context())
            source.bdp_client = mock_client
    
            start_time = time.time()
            list(source.schema_processor._get_data_structures_filtered())
            parallel_time = time.time() - start_time
    
        # Performance assertions
        speedup = sequential_time / parallel_time
        print("\nPerformance Results:")
        print(f"  Sequential time: {sequential_time:.2f}s")
        print(f"  Parallel time: {parallel_time:.2f}s")
        print(f"  Speedup: {speedup:.2f}x")
    
        # Parallel should be at least 3x faster with 10 workers
        # (100 schemas / 10 workers = ~10 sequential batches vs 100 sequential calls)
>       assert parallel_time < sequential_time / 3, (
            f"Parallel fetching should be at least 3x faster (got {speedup:.2f}x)"
        )
E       AssertionError: Parallel fetching should be at least 3x faster (got 0.41x)
E       assert 2.542221784591675 < (1.045442819595337 / 3)

.../integration/snowplow/test_snowplow_performance.py:180: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}