[TASK-263] Remove async from queueing while writing#271
Merged
Conversation
Member
Author
|
@luoyuxia @leekeiabstraction |
There was a problem hiding this comment.
Pull request overview
This PR removes unnecessary async from the write queueing path by switching internal queue synchronization to parking_lot::Mutex, making upsert(), delete(), and append() queue synchronously while still allowing async acknowledgement via a returned handle/future; async behavior is retained for flush() since it performs network I/O.
Changes:
- Replace async queueing (
tokio::sync::Mutex) with synchronous locking (parking_lot::Mutex) in the core write accumulator and writer client. - Update Rust API call sites (tests/examples/docs) to remove the first
awaiton queueing methods while keeping per-record ack via awaiting the returned handle. - Apply the same synchronous-queue + async-ack pattern to Python and C++ bindings (including a new Python
WriteResultHandle).
Reviewed changes
Copilot reviewed 18 out of 18 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/fluss/src/client/write/accumulator.rs | Switch queue locks to parking_lot::Mutex and make queue operations synchronous (append/ready/drain/re_enqueue). |
| crates/fluss/src/client/write/writer_client.rs | Make send() synchronous and remove awaits on accumulation. |
| crates/fluss/src/client/write/sender.rs | Adjust sender to use synchronous accumulator APIs; keep async for network/metadata updates. |
| crates/fluss/src/client/write/mod.rs | Update docs for new call pattern around WriteResultFuture. |
| crates/fluss/src/client/table/append.rs | Make append/append_arrow_batch synchronous queueing returning WriteResultFuture. |
| crates/fluss/src/client/table/upsert.rs | Make upsert/delete synchronous queueing returning WriteResultFuture. |
| crates/fluss/tests/integration/log_table.rs | Remove await from queueing calls to match the new synchronous API. |
| crates/fluss/tests/integration/kv_table.rs | Remove await from queueing calls and keep per-record ack via awaiting the returned handle. |
| crates/examples/src/example_table.rs | Update example to use synchronous queueing + async flush. |
| crates/examples/src/example_kv_table.rs | Update KV example for synchronous queueing + optional per-record ack. |
| crates/examples/src/example_partitioned_kv_table.rs | Update partitioned KV example for synchronous queueing + optional per-record ack. |
| bindings/python/src/write_handle.rs | Add Python WriteResultHandle with wait() for per-record ack. |
| bindings/python/src/upsert.rs | Make Python upsert/delete synchronous and return WriteResultHandle; keep flush() async. |
| bindings/python/src/table.rs | Make Python append APIs return WriteResultHandle; convert flush() to async/awaitable. |
| bindings/python/src/lib.rs | Export/register WriteResultHandle. |
| bindings/python/example/example.py | Update Python example to reflect synchronous queueing and awaiting flush()/handle.wait(). |
| bindings/cpp/src/lib.rs | Remove runtime blocking for append queueing; keep blocking for waiting/flush. |
| README.md | Update Rust snippet for synchronous queueing and add flush. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Member
Author
|
fixed failing tests, should be fine now |
cefd4b7 to
0802788
Compare
luoyuxia
approved these changes
Feb 7, 2026
Contributor
luoyuxia
left a comment
There was a problem hiding this comment.
@fresh-borzoni Thanks for the pr. LGTM!
This file contains hidden or 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
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.
Summary
closes #263
tokio::sync::Mutexwithparking_lot::Mutexin the core accumulator and writer clientupsert(),delete(),append()now queue writes synchronously (noawaitneeded) and return aWriteResultHandleWriteResultHandlesupports both fire-and-forget (ignore the handle) and per-record ack (await handle.wait())flush()remains async (real network I/O)