Repository navigation
π Source Salesforce: make paging work, if cursor_field is not changed inside one page - #17615
Conversation
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
|
/test connector=connectors/source-salesforce
Build FailedTest summary info: |
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
|
/test connector=connectors/source-salesforce
Build FailedTest summary info: |
|
/test connector=connectors/source-salesforce
Build PassedTest summary info: |
| order_by_fields = [self.cursor_field] | ||
| if self.primary_key: | ||
| order_by_fields.append(self.primary_key) | ||
| query += f"ORDER BY {','.join(order_by_fields)} ASC LIMIT {self.page_size}" |
There was a problem hiding this comment.
Could there be a case where:
date=2 primarykey=2
date=3 primarykey=1
If so, wouldn't we miss the second entry?
There was a problem hiding this comment.
yes it seems you are right
it seems we need only add primarykey to filtering if we inside that same date batch
There was a problem hiding this comment.
OMG things become even more complex
There was a problem hiding this comment.
Yes, because the order by date is mandatory to make the incremental process work. (since we need the most recent date to be saved in the state)
There was a problem hiding this comment.
I'm under the impression this discussion is resolved with the upstream logic. Isn't it?
cursor_field not changed in one page
cursor_field not changed in one pagecursor_field is not changed inside one page
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
|
/test connector=connectors/source-salesforce
Build PassedTest summary info: |
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
|
/test connector=connectors/source-salesforce
Build PassedTest summary info: |
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
|
/test connector=connectors/source-salesforce
Build FailedTest summary info: |
|
/test connector=connectors/source-salesforce
Build FailedTest summary info: |
| query = f"SELECT {','.join(selected_properties.keys())} FROM {self.name} " | ||
| if start_date: | ||
| query += f"WHERE {self.cursor_field} >= {start_date} " | ||
| if primary_key and self.name not in UNSUPPORTED_FILTERING_STREAMS: |
There was a problem hiding this comment.
I understand why we need to check primary_key here, but what is the purpose of self.name not in UNSUPPORTED_FILTERING_STREAMS as part of the condition? Is this intended to dictate whether or not the stream supports using a WHERE clause in query?
This block feels very confusing to read with a lot of repeated conditions which makes it a little hard to reason about and I'm trying to think of ways we can streamline this query builder a bit
There was a problem hiding this comment.
Yes, good point. I agree it's not very obvious.
In really I just try to be consistent with this part of code, look at master branch:
Because my "improved" query strictly relaies on ORDER BY
if ORDER BY is not used I also skip my "improved" query
There was a problem hiding this comment.
Brian Lai (@brianjlai) Searching the documentation, some of those UNSUPPORTED_FILTERING_STREAMS fields support only the "where" filter which is consistent with the master logic. Though, I am not sure about it (since for those fields without the "order" the next date can be random ?).
I tried to find any refactors to the code to maintain that consistency but failed to do so.
There was a problem hiding this comment.
I tried to find any refactors to the code to maintain that consistency but failed to do so.
We could add a supports_filtering attribute to BulkIncrementalSalesforceStream and set this attribute on instantiation after checking if name in UNSUPPORTED_FILTERING_STREAMS
|
Serhii Chvaliuk (@grubberr) feel free to ask for a review from my side once you have implemented Brian Lai (@brianjlai) and Ana Loureiro (@anamargaridarl) suggestions. |
|
Serhii Chvaliuk (@grubberr) The PR can be reviewed by Augustin (@alafanechere) from my side |
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
Augustin (alafanechere)
left a comment
There was a problem hiding this comment.
Nice job, thank you! Minor suggestions
| order_by_fields = [self.cursor_field] | ||
| if self.primary_key: | ||
| order_by_fields.append(self.primary_key) | ||
| query += f"ORDER BY {','.join(order_by_fields)} ASC LIMIT {self.page_size}" |
There was a problem hiding this comment.
I'm under the impression this discussion is resolved with the upstream logic. Isn't it?
| {"Field1": "test", "Id": 3, "LastModifiedDate": LastModifiedDate1}, | ||
| {"Field1": "test", "Id": 5, "LastModifiedDate": LastModifiedDate1}, | ||
| {"Field1": "test", "Id": 2, "LastModifiedDate": LastModifiedDate2}, | ||
| {"Field1": "test", "Id": 2, "LastModifiedDate": LastModifiedDate2}, # duplicate record |
There was a problem hiding this comment.
Why did you add a duplicate to the test?
There was a problem hiding this comment.
I just want to show here that duplicates are possible
it's not related to my changes
There was a problem hiding this comment.
All right, is it something you've already observed with the API?
There was a problem hiding this comment.
it's possible because SF can store multiple records for the same cursor field.
Because we query using >= on subsequent queries we can grab duplicates
I mean 2-nd query can produce some result which was on 1-st query
There was a problem hiding this comment.
Understood, can we deduplicate downstream in the connector or is it too complex? Do you mind opening an issue if you think it's something we can address later?
There was a problem hiding this comment.
yes I think it's possible for incremental streams which have primary_key
we can play with ORDER BY cursor_field, primary_key
or maybe store in memory all primary_key for the latest cursor
Anyway, I think it is posible but I would prefer to do it in another issue
because it can take some time
Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
req -> get_query Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
|
/test connector=connectors/source-salesforce
Build PassedTest summary info: |
|
Augustin (@alafanechere) what you do think can I publish and merge this PR ? |
|
/publish connector=connectors/source-salesforce
if you have connectors that successfully published but failed definition generation, follow step 4 here |
β¦ed inside one page (airbytehq#17615) Signed-off-by: Sergey Chvalyuk <grubberr@gmail.com>
What
This PR also related to this oncall https://github.com/airbytehq/oncall/issues/735
How
Data:
cursor_field=2022-10-01, Pk=1
cursor_field=2022-10-01, Pk=3
cursor_field=2022-10-01, Pk=5
cursor_field=2022-10-02, Pk=2
cursor_field=2022-10-02, Pk=4
cursor_field=2022-10-02, Pk=6
New querying with page_size=2
SELECT * FROM table WHERE cursor_field >= 2022-10-01 ORDER BY cursor_field, pk LIMIT 2-> cursor_field=2022-10-01, Pk=1
-> cursor_field=2022-10-01, Pk=3
SELECT * FROM table WHERE (cursor_field == 2022-10-01 AND pk > 3) OR (cursor_field > 2022-10-01) ORDER BY cursor_field, pk LIMIT 2-> cursor_field=2022-10-01, Pk=5
-> cursor_field=2022-10-02, Pk=2
SELECT * FROM table WHERE cursor_field >= 2022-10-02 ORDER BY cursor_field, pk LIMIT 2-> cursor_field=2022-10-02, Pk=2 (duplicate, it was before this PR too)
-> cursor_field=2022-10-02, Pk=4
SELECT * FROM table WHERE (cursor_field == 2022-10-02 AND pk > 4) OR (cursor_field > 2022-10-02) ORDER BY cursor_field, pk LIMIT 2-> cursor_field=2022-10-02, Pk=6
Recommended reading order
x.javay.pythonπ¨ User Impact π¨
Are there any breaking changes? What is the end result perceived by the user? If yes, please merge this PR with the π¨π¨ emoji so changelog authors can further highlight this if needed.
Pre-merge Checklist
Expand the relevant checklist and delete the others.
Updating a connector
Community member or Airbyter
airbyte_secret./gradlew :airbyte-integrations:connectors:<name>:integrationTest.README.mdbootstrap.md. See description and examplesdocs/integrations/<source or destination>/<name>.mdincluding changelog. See changelog exampleAirbyter
If this is a community PR, the Airbyte engineer reviewing this PR is responsible for the below items.
/test connector=connectors/<name>command is passing/publishcommand described hereTests
Unit
Put your unit tests output here.
Integration
Put your integration tests output here.
Acceptance
Put your acceptance tests output here.