You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "leekeiabstraction (via GitHub)" <gi...@apache.org> on 2026/01/20 14:37:17 UTC
[PR] Sender wiring [fluss-rust]
leekeiabstraction opened a new pull request, #184:
URL: https://github.com/apache/fluss-rust/pull/184
### Purpose
Linked issue: close #183
Wire up KvWriteBatch within Sender
Tested using example Kv table in: PR181
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709450885
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -211,87 +212,161 @@ impl Sender {
records_by_bucket.insert(request_batch.table_bucket.clone(), request_batch);
}
- let response = match connection.request(request).await {
- Ok(response) => response,
- Err(e) => {
- self.handle_batches_with_error(
- table_buckets
- .iter()
- .filter_map(|bucket| records_by_bucket.remove(bucket))
- .collect(),
- FlussError::NetworkException,
- format!("Failed to send produce request: {e}"),
- )
- .await?;
- continue;
- }
- };
-
- self.handle_produce_response(
+ self.send_and_handle_response(
+ &connection,
+ write_request,
table_id,
&table_buckets,
&mut records_by_bucket,
- response,
)
.await?;
}
Ok(())
}
- async fn handle_produce_response(
+ fn build_write_request(
+ table_id: i64,
+ acks: i16,
+ timeout_ms: i32,
+ request_batches: &mut [ReadyWriteBatch],
+ ) -> Result<Option<WriteRequest>> {
+ let first_batch = match request_batches.first() {
+ None => return Ok(None),
+ Some(b) => &b.write_batch,
+ };
+
+ let request = match first_batch {
+ WriteBatch::ArrowLog(_) => {
+ let req = ProduceLogRequest::new(table_id, acks, timeout_ms, request_batches)?;
+ WriteRequest::ProduceLog(req)
+ }
+ WriteBatch::Kv(kv_write_batch) => {
+ let target_columns = kv_write_batch.target_columns.clone();
+ if let Some(batch) = request_batches.iter().next() {
Review Comment:
Very well spotted!
Eyes / brain fatigue from looking at code all day :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709450885
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -211,87 +212,161 @@ impl Sender {
records_by_bucket.insert(request_batch.table_bucket.clone(), request_batch);
}
- let response = match connection.request(request).await {
- Ok(response) => response,
- Err(e) => {
- self.handle_batches_with_error(
- table_buckets
- .iter()
- .filter_map(|bucket| records_by_bucket.remove(bucket))
- .collect(),
- FlussError::NetworkException,
- format!("Failed to send produce request: {e}"),
- )
- .await?;
- continue;
- }
- };
-
- self.handle_produce_response(
+ self.send_and_handle_response(
+ &connection,
+ write_request,
table_id,
&table_buckets,
&mut records_by_bucket,
- response,
)
.await?;
}
Ok(())
}
- async fn handle_produce_response(
+ fn build_write_request(
+ table_id: i64,
+ acks: i16,
+ timeout_ms: i32,
+ request_batches: &mut [ReadyWriteBatch],
+ ) -> Result<Option<WriteRequest>> {
+ let first_batch = match request_batches.first() {
+ None => return Ok(None),
+ Some(b) => &b.write_batch,
+ };
+
+ let request = match first_batch {
+ WriteBatch::ArrowLog(_) => {
+ let req = ProduceLogRequest::new(table_id, acks, timeout_ms, request_batches)?;
+ WriteRequest::ProduceLog(req)
+ }
+ WriteBatch::Kv(kv_write_batch) => {
+ let target_columns = kv_write_batch.target_columns.clone();
+ if let Some(batch) = request_batches.iter().next() {
Review Comment:
Very well spotted!
Eyes / brain fatigue from looking at code all day :D
Addressed in latest commit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709584004
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -211,87 +215,159 @@ impl Sender {
records_by_bucket.insert(request_batch.table_bucket.clone(), request_batch);
}
- let response = match connection.request(request).await {
- Ok(response) => response,
- Err(e) => {
- self.handle_batches_with_error(
- table_buckets
- .iter()
- .filter_map(|bucket| records_by_bucket.remove(bucket))
- .collect(),
- FlussError::NetworkException,
- format!("Failed to send produce request: {e}"),
- )
- .await?;
- continue;
- }
- };
-
- self.handle_produce_response(
+ self.send_and_handle_response(
+ &connection,
+ write_request,
table_id,
&table_buckets,
&mut records_by_bucket,
- response,
)
.await?;
}
Ok(())
}
- async fn handle_produce_response(
+ fn build_write_request(
+ table_id: i64,
+ acks: i16,
+ timeout_ms: i32,
+ request_batches: &mut [ReadyWriteBatch],
+ ) -> Result<WriteRequest> {
+ let first_batch = &request_batches.first().unwrap().write_batch;
+
+ let request = match first_batch {
Review Comment:
Should we symmetrically validate ArrowLog as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #184:
URL: https://github.com/apache/fluss-rust/pull/184
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#issuecomment-3774202447
@fresh-borzoni @luoyuxia Appreciate the review. Addressed your comments. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "AndreaBozzo (via GitHub)" <gi...@apache.org>.
AndreaBozzo commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2710121664
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -211,87 +215,159 @@ impl Sender {
records_by_bucket.insert(request_batch.table_bucket.clone(), request_batch);
}
- let response = match connection.request(request).await {
- Ok(response) => response,
- Err(e) => {
- self.handle_batches_with_error(
- table_buckets
- .iter()
- .filter_map(|bucket| records_by_bucket.remove(bucket))
- .collect(),
- FlussError::NetworkException,
- format!("Failed to send produce request: {e}"),
- )
- .await?;
- continue;
- }
- };
-
- self.handle_produce_response(
+ self.send_and_handle_response(
+ &connection,
+ write_request,
table_id,
&table_buckets,
&mut records_by_bucket,
- response,
)
.await?;
}
Ok(())
}
- async fn handle_produce_response(
+ fn build_write_request(
+ table_id: i64,
+ acks: i16,
+ timeout_ms: i32,
+ request_batches: &mut [ReadyWriteBatch],
+ ) -> Result<WriteRequest> {
+ let first_batch = &request_batches.first().unwrap().write_batch;
+
+ let request = match first_batch {
+ WriteBatch::ArrowLog(_) => {
+ let req = ProduceLogRequest::new(table_id, acks, timeout_ms, request_batches)?;
+ WriteRequest::ProduceLog(req)
+ }
+ WriteBatch::Kv(kv_write_batch) => {
+ let target_columns = kv_write_batch.target_columns();
+ for batch in request_batches.iter().skip(1) {
+ match &batch.write_batch {
+ WriteBatch::ArrowLog(_) => {
+ return Err(UnexpectedError {
+ message: "Expecting KvWriteBatch but found ArrowLogWriteBatch"
+ .to_string(),
+ source: None,
+ });
+ }
+ WriteBatch::Kv(kvb) => {
+ if target_columns != kvb.target_columns() {
+ return Err(UnexpectedError {
+ message: format!(
+ "All the write batches to make put kv request should have the same target columns, but got {:?} and {:?}.",
+ target_columns,
+ kvb.target_columns()
+ ),
+ source: None,
+ });
+ }
+ }
+ }
+ }
+ let cols = target_columns
+ .map(|arc| arc.iter().map(|&c| c as i32).collect())
+ .unwrap_or_default();
+ let req = PutKvRequest::new(table_id, acks, timeout_ms, cols, request_batches)?;
+ WriteRequest::PutKv(req)
+ }
+ };
+
+ Ok(request)
+ }
+
+ async fn send_and_handle_response(
+ &self,
+ connection: &ServerConnection,
+ write_request: WriteRequest,
+ table_id: i64,
+ table_buckets: &[TableBucket],
+ records_by_bucket: &mut HashMap<TableBucket, ReadyWriteBatch>,
+ ) -> Result<()> {
+ macro_rules! send {
+ ($request:expr) => {
+ match connection.request($request).await {
+ Ok(response) => {
+ self.handle_write_response(
+ table_id,
+ table_buckets,
+ records_by_bucket,
+ response,
Review Comment:
cool macro, but i wonder if this can become an async trait
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709019786
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -182,23 +187,19 @@ impl Sender {
.iter()
.filter_map(|bucket| records_by_bucket.remove(bucket))
.collect();
- if request_batches.is_empty() {
- continue;
- }
- let request = match ProduceLogRequest::new(
+
+ let write_request = match Self::build_write_request(
Review Comment:
I still prefer the previous code where we explicitly check if request_batches.is_empty() { continue; }. Making build_write_request return None feels a bit clunky because it hides the skip logic inside the builder. Keeping the check in the main loop makes the execution flow more transparent at a glance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#issuecomment-3773751046
@luoyuxia ack on integration test.
There's still one more bug I'm chasing down on lookup side before writing the integ test. Will raise the PRs when I'm done fixing that lookup bug
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709459980
##########
crates/fluss/src/client/write/batch.rs:
##########
@@ -250,7 +250,7 @@ impl ArrowLogWriteBatch {
pub struct KvWriteBatch {
write_batch: InnerWriteBatch,
kv_batch_builder: KvRecordBatchBuilder,
- target_columns: Option<Arc<Vec<usize>>>,
+ pub target_columns: Option<Arc<Vec<usize>>>,
Review Comment:
Addressed in latest commit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709622733
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -211,87 +215,159 @@ impl Sender {
records_by_bucket.insert(request_batch.table_bucket.clone(), request_batch);
}
- let response = match connection.request(request).await {
- Ok(response) => response,
- Err(e) => {
- self.handle_batches_with_error(
- table_buckets
- .iter()
- .filter_map(|bucket| records_by_bucket.remove(bucket))
- .collect(),
- FlussError::NetworkException,
- format!("Failed to send produce request: {e}"),
- )
- .await?;
- continue;
- }
- };
-
- self.handle_produce_response(
+ self.send_and_handle_response(
+ &connection,
+ write_request,
table_id,
&table_buckets,
&mut records_by_bucket,
- response,
)
.await?;
}
Ok(())
}
- async fn handle_produce_response(
+ fn build_write_request(
+ table_id: i64,
+ acks: i16,
+ timeout_ms: i32,
+ request_batches: &mut [ReadyWriteBatch],
+ ) -> Result<WriteRequest> {
+ let first_batch = &request_batches.first().unwrap().write_batch;
+
+ let request = match first_batch {
Review Comment:
Ah, I see, It's a bit hidden invariant at work, I think we may assume that all the batches by table_id are of the same type, so first batch is fine here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#issuecomment-3773210757
@luoyuxia @fresh-borzoni Would appreciate review here.
Found this gap when running the kv table example
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709135410
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -211,87 +212,161 @@ impl Sender {
records_by_bucket.insert(request_batch.table_bucket.clone(), request_batch);
}
- let response = match connection.request(request).await {
- Ok(response) => response,
- Err(e) => {
- self.handle_batches_with_error(
- table_buckets
- .iter()
- .filter_map(|bucket| records_by_bucket.remove(bucket))
- .collect(),
- FlussError::NetworkException,
- format!("Failed to send produce request: {e}"),
- )
- .await?;
- continue;
- }
- };
-
- self.handle_produce_response(
+ self.send_and_handle_response(
+ &connection,
+ write_request,
table_id,
&table_buckets,
&mut records_by_bucket,
- response,
)
.await?;
}
Ok(())
}
- async fn handle_produce_response(
+ fn build_write_request(
+ table_id: i64,
+ acks: i16,
+ timeout_ms: i32,
+ request_batches: &mut [ReadyWriteBatch],
+ ) -> Result<Option<WriteRequest>> {
+ let first_batch = match request_batches.first() {
+ None => return Ok(None),
+ Some(b) => &b.write_batch,
+ };
+
+ let request = match first_batch {
+ WriteBatch::ArrowLog(_) => {
+ let req = ProduceLogRequest::new(table_id, acks, timeout_ms, request_batches)?;
+ WriteRequest::ProduceLog(req)
+ }
+ WriteBatch::Kv(kv_write_batch) => {
+ let target_columns = kv_write_batch.target_columns.clone();
+ if let Some(batch) = request_batches.iter().next() {
Review Comment:
why do we compare first batch with itself?
Should it be cycle here and skip first batch
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709162018
##########
crates/fluss/src/client/write/batch.rs:
##########
@@ -250,7 +250,7 @@ impl ArrowLogWriteBatch {
pub struct KvWriteBatch {
write_batch: InnerWriteBatch,
kv_batch_builder: KvRecordBatchBuilder,
- target_columns: Option<Arc<Vec<usize>>>,
+ pub target_columns: Option<Arc<Vec<usize>>>,
Review Comment:
nit: better to hide it and not expose directly
``` rust
pub fn target_columns(&self) -> Option<&Arc<Vec<usize>>> {
self.target_columns.as_ref()
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KvWriteBatch wiring in Sender [fluss-rust]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #184:
URL: https://github.com/apache/fluss-rust/pull/184#discussion_r2709461039
##########
crates/fluss/src/client/write/sender.rs:
##########
@@ -182,23 +187,19 @@ impl Sender {
.iter()
.filter_map(|bucket| records_by_bucket.remove(bucket))
.collect();
- if request_batches.is_empty() {
- continue;
- }
- let request = match ProduceLogRequest::new(
+
+ let write_request = match Self::build_write_request(
Review Comment:
Addressed in latest commit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org