feat(ingest/dynamodb): add S3 export lineage and Glue DynamoDB job URNs

[acrylJonny]

Summary

• Add opt-in include_s3_export_lineage on the DynamoDB source to discover existing Export to S3 jobs (ListExports / DescribeExport) and emit COPY lineage to s3://bucket/prefix without starting exports.
• Teach the Glue source to resolve connection_type: dynamodb job DAG nodes to DynamoDB dataset URNs so ETL jobs into Glue catalog / Iceberg / Parquet join correctly.
• Document the two-hop pattern: native DynamoDB export (JSON/Ion) vs Glue conversion to Parquet/Iceberg, plus IAM and target_platform_configs.dynamodb alignment.

Fixes ING-3362

Test plan

• Unit tests for DynamoDB S3 export lineage discovery / aggregation / error handling
• Unit tests for Glue DynamoDB ETL connector, table ARN, target_platform_configs, and missing-table cases
• Optional: run DynamoDB + Glue ingestion against an account with a completed export and a DynamoDB→Iceberg Glue job

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
18549	1	18548	71

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

tests.integration.snowplow.test_snowplow_performance::test_parallel_fetching_performance

Stack Traces | 2.69s run time

pytestconfig = <_pytest.config.Config object at 0x7f658e449550>
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.65x)
E       assert 1.611966609954834 < (1.053229570388794 / 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.}

[datahub-connector-tests]

Connector Tests Results

All connector tests passed for commit 75bfd5d

View full test logs →

To skip connector tests, add the skip-connector-tests label (org members only).

Autogenerated by the connector-tests CI pipeline.

[cursor]

PR Summary

Overview
Adds opt-in DynamoDB Export to S3 lineage and Glue job URNs that match the DynamoDB source, so native exports and Glue ETL into catalog/Iceberg/Parquet join in the graph.

The DynamoDB connector can set include_s3_export_lineage to read existing ListExports/DescribeExport metadata (no export creation) and emit table-level COPY edges to s3://bucket/prefix. Optional include_s3_export_column_lineage maps inferred field paths onto the S3 dataset; it stays off because native JSON/Ion layouts often do not match. Multiple tables sharing a prefix are aggregated onto one S3 UpstreamLineage aspect. Failures become warnings.

Glue now resolves connection_type: dynamodb DAG nodes (ETL table name or tableArn) to {region}.{table} URNs with platform_instance defaulting to account id (catalog_id then STS). Align custom instances via target_platform_configs.dynamodb. Docs cover IAM (ListExports/DescribeExport) and the 90-day export-metadata limit.

^{Reviewed by Cursor Bugbot for commit a7e013b. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 75bfd5d. Configure here.}

[cubic-dev-ai]

1 issue found across 9 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="metadata-ingestion/tests/unit/glue/test_glue_source.py">

<violation number="1" location="metadata-ingestion/tests/unit/glue/test_glue_source.py:3003">
P2: The expected URN in this assertion bakes the account id into the dataset name ("123456789012.us-east-1.customers"), but _process_dynamodb_node passes it as a separate platform_instance argument (account_id) to make_dataset_urn_with_platform_instance with name="us-east-1.customers". That serializes the account id as its own tuple field, so the emitted URN is urn:li:dataset:(urn:li:dataPlatform:dynamodb,us-east-1.customers,123456789012,PROD). Update the assertion to the 4-field form (or fix the source if the intended convention really is to prepend the account), or the test fails against the current implementation.</violation>
</file>

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[acrylJonny]

Linked to ING-3362.

[cubic-dev-ai]

All reported issues were addressed across 2 files (changes from recent commits).

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[cubic-dev-ai]

1 issue found across 5 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="metadata-ingestion/src/datahub/ingestion/source/dynamodb/dynamodb.py">

<violation number="1" location="metadata-ingestion/src/datahub/ingestion/source/dynamodb/dynamodb.py:150">
P2: Custom agent: **Enforce Pragmatic Test Coverage**

When `include_s3_export_column_lineage` is omitted, the new `default=False` changes emitted lineage, but no successful export test verifies that default. Add a test that omits the option and asserts table-level COPY lineage without fine-grained lineage.</violation>
</file>

<sub>Tip: Review your code locally with the cubic CLI to iterate faster.

Re-trigger cubic</sub>

[cubic-dev-ai]

P2: Custom agent: Enforce Pragmatic Test Coverage

When include_s3_export_column_lineage is omitted, the new default=False changes emitted lineage, but no successful export test verifies that default. Add a test that omits the option and asserts table-level COPY lineage without fine-grained lineage.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At metadata-ingestion/src/datahub/ingestion/source/dynamodb/dynamodb.py, line 150:

<comment>When `include_s3_export_column_lineage` is omitted, the new `default=False` changes emitted lineage, but no successful export test verifies that default. Add a test that omits the option and asserts table-level COPY lineage without fine-grained lineage.</comment>

<file context>
@@ -147,11 +147,12 @@ class DynamoDBConfig(
     )
     include_s3_export_column_lineage: bool = Field(
-        default=True,
+        default=False,
         description=(
             "When `include_s3_export_lineage` is enabled, also emit column-level COPY lineage "
</file context>

[acrylJonny]

Added test_column_lineage_off_by_default: omits include_s3_export_column_lineage, asserts the default is False and that emit produces a table-level COPY upstream with no fineGrainedLineages.