fix(ingestion): salesforce connector + dbt - #1288
Conversation
|
Warning Review limit reached
More reviews will be available in 34 minutes and 1 second. Learn how PR review limits work. To continue reviewing without waiting, enable usage-based billing in the billing tab. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan refill rate. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, the refill rate gradually slows as usage increases. The highest same-day bursts are limited more strictly. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📥 CommitsReviewing files that changed from the base of the PR and between e3847208b1e57581b1db9596e654ce14ea8a42d3 and 8fcc767. 📒 Files selected for processing (24)
📝 WalkthroughWalkthroughThe Salesforce connector shifts from a hybrid REST + Bulk API 2.0 approach to a REST-only ChangesREST-only Salesforce Connector Refactor
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py (1)
476-616:⚠️ Potential issue | 🔴 Critical | 🏗️ Heavy liftCritical: State regression bug in concurrent slice processing.
The
_get_updated_statemethod (lines 582-616) has a state-management hazard in concurrent slicing scenarios. When a record slips past its slice boundary (latest_record_value > slice_max_value) and a later slice has already advanced global state beyond that slice's maximum, lines 600-602 returnmin(latest_record_value, slice_max_value)without comparing tocurrent_stream_state, causing state to move backward.Scenario:
- Slice B
[2024-06-10, 2024-06-20)completes first → state advances to2024-06-19- Slice A
[2024-06-01, 2024-06-10)processes a stray record at2024-06-11(past its boundary)- Logic:
latest=06-11 > slice_max=06-10→ returnsmin(06-11, 06-10) = 06-10- State regresses from
06-19→06-10, causing data loss on next sync (records in[06-10, 06-19)are re-read or skipped depending on cursor semantics)Root cause: Line 602 ignores
current_stream_state[self.cursor_field]when clamping.🔧 Proposed fix
slice_end = (self._slice or {}).get("end_date") if slice_end: slice_max_value: pendulum.DateTime = pendulum.parse(slice_end) max_possible_value = min(latest_record_value, slice_max_value) if current_stream_state.get(self.cursor_field): + existing_value = pendulum.parse(current_stream_state[self.cursor_field]) if latest_record_value > slice_max_value: - return {self.cursor_field: max_possible_value.isoformat()} + # Clamp to slice boundary, but never regress state from concurrent slices + return {self.cursor_field: max(max_possible_value, existing_value).isoformat()} max_possible_value = max( latest_record_value, - pendulum.parse(current_stream_state[self.cursor_field]), + existing_value, ) return {self.cursor_field: max_possible_value.isoformat()}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py` around lines 476 - 616, The _get_updated_state method can regress global state when clamping a latest_record_value > slice_max_value because it returns slice_max_value without considering current_stream_state; modify _get_updated_state (in the IncrementalRestSalesforceStream class) so that when slice_end/slice_max_value exists and latest_record_value > slice_max_value you compute the clamped_value = min(latest_record_value, slice_max_value) and then compare it to current_stream_state.get(self.cursor_field) (if present) and return the max of those two (as isoformat) to avoid moving state backwards; ensure all other branches preserve existing logic and that comparisons use pendulum-parsed datetimes.
🧹 Nitpick comments (2)
src/ingestion/connectors/crm/salesforce/source_salesforce/source.py (1)
182-210: ⚡ Quick winUpdate stale comment to reflect REST-only implementation.
The comment on line 183 says "Choose proper stream class: syncMode (full_refresh/incremental), REST API, SubStream", listing "REST API" as if it's one of multiple API choices. Since the refactor removed Bulk API and now exclusively uses REST
/queryAll, this phrasing is outdated.📝 Suggested comment update
- def prepare_stream(self, stream_name: str, json_schema, sobject_options, sf_object, authenticator, config): - """Choose proper stream class: syncMode (full_refresh/incremental), REST API, SubStream.""" + def prepare_stream(self, stream_name: str, json_schema, sobject_options, sf_object, authenticator, config): + """Choose proper stream class based on sync mode (full_refresh/incremental) and whether it's a substream."""🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/source_salesforce/source.py` around lines 182 - 210, Update the stale docstring in prepare_stream to reflect the current REST-only implementation: replace "Choose proper stream class: syncMode (full_refresh/incremental), REST API, SubStream" with a concise description mentioning only selecting the stream class (full_refresh/incremental) and that the connector uses the REST /queryAll API (no Bulk API). Ensure the updated comment references prepare_stream (and that stream selection still uses _get_stream_type and UNSUPPORTED_FILTERING_STREAMS) so readers understand the function's purpose and REST-only behavior.src/ingestion/connectors/crm/salesforce/source_salesforce/rate_limiting.py (1)
200-223: 💤 Low valueMinor inconsistency: mixing
Tupleimport with lowercasetupleannotation.Line 14 imports
Tuplefromtyping, but line 203 uses the built-intuple[...]annotation (Python 3.9+ syntax). While both work in Python 3.9+, consider using consistent style throughout the file.- ) -> tuple[Optional[str], str]: + ) -> Tuple[Optional[str], str]:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/ingestion/connectors/crm/salesforce/source_salesforce/rate_limiting.py` around lines 200 - 223, The type annotation for _extract_error_code_and_message mixes typing.Tuple with the built-in tuple[...] syntax; make them consistent by switching the function return annotation to use typing.Tuple[Optional[str], str] (or alternatively remove the Tuple import and use the built-in tuple[Optional[str], str]) and update the imports accordingly so the file consistently uses one style (refer to the _extract_error_code_and_message function and the Tuple import).
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/ingestion/dbt/macros/drop_silver_placeholders_at_start.sql`:
- Around line 6-7: Update the stale comment in dbt_project.yml to reflect the
actual behavior of the macro drop_silver_placeholders_at_start.sql: replace the
phrase "Three-factor detection (placeholder marker + 0 rows + at least one
staging model materialised...)" with "Two-factor detection (placeholder marker +
staging materialised for the matching `silver:<id>` tag)". Apply this same
wording adjustment to the other outdated comment blocks mentioned (around the
provided ranges) so the on-run-start documentation matches the macro's
two-factor detection logic.
---
Outside diff comments:
In `@src/ingestion/connectors/crm/salesforce/source_salesforce/streams.py`:
- Around line 476-616: The _get_updated_state method can regress global state
when clamping a latest_record_value > slice_max_value because it returns
slice_max_value without considering current_stream_state; modify
_get_updated_state (in the IncrementalRestSalesforceStream class) so that when
slice_end/slice_max_value exists and latest_record_value > slice_max_value you
compute the clamped_value = min(latest_record_value, slice_max_value) and then
compare it to current_stream_state.get(self.cursor_field) (if present) and
return the max of those two (as isoformat) to avoid moving state backwards;
ensure all other branches preserve existing logic and that comparisons use
pendulum-parsed datetimes.
---
Nitpick comments:
In `@src/ingestion/connectors/crm/salesforce/source_salesforce/rate_limiting.py`:
- Around line 200-223: The type annotation for _extract_error_code_and_message
mixes typing.Tuple with the built-in tuple[...] syntax; make them consistent by
switching the function return annotation to use typing.Tuple[Optional[str], str]
(or alternatively remove the Tuple import and use the built-in
tuple[Optional[str], str]) and update the imports accordingly so the file
consistently uses one style (refer to the _extract_error_code_and_message
function and the Tuple import).
In `@src/ingestion/connectors/crm/salesforce/source_salesforce/source.py`:
- Around line 182-210: Update the stale docstring in prepare_stream to reflect
the current REST-only implementation: replace "Choose proper stream class:
syncMode (full_refresh/incremental), REST API, SubStream" with a concise
description mentioning only selecting the stream class
(full_refresh/incremental) and that the connector uses the REST /queryAll API
(no Bulk API). Ensure the updated comment references prepare_stream (and that
stream selection still uses _get_stream_type and UNSUPPORTED_FILTERING_STREAMS)
so readers understand the function's purpose and REST-only behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 96dfd7a5-cb5a-4036-9b41-acefc0327ecf
📒 Files selected for processing (23)
src/ingestion/connectors/crm/salesforce/README.mdsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_accounts.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_activities.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_contacts.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_deals.sqlsrc/ingestion/connectors/crm/salesforce/dbt/salesforce__crm_users.sqlsrc/ingestion/connectors/crm/salesforce/pyproject.tomlsrc/ingestion/connectors/crm/salesforce/source_salesforce/api.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/availability_strategy.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/constants.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/exceptions.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/rate_limiting.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/source.pysrc/ingestion/connectors/crm/salesforce/source_salesforce/spec.jsonsrc/ingestion/connectors/crm/salesforce/source_salesforce/streams.pysrc/ingestion/dbt/macros/drop_silver_placeholders_at_start.sqlsrc/ingestion/reconcile-connectors/lib/cdk-build.shsrc/ingestion/secrets/connectors/salesforce.yaml.examplesrc/ingestion/silver/crm/class_crm_accounts.sqlsrc/ingestion/silver/crm/class_crm_activities.sqlsrc/ingestion/silver/crm/class_crm_contacts.sqlsrc/ingestion/silver/crm/class_crm_deals.sqlsrc/ingestion/silver/crm/class_crm_users.sql
💤 Files with no reviewable changes (2)
- src/ingestion/connectors/crm/salesforce/source_salesforce/availability_strategy.py
- src/ingestion/connectors/crm/salesforce/source_salesforce/exceptions.py
Code review notesSolid, well-motivated surgery — the Bulk→REST removal is clean (no dangling references to 1.
|
d6f881f to
e384720
Compare
e384720 to
78d5094
Compare
- per-request token refresh: syncs longer than the SF session timeout no longer fail on expired tokens - num_workers typed as string (K8s Secret stringData), default 1 - drop salesforce_streams override; CRM_STREAMS is source of truth - stop_sync_on_stream_failure=false: one stream fail no longer aborts the rest - pin airbyte-cdk <8 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
- map only 003-prefixed WhoId to contact_id; leads polluted half the links - per tenant+source incremental watermark in salesforce staging and crm silver classes; global max dropped late-syncing sources' rows - drop populated silver placeholders: row-count guard locked in stale schemas Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
… macro doc - Remove unused IncrementalRestSalesforceStream._get_updated_state: stream state is owned by ConcurrentCursor, and the method read a never-assigned self._slice attribute. - Raise airbyte-cdk floor to >=7.23.1 and requests to >=2.34.2 (CVE-2024-35195). - Fix the on-run-start comment in dbt_project.yml to describe the macro's two-factor placeholder detection (marker + staging materialised). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Aleksandr Barkhatov <pm@aleks.bar>
78d5094 to
8fcc767
Compare
2 participants
/queryAllonly; remove the Bulk APIpath.
num_workerstyped as string (K8s Secret stringData), default 1;stop_sync_on_stream_failure=false; dropsalesforce_streamsoverride.(DateTime64/Date32) and drop User fields the org never exposes.
contact_id.classes; drop populated silver placeholders (row-count guard locked in stale
schemas).
Summary by CodeRabbit
Changes
/queryAllby default (Bulk API 2.0 behavior removed).Bug Fixes / Reliability
Documentation