feat(low-code): add UnionPartitionRouter component

[Daryna Ishchenko (darynaishchenko)]

Summary

Adds a new first-class declarative component, UnionPartitionRouter, to unblock the source-github migration to manifest-only (see https://github.com/airbytehq/airbyte-internal-issues/issues/16492, https://github.com/airbytehq/airbyte-internal-issues/issues/16512 and airbytehq/airbyte#81428). It yields the deduplicated union of its child partition routers' slices, normalized to a single stable partition field:

partition_router:
  type: UnionPartitionRouter
  partition_field: repository
  partition_routers:
    - type: SubstreamPartitionRouter   # repos from parent stream(s)
      parent_stream_configs: [...]
    - type: ListPartitionRouter        # repos from config
      cursor_field: repository
      values: "{{ config['repositories'] }}"

Behavior of UnionPartitionRouter.stream_slices():

• Iterates children in order; first occurrence of a partition_field value wins, later duplicates are skipped.
• Every emitted slice's partition is exactly {partition_field: value}; any other partition keys from a child (e.g. SubstreamPartitionRouter's parent_slice) are moved into extra_fields, so per-partition state keys stay stable regardless of which child produced the slice.
• get_stream_state() merges children's parent-stream states; request option methods return {} (children declaring request_option are rejected at factory time — consume partition values via stream_partition interpolation instead). Note: CustomPartitionRouter children are opaque, so the request-option rejection cannot cover them.

Wiring beyond the runtime class and schema:

• create_union_partition_router in ModelToComponentFactory, registered in PYDANTIC_MODEL_TO_CONSTRUCTOR; schema registered in SimpleRetriever/AsyncRetriever partition_router anyOf, with self-nesting allowed (minItems: 2).
• LegacyToPerPartitionStateMigration accepts UnionPartitionRouter: _get_partition_field reads partition_field directly, and the single-parent-stream check in should_migrate doesn't apply.
• Parent-stream use_cache auto-enable (_initialize_cache_for_parent_streams) and parent/ancestor dependency-graph collection (_collect_all_parent_stream_names, _collect_all_ancestor_names) now recurse into composed routers (partition_routers / underlying_partition_router).

Tests:

• Runtime unit tests (dedup, normalization, first-wins, nesting, state merge), factory tests (construction + request_option rejection), and legacy→per-partition state migration tests with a union router.
• Concurrency-graph/caching coverage: use_cache force-enabled on all parents referenced through a union (including nested parent stream instances), and parent/ancestor collection detects deadlocks through UnionPartitionRouter children, incl. a union nested inside GroupingPartitionRouter.
• Runtime round-trip: a read resuming from legacy state through a union-partitioned incremental stream asserts the migrated per-partition keys exactly match the runtime per-partition state keys emitted during the read (and that duplicated partitions across parents are read only once).

Note on generated models: poetry run poe assemble produces the same ~350-line style drift (conint/confloat, class reordering) on a clean main checkout in this environment, so declarative_component_schema.py was hand-patched with only the UnionPartitionRouter additions to preserve the committed generated style.

Link to Devin session: https://app.devin.ai/sessions/38753ed2d1f346e6a370888444676f4f
Requested by: Daryna Ishchenko (@darynaishchenko)

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1783689479-union-partition-router#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1783689479-union-partition-router

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

4 156 tests  +29   4 144 ✅ +29   8m 1s ⏱️ +20s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 36b06ad. ± Comparison against base commit dd9558c.

This pull request removes 1 and adds 30 tests. Note that renamed tests count towards both.

unit_tests.sources.declarative.migrations.test_legacy_to_per_partition_migration ‑ test_create_legacy_to_per_partition_state_migration[SimpleRetriever-None-False-ValueError-LegacyToPerPartitionStateMigrations can only be applied on a SimpleRetriever with a Substream partition router. Got <class 'NoneType'>]

unit_tests.sources.declarative.migrations.test_legacy_to_per_partition_migration ‑ test_create_legacy_to_per_partition_state_migration[SimpleRetriever-None-False-ValueError-LegacyToPerPartitionStateMigrations can only be applied on a SimpleRetriever with a SubstreamPartitionRouter, UnionPartitionRouter or CustomPartitionRouter. Got <class 'NoneType'>]
unit_tests.sources.declarative.migrations.test_legacy_to_per_partition_migration ‑ test_migrate_a_valid_legacy_state_to_per_partition_with_union_partition_router
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_interpolated_partition_field
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_mismatched_partition_field[list_child_with_mismatched_cursor_field]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_mismatched_partition_field[substream_child_with_mismatched_partition_field]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_request_option[list_child_with_request_option]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_request_option[substream_child_with_request_option]
unit_tests.sources.declarative.partition_routers.test_union_partition_router ‑ test_get_stream_state_merges_children_states[all_children_without_state]
unit_tests.sources.declarative.partition_routers.test_union_partition_router ‑ test_get_stream_state_merges_children_states[merges_states_of_all_children]
…

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 159 tests  +29   4 147 ✅ +29   12m 20s ⏱️ + 5m 5s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit 36b06ad. ± Comparison against base commit dd9558c.

This pull request removes 1 and adds 30 tests. Note that renamed tests count towards both.

unit_tests.sources.declarative.migrations.test_legacy_to_per_partition_migration ‑ test_create_legacy_to_per_partition_state_migration[SimpleRetriever-None-False-ValueError-LegacyToPerPartitionStateMigrations can only be applied on a SimpleRetriever with a Substream partition router. Got <class 'NoneType'>]

unit_tests.sources.declarative.migrations.test_legacy_to_per_partition_migration ‑ test_create_legacy_to_per_partition_state_migration[SimpleRetriever-None-False-ValueError-LegacyToPerPartitionStateMigrations can only be applied on a SimpleRetriever with a SubstreamPartitionRouter, UnionPartitionRouter or CustomPartitionRouter. Got <class 'NoneType'>]
unit_tests.sources.declarative.migrations.test_legacy_to_per_partition_migration ‑ test_migrate_a_valid_legacy_state_to_per_partition_with_union_partition_router
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_interpolated_partition_field
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_mismatched_partition_field[list_child_with_mismatched_cursor_field]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_mismatched_partition_field[substream_child_with_mismatched_partition_field]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_request_option[list_child_with_request_option]
unit_tests.sources.declarative.parsers.test_model_to_component_factory ‑ test_create_union_partition_router_with_request_option[substream_child_with_request_option]
unit_tests.sources.declarative.partition_routers.test_union_partition_router ‑ test_get_stream_state_merges_children_states[all_children_without_state]
unit_tests.sources.declarative.partition_routers.test_union_partition_router ‑ test_get_stream_state_merges_children_states[merges_states_of_all_children]
…

♻️ This comment has been updated with latest results.

[coderabbitai]

Caution

Review failed

An error occurred during the review process. Please try again later.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch devin/1783689479-union-partition-router

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}