feat(data-warehouse): Added incremental syncs for postgres #23145
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Gilbert09
commented
Jun 21, 2024
Comment on lines
+13
to
+17
| class IncrementalField(TypedDict): | ||
| label: str # Label shown in the UI | ||
| type: IncrementalFieldType # Field type shown in the UI | ||
| field: str # Actual DB field accessed | ||
| field_type: IncrementalFieldType # Actual DB type of the field |
There was a problem hiding this comment.
We have the extra fields for sources where the incremental field is actually masked by an ast.ExpressionField - e.g. in Stripe tables, we expose a created_at datetime, but the underlying data and Stripe API is actually created and integer.
|
Size Change: +197 B (+0.02%) Total Size: 1.06 MB ℹ️ View Unchanged
|
EDsCODE
approved these changes
Jun 24, 2024
| except Exception as e: | ||
| logger.exception(f"Could not clean up data import folder: {job.folder_path}", exc_info=e) | ||
| logger.exception(f"Could not clean up data import folder: {job.folder_path()}", exc_info=e) |
There was a problem hiding this comment.
What was the reason for shifting these from property to function?
There was a problem hiding this comment.
It needed to be wrapped in a sync_to_async call in a test because calling the property was failing from async issues - couldn't figure out how to do the wrapping as a property, and so just changed it to a func instead
| for name, field_type in filter_postgres_incremental_fields(columns) | ||
| ] | ||
| elif source.source_type == ExternalDataSource.Type.SNOWFLAKE: | ||
| # TODO(@Gilbert09): Move all this into a util and replace elsewhere |
Suspect IssuesThis pull request was deployed and Sentry observed the following issues:
Did you find this useful? React with a 👍 or 👎 |
timgl
pushed a commit
that referenced
this pull request
Jun 27, 2024
* Added incremental syncs for postgres * Tests * Fixed mypy * Added extra protection on the sync form * Set the correct write disposition for SQL * PR fixes * Fixed tests * Fixed tests * Fixed tests
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Firstly, sorry for the huge PR - this kinda snowballed a fair amount
Problem
Changes
date,timestamp, andintfields as the incremental fieldExternalDataSchemaand added logic there for triggering refreshes etcsync_type_payloadonto the schema object. We'll be storingincremental_fieldandincremental_field_typehere.reset_pipelinevalue to thesource'sjob_inputswhich then gets reset at the end of the sync. This is to tell DLT to do a full reset of the tableScreen.Recording.2024-06-21.at.13.56.26.mov
Screen.Recording.2024-06-21.at.13.57.29.mov
Does this work well for both Cloud and self-hosted?
Yup
How did you test this code?