Concurrent CDK: support partitioned states

[maxi297]

What

Addresses https://github.com/airbytehq/airbyte-internal-issues/issues/6865

How

By having the CDK support partitioned states. This implies a couple of changes:

• the cursor should generate the slices (see ConcurrentCursor.generate_slices)
• sequential state is not a given anymore and should be slowly removed from the concurrent cursor (see various changes to state_converters and ConcurrentCursor)

Once this is done, we have that errors should not break the sync

Note: this PR serves as a CDK feature implementation and the changes on source-salesforce are only to show usage. The changes to source-salesforce will be moved to another PR

Migration

Following the point above: Right now if there is a state, we can only assume that everything from self._start to <cursor value> was synced. If there is a state, we therefore need to create a slice from {"start": self._start, "end": }. I think we had already thought about this logic but I didn't see it in action in my round of testing today (2024-04-05)

[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		Apr 9, 2024 8:32pm

[maxi297]

I really hate this but for full refresh, we create a FinalStateCursor and not a ConcurrentCursor. It probably means that the generation of slices based on a state is not valid. In other words, with or without a state, we generate slices that relies on incrementality and the fact that the stream is full_refresh is irrelevant. I think this will be more true with Resumable Full Refresh but it is a bit annoying right now

[girarda]

why?

[maxi297]

I don't know. It indeed does not make sense as we are fully owner of LOOKBACK_SECONDS. This is code that was there before.

[maxi297]

Adding to that, some tests were patching this constant with invalid values. I just removed them as it is not a real case scenario

[girarda]

nit: let's add a docstring

[maxi297]

Actually, this does not always make sense. I think the typing is kind of wrong for this class. I think we need a generic with two "variables": one that represents cursor values and the other that represents gap between those. For a cursor where values are datetimes, the first would be datetime and the second would be timedelta

[maxi297]

🚨 🚨 🚨 Breaking change alert: The type was changed for start as it was either a Python object when being passed a partitioned state or the serialized value when being passed a sequential state. As I'm trying to remove sequential state from this class, we needed to remove the second case

[clnoll]

@maxi297 I'm not seeing anything that I think would pose a problem for file-based sources. I can do a follow up to update the S3 connector after this is deployed for Salesforce if that would be helpful.

What I'm not clear on is why the cursor needs to be updated in the same PR that's changing the thread shutdown logic, which I thought was the primary issue that we were trying to solve in this PR. Can you help me understand that?