feat(snowflake-sink): change to streaming upload instead of batched bulk load

[xzhseh]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

as titled.

related: #15429 (comment).

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[xzhseh]

note that some difference when changing to streaming upload, e.g.

• previously data is evenly distributed by parallel writer(s) and sink directly to the intermediate s3.
• using streaming upload under the context example use-case (i.e., the one specified in integration_tests/snowflake-sink) would lead to intermediate file(s) in s3 being further (evenly) split to multiple smaller files - due to the streaming nature.

as an example, using my personal s3 bucket and the example source table / mv in integration_tests/snowflake-sink/..., 25 KB * 4 files will appear intermediately in previous approach, ~19.1 KB * 4 + ~2.1 KB * 4 files will appear intemediately when changing to streaming upload, just as shown below.

[xzhseh]

plus, do we need special parameters tuning for the configurations?

[fuyufjh]

Not necessarily, but it would be better to allow users to pass parameters to the S3 client (OpenDAL) if possible. Just for illustration, any parameters start with opendal.param_name = param_val with be passed to OpenDAL.

[xzhseh]

but it would be better to allow users to pass parameters to the S3 client (OpenDAL) if possible.

agree, I could put the support of extra configurations for s3 client in subsequent pr.

[wenym1]

Actually the idea of using streaming upload is not to simply change the way to upload a complete Vec<u8> buffer, but to replace the buffer with the streaming uploader. The ideal implementation will be,

1. at the very beginning, we create a streaming uploader.
2. when new data comes, we do some conversion and write the data to the streaming uploader
3. when checkpoint barrier comes, we close the writer and commit the file to snowflake
4. when the first next data comes, initialize a new streaming uploader

[xzhseh]

according to #16269 (comment), I refactor the SnowflakeSinkWriter and now we don't need to batch intermediate in-memory data in payload, instead we'll write them to streaming_uploader, which will be initialized / updated per epoch, and whenever a ckpt barrier comes (and there indeed has some data in the current streaming_uploader) we will finalize the entire streaming upload, and notify snowflake pipe about this.

and as a result - we don't need configuration like snowflake.max_batch_row_num anymore, everything is handled internally and transparent to users.

cc @wenym1 @fuyufjh @xxhZs.

[fuyufjh]

LGTM for the rest

[wenym1]

Rest LGTM! Thanks for the efforts to update the PR!

[wenym1]

LGTM!