snowpipe: exactly once semantics #3060
Merged
+1,739
−482
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
rockwotj
force-pushed
the
snow-once
branch
4 times, most recently
from
e5423ce to
10a350f
Compare
rockwotj
force-pushed
the
snow-once
branch
5 times, most recently
from
f4c4952 to
afe38a5
Compare
mihaitodor
approved these changes
Jan 6, 2025
There was a problem hiding this comment.
Nice one @rockwotj! 🏆 I spotted a few small typos and such, but looks OK otherwise. I didn't try to play with it locally, but please let me know if you'd like me to test it.
this is what is required for exactly once. We don't yet use it.
NOTE that since we can't ensure certain messages can go to a specific channel at the moment, this only really works with max_in_flight=1, which is probably fine for postgres, but another commit will support channel_name properly, so one can specify explicitly the mapping from data to channel.
This will help to re-use all this logic when we create the new output that specifies channel names explicitly.
To a seperate function so it can be used between different outputs.
To clarify it, instead of spreading it out all over, this also means the schema migration function can now be a free function
One that is responsible for coordination of schema evolution and other small pieces (like custom mappings). The purpose of this is to allow for another kind of inner output that can allow for a user to specifically set the channel name (instead of using a pool).
I'm not sure if this is 100% correct, but it will work for most cases.
See the examples on what this enables with a Redpanda/Kafka input (but not kafka_franz!).
This seems a bit clearer and has nice duality with the indexed pool
By holding a lock when doing this during WriteBatch, and not having the framework call Connect outside of pipeline creation, just handle it internally.
I think this is what was missing...
We should try not to always run a SQL query everytime we startup for cost reasons. Instead of running a query (which is likely flaky because of identifier normalization anyways), just open the channel lazily and catch the specific error for the table not existing, then create the table and retry.
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.
Support 2 new properties in
snowflake_streaming:offset_token: A new property to support exactly once delivery: https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-overview#offset-tokenschannel_name: The ability to explicitly assign a batch to a channel. The currentchannel_prefixoption doesn't support explicitly picking a channel, this allows exactly once from Kafka.