Source s3 cursor with history adapter by clnoll · Pull Request #29028 · airbytehq/airbyte

Catherine Noll (clnoll) · 2023-08-03T03:08:20Z

Creates a Cursor object for source S3 that adapts state message in the old format to the new format.

Depends on #29027.

github-actions · 2023-08-03T03:08:40Z

Before Merging a Connector Pull Request

Wow! What a great pull request you have here! 🎉

To merge this PR, ensure the following has been done/considered for each connector added or updated:

PR name follows PR naming conventions
Breaking changes are considered. If a Breaking Change is being introduced, ensure an Airbyte engineer has created a Breaking Change Plan.
Connector version has been incremented in the Dockerfile and metadata.yaml according to our Semantic Versioning for Connectors guidelines
You've updated the connector's metadata.yaml file any other relevant changes, including a breakingChanges entry for major version bumps. See metadata.yaml docs
Secrets in the connector's spec are annotated with airbyte_secret
All documentation files are up to date. (README.md, bootstrap.md, docs.md, etc...)
Changelog updated in docs/integrations/<source or destination>/<name>.md with an entry for the new version. See changelog example
Migration guide updated in docs/integrations/<source or destination>/<name>-migrations.md with an entry for the new version, if the version is a breaking change. See migration guide example
If set, you've ensured the icon is present in the platform-internal repo. (Docs)

If the checklist is complete, but the CI check is failing,

Check for hidden checklists in your PR description
Toggle the github label checklist-action-run on/off to re-run the checklist CI.

Octavia Squidington III (octavia-squidington-iii) · 2023-08-03T03:28:23Z

source-s3 test report (commit `e462779b5e`) - ❌

⏲️ Total pipeline duration: 17mn43s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
QA checks	✅
Code format checks	❌
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	❌

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-05T20:17:00Z

source-s3 test report (commit `5dd8d066f5`) - ❌

⏲️ Total pipeline duration: 21mn53s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	❌
QA checks	✅
Code format checks	❌
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-06T02:40:33Z

source-s3 test report (commit `ebe3cd975a`) - ❌

⏲️ Total pipeline duration: 01mn39s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	❌
Code format checks	❌
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	❌

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-06T03:15:26Z

source-s3 test report (commit `7a6a6b973d`) - ❌

⏲️ Total pipeline duration: 20mn35s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	❌
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-06T10:28:21Z

source-s3 test report (commit `10ce5fe70d`) - ❌

⏲️ Total pipeline duration: 21mn24s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	❌
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-06T21:04:35Z

source-s3 test report (commit `0a23e96e08`) - ❌

⏲️ Total pipeline duration: 21mn32s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	❌
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Octavia Squidington III (octavia-squidington-iii) · 2023-08-07T00:09:33Z

source-s3 test report (commit `fb0cf97942`) - ✅

⏲️ Total pipeline duration: 22mn09s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	✅
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Maxime Carbonneau-Leclerc (maxi297)

I feel like test coverage does not guarantee the behavior we would like. Can we test public methods instead?

Maxime Carbonneau-Leclerc (maxi297) · 2023-08-07T16:27:31Z

+                return True
+            else:
+                return False
+        if cursor := stream_state.get(self.cursor_field):


This seems like an edge case but if the cursor_field change, the state will be considered as V3. Is that ok?

Good call-out, and I'm okay with that since it's an edge case, and we will hopefully be deprecating v3 before too long.

Maxime Carbonneau-Leclerc (maxi297) · 2023-08-07T16:32:52Z

+    ],
+)
+def test_convert_dates(input_history: MutableMapping[str, Any], expected_output: MutableMapping[str, Any]) -> None:
+    assert Cursor._convert_legacy_history(input_history) == expected_output


It seems like this test does not guarantee that the public interface is working as expected i.e. set_initial_state could do something completely unexpected and we won't know with this coverage. Should we test set_initial_state instead?

That's a good call, updated tests to use that.

Btw this change caught a bug. Updated code in airbyte-integrations/connectors/source-s3/source_s3/v4/cursor.py if you want to take another look.

Alexandre Girard (girarda) · 2023-08-07T17:18:03Z

+            timestamp_millis = stream_state[self.cursor_field].split("_")[0]
+            converted_state[self.cursor_field] = self._get_ts_from_millis_ts(timestamp_millis, "%Y-%m-%dT%H:%M:%SZ")
+        if "history" in stream_state:
+            converted_state["history"] = converted_history


nit: I think we can always set converted_state["history"] = converted_history. It'll just be an empty object if the stream_state doesn't have a history field

Brian Lai (brianjlai) · 2023-08-07T18:01:49Z

