fix(low-code CDK): fix checkpointing for declarative streams #177
Conversation
📝 Walkthrough📝 WalkthroughWalkthroughThe changes focus on refining the state management and checkpointing mechanism within the Changes
Sequence Diagram(s)sequenceDiagram
participant Stream
participant StateManager
participant Logger
Stream->>StateManager: Retrieve checkpoint
StateManager-->>Stream: Return checkpoint
Stream->>Stream: Check checkpoint existence
alt Checkpoint exists
Stream->>Logger: Yield state message
else No checkpoint
Stream->>Stream: Continue reading
end
Possibly related PRs
Suggested labels
Suggested reviewers
Tip CodeRabbit's docstrings feature is now available as part of our Early Access Program! Simply use the command Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
airbyte_cdk/sources/streams/core.py (1)
231-236: The checkpoint emission logic looks good, but should we add some debug logging? 🤔The changes correctly ensure that state messages are only emitted when there's an actual checkpoint to emit. This fixes the issue where state messages weren't being emitted when records shared the same cursor value.
Would it be helpful to add some debug logging here to help troubleshoot state emission issues? Something like this, wdyt?
checkpoint = checkpoint_reader.get_checkpoint() if checkpoint: + logger.debug(f"Emitting state message for checkpoint: {checkpoint}") airbyte_state_message = self._checkpoint_state( checkpoint, state_manager=state_manager ) yield airbyte_state_message
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
airbyte_cdk/sources/streams/core.py(1 hunks)
🔇 Additional comments (1)
airbyte_cdk/sources/streams/core.py (1)
231-236: Should we verify the checkpoint structure? 🧐
Since this is a critical part of the state management logic, it might be worth adding a verification step to ensure the checkpoint has the expected structure before emitting it. This could help catch any potential issues early.
|
Should there be a test that is being updated if we change the behavior? |
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (4)
unit_tests/sources/declarative/test_manifest_declarative_source.py (4)
1906-1907: Add a brief docstring for the new parameterized test?It might be helpful to add a concise docstring at the start of this new test to clarify its purpose, e.g., "Ensures that slice-based checkpointing emits the expected number of states for pagination and partition routers." Wdyt?
1972-1981: Verify incremental sync config test coverage.You’re passing a new "incremental_sync" structure in your manifest. Do you want to add a test scenario that checks what happens when "start_date" is not present or is an invalid format? That might future-proof your coverage. Wdyt?
2011-2036: Confirm presence of a second partition scenario.You're partitioning with two values ["0", "1"], and your pages are arranged to produce two distinct sets of records. Do you want to add a partition scenario beyond these two (e.g., three or four partitions) to ensure the checkpoint logic holds under multiple partitions? Wdyt?
2044-2045: Check for more robust state validation.Here you’re only verifying the total count of states. Would verifying the exact state content and timestamps provide additional confidence? Wdyt?
What
With this PR, the get_checkpointing method has been modified so that the state will be emitted only if the state value is not equal to the previous state value. However, declarative streams do not support record-level checkpointing and instead perform slice checkpointing. In cases where a slice ends with multiple records having the same cursor value, the state message will not be emitted because _previous_state in the checkpoint reader is already updated and matches the current state.
How
Moved the record checkpoint definition after the condition that checks if checkpointing is needed, ensuring that checkpoint_interval exists and
record_counter % checkpoint_interval == 0.Summary by CodeRabbit
Bug Fixes
New Features
ManifestDeclarativeSource.start_dateandincremental_sync.