You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/07/14 02:57:35 UTC
[PR] [rust] Hash-distribute log-table appends by declared bucket key [fluss]
fresh-borzoni opened a new pull request, #3643:
URL: https://github.com/apache/fluss/pull/3643
Log tables with a declared bucket key weren't hashing appends by it and every row in a flush landed in one bucket (sticky), unlike the KV/upsert path and the Java client, so bucket-key co-location was silently broken.
Now the append path hashes by the bucket key like Java: `append(row)` hashes each row, and `append_arrow_batch` splits a batch across buckets (single-bucket batches send as-is, mixed batches fan out per bucket).
Tables without a bucket key are unchanged.
--
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] [rust] Hash-distribute log-table appends by declared bucket key [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #3643:
URL: https://github.com/apache/fluss/pull/3643#issuecomment-4965004945
@charlesdong1991 @leekeiabstraction 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] [rust] Hash-distribute log-table appends by declared bucket key [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #3643:
URL: https://github.com/apache/fluss/pull/3643#issuecomment-5028985888
@charlesdong1991 Thank you for the review, addressed, 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] [rust] Hash-distribute log-table appends by declared bucket key [fluss]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #3643:
URL: https://github.com/apache/fluss/pull/3643#discussion_r3610794369
##########
fluss-rust/crates/fluss/src/client/table/append.rs:
##########
@@ -141,12 +186,61 @@ impl AppendWriter {
Arc::new(PhysicalTablePath::of(Arc::clone(&self.table_path)))
};
+ let Some(router) = self.bucket_router.as_ref() else {
+ return self.send_arrow_batch(batch, physical_table_path, None);
+ };
+ if batch.num_rows() == 0 {
+ return self.send_arrow_batch(batch, physical_table_path, None);
Review Comment:
so empty batch will result in sticky assigner, but if the first batch is empty, will this mistake all following keyed appends?
maybe we can add a test to see how it behaves?
##########
fluss-rust/crates/fluss/src/client/table/append.rs:
##########
@@ -141,12 +186,61 @@ impl AppendWriter {
Arc::new(PhysicalTablePath::of(Arc::clone(&self.table_path)))
};
+ let Some(router) = self.bucket_router.as_ref() else {
+ return self.send_arrow_batch(batch, physical_table_path, None);
+ };
+ if batch.num_rows() == 0 {
+ return self.send_arrow_batch(batch, physical_table_path, None);
+ }
+
+ // Group rows by bucket, keeping one key per bucket (it hashes back there).
+ let num_buckets = self.table_info.get_num_buckets();
+ let batch_arc = Arc::new(batch.clone());
+ let row_type = Arc::new(self.table_info.row_type.clone());
+ let mut groups: HashMap<i32, (Vec<u32>, Bytes)> = HashMap::new();
+ for i in 0..batch.num_rows() {
+ let row = ColumnarRow::new(Arc::clone(&batch_arc), Arc::clone(&row_type), i, None)?;
+ let (bucket, key) = router.bucket_of(&row, num_buckets)?;
Review Comment:
Nit: it might not be very efficient when row numbers are large given this will do full typed-batch builds on each iteration, because it will also downcast each column? maybe can utilize set_row_id in ColumnarRow and move the cursor instead?
--
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] [rust] Hash-distribute log-table appends by declared bucket key [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3643:
URL: https://github.com/apache/fluss/pull/3643#discussion_r3618659419
##########
fluss-rust/crates/fluss/src/client/table/append.rs:
##########
@@ -141,12 +186,61 @@ impl AppendWriter {
Arc::new(PhysicalTablePath::of(Arc::clone(&self.table_path)))
};
+ let Some(router) = self.bucket_router.as_ref() else {
+ return self.send_arrow_batch(batch, physical_table_path, None);
+ };
+ if batch.num_rows() == 0 {
+ return self.send_arrow_batch(batch, physical_table_path, None);
Review Comment:
Good catch, it's actually a bug, thank you for spotting, fixed
--
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] [rust] Hash-distribute log-table appends by declared bucket key [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3643:
URL: https://github.com/apache/fluss/pull/3643#discussion_r3618661667
##########
fluss-rust/crates/fluss/src/client/table/append.rs:
##########
@@ -141,12 +186,61 @@ impl AppendWriter {
Arc::new(PhysicalTablePath::of(Arc::clone(&self.table_path)))
};
+ let Some(router) = self.bucket_router.as_ref() else {
+ return self.send_arrow_batch(batch, physical_table_path, None);
+ };
+ if batch.num_rows() == 0 {
+ return self.send_arrow_batch(batch, physical_table_path, None);
+ }
+
+ // Group rows by bucket, keeping one key per bucket (it hashes back there).
+ let num_buckets = self.table_info.get_num_buckets();
+ let batch_arc = Arc::new(batch.clone());
+ let row_type = Arc::new(self.table_info.row_type.clone());
+ let mut groups: HashMap<i32, (Vec<u32>, Bytes)> = HashMap::new();
+ for i in 0..batch.num_rows() {
+ let row = ColumnarRow::new(Arc::clone(&batch_arc), Arc::clone(&row_type), i, None)?;
+ let (bucket, key) = router.bucket_of(&row, num_buckets)?;
Review Comment:
Good idea 👍
--
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] [rust] Hash-distribute log-table appends by declared bucket key [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #3643:
URL: https://github.com/apache/fluss/pull/3643
--
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