🐛 follow up to #35471: update the cartesian stream slicer

[girarda]

What

• I forgot to update the cartesian stream slicer to return typed StreamSlice in 🐛 low-code: Fix incremental substreams #35471

This caused connectors to fail because dicts are returned instead of StreamSlices

{"type": "TRACE", "trace": {"type": "ERROR", "emitted_at": 1709764005297.951, "error": {"message": "Something went wrong in the connector. See the logs for more details.", "internal_message": "'dict' object has no attribute 'partition'", "stack_trace": "Traceback (most recent call last):\n  File \"/Users/alex/code/airbyte/airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py\", line 129, in read\n    yield from self._read_stream(\n  File \"/Users/alex/code/airbyte/airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py\", line 229, in _read_stream\n    for record_data_or_message in record_iterator:\n  File \"/Users/alex/code/airbyte/airbyte-cdk/python/airbyte_cdk/sources/streams/core.py\", line 135, in read\n    for _slice in slices:\n  File \"/Users/alex/code/airbyte/airbyte-cdk/python/airbyte_cdk/sources/declarative/incremental/per_partition_cursor.py\", line 74, in stream_slices\n    cursor = self._cursor_per_partition.get(self._to_partition_key(partition.partition))\nAttributeError: 'dict' object has no attribute 'partition'\n", "failure_type": "system_error", "stream_descriptor": {"name": "billable_rates"}}}}

How

• Create a StreamSlice by ChainMapping the partitions and the cursor_slice separately

Testing

I tested this bugfix with Patrick's harvest low-code connector and confirmed we were able to pull the records without failing

{"type": "LOG", "log": {"level": "INFO", "message": "Read 4 records from billable_rates stream"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Marking stream billable_rates as STOPPED"}}
{"type": "TRACE", "trace": {"type": "STREAM_STATUS", "emitted_at": 1709763953537.6611, "stream_status": {"stream_descriptor": {"name": "billable_rates", "namespace": null}, "status": "COMPLETE"}}}
{"type": "LOG", "log": {"level": "INFO", "message": "Finished syncing billable_rates"}}
{"type": "LOG", "log": {"level": "INFO", "message": "SourceHarvest runtimes:\nSyncing stream billable_rates 0:00:08.429028"}}
{"type": "LOG", "log": {"level": "INFO", "message": "Finished syncing SourceHarvest"}

Recommended reading order

1. airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/cartesian_product_stream_slicer.py
2. airbyte-cdk/python/unit_tests/sources/declarative/stream_slicers/test_cartesian_product_stream_slicer.py

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Ignored Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Mar 7, 2024 3:59pm

[pnilan]

Thanks for tackling this so quickly.

[maxi297]

Should we also have this check if the list has only one element?

[maxi297]

It feels weird that we would have multiple cursor here but I guess it has always been like that. I'm just wondering if we should fail loudly because we don't understand this business case or just let it happen and have an undetermined behavior

[girarda]

agreed. fixed

[girarda]

Should we also have this check if the list has only one element?

I'd prefer avoiding a breaking change as part of a bug fix, but happy to revisit