+def get_source(args: List[str]):
+    catalog_path = AirbyteEntrypoint.extract_catalog(args)
+    try:
+        return FileBasedSource(SourceS3StreamReader(), Config, catalog_path, cursor_cls=Cursor)


can you elaborate on why or what scenario we need this additional try/catch a opposed to just initializing it the way we had been doing before?

Yes so I added this in order to make the CATs pass. When check errors, they expect the error to be output as an AirbyteMessage. But if we initialize FileBasedSource outside of the try that won't happen.

Octavia Squidington III (octavia-squidington-iii) · 2023-08-08T04:08:44Z

source-s3 test report (commit `0127a8f6db`) - ✅

⏲️ Total pipeline duration: 22mn12s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	✅
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Alexandre Girard (girarda)

🚢 🚢 🚢

Octavia Squidington III (octavia-squidington-iii) · 2023-08-09T22:45:06Z

source-s3 test report (commit `78dcbe85eb`) - ❌

⏲️ Total pipeline duration: 24mn22s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	✅
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	❌
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

Brian Lai (brianjlai)

just small nits, nothing major so looks good!

Brian Lai (brianjlai) · 2023-08-10T18:54:01Z

+        if not timestamp:
+            return timestamp
+        try:
+            timestamp_millis = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ")


nit: but since this is used in mroe than one place in the file (also on line 460), can we make this a constant

Brian Lai (brianjlai) · 2023-08-10T18:54:56Z

+
+            for filename in filenames:
+                if filename in converted_history:
+                    if date_obj > datetime.strptime(converted_history[filename], "%Y-%m-%dT%H:%M:%S.%fZ"):


same here let's use a constant since its used 3 times

Octavia Squidington III (octavia-squidington-iii) · 2023-08-10T23:47:43Z

source-s3 test report (commit `c14d044e66`) - ✅

⏲️ Total pipeline duration: 24mn28s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	✅
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

…sages

Octavia Squidington III (octavia-squidington-iii) · 2023-08-11T00:32:00Z

source-s3 test report (commit `5e32482768`) - ✅

⏲️ Total pipeline duration: 24mn34s

Step	Result
Validate airbyte-integrations/connectors/source-s3/metadata.yaml	✅
Connector version semver check	✅
Connector version increment check	✅
QA checks	✅
Code format checks	✅
Connector package install	✅
Build source-s3 docker image for platform linux/x86_64	✅
Unit tests	✅
Integration tests	✅
Acceptance tests	✅

🔗 View the logs here

☁️ View runs for commit in Dagger Cloud

Please note that tests are only run on PR ready for review. Please set your PR to draft mode to not flood the CI engine and upstream service on following commits.
You can run the same pipeline locally on this branch with the airbyte-ci tool with the following command

airbyte-ci connectors --name=source-s3 test

…sages (airbytehq#29028)

source-s3 previously used the legacy DefaultFileBasedCursor via the source-s3-specific Cursor subclass. That cursor goes through the legacy non-concurrent file-based read path, which emits a state message per processed file. With ~10 k files in a stream, the platform/orchestrator buffers ~10 k state messages (each carrying the growing history dict) waiting for the destination to ACK them; on slow destinations the orchestrator pod OOMs, the source pod is torn down before emitting terminal stream status, and the destination fails with TransientErrorException("Input was fully read, but some streams did not receive a terminal stream status message"). Move source-s3 onto FileBasedConcurrentCursor (CDK) which: - emits state at most once per 600s during a sync (state throttle added in airbytehq/airbyte-python-cdk#1032); - always force-emits a final state via ensure_at_least_one_state_emitted; - runs the read on the concurrent file-based path with up to DEFAULT_CONCURRENCY (100) workers. Verified locally against a customer connection with ~9 k files: state messages dropped from 9 200 to 2, runtime from 52:46 to 4:24, stdout volume from 4.1 GB to 922 MB, sync completes successfully with a STREAM_STATUS COMPLETE. The legacy source-s3 Cursor class only existed to host the v3-to-v4 state migration shipped in 2023 (#29028). 2.5 years on, all active customers have completed the migration on their first v4 sync, so the subclass and its tests are removed. A hypothetical customer with stale v3 state would trigger one full re-sync, after which they'd be on v4. Pinning to airbyte-cdk 7.19.2.post3.dev26244645194, a prerelease that includes both the state throttle and a fallback-branch dispatch fix for the file-based source so connectors using FileBasedConcurrentCursor do not crash on partial-catalog reads / check / discover.

Catherine Noll (clnoll) requested a review from a team August 3, 2023 03:08

Octavia Squidington III (octavia-squidington-iii) added area/connectors Connector related issues CDK Connector Development Kit connectors/source/s3 labels Aug 3, 2023

Catherine Noll (clnoll) requested review from Brian Lai (brianjlai), Alexandre Girard (girarda) and Maxime Carbonneau-Leclerc (maxi297) August 3, 2023 03:08

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from e462779 to 5dd8d06 Compare August 5, 2023 19:53

Octavia Squidington III (octavia-squidington-iii) removed the CDK Connector Development Kit label Aug 5, 2023

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from ebe3cd9 to 7a6a6b9 Compare August 6, 2023 02:53

Octavia Squidington III (octavia-squidington-iii) added the area/documentation Improvements or additions to documentation label Aug 6, 2023

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from efd77d9 to 10ce5fe Compare August 6, 2023 10:05

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from 10ce5fe to 0a23e96 Compare August 6, 2023 20:41

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from 0a23e96 to fb0cf97 Compare August 6, 2023 23:45

Maxime Carbonneau-Leclerc (maxi297) reviewed Aug 7, 2023

View reviewed changes

Alexandre Girard (girarda) reviewed Aug 7, 2023

View reviewed changes

Brian Lai (brianjlai) reviewed Aug 7, 2023

View reviewed changes

Octavia Squidington III (octavia-squidington-iii) added the CDK Connector Development Kit label Aug 8, 2023

Catherine Noll (clnoll) requested review from Brian Lai (brianjlai), Alexandre Girard (girarda) and Maxime Carbonneau-Leclerc (maxi297) August 9, 2023 21:48

Alexandre Girard (girarda) approved these changes Aug 9, 2023

View reviewed changes

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from 0127a8f to 78dcbe8 Compare August 9, 2023 22:18

Brian Lai (brianjlai) approved these changes Aug 10, 2023

View reviewed changes

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from 78dcbe8 to c14d044 Compare August 10, 2023 23:21

Source S3: maintain backwards compatibility between V3 & V4 state mes…

5e32482

…sages

Catherine Noll (clnoll) force-pushed the source-s3-cursor-with-history-adapter branch from c14d044 to 5e32482 Compare August 11, 2023 00:05

Catherine Noll (clnoll) merged commit 6946052 into master Aug 11, 2023

Catherine Noll (clnoll) deleted the source-s3-cursor-with-history-adapter branch August 11, 2023 15:38

Harry Touloupas (harrytou) pushed a commit to KYVENetwork/airbyte that referenced this pull request Sep 1, 2023

Source S3: maintain backwards compatibility between V3 & V4 state mes…

628c1c8

…sages (airbytehq#29028)

Conversation

Catherine Noll (clnoll) commented Aug 3, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Aug 3, 2023

Before Merging a Connector Pull Request

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 3, 2023

source-s3 test report (commit e462779b5e) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 5, 2023

source-s3 test report (commit 5dd8d066f5) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 6, 2023

source-s3 test report (commit ebe3cd975a) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 6, 2023

source-s3 test report (commit 7a6a6b973d) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 6, 2023

source-s3 test report (commit 10ce5fe70d) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 6, 2023

source-s3 test report (commit 0a23e96e08) - ❌

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 7, 2023

source-s3 test report (commit fb0cf97942) - ✅

Uh oh!

Maxime Carbonneau-Leclerc (maxi297) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 8, 2023

source-s3 test report (commit 0127a8f6db) - ✅

Uh oh!

Alexandre Girard (girarda) left a comment

Choose a reason for hiding this comment

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 9, 2023

source-s3 test report (commit 78dcbe85eb) - ❌

Uh oh!

Brian Lai (brianjlai) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 10, 2023

source-s3 test report (commit c14d044e66) - ✅

Uh oh!

Octavia Squidington III (octavia-squidington-iii) commented Aug 11, 2023

source-s3 test report (commit 5e32482768) - ✅

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Catherine Noll (clnoll) commented Aug 3, 2023 •

edited

Loading

source-s3 test report (commit `e462779b5e`) - ❌

source-s3 test report (commit `5dd8d066f5`) - ❌

source-s3 test report (commit `ebe3cd975a`) - ❌

source-s3 test report (commit `7a6a6b973d`) - ❌

source-s3 test report (commit `10ce5fe70d`) - ❌

source-s3 test report (commit `0a23e96e08`) - ❌

source-s3 test report (commit `fb0cf97942`) - ✅

source-s3 test report (commit `0127a8f6db`) - ✅

source-s3 test report (commit `78dcbe85eb`) - ❌

source-s3 test report (commit `c14d044e66`) - ✅

source-s3 test report (commit `5e32482768`) - ✅