fix: ConfiguredAirbyteCatalog schema violation (cursor_field)

[nicob3y]

"cursor_field" property from an ConfiguredAirbyteCatalog is not properly set, leaving its value to Null while it's not allowed. Fixes #567.

Summary by CodeRabbit

• New Features
  • Enhanced data stream configuration now supports incremental processing with a dedicated tracking field for efficient data synchronization. Existing record retrieval behaviors remain unchanged.

[coderabbitai[bot]]

📝 Walkthrough

Walkthrough

The changes modify the get_configured_catalog method in the Source class to include a new parameter, cursor_field, when instantiating a ConfiguredAirbyteStream. This parameter, set as stream.default_cursor_field, is now part of the configured catalog to support incremental data processing. Additionally, a minor change in the get_records method maintains the previous behavior without the new parameter.
wdyt?

Changes

File	Change Summary
airbyte/sources/base.py	Modified get_configured_catalog to pass cursor_field=stream.default_cursor_field when creating a ConfiguredAirbyteStream. Kept get_records unchanged for this aspect.

Sequence Diagram(s)

sequenceDiagram
    participant Client
    participant Source
    participant Stream
    participant ConfiguredStream

    Client->>Source: get_configured_catalog(streams)
    Source->>Stream: Iterate through each stream
    Stream-->>Source: Provide default_cursor_field
    Source->>ConfiguredStream: Instantiate ConfiguredAirbyteStream(cursor_field=default_cursor_field)
    Source-->>Client: Return ConfiguredAirbyteCatalog

Would you like any further diagram details or adjustments?

---

📜 Recent review details

Configuration used: CodeRabbit UI Review profile: CHILL Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between d064d42fdc9ae44681c1eb710fbcd4fc8c7dfeaf and b51c06027acbf6d7238285bb263bfcf82aff1a32.

📒 Files selected for processing (1)

• airbyte/sources/base.py (1 hunks)

⏰ Context from checks skipped due to timeout of 90000ms (2)

• GitHub Check: Pytest (No Creds)
• GitHub Check: Pytest (Fast)

🔇 Additional comments (2)

airbyte/sources/base.py (2)

371-383:

❓ Verification inconclusive

Consider handling empty default_cursor_field.

The current implementation assumes that stream.default_cursor_field will always be valid. Should we add a fallback value or validation check for cases where default_cursor_field might be empty or None, wdyt?

Let's check if there are any existing patterns for handling empty cursor fields:

---

🏁 Script executed:

#!/bin/bash
# Search for patterns handling default_cursor_field
rg -A 3 "default_cursor_field" 

Length of output: 776

---

Suggestion: Add a Fallback for Empty default_cursor_field?

It seems that in airbyte/sources/base.py (lines 371–383) we directly assign cursor_field=stream.default_cursor_field without a guard against an empty or None value. Our tests in tests/integration_tests/fixtures/source-test/source_test/run.py always supply a valid default_cursor_field (e.g., ["Column1"]), but this might not cover all runtime scenarios.

Would you consider adding a fallback (for example, defaulting to an empty list or another appropriate value) or a validation check to handle cases when default_cursor_field is empty? wdyt?

---

378-378:

✅ Verification successful

LGTM! The change addresses the schema violation.

The addition of cursor_field=stream.default_cursor_field in the ConfiguredAirbyteStream instantiation aligns with the PR objective. This change ensures that the cursor field is properly set, preventing the schema violation where it was previously left as Null.

Let's verify that this change complies with the Airbyte protocol schema:

---

🌐 Web query:

What is the schema definition for cursor_field in ConfiguredAirbyteStream?

💡 Result:

The cursor_field in ConfiguredAirbyteStream defines the path to the field used for incremental synchronization. Here's its schema definition and key characteristics:

Schema Definition

• Type: Array of strings

  "cursor_field": {
    "type": "array",
    "items": {
      "type": "string"
    }
  }
  

• Required: Only when sync_mode is set to INCREMENTAL[1][3][7]

Key Properties

1. Path Structure:

   • Represents nested fields using dot notation (e.g., ["nested", "field"] translates to nested.field)[6][10]
   • Must match an existing field in the stream's JSON Schema[5][8]

2. Incremental Sync Requirements:

   • Sources must declare if they define the cursor via source_defined_cursor[1][6]
   • If source-defined, user-provided cursors are ignored[1][6]

3. Data Type Constraints:

   • Typically uses timestamp fields (e.g., updated_at) for reliable incremental syncs[4][5]
   • Should have sufficient granularity (at least second precision recommended)[4]

Example Configuration

{
  "ConfiguredAirbyteStream": {
    "cursor_field": ["updated_at"],
    "sync_mode": "INCREMENTAL",
    "stream": {
      "name": "users",
      "json_schema": {
        "properties": {
          "updated_at": {"type": "string", "format": "date-time"}
        }
      }
    }
  }
}

Limitations

• Nested cursors require explicit path declaration (e.g., ["object", "subfield"])[10]
• Array-type fields cannot be used as cursors[4][10]
• Sources must ensure cursor fields are always updated on record changes[4][5]

Citations:

• 1: https://pkg.go.dev/github.com/datakit-dev/airbyte-protocol-go/pkg/airbyte
• 2: https://www.restack.io/docs/airbyte-knowledge-airbyte-protocol-guide
• 3: https://docs.airbyte.com/understanding-airbyte/beginners-guide-to-catalog
• 4: https://www.restack.io/docs/airbyte-knowledge-airbyte-cursor-field-guide
• 5: https://www.restack.io/docs/airbyte-knowledge-airbyte-json-schema-guide
• 6: https://gist.github.com/aaronsteers/ba41035bebd526d7aa30cbb3c43bd661
• 7: https://github.com/airbytehq/airbyte/blob/master/docs/understanding-airbyte/beginners-guide-to-catalog.md
• 8: https://docs.airbyte.com/connector-development/cdk-python/schemas
• 9: https://github.com/airbytehq/PyAirbyte/issues/575
• 10: https://discuss.airbyte.io/t/acceptance-test-is-failing-on-a-schema-with-a-nested-cursor/788

---

LGTM! The change correctly sets the cursor field using stream.default_cursor_field, ensuring that the value complies with the Airbyte protocol schema (i.e., an array of strings as required for incremental sync).

• The update in the ConfiguredAirbyteStream instantiation now guarantees that the cursor field is properly initialized, mitigating the schema violation.
• With this change, incremental sync functionality should reliably use the expected field path (array of strings) as defined in the schema.

Does this look good to you, wdyt?

✨ Finishing Touches

• [ ] 📝 Generate Docstrings (Beta)

---

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?

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

🪧 Tips

Chat

There are 3 ways to chat with CodeRabbit:

‼️ IMPORTANT Auto-reply has been disabled for this repository in the CodeRabbit settings. The CodeRabbit bot will not respond to your replies unless it is explicitly tagged.

• Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
  • @coderabbitai generate unit testing code for this file.
  • @coderabbitai modularize this function.
• PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
  • @coderabbitai gather interesting stats about this repository and render them as a table. Additionally, render a pie chart showing the language distribution in the codebase.
  • @coderabbitai read src/utils.ts and generate unit testing code.
  • @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.
  • @coderabbitai help me debug CodeRabbit configuration file.

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)

• @coderabbitai pause to pause the reviews on a PR.
• @coderabbitai resume to resume the paused reviews.
• @coderabbitai review to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
• @coderabbitai full review to do a full review from scratch and review all the files again.
• @coderabbitai summary to regenerate the summary of the PR.
• @coderabbitai generate docstrings to generate docstrings for this PR. (Beta)
• @coderabbitai resolve resolve all the CodeRabbit review comments.
• @coderabbitai configuration to show the current CodeRabbit configuration for the repository.
• @coderabbitai help to get help.

Other keywords and placeholders

• Add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.
• Add @coderabbitai summary to generate the high-level summary at a specific location in the PR description.
• Add @coderabbitai or @coderabbitai title anywhere in the PR title to generate the title automatically.

CodeRabbit Configuration File (.coderabbit.yaml)

• You can programmatically configure CodeRabbit by adding a .coderabbit.yaml file to the root of your repository.
• Please see the configuration documentation for more information.
• If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json

Documentation and Community

• Visit our Documentation for detailed information on how to use CodeRabbit.
• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[aaronsteers]

/test-pr

PR test job started... Check job output.

❌ Tests failed.

[aaronsteers]

@nicob3y - Thank you very much for contributing. Closing this old PR, as the logic now exists in the main branch:

                    cursor_field=(
                        [self._cursor_key_overrides[stream.name.lower()]]
                        if stream.name.lower() in self._cursor_key_overrides
                        else stream.default_cursor_field
                    ),