(paimon-rust) branch main updated: fix(read): honor read.batch-size for primary-key sort-merge output (#1239)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git


The following commit(s) were added to refs/heads/main by this push:
     new dac0b1a6 fix(read): honor read.batch-size for primary-key sort-merge 
output (#1239)
dac0b1a6 is described below

commit dac0b1a68e67132941c9c57a6ab82db1fa9ffd3d
Author: ZENOTME <st810918...@gmail.com>
AuthorDate: Wed Oct 28 17:49:56 2026 +0800

    fix(read): honor read.batch-size for primary-key sort-merge output (#1239)
---
 crates/paimon/src/table/kv_file_reader.rs | 11 ++++++++---
 crates/paimon/src/table/sort_merge.rs     |  1 -
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/crates/paimon/src/table/kv_file_reader.rs 
b/crates/paimon/src/table/kv_file_reader.rs
index e5486e9e..d14b9f12 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -832,6 +832,7 @@ impl KeyValueFileReader {
                         !CoreOptions::new(&config.table_options)
                             .sequence_field_sort_order_is_ascending(),
                     )
+                    .with_batch_size(config.read_batch_size)
                     .build()?;
 
                     while let Some(batch) = merge_stream.next().await {
@@ -2393,7 +2394,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn 
kv_input_decode_honors_read_batch_size_without_changing_merge_batching() {
+    async fn kv_input_decode_and_merge_output_honor_read_batch_size() {
         let file_io = test_file_io();
         let table_path = "memory:/kv_read_batch_size";
         setup_dirs(&file_io, table_path).await;
@@ -2456,8 +2457,12 @@ mod tests {
         let mut decoded_batch_sizes = 
input_batch_sizes.lock().unwrap().clone();
         decoded_batch_sizes.sort_unstable();
         assert_eq!(decoded_batch_sizes, vec![1, 1, 2, 2, 2, 2]);
-        assert_eq!(batches.len(), 1, "merge output batching stays 
independent");
-        assert_eq!(batches[0].num_rows(), 5);
+        let output_batch_sizes: Vec<usize> = batches.iter().map(|b| 
b.num_rows()).collect();
+        assert_eq!(
+            output_batch_sizes,
+            vec![2, 2, 1],
+            "merge output follows read.batch-size"
+        );
         assert_eq!(int_column(&batches, "value"), vec![11, 21, 31, 41, 51]);
     }
 
diff --git a/crates/paimon/src/table/sort_merge.rs 
b/crates/paimon/src/table/sort_merge.rs
index 579ca39b..b0cdec9f 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -1252,7 +1252,6 @@ impl SortMergeReaderBuilder {
         }
     }
 
-    #[cfg(test)]
     pub(crate) fn with_batch_size(mut self, batch_size: usize) -> Self {
         self.batch_size = batch_size;
         self