Bug - incremental_dependency: true triggers full refresh on child streams due to context loss in SubstreamPartitionRouter (Concurrent CDK) #1067

[Bug] incremental_dependency: true triggers full refresh on child streams due to context loss in SubstreamPartitionRouter (Concurrent CDK)

Note: This investigation stems from Issue #61567 already reported in the main Airbyte repository. To identify the root cause, I leveraged GitHub Copilot and Gemini Pro to analyze the current architecture of the airbyte-python-cdk repository. The findings point to a specific architectural gap causing the bug.

Problem Summary

There is a severe performance degradation in the Concurrent CDK affecting any connector built with the Declarative CDK / Connector Builder that uses hierarchical relationships with incremental_dependency: true.

• Initial Sync (Sync 1): Works correctly. The streamState is persisted with the parent's cursor checkpoints.
• Subsequent Syncs (Sync 2+): The child stream loses its state context, forcing the parent stream to ignore its saved cursor and restart data extraction from the beginning (start_date).
• Impact: Delta syncs are 100x to 1000x slower. This affects production connectors such as source-gmail, source-mixpanel, source-instagram, source-typeform, and custom implementations (e.g., source-dolibarr).

Root Cause Analysis (Code Investigation)

The issue lies in how the concurrent engine fails to propagate the state context within the SubstreamPartitionRouter.

Target File: airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

1. Context loss in cursor_slice (Line ~214):
Inside the stream_slices generator, when iterating over partitions (partition.read()), the StreamSlice is built and yielded with a hardcoded empty dictionary for the cursor:

yield StreamSlice(
    partition={
        partition_field: partition_value,
        "parent_slice": parent_partition or {},
    },
    cursor_slice={},  # <-- Parent's cursor context is lost here
    extra_fields=extracted_extra_fields,
)

By yielding an empty cursor_slice, downstream components that rely on this slice to maintain the parent's checkpoint fail to associate the exact cursor bounds.

2. Brittle state migration (_migrate_child_state_to_parent_state - Line ~261):
The CDK attempts to mitigate this by rebuilding the parent state. However, the logic assumes very simple structures (strictly looking for a "state" key) and fails to rebuild the nested parent_state envelope required by child streams when multiple cursors or complex dates are involved.

State Structure Breakdown:
The state JSON collapses from its required nested format to an unusable flat structure:
Required: {"parent_state": {"parent_stream_name": {"date_modification": "2024-08-01T00:00:00Z"}}}
Generated (Broken): {"date_modification": "2024-08-01T00:00:00Z"}

Proposed Core CDK Solution

To resolve this natively in the concurrent framework, stream_slices and read_parent_stream should be modified to optionally accept stream_state. The parent state should be extracted inline and injected into the cursor_slice rather than forcing an empty dictionary. Additionally, _migrate_child_state_to_parent_state needs to be strengthened to respect nested parent_state structures.

Temporary Connector-Level Workaround (Proof of Concept)

While a core solution is evaluated, it is possible to bypass this limitation in affected connectors by implementing a custom StateMigration that rebuilds the JSON before the CDK parses it. This makes the workaround idempotent and works without breaking the concurrent engine:

1. components.py

from dataclasses import dataclass
from typing import Any, Mapping
from airbyte_cdk.sources.declarative.state_migration import StateMigration

@dataclass
class IncrementalSyncStateMigration(StateMigration):
    def should_migrate(self, stream_state: Mapping[str, Any]) -> bool:
        return "parent_state" not in stream_state and stream_state and len(stream_state) > 0

    def migrate(self, stream_state: Mapping[str, Any]) -> Mapping[str, Any]:
        if not stream_state:
            return stream_state

        migrated_parent_state = {}
        for key, value in stream_state.items():
            if key != "states" and isinstance(value, dict) and any(
                cursor in value for cursor in ["date_modification", "updated_at", "timestamp", "tms"]
            ):
                migrated_parent_state[key] = value

        return {
            "parent_state": migrated_parent_state,
            "states": [],
            "lookback_window": 0,
        }

(Integrated by registering the state_migration class directly in the definitions block of manifest.yaml).

Steps to reproduce and verify:

1. Create a connector with a parent and child stream using incremental_dependency: true.
2. Run the initial sync and save the STATE output.
3. Inject the saved STATE into a second run (docker-compose run connector read config.json catalog.json <<< "$STATE").
4. Expected result: Fast delta sync.
5. Actual result: Logs show re-extraction of all parent records.

I would highly appreciate it if the maintainers could review this state propagation architecture. Official connectors (like source-intercom and source-jira) have already had to patch this locally. Does fixing cursor_slice in the main router align with the vision for the concurrent CDK?