feat(sink): support async for mongodb dynamodb

[xxhZs]

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

What's changed and what's your intention?

as title and fix #17572

bench mongo db
before async

[avg: 93830 rows/s
p90: 98304 rows/s
p95: 98304 rows/s
p99: 100352 rows/s]

after async

avg: 146075 rows/s
p90: 163840 rows/s
p95: 167936 rows/s
p99: 169984 rows/s

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

We remove mogodb's option bulk_write_max_entries, and dynamodb's option default_max_batch_rows
We Add dynamodb's option max_batch_item_nums max_future_send_nums
The max_batch_item_nums is the max num of items in a batch_write_item, which should be >1 and <=25, and we set the default value to 25.
https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html
The max_future_send_nums is the num of write futures that exist at the same time, he is related to the max parallelism units set by the user in dynamodb, this value is theoretically equal to max_parallelism_units /(stream_chunk_size /max_batch_item_nums), the default value of max parallelism units is 40000, so the default value of this value should be <360, here we default value is 256

We also need to prompt the user to select the appropriate max parallelism units for dynamodb, when the throughout of RisingWave writes > the max parallelism units set by dynamodb, an error will be reported

After this pr sink_douple is default for mongodb , dynamodb , redis

[wenym1]

Rest LGTM

[wenym1]

Remove the dead code.

[wenym1]

The flush_insert can change to consume the ownership of insert_builder. So can flush_upsert.

[wenym1]

We can change to return try_join_all(...).boxed() here so that we can avoid allocating an unnecessary intermediate vec.

We can collect the future by converting the builder into an iterator of future by code like the following

try_join_all(upsert_builder.into_iter().flat_map(|(ns, builder)| {
    let (upsert, delete) = builder.build();
    let db = self.client.database(&ns.0);
    upsert
        .map(|upsert| Self::send_bulk_write_command(db.clone(), upsert))
        .into_iter()
        .chain(delete.map(|delete| Self::send_bulk_write_command(db, delete)))
}))

So as flush_insert.

[wenym1]

I just realize that both flush_upsert and flush_insert are returning a vector of the same future generated by this method.

If so, we can make the return type of flush_upsert and flush_insert to be the same, and then we don't need to add a boxed to the try_join_all future.

This method can be defined in this way

type SendBulkWriteCommandFuture = impl Future<Output = Result<()>> + 'static;

fn send_bulk_write_command(db: Database, command: Document) -> SendBulkWriteCommandFuture {
   async move {
        ...
    }
}

And then both flush_upsert and flush_insert can change to return TryJoinAll<SendBulkWriteCommandFuture>, and in write_chunk we don't need an extra boxed to unify the type.

[wenym1]

Rest LGTM. Thanks for the PR.