From 4b164648bfb536a5d00d6b9619ec5af85a802723 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Mon, 17 Aug 2026 18:26:56 +0800
Subject: [PATCH 01/16] perf(vindex): add detailed build timing logs

---
 .../paimon/src/table/data_evolution_reader.rs |  18 +-
 crates/paimon/src/table/data_file_reader.rs   |  86 +++++-
 crates/paimon/src/table/table_read.rs         |  24 +-
 .../src/table/vindex_index_build_builder.rs   | 252 +++++++++++++++---
 4 files changed, 329 insertions(+), 51 deletions(-)

diff --git a/crates/paimon/src/table/data_evolution_reader.rs b/crates/paimon/src/table/data_evolution_reader.rs
index f3f10a25..11a4cb96 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -20,7 +20,7 @@ mod blob_fallback;
 use super::blob_resolver::{BlobReadLimiter, BLOB_DESCRIPTOR_READ_CONCURRENCY};
 use super::data_file_reader::{
     append_null_row_id_column, attach_row_id, expand_selected_row_ids, insert_column_at,
-    DataFileReader,
+    DataFileReadTiming, DataFileReader,
 };
 use crate::arrow::format::FilePredicates;
 use crate::arrow::{build_target_arrow_schema, ParquetReadBudget};
@@ -114,6 +114,7 @@ pub(crate) struct DataEvolutionReader {
     blob_read_limiter: BlobReadLimiter,
     batch_size: Option<usize>,
     parquet_read_budget: Option<Arc<ParquetReadBudget>>,
+    read_timing: Option<Arc<DataFileReadTiming>>,
 }
 
 impl DataEvolutionReader {
@@ -191,6 +192,7 @@ impl DataEvolutionReader {
             blob_read_limiter: BlobReadLimiter::new(),
             batch_size: None,
             parquet_read_budget: None,
+            read_timing: None,
         })
     }
 
@@ -207,6 +209,11 @@ impl DataEvolutionReader {
         self
     }
 
+    pub(crate) fn with_read_timing(mut self, read_timing: Option<Arc<DataFileReadTiming>>) -> Self {
+        self.read_timing = read_timing;
+        self
+    }
+
     /// Read data files in data evolution mode.
     pub fn read(self, data_splits: &[DataSplit]) -> crate::Result<ArrowRecordBatchStream> {
         let splits: Vec<DataSplit> = data_splits.to_vec();
@@ -248,7 +255,8 @@ impl DataEvolutionReader {
                 },
             )
             .with_batch_size(self.batch_size)
-            .with_parquet_read_budget(self.parquet_read_budget.clone());
+            .with_parquet_read_budget(self.parquet_read_budget.clone())
+            .with_read_timing(self.read_timing.clone());
 
             for split in splits {
                 let row_ranges = split.row_ranges().map(|r| r.to_vec());
@@ -611,6 +619,7 @@ impl DataEvolutionReader {
         let blob_as_descriptor = self.blob_as_descriptor;
         let batch_size = self.batch_size;
         let parquet_read_budget = self.parquet_read_budget.clone();
+        let read_timing = self.read_timing.clone();
         let anchor_deletion_vector = anchor_deletion_vector.clone();
         // Batch size for column-merge output. Matches the default Parquet reader batch size.
         const MERGE_BATCH_SIZE: usize = 1024;
@@ -697,6 +706,7 @@ impl DataEvolutionReader {
                             batch_size,
                             blob_as_descriptor,
                             source_parquet_read_budget.clone(),
+                            read_timing.clone(),
                             anchor_deletion_vector.as_ref(),
                         )
                         .map(Some)
@@ -1228,6 +1238,7 @@ fn open_source_stream(
     batch_size: Option<usize>,
     blob_as_descriptor: bool,
     parquet_read_budget: Option<Arc<ParquetReadBudget>>,
+    read_timing: Option<Arc<DataFileReadTiming>>,
     anchor_deletion_vector: Option<&DeletionVectorContext>,
 ) -> crate::Result<ArrowRecordBatchStream> {
     let mut row_ranges = row_ranges;
@@ -1292,7 +1303,8 @@ fn open_source_stream(
     )
     .with_batch_size(batch_size)
     .with_blob_as_descriptor(blob_as_descriptor)
-    .with_parquet_read_budget(parquet_read_budget);
+    .with_parquet_read_budget(parquet_read_budget)
+    .with_read_timing(read_timing);
 
     match source {
         FieldSource::DataFile {
diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs
index 5954984a..c9e2e1a7 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -20,7 +20,7 @@ use crate::arrow::format::create_format_reader_with_budget;
 use crate::arrow::schema_evolution::{create_index_mapping, NULL_FIELD_INDEX};
 use crate::arrow::ParquetReadBudget;
 use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
-use crate::io::FileIO;
+use crate::io::{FileIO, FileRead};
 use crate::spec::{
     is_variant_extraction_row_type, DataField, DataFileMeta, DataType, Predicate, ROW_ID_FIELD_NAME,
 };
@@ -33,7 +33,51 @@ use arrow_cast::cast;
 
 use async_stream::try_stream;
 use futures::StreamExt;
+use std::ops::Range;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+#[derive(Debug, Default)]
+pub(crate) struct DataFileReadTiming {
+    file_read_nanos: AtomicU64,
+    parquet_decode_nanos: AtomicU64,
+}
+
+impl DataFileReadTiming {
+    fn add_file_read(&self, duration: Duration) {
+        self.file_read_nanos
+            .fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
+    }
+
+    fn add_parquet_decode(&self, duration: Duration) {
+        self.parquet_decode_nanos
+            .fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
+    }
+
+    pub(crate) fn file_read(&self) -> Duration {
+        Duration::from_nanos(self.file_read_nanos.load(Ordering::Relaxed))
+    }
+
+    pub(crate) fn parquet_decode(&self) -> Duration {
+        Duration::from_nanos(self.parquet_decode_nanos.load(Ordering::Relaxed))
+    }
+}
+
+struct TimedFileRead {
+    inner: Box<dyn FileRead>,
+    timing: Arc<DataFileReadTiming>,
+}
+
+#[async_trait::async_trait]
+impl FileRead for TimedFileRead {
+    async fn read(&self, range: Range<u64>) -> crate::Result<bytes::Bytes> {
+        let start = Instant::now();
+        let result = self.inner.read(range).await;
+        self.timing.add_file_read(start.elapsed());
+        result
+    }
+}
 
 /// Reads data from Parquet files.
 #[derive(Clone)]
@@ -48,6 +92,7 @@ pub(crate) struct DataFileReader {
     blob_as_descriptor: bool,
     batch_size: Option<usize>,
     parquet_read_budget: Option<Arc<ParquetReadBudget>>,
+    read_timing: Option<Arc<DataFileReadTiming>>,
 }
 
 impl DataFileReader {
@@ -70,6 +115,7 @@ impl DataFileReader {
             blob_as_descriptor: false,
             batch_size: None,
             parquet_read_budget: None,
+            read_timing: None,
         }
     }
 
@@ -91,6 +137,11 @@ impl DataFileReader {
         self
     }
 
+    pub(crate) fn with_read_timing(mut self, read_timing: Option<Arc<DataFileReadTiming>>) -> Self {
+        self.read_timing = read_timing;
+        self
+    }
+
     pub(crate) fn with_row_filter_factory(
         mut self,
         factory: Arc<dyn crate::arrow::RowFilterFactory>,
@@ -291,6 +342,7 @@ impl DataFileReader {
         let blob_as_descriptor = self.blob_as_descriptor;
         let batch_size = self.batch_size;
         let parquet_read_budget = self.parquet_read_budget.clone();
+        let read_timing = self.read_timing.clone();
 
         let target_schema = build_target_arrow_schema(&read_type)?;
         let file_fields = data_fields.clone().unwrap_or_else(|| table_fields.clone());
@@ -344,7 +396,19 @@ impl DataFileReader {
                 parquet_read_budget,
             )?;
             let input_file = file_io.new_input(&path_to_read)?;
+            let open_start = read_timing.as_ref().map(|_| Instant::now());
             let file_reader = input_file.reader().await?;
+            if let (Some(timing), Some(start)) = (read_timing.as_ref(), open_start) {
+                timing.add_file_read(start.elapsed());
+            }
+            let file_reader: Box<dyn FileRead> = match read_timing.as_ref() {
+                Some(timing) => Box::new(TimedFileRead {
+                    inner: Box::new(file_reader),
+                    timing: Arc::clone(timing),
+                }),
+                None => Box::new(file_reader),
+            };
+            let is_parquet = path_to_read.to_ascii_lowercase().ends_with(".parquet");
             let local_ranges = row_ranges.as_ref().map(|ranges| {
                 to_local_row_ranges(ranges, file_meta.first_row_id.unwrap_or(0), file_meta.row_count)
             });
@@ -364,7 +428,7 @@ impl DataFileReader {
             let mut row_id_offset = 0usize;
 
             let mut batch_stream = format_reader.read_batch_stream(
-                Box::new(file_reader),
+                file_reader,
                 file_meta.file_size as u64,
                 &format_read_fields,
                 file_predicates.as_ref(),
@@ -372,7 +436,23 @@ impl DataFileReader {
                 row_selection,
             ).await?;
 
-            while let Some(batch) = batch_stream.next().await {
+            loop {
+                let batch = if is_parquet {
+                    if let Some(timing) = read_timing.as_ref() {
+                        std::future::poll_fn(|cx| {
+                            let start = Instant::now();
+                            let batch = batch_stream.as_mut().poll_next(cx);
+                            timing.add_parquet_decode(start.elapsed());
+                            batch
+                        })
+                        .await
+                    } else {
+                        batch_stream.next().await
+                    }
+                } else {
+                    batch_stream.next().await
+                };
+                let Some(batch) = batch else { break };
                 let batch = batch?;
                 let num_rows = batch.num_rows();
                 let batch_schema = batch.schema();
diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs
index b36af841..4ff0c5a0 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -16,7 +16,7 @@
 // under the License.
 
 use super::data_evolution_reader::DataEvolutionReader;
-use super::data_file_reader::DataFileReader;
+use super::data_file_reader::{DataFileReadTiming, DataFileReader};
 use super::format_table_read::FormatTableRead;
 use super::incremental_scan::{IncrementalPlan, IncrementalScanMode, IncrementalSplit};
 use super::kv_file_reader::{KeyValueFileReader, KeyValueReadConfig};
@@ -158,6 +158,15 @@ impl<'a> TableRead<'a> {
         }
     }
 
+    pub(crate) fn with_data_file_read_timing(self, timing: Arc<DataFileReadTiming>) -> Self {
+        match self.0 {
+            TableReadKind::Paimon(read) => Self(TableReadKind::Paimon(
+                read.with_data_file_read_timing(timing),
+            )),
+            TableReadKind::Format(read) => Self(TableReadKind::Format(read)),
+        }
+    }
+
     /// Returns an [`ArrowRecordBatchStream`].
     pub fn to_arrow(&self, data_splits: &[DataSplit]) -> crate::Result<ArrowRecordBatchStream> {
         match &self.0 {
@@ -216,6 +225,7 @@ struct PaimonTableRead<'a> {
     data_predicates: Vec<Predicate>,
     row_filter_factory: Option<Arc<dyn crate::arrow::RowFilterFactory>>,
     parquet_read_budget: Option<Arc<ParquetReadBudget>>,
+    data_file_read_timing: Option<Arc<DataFileReadTiming>>,
 }
 
 impl<'a> PaimonTableRead<'a> {
@@ -231,6 +241,7 @@ impl<'a> PaimonTableRead<'a> {
             data_predicates,
             row_filter_factory: None,
             parquet_read_budget: None,
+            data_file_read_timing: None,
         }
     }
 
@@ -274,6 +285,11 @@ impl<'a> PaimonTableRead<'a> {
         self
     }
 
+    fn with_data_file_read_timing(mut self, timing: Arc<DataFileReadTiming>) -> Self {
+        self.data_file_read_timing = Some(timing);
+        self
+    }
+
     fn parquet_read_budget(&self) -> crate::Result<Arc<ParquetReadBudget>> {
         match &self.parquet_read_budget {
             Some(budget) => Ok(Arc::clone(budget)),
@@ -856,7 +872,8 @@ impl<'a> PaimonTableRead<'a> {
             self.table.rest_env().cloned(),
         )?
         .with_batch_size(Some(core_options.read_batch_size()?))
-        .with_parquet_read_budget(Some(self.parquet_read_budget()?));
+        .with_parquet_read_budget(Some(self.parquet_read_budget()?))
+        .with_read_timing(self.data_file_read_timing.clone());
         reader.read(data_splits)
     }
 
@@ -875,7 +892,8 @@ impl<'a> PaimonTableRead<'a> {
             self.data_predicates.clone(),
         )
         .with_batch_size(Some(self.table.schema().core_options().read_batch_size()?))
-        .with_parquet_read_budget(Some(self.parquet_read_budget()?));
+        .with_parquet_read_budget(Some(self.parquet_read_budget()?))
+        .with_read_timing(self.data_file_read_timing.clone());
         // The engine decoder filter is safe only on the plain append/raw path.
         // This constructor is also used by raw-convertible primary-key splits,
         // where positional merge semantics must remain untouched.
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index 7638f28c..86f36e99 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -19,6 +19,7 @@ use crate::spec::{
     bucket_dir_name, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, FileKind,
     GlobalIndexMeta, IndexFileMeta, ROW_ID_FIELD_NAME,
 };
+use crate::table::data_file_reader::DataFileReadTiming;
 use crate::table::source::exclude_row_ranges;
 use crate::table::{
     CommitMessage, DataSplit, DataSplitBuilder, RowRange, SnapshotManager, Table, TableCommit,
@@ -28,15 +29,89 @@ use crate::{Error, Result};
 use arrow_array::{Array, FixedSizeListArray, Float32Array, Int64Array, ListArray, RecordBatch};
 use arrow_buffer::MutableBuffer;
 use futures::TryStreamExt;
+use paimon_vindex_core::autotune::default_training_vector_count;
 use paimon_vindex_core::index::{VectorIndexTrainer, VectorIndexWriter};
 use paimon_vindex_core::io::PosWriter;
 use std::collections::HashMap;
 use std::io::{Read, Seek, SeekFrom};
+use std::sync::{Arc, OnceLock};
+use std::time::{Duration, Instant};
 use tokio::io::AsyncWriteExt;
 use tokio_util::io::SyncIoBridge;
 
 const INDEX_DIR: &str = "index";
 const VECTOR_BUFFER_BYTES: usize = 8 * 1024 * 1024;
+const VECTOR_INDEX_BUILD_TIMING_ENV: &str = "PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING";
+
+fn vector_index_build_timing_enabled() -> bool {
+    static ENABLED: OnceLock<bool> = OnceLock::new();
+    *ENABLED.get_or_init(|| {
+        std::env::var_os(VECTOR_INDEX_BUILD_TIMING_ENV).is_some_and(|value| value == "1")
+    })
+}
+
+struct VectorIndexBuildTiming {
+    total_without_commit: Duration,
+    source_batch_wait: Duration,
+    oss_read: Duration,
+    parquet_decode: Duration,
+    raw_temp_write: Duration,
+    train_finish: Duration,
+    raw_temp_reread: Duration,
+    index_add: Duration,
+    serialize_upload: Duration,
+    rows: usize,
+    training_rows_seen: usize,
+    training_rows_retained: usize,
+    batch_count: usize,
+    raw_temp_bytes: usize,
+    index_bytes: u64,
+    data_file_count: usize,
+    file_name: String,
+}
+
+impl VectorIndexBuildTiming {
+    fn log(self, index_type: &str, commit: Duration) {
+        let total = self.total_without_commit.saturating_add(commit);
+        let accounted = self
+            .source_batch_wait
+            .saturating_add(self.raw_temp_write)
+            .saturating_add(self.train_finish)
+            .saturating_add(self.raw_temp_reread)
+            .saturating_add(self.index_add)
+            .saturating_add(self.serialize_upload)
+            .saturating_add(commit);
+        let unattributed = total.saturating_sub(accounted);
+        eprintln!(
+            "event=paimon_vector_index_build index_type={} file={} rows={} training_rows_seen={} training_rows_retained={} batch_count={} raw_temp_bytes={} index_bytes={} source_batch_wait_ms={:.3} oss_read_ms={:.3} parquet_decode_ms={:.3} raw_temp_write_ms={:.3} train_finish_ms={:.3} raw_temp_reread_ms={:.3} index_add_ms={:.3} serialize_upload_ms={:.3} commit_ms={:.3} sample_read_ms=0.000 full_scan_add_ms=0.000 pipeline_blocked_ms=0.000 producer_blocked_ms=0.000 consumer_add_ms=0.000 data_file_count={} data_file_read_concurrency=1 peak_ready_batches=0 total_ms={:.3} unattributed_ms={:.3}",
+            index_type,
+            self.file_name,
+            self.rows,
+            self.training_rows_seen,
+            self.training_rows_retained,
+            self.batch_count,
+            self.raw_temp_bytes,
+            self.index_bytes,
+            self.source_batch_wait.as_secs_f64() * 1000.0,
+            self.oss_read.as_secs_f64() * 1000.0,
+            self.parquet_decode.as_secs_f64() * 1000.0,
+            self.raw_temp_write.as_secs_f64() * 1000.0,
+            self.train_finish.as_secs_f64() * 1000.0,
+            self.raw_temp_reread.as_secs_f64() * 1000.0,
+            self.index_add.as_secs_f64() * 1000.0,
+            self.serialize_upload.as_secs_f64() * 1000.0,
+            commit.as_secs_f64() * 1000.0,
+            self.data_file_count,
+            total.as_secs_f64() * 1000.0,
+            unattributed.as_secs_f64() * 1000.0,
+        );
+    }
+}
+
+struct BuiltIndexFile {
+    meta: IndexFileMeta,
+    timing: Option<VectorIndexBuildTiming>,
+}
 
 pub struct VindexIndexBuildBuilder<'a> {
     table: &'a Table,
@@ -172,8 +247,9 @@ impl<'a> VindexIndexBuildBuilder<'a> {
         );
         let shard_count = shards.len();
         let mut messages = Vec::with_capacity(shard_count);
+        let mut timings = Vec::with_capacity(shard_count);
         for shard in shards {
-            let index_file = match self
+            let built = match self
                 .build_index_file(
                     &shard,
                     index_column,
@@ -191,13 +267,23 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 }
             };
             let mut message = CommitMessage::new(shard.partition_bytes.clone(), 0, vec![]);
-            message.new_index_files = vec![index_file];
+            message.new_index_files = vec![built.meta];
             messages.push(message);
+            if let Some(timing) = built.timing {
+                timings.push(timing);
+            }
         }
 
+        let commit_start = vector_index_build_timing_enabled().then(Instant::now);
         commit
             .commit_if_latest_snapshot(messages, snapshot.id())
             .await?;
+        if let Some(commit_start) = commit_start {
+            let commit = commit_start.elapsed();
+            for timing in timings {
+                timing.log(&self.index_type, commit);
+            }
+        }
 
         Ok(shard_count)
     }
@@ -210,7 +296,13 @@ impl<'a> VindexIndexBuildBuilder<'a> {
         index_field_id: i32,
         options: &VindexVectorIndexOptions,
         index_meta: Vec<u8>,
-    ) -> Result<IndexFileMeta> {
+    ) -> Result<BuiltIndexFile> {
+        let timing_enabled = vector_index_build_timing_enabled();
+        let total_start = timing_enabled.then(Instant::now);
+        let mut source_batch_wait = Duration::ZERO;
+        let mut raw_temp_write = Duration::ZERO;
+        let read_timing = timing_enabled.then(|| Arc::new(DataFileReadTiming::default()));
+        let mut batch_count = 0usize;
         let row_count = checked_row_count(shard.row_range_start, shard.row_range_end)?;
         let row_count_usize = usize::try_from(row_count).map_err(|e| Error::DataInvalid {
             message: format!("Invalid vindex row count: {row_count}"),
@@ -252,6 +344,10 @@ impl<'a> VindexIndexBuildBuilder<'a> {
         let mut read_builder = self.table.new_read_builder();
         read_builder.with_projection(&[index_column, ROW_ID_FIELD_NAME])?;
         let read = read_builder.new_read()?;
+        let read = match read_timing.as_ref() {
+            Some(timing) => read.with_data_file_read_timing(Arc::clone(timing)),
+            None => read,
+        };
         let mut batches = read.to_arrow(&[split])?;
         let mut expected_row_id = shard.row_range_start;
         let mut rows_seen = 0usize;
@@ -259,7 +355,14 @@ impl<'a> VindexIndexBuildBuilder<'a> {
         let mut next_training_sample = 0usize;
         let mut training_buffer = Vec::with_capacity(training_buffer_floats);
 
-        while let Some(batch) = batches.try_next().await? {
+        loop {
+            let source_start = timing_enabled.then(Instant::now);
+            let batch = batches.try_next().await?;
+            if let Some(source_start) = source_start {
+                source_batch_wait = source_batch_wait.saturating_add(source_start.elapsed());
+            }
+            let Some(batch) = batch else { break };
+            batch_count += 1;
             let vectors =
                 validate_vector_batch(&batch, index_column, dimension_usize, &mut expected_row_id)?;
             let batch_end =
@@ -306,6 +409,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 }
             }
 
+            let raw_write_start = timing_enabled.then(Instant::now);
             raw_file
                 .write_all(vectors.bytes)
                 .await
@@ -313,6 +417,9 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                     message: format!("Failed to spill vindex vectors: {e}"),
                     source: Some(Box::new(e)),
                 })?;
+            if let Some(raw_write_start) = raw_write_start {
+                raw_temp_write = raw_temp_write.saturating_add(raw_write_start.elapsed());
+            }
             bytes_written = bytes_written
                 .checked_add(vectors.bytes.len())
                 .ok_or_else(|| Error::DataInvalid {
@@ -350,10 +457,14 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 source: None,
             });
         }
+        let raw_write_start = timing_enabled.then(Instant::now);
         raw_file.flush().await.map_err(|e| Error::UnexpectedError {
             message: format!("Failed to flush temporary vindex vector file: {e}"),
             source: Some(Box::new(e)),
         })?;
+        if let Some(raw_write_start) = raw_write_start {
+            raw_temp_write = raw_temp_write.saturating_add(raw_write_start.elapsed());
+        }
         let raw_file_len = raw_file
             .metadata()
             .await
@@ -371,42 +482,71 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             });
         }
         let raw_file = raw_file.into_std().await;
-
-        let writer = tokio::task::spawn_blocking(move || -> std::io::Result<VectorIndexWriter> {
-            let training = trainer.finish()?;
-            let mut writer = VectorIndexWriter::new(training);
-            let mut raw_file = raw_file;
-            raw_file.seek(SeekFrom::Start(0))?;
-            let batch_rows = training_buffer_rows.min(row_count_usize);
-            let batch_bytes = checked_std_vector_bytes(batch_rows, dimension_usize)?;
-            let mut buffer = MutableBuffer::new(batch_bytes);
-            let mut ids = Vec::with_capacity(batch_rows);
-            let mut rows_added = 0usize;
-            while rows_added < row_count_usize {
-                let rows = batch_rows.min(row_count_usize - rows_added);
-                buffer.resize(checked_std_vector_bytes(rows, dimension_usize)?, 0);
-                raw_file.read_exact(buffer.as_slice_mut())?;
-                ids.clear();
-                for row in rows_added..rows_added + rows {
-                    ids.push(i64::try_from(row).map_err(|_| {
-                        std::io::Error::new(
-                            std::io::ErrorKind::InvalidData,
-                            "vindex row id does not fit i64",
-                        )
-                    })?);
+        let training_rows_retained =
+            default_training_vector_count(training_vector_count, options.config.nlist()).map_err(
+                |e| Error::DataInvalid {
+                    message: format!("Failed to determine retained vindex training vectors: {e}"),
+                    source: Some(Box::new(e)),
+                },
+            )?;
+
+        let (writer, train_finish, raw_temp_reread, index_add) = tokio::task::spawn_blocking(
+            move || -> std::io::Result<(VectorIndexWriter, Duration, Duration, Duration)> {
+                let train_start = timing_enabled.then(Instant::now);
+                let training = trainer.finish()?;
+                let train_finish = train_start.map_or(Duration::ZERO, |start| start.elapsed());
+                let mut writer = VectorIndexWriter::new(training);
+                let mut raw_temp_reread = Duration::ZERO;
+                let mut index_add = Duration::ZERO;
+                let mut raw_file = raw_file;
+                let reread_start = timing_enabled.then(Instant::now);
+                raw_file.seek(SeekFrom::Start(0))?;
+                if let Some(start) = reread_start {
+                    raw_temp_reread = raw_temp_reread.saturating_add(start.elapsed());
                 }
-                writer.add_vectors(&ids, buffer.typed_data::<f32>(), rows)?;
-                rows_added += rows;
-            }
-            let mut trailing = [0u8; 1];
-            if raw_file.read(&mut trailing)? != 0 {
-                return Err(std::io::Error::new(
-                    std::io::ErrorKind::InvalidData,
-                    "temporary vindex vector file contains trailing bytes",
-                ));
-            }
-            Ok(writer)
-        })
+                let batch_rows = training_buffer_rows.min(row_count_usize);
+                let batch_bytes = checked_std_vector_bytes(batch_rows, dimension_usize)?;
+                let mut buffer = MutableBuffer::new(batch_bytes);
+                let mut ids = Vec::with_capacity(batch_rows);
+                let mut rows_added = 0usize;
+                while rows_added < row_count_usize {
+                    let rows = batch_rows.min(row_count_usize - rows_added);
+                    buffer.resize(checked_std_vector_bytes(rows, dimension_usize)?, 0);
+                    let reread_start = timing_enabled.then(Instant::now);
+                    raw_file.read_exact(buffer.as_slice_mut())?;
+                    if let Some(start) = reread_start {
+                        raw_temp_reread = raw_temp_reread.saturating_add(start.elapsed());
+                    }
+                    ids.clear();
+                    for row in rows_added..rows_added + rows {
+                        ids.push(i64::try_from(row).map_err(|_| {
+                            std::io::Error::new(
+                                std::io::ErrorKind::InvalidData,
+                                "vindex row id does not fit i64",
+                            )
+                        })?);
+                    }
+                    let add_start = timing_enabled.then(Instant::now);
+                    writer.add_vectors(&ids, buffer.typed_data::<f32>(), rows)?;
+                    if let Some(start) = add_start {
+                        index_add = index_add.saturating_add(start.elapsed());
+                    }
+                    rows_added += rows;
+                }
+                let mut trailing = [0u8; 1];
+                let reread_start = timing_enabled.then(Instant::now);
+                if raw_file.read(&mut trailing)? != 0 {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::InvalidData,
+                        "temporary vindex vector file contains trailing bytes",
+                    ));
+                }
+                if let Some(start) = reread_start {
+                    raw_temp_reread = raw_temp_reread.saturating_add(start.elapsed());
+                }
+                Ok((writer, train_finish, raw_temp_reread, index_add))
+            },
+        )
         .await
         .map_err(|e| Error::UnexpectedError {
             message: format!("vindex training task failed: {e}"),
@@ -417,6 +557,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             source: Some(Box::new(e)),
         })?;
 
+        let serialize_upload_start = timing_enabled.then(Instant::now);
         self.table
             .file_io()
             .mkdirs(&format!(
@@ -466,9 +607,11 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 return Err(error);
             }
         };
-        Ok(IndexFileMeta {
+        let serialize_upload =
+            serialize_upload_start.map_or(Duration::ZERO, |start| start.elapsed());
+        let meta = IndexFileMeta {
             index_type: self.index_type.clone(),
-            file_name,
+            file_name: file_name.clone(),
             file_size: checked_i64(
                 status.size,
                 "Index file is too large for Rust IndexFileMeta",
@@ -483,7 +626,32 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 source_meta: None,
                 index_meta: Some(index_meta),
             }),
-        })
+        };
+        let (oss_read, parquet_decode) = read_timing
+            .as_ref()
+            .map_or((Duration::ZERO, Duration::ZERO), |timing| {
+                (timing.file_read(), timing.parquet_decode())
+            });
+        let timing = total_start.map(|start| VectorIndexBuildTiming {
+            total_without_commit: start.elapsed(),
+            source_batch_wait,
+            oss_read,
+            parquet_decode,
+            raw_temp_write,
+            train_finish,
+            raw_temp_reread,
+            index_add,
+            serialize_upload,
+            rows: row_count_usize,
+            training_rows_seen: training_vector_count,
+            training_rows_retained,
+            batch_count,
+            raw_temp_bytes: bytes_written,
+            index_bytes: status.size,
+            data_file_count: shard.files.len(),
+            file_name,
+        });
+        Ok(BuiltIndexFile { meta, timing })
     }
 }
 

From ed768cb1e690f9414fb70f3f9e74c10520a00e51 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Tue, 18 Aug 2026 14:43:09 +0800
Subject: [PATCH 02/16] fix(vindex): gate training_rows_retained diagnostics on
 timing flag

The default_training_vector_count call was only used to populate a
timing log field, but it ran unconditionally and propagated errors,
introducing a new build failure path even when timing diagnostics
were disabled. Compute it only when timing is enabled and fall back
to 0 on error instead of failing the build.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 .../paimon/src/table/vindex_index_build_builder.rs | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index 86f36e99..a7ef0897 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -482,13 +482,13 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             });
         }
         let raw_file = raw_file.into_std().await;
-        let training_rows_retained =
-            default_training_vector_count(training_vector_count, options.config.nlist()).map_err(
-                |e| Error::DataInvalid {
-                    message: format!("Failed to determine retained vindex training vectors: {e}"),
-                    source: Some(Box::new(e)),
-                },
-            )?;
+        // Diagnostics only: never fail the build for a timing log field.
+        let training_rows_retained = if timing_enabled {
+            default_training_vector_count(training_vector_count, options.config.nlist())
+                .unwrap_or(0)
+        } else {
+            0
+        };
 
         let (writer, train_finish, raw_temp_reread, index_add) = tokio::task::spawn_blocking(
             move || -> std::io::Result<(VectorIndexWriter, Duration, Duration, Duration)> {

From ef692e2edbd867083df0782bb1b50709d203fc21 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Tue, 18 Aug 2026 16:11:40 +0800
Subject: [PATCH 03/16] build(vindex): use core 0.4.0

---
 Cargo.lock               | 4 +---
 bindings/c/Cargo.toml    | 2 +-
 crates/paimon/Cargo.toml | 2 +-
 3 files changed, 3 insertions(+), 5 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index a8e8270c..72d81d78 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4750,9 +4750,7 @@ dependencies = [
 
 [[package]]
 name = "paimon-vindex-core"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b2c67c916596e578ed09f78ace8b4c54fc45d262933c658e1ca7fdb61b722635"
+version = "0.4.0"
 dependencies = [
  "half",
  "matrixmultiply",
diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml
index af0a5121..51e3731c 100644
--- a/bindings/c/Cargo.toml
+++ b/bindings/c/Cargo.toml
@@ -43,4 +43,4 @@ serde_json = "1.0.120"
 # Test-only: the vector-search integration tests build a real primary-key vindex
 # IVF-flat ANN segment fixture in-process. Versions match crates/paimon.
 bytes = "1.7.1"
-paimon-vindex-core = "0.3.0"
+paimon-vindex-core = "0.4.0"
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 4c0bd8dd..4ed26f97 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -131,7 +131,7 @@ urlencoding = "2.1"
 paimon-mosaic-core = "0.2.0"
 paimon-ftindex-core = { version = "0.1.0", optional = true }
 tempfile = "3"
-paimon-vindex-core = "0.3.0"
+paimon-vindex-core = "0.4.0"
 vortex = { version = "0.75.0", features = ["tokio"], optional = true }
 libloading = "0.9"
 log = "0.4"

From 1b3128040191cc3609e6c8a1ecfe2f8fa3c7e49a Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Tue, 18 Aug 2026 16:11:52 +0800
Subject: [PATCH 04/16] perf(vindex): enlarge index add batches

---
 .../src/table/vindex_index_build_builder.rs   | 43 +++++++++++++++++--
 1 file changed, 40 insertions(+), 3 deletions(-)

diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index a7ef0897..4905f7df 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -41,6 +41,7 @@ use tokio_util::io::SyncIoBridge;
 
 const INDEX_DIR: &str = "index";
 const VECTOR_BUFFER_BYTES: usize = 8 * 1024 * 1024;
+const INDEX_ADD_BATCH_ROWS: usize = 32 * 1024;
 const VECTOR_INDEX_BUILD_TIMING_ENV: &str = "PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING";
 
 fn vector_index_build_timing_enabled() -> bool {
@@ -504,13 +505,14 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 if let Some(start) = reread_start {
                     raw_temp_reread = raw_temp_reread.saturating_add(start.elapsed());
                 }
-                let batch_rows = training_buffer_rows.min(row_count_usize);
+                let batch_rows = INDEX_ADD_BATCH_ROWS.min(row_count_usize);
                 let batch_bytes = checked_std_vector_bytes(batch_rows, dimension_usize)?;
                 let mut buffer = MutableBuffer::new(batch_bytes);
                 let mut ids = Vec::with_capacity(batch_rows);
                 let mut rows_added = 0usize;
                 while rows_added < row_count_usize {
                     let rows = batch_rows.min(row_count_usize - rows_added);
+                    let batch_end = checked_index_add_batch_end(rows_added, rows)?;
                     buffer.resize(checked_std_vector_bytes(rows, dimension_usize)?, 0);
                     let reread_start = timing_enabled.then(Instant::now);
                     raw_file.read_exact(buffer.as_slice_mut())?;
@@ -518,7 +520,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                         raw_temp_reread = raw_temp_reread.saturating_add(start.elapsed());
                     }
                     ids.clear();
-                    for row in rows_added..rows_added + rows {
+                    for row in rows_added..batch_end {
                         ids.push(i64::try_from(row).map_err(|_| {
                             std::io::Error::new(
                                 std::io::ErrorKind::InvalidData,
@@ -531,7 +533,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                     if let Some(start) = add_start {
                         index_add = index_add.saturating_add(start.elapsed());
                     }
-                    rows_added += rows;
+                    rows_added = batch_end;
                 }
                 let mut trailing = [0u8; 1];
                 let reread_start = timing_enabled.then(Instant::now);
@@ -1119,6 +1121,15 @@ fn checked_std_vector_bytes(row_count: usize, dimension: usize) -> std::io::Resu
         })
 }
 
+fn checked_index_add_batch_end(start: usize, rows: usize) -> std::io::Result<usize> {
+    start.checked_add(rows).ok_or_else(|| {
+        std::io::Error::new(
+            std::io::ErrorKind::InvalidInput,
+            "vindex index-add batch end overflows usize",
+        )
+    })
+}
+
 fn checked_training_vector_count(row_count: usize, ratio: f64) -> Result<usize> {
     if row_count == 0 || !(ratio > 0.0 && ratio <= 1.0) {
         return Err(Error::DataInvalid {
@@ -1482,6 +1493,32 @@ mod tests {
         assert!(checked_training_sample_index(usize::MAX, usize::MAX, 1).is_err());
     }
 
+    #[test]
+    fn test_index_add_batch_ranges() {
+        let ranges = |row_count| {
+            let mut result = Vec::new();
+            let mut start = 0;
+            while start < row_count {
+                let rows = INDEX_ADD_BATCH_ROWS.min(row_count - start);
+                let end = checked_index_add_batch_end(start, rows).unwrap();
+                result.push(start..end);
+                start = end;
+            }
+            result
+        };
+
+        assert_eq!(ranges(10), vec![0..10]);
+        assert_eq!(ranges(INDEX_ADD_BATCH_ROWS), vec![0..INDEX_ADD_BATCH_ROWS]);
+        assert_eq!(
+            ranges(INDEX_ADD_BATCH_ROWS + 7),
+            vec![
+                0..INDEX_ADD_BATCH_ROWS,
+                INDEX_ADD_BATCH_ROWS..INDEX_ADD_BATCH_ROWS + 7
+            ]
+        );
+        assert!(checked_index_add_batch_end(usize::MAX, 1).is_err());
+    }
+
     fn test_table_with_io(file_io: FileIO, table_path: &str, schema: Schema) -> Table {
         Table::new(
             file_io,

From 0cff10a7ecb04b33c022cd77c276bbca1f512394 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Tue, 18 Aug 2026 18:46:00 +0800
Subject: [PATCH 05/16] feat: diagnose parquet row group reads

---
 crates/paimon/src/arrow/format/parquet.rs     |  29 ++--
 .../paimon/src/arrow/parquet_read_budget.rs   | 136 ++++++++++++++++++
 .../src/table/vindex_index_build_builder.rs   |  32 ++++-
 3 files changed, 185 insertions(+), 12 deletions(-)

diff --git a/crates/paimon/src/arrow/format/parquet.rs b/crates/paimon/src/arrow/format/parquet.rs
index 76c76892..7c26cb73 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -490,29 +490,36 @@ impl FormatFileReader for ParquetFormatReader {
         // preserving positional `_ROW_ID`, sort order, and batch backpressure. Reads
         // with predicates or an explicit row selection retain the original
         // single-stream path until their selections are split per row group.
-        let row_group_parallelism = self
-            .read_budget
-            .as_ref()
-            .filter(|_| preds.is_empty() && row_filter_factory.is_none() && row_selection.is_none())
+        let read_budget = self.read_budget.as_ref().filter(|_| {
+            preds.is_empty() && row_filter_factory.is_none() && row_selection.is_none()
+        });
+        let row_group_parallelism = read_budget
             .map(|budget| {
                 budget
                     .parallelism()
                     .min(batch_stream_builder.metadata().num_row_groups())
             })
             .unwrap_or(1);
+        let projected_bytes = read_budget
+            .filter(|budget| row_group_parallelism > 1 || budget.diagnostics_enabled())
+            .map(|budget| {
+                let projected_bytes = batch_stream_builder
+                    .metadata()
+                    .row_groups()
+                    .iter()
+                    .map(|row_group| projected_row_group_bytes(row_group, &mask))
+                    .collect::<Vec<_>>();
+                budget.record_projected_row_groups(&projected_bytes);
+                projected_bytes
+            });
         if row_group_parallelism > 1 {
             let row_group_count = batch_stream_builder.metadata().num_row_groups();
             let reader_metadata = ArrowReaderMetadata::try_new(
                 batch_stream_builder.metadata().clone(),
                 ArrowReaderOptions::new(),
             )?;
-            let projected_bytes = batch_stream_builder
-                .metadata()
-                .row_groups()
-                .iter()
-                .map(|row_group| projected_row_group_bytes(row_group, &mask))
-                .collect::<Vec<_>>();
-            let read_budget = Arc::clone(self.read_budget.as_ref().expect("checked above"));
+            let projected_bytes = projected_bytes.expect("parallel row-group reads need sizes");
+            let read_budget = Arc::clone(read_budget.expect("checked above"));
             let (row_group_tx, mut row_group_rx) = mpsc::channel(row_group_parallelism);
             tokio::spawn(async move {
                 for (row_group_index, projected_bytes) in projected_bytes.into_iter().enumerate() {
diff --git a/crates/paimon/src/arrow/parquet_read_budget.rs b/crates/paimon/src/arrow/parquet_read_budget.rs
index e0f6e5cc..3e60129e 100644
--- a/crates/paimon/src/arrow/parquet_read_budget.rs
+++ b/crates/paimon/src/arrow/parquet_read_budget.rs
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
@@ -30,6 +31,42 @@ pub struct ParquetReadBudget {
     row_groups: Arc<Semaphore>,
     bytes: Arc<Semaphore>,
     byte_permits: u32,
+    diagnostics: Arc<ParquetReadDiagnostics>,
+}
+
+#[derive(Debug)]
+struct ParquetReadDiagnostics {
+    enabled: AtomicBool,
+    row_group_count: AtomicU64,
+    projected_bytes_min: AtomicU64,
+    projected_bytes_max: AtomicU64,
+    projected_bytes_total: AtomicU64,
+    current_inflight: AtomicUsize,
+    peak_inflight: AtomicUsize,
+}
+
+impl Default for ParquetReadDiagnostics {
+    fn default() -> Self {
+        Self {
+            enabled: AtomicBool::new(false),
+            row_group_count: AtomicU64::new(0),
+            projected_bytes_min: AtomicU64::new(u64::MAX),
+            projected_bytes_max: AtomicU64::new(0),
+            projected_bytes_total: AtomicU64::new(0),
+            current_inflight: AtomicUsize::new(0),
+            peak_inflight: AtomicUsize::new(0),
+        }
+    }
+}
+
+#[derive(Debug, Default, PartialEq, Eq)]
+pub(crate) struct ParquetReadDiagnosticsSnapshot {
+    pub(crate) row_group_count: u64,
+    pub(crate) projected_bytes_min: u64,
+    pub(crate) projected_bytes_max: u64,
+    pub(crate) projected_bytes_total: u64,
+    pub(crate) current_inflight: usize,
+    pub(crate) peak_inflight: usize,
 }
 
 impl ParquetReadBudget {
@@ -59,6 +96,7 @@ impl ParquetReadBudget {
             row_groups: Arc::new(Semaphore::new(parallelism)),
             bytes: Arc::new(Semaphore::new(byte_permits as usize)),
             byte_permits,
+            diagnostics: Arc::new(ParquetReadDiagnostics::default()),
         })
     }
 
@@ -66,6 +104,57 @@ impl ParquetReadBudget {
         self.parallelism
     }
 
+    pub(crate) fn enable_diagnostics(&self) {
+        self.diagnostics.enabled.store(true, Ordering::Relaxed);
+    }
+
+    pub(crate) fn diagnostics_enabled(&self) -> bool {
+        self.diagnostics.enabled.load(Ordering::Relaxed)
+    }
+
+    pub(crate) fn record_projected_row_groups(&self, projected_bytes: &[u64]) {
+        if !self.diagnostics_enabled() || projected_bytes.is_empty() {
+            return;
+        }
+        self.diagnostics
+            .row_group_count
+            .fetch_add(projected_bytes.len() as u64, Ordering::Relaxed);
+        self.diagnostics.projected_bytes_min.fetch_min(
+            *projected_bytes.iter().min().expect("checked non-empty"),
+            Ordering::Relaxed,
+        );
+        self.diagnostics.projected_bytes_max.fetch_max(
+            *projected_bytes.iter().max().expect("checked non-empty"),
+            Ordering::Relaxed,
+        );
+        self.diagnostics.projected_bytes_total.fetch_add(
+            projected_bytes
+                .iter()
+                .copied()
+                .fold(0u64, u64::saturating_add),
+            Ordering::Relaxed,
+        );
+    }
+
+    pub(crate) fn diagnostics(&self) -> ParquetReadDiagnosticsSnapshot {
+        let row_group_count = self.diagnostics.row_group_count.load(Ordering::Relaxed);
+        ParquetReadDiagnosticsSnapshot {
+            row_group_count,
+            projected_bytes_min: if row_group_count == 0 {
+                0
+            } else {
+                self.diagnostics.projected_bytes_min.load(Ordering::Relaxed)
+            },
+            projected_bytes_max: self.diagnostics.projected_bytes_max.load(Ordering::Relaxed),
+            projected_bytes_total: self
+                .diagnostics
+                .projected_bytes_total
+                .load(Ordering::Relaxed),
+            current_inflight: self.diagnostics.current_inflight.load(Ordering::Relaxed),
+            peak_inflight: self.diagnostics.peak_inflight.load(Ordering::Relaxed),
+        }
+    }
+
     pub(crate) async fn acquire(
         &self,
         projected_uncompressed_bytes: u64,
@@ -88,9 +177,21 @@ impl ParquetReadBudget {
                 message: "Parquet byte read budget was closed".to_string(),
                 source: Some(Box::new(error)),
             })?;
+        let diagnostics = self.diagnostics_enabled().then(|| {
+            let current = self
+                .diagnostics
+                .current_inflight
+                .fetch_add(1, Ordering::Relaxed)
+                + 1;
+            self.diagnostics
+                .peak_inflight
+                .fetch_max(current, Ordering::Relaxed);
+            Arc::clone(&self.diagnostics)
+        });
         Ok(ParquetReadPermit {
             _row_group: row_group,
             _bytes: bytes,
+            diagnostics,
         })
     }
 }
@@ -106,6 +207,15 @@ impl Default for ParquetReadBudget {
 pub(crate) struct ParquetReadPermit {
     _row_group: OwnedSemaphorePermit,
     _bytes: OwnedSemaphorePermit,
+    diagnostics: Option<Arc<ParquetReadDiagnostics>>,
+}
+
+impl Drop for ParquetReadPermit {
+    fn drop(&mut self) {
+        if let Some(diagnostics) = &self.diagnostics {
+            diagnostics.current_inflight.fetch_sub(1, Ordering::Relaxed);
+        }
+    }
 }
 
 #[cfg(test)]
@@ -132,6 +242,32 @@ mod tests {
             .unwrap();
     }
 
+    #[tokio::test]
+    async fn diagnostics_aggregate_shared_row_group_reads() {
+        let budget = Arc::new(ParquetReadBudget::new(2, 2 * BYTE_PERMIT_UNIT).unwrap());
+        budget.enable_diagnostics();
+        budget.record_projected_row_groups(&[300, 100, 200]);
+
+        let first = budget.acquire(1).await.unwrap();
+        let second = budget.acquire(1).await.unwrap();
+        assert_eq!(
+            budget.diagnostics(),
+            ParquetReadDiagnosticsSnapshot {
+                row_group_count: 3,
+                projected_bytes_min: 100,
+                projected_bytes_max: 300,
+                projected_bytes_total: 600,
+                current_inflight: 2,
+                peak_inflight: 2,
+            }
+        );
+
+        drop(first);
+        drop(second);
+        assert_eq!(budget.diagnostics().current_inflight, 0);
+        assert_eq!(budget.diagnostics().peak_inflight, 2);
+    }
+
     #[test]
     fn rejects_invalid_limits() {
         assert!(ParquetReadBudget::new(0, BYTE_PERMIT_UNIT).is_err());
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index 4905f7df..4aba438f 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -21,6 +21,7 @@ use crate::spec::{
 };
 use crate::table::data_file_reader::DataFileReadTiming;
 use crate::table::source::exclude_row_ranges;
+use crate::table::table_read::configured_parquet_read_budget;
 use crate::table::{
     CommitMessage, DataSplit, DataSplitBuilder, RowRange, SnapshotManager, Table, TableCommit,
 };
@@ -56,6 +57,11 @@ struct VectorIndexBuildTiming {
     source_batch_wait: Duration,
     oss_read: Duration,
     parquet_decode: Duration,
+    parquet_row_group_count: u64,
+    parquet_projected_bytes_min: u64,
+    parquet_projected_bytes_max: u64,
+    parquet_projected_bytes_total: u64,
+    parquet_peak_inflight_row_groups: usize,
     raw_temp_write: Duration,
     train_finish: Duration,
     raw_temp_reread: Duration,
@@ -84,7 +90,7 @@ impl VectorIndexBuildTiming {
             .saturating_add(commit);
         let unattributed = total.saturating_sub(accounted);
         eprintln!(
-            "event=paimon_vector_index_build index_type={} file={} rows={} training_rows_seen={} training_rows_retained={} batch_count={} raw_temp_bytes={} index_bytes={} source_batch_wait_ms={:.3} oss_read_ms={:.3} parquet_decode_ms={:.3} raw_temp_write_ms={:.3} train_finish_ms={:.3} raw_temp_reread_ms={:.3} index_add_ms={:.3} serialize_upload_ms={:.3} commit_ms={:.3} sample_read_ms=0.000 full_scan_add_ms=0.000 pipeline_blocked_ms=0.000 producer_blocked_ms=0.000 consumer_add_ms=0.000 data_file_count={} data_file_read_concurrency=1 peak_ready_batches=0 total_ms={:.3} unattributed_ms={:.3}",
+            "event=paimon_vector_index_build index_type={} file={} rows={} training_rows_seen={} training_rows_retained={} batch_count={} raw_temp_bytes={} index_bytes={} source_batch_wait_ms={:.3} oss_read_ms={:.3} parquet_decode_ms={:.3} parquet_row_group_count={} parquet_projected_bytes_min={} parquet_projected_bytes_max={} parquet_projected_bytes_total={} parquet_peak_inflight_row_groups={} raw_temp_write_ms={:.3} train_finish_ms={:.3} raw_temp_reread_ms={:.3} index_add_ms={:.3} serialize_upload_ms={:.3} commit_ms={:.3} sample_read_ms=0.000 full_scan_add_ms=0.000 pipeline_blocked_ms=0.000 producer_blocked_ms=0.000 consumer_add_ms=0.000 data_file_count={} data_file_read_concurrency=1 peak_ready_batches=0 total_ms={:.3} unattributed_ms={:.3}",
             index_type,
             self.file_name,
             self.rows,
@@ -96,6 +102,11 @@ impl VectorIndexBuildTiming {
             self.source_batch_wait.as_secs_f64() * 1000.0,
             self.oss_read.as_secs_f64() * 1000.0,
             self.parquet_decode.as_secs_f64() * 1000.0,
+            self.parquet_row_group_count,
+            self.parquet_projected_bytes_min,
+            self.parquet_projected_bytes_max,
+            self.parquet_projected_bytes_total,
+            self.parquet_peak_inflight_row_groups,
             self.raw_temp_write.as_secs_f64() * 1000.0,
             self.train_finish.as_secs_f64() * 1000.0,
             self.raw_temp_reread.as_secs_f64() * 1000.0,
@@ -303,6 +314,13 @@ impl<'a> VindexIndexBuildBuilder<'a> {
         let mut source_batch_wait = Duration::ZERO;
         let mut raw_temp_write = Duration::ZERO;
         let read_timing = timing_enabled.then(|| Arc::new(DataFileReadTiming::default()));
+        let parquet_read_budget = if timing_enabled {
+            let budget = configured_parquet_read_budget(self.table)?;
+            budget.enable_diagnostics();
+            Some(budget)
+        } else {
+            None
+        };
         let mut batch_count = 0usize;
         let row_count = checked_row_count(shard.row_range_start, shard.row_range_end)?;
         let row_count_usize = usize::try_from(row_count).map_err(|e| Error::DataInvalid {
@@ -349,6 +367,10 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             Some(timing) => read.with_data_file_read_timing(Arc::clone(timing)),
             None => read,
         };
+        let read = match parquet_read_budget.as_ref() {
+            Some(budget) => read.with_parquet_read_budget(Arc::clone(budget)),
+            None => read,
+        };
         let mut batches = read.to_arrow(&[split])?;
         let mut expected_row_id = shard.row_range_start;
         let mut rows_seen = 0usize;
@@ -634,11 +656,19 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             .map_or((Duration::ZERO, Duration::ZERO), |timing| {
                 (timing.file_read(), timing.parquet_decode())
             });
+        let parquet_diagnostics = parquet_read_budget
+            .as_ref()
+            .map_or_else(Default::default, |budget| budget.diagnostics());
         let timing = total_start.map(|start| VectorIndexBuildTiming {
             total_without_commit: start.elapsed(),
             source_batch_wait,
             oss_read,
             parquet_decode,
+            parquet_row_group_count: parquet_diagnostics.row_group_count,
+            parquet_projected_bytes_min: parquet_diagnostics.projected_bytes_min,
+            parquet_projected_bytes_max: parquet_diagnostics.projected_bytes_max,
+            parquet_projected_bytes_total: parquet_diagnostics.projected_bytes_total,
+            parquet_peak_inflight_row_groups: parquet_diagnostics.peak_inflight,
             raw_temp_write,
             train_finish,
             raw_temp_reread,

From 85b906aaf0df1ae2f80925e72333a4b39b65a220 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Thu, 20 Aug 2026 10:10:19 +0800
Subject: [PATCH 06/16] perf(vindex): diagnose per-file read waits

---
 crates/paimon/src/table/data_file_reader.rs   | 61 +++++++++++++++++++
 .../src/table/vindex_index_build_builder.rs   | 16 ++++-
 2 files changed, 76 insertions(+), 1 deletion(-)

diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs
index c9e2e1a7..bf261a2a 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -42,6 +42,9 @@ use std::time::{Duration, Instant};
 pub(crate) struct DataFileReadTiming {
     file_read_nanos: AtomicU64,
     parquet_decode_nanos: AtomicU64,
+    file_schema_open_nanos: AtomicU64,
+    first_batch_wait_nanos: AtomicU64,
+    remaining_batch_wait_nanos: AtomicU64,
 }
 
 impl DataFileReadTiming {
@@ -55,6 +58,20 @@ impl DataFileReadTiming {
             .fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
     }
 
+    fn add_file_schema_open(&self, duration: Duration) {
+        self.file_schema_open_nanos
+            .fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
+    }
+
+    fn add_batch_wait(&self, duration: Duration, first: bool) {
+        let target = if first {
+            &self.first_batch_wait_nanos
+        } else {
+            &self.remaining_batch_wait_nanos
+        };
+        target.fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
+    }
+
     pub(crate) fn file_read(&self) -> Duration {
         Duration::from_nanos(self.file_read_nanos.load(Ordering::Relaxed))
     }
@@ -62,6 +79,14 @@ impl DataFileReadTiming {
     pub(crate) fn parquet_decode(&self) -> Duration {
         Duration::from_nanos(self.parquet_decode_nanos.load(Ordering::Relaxed))
     }
+
+    pub(crate) fn file_waits(&self) -> (Duration, Duration, Duration) {
+        (
+            Duration::from_nanos(self.file_schema_open_nanos.load(Ordering::Relaxed)),
+            Duration::from_nanos(self.first_batch_wait_nanos.load(Ordering::Relaxed)),
+            Duration::from_nanos(self.remaining_batch_wait_nanos.load(Ordering::Relaxed)),
+        )
+    }
 }
 
 struct TimedFileRead {
@@ -225,7 +250,13 @@ impl DataFileReader {
                     );
 
                     // Load data file's schema if it differs from the table schema.
+                    let schema_start = reader.read_timing.as_ref().map(|_| Instant::now());
                     let data_fields = reader.derive_data_fields(&file_meta).await?;
+                    if let (Some(timing), Some(start)) =
+                        (reader.read_timing.as_ref(), schema_start)
+                    {
+                        timing.add_file_schema_open(start.elapsed());
+                    }
 
                     let mut stream = reader.read_single_file_stream(
                         &split,
@@ -388,6 +419,7 @@ impl DataFileReader {
         };
 
         Ok(try_stream! {
+            let schema_open_start = read_timing.as_ref().map(|_| Instant::now());
             let path_to_read = split.data_file_path(&file_meta);
             let format_reader = create_format_reader_with_budget(
                 &path_to_read,
@@ -435,8 +467,13 @@ impl DataFileReader {
                 batch_size,
                 row_selection,
             ).await?;
+            if let (Some(timing), Some(start)) = (read_timing.as_ref(), schema_open_start) {
+                timing.add_file_schema_open(start.elapsed());
+            }
+            let mut first_batch = true;
 
             loop {
+                let batch_wait_start = read_timing.as_ref().map(|_| Instant::now());
                 let batch = if is_parquet {
                     if let Some(timing) = read_timing.as_ref() {
                         std::future::poll_fn(|cx| {
@@ -452,7 +489,11 @@ impl DataFileReader {
                 } else {
                     batch_stream.next().await
                 };
+                if let (Some(timing), Some(start)) = (read_timing.as_ref(), batch_wait_start) {
+                    timing.add_batch_wait(start.elapsed(), first_batch);
+                }
                 let Some(batch) = batch else { break };
+                first_batch = false;
                 let batch = batch?;
                 let num_rows = batch.num_rows();
                 let batch_schema = batch.schema();
@@ -1413,6 +1454,26 @@ mod tests {
     use roaring::RoaringBitmap;
     use std::io;
 
+    #[test]
+    fn test_data_file_read_timing_aggregates_file_waits() {
+        let timing = DataFileReadTiming::default();
+        timing.add_file_schema_open(Duration::from_millis(2));
+        timing.add_batch_wait(Duration::from_millis(5), true);
+        timing.add_batch_wait(Duration::from_millis(7), false);
+        timing.add_file_schema_open(Duration::from_millis(3));
+        timing.add_batch_wait(Duration::from_millis(11), true);
+        timing.add_batch_wait(Duration::from_millis(13), false);
+
+        assert_eq!(
+            timing.file_waits(),
+            (
+                Duration::from_millis(5),
+                Duration::from_millis(16),
+                Duration::from_millis(20),
+            )
+        );
+    }
+
     #[test]
     fn test_accessors_expose_read_type_and_row_filtering_predicate() {
         use crate::spec::{DataField, DataType, IntType};
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index 4aba438f..5b34d03c 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -57,6 +57,9 @@ struct VectorIndexBuildTiming {
     source_batch_wait: Duration,
     oss_read: Duration,
     parquet_decode: Duration,
+    file_schema_open: Duration,
+    first_batch_wait: Duration,
+    remaining_batch_wait: Duration,
     parquet_row_group_count: u64,
     parquet_projected_bytes_min: u64,
     parquet_projected_bytes_max: u64,
@@ -90,7 +93,7 @@ impl VectorIndexBuildTiming {
             .saturating_add(commit);
         let unattributed = total.saturating_sub(accounted);
         eprintln!(
-            "event=paimon_vector_index_build index_type={} file={} rows={} training_rows_seen={} training_rows_retained={} batch_count={} raw_temp_bytes={} index_bytes={} source_batch_wait_ms={:.3} oss_read_ms={:.3} parquet_decode_ms={:.3} parquet_row_group_count={} parquet_projected_bytes_min={} parquet_projected_bytes_max={} parquet_projected_bytes_total={} parquet_peak_inflight_row_groups={} raw_temp_write_ms={:.3} train_finish_ms={:.3} raw_temp_reread_ms={:.3} index_add_ms={:.3} serialize_upload_ms={:.3} commit_ms={:.3} sample_read_ms=0.000 full_scan_add_ms=0.000 pipeline_blocked_ms=0.000 producer_blocked_ms=0.000 consumer_add_ms=0.000 data_file_count={} data_file_read_concurrency=1 peak_ready_batches=0 total_ms={:.3} unattributed_ms={:.3}",
+            "event=paimon_vector_index_build index_type={} file={} rows={} training_rows_seen={} training_rows_retained={} batch_count={} raw_temp_bytes={} index_bytes={} source_batch_wait_ms={:.3} oss_read_ms={:.3} parquet_decode_ms={:.3} file_schema_open_ms={:.3} first_batch_wait_ms={:.3} remaining_batch_wait_ms={:.3} parquet_row_group_count={} parquet_projected_bytes_min={} parquet_projected_bytes_max={} parquet_projected_bytes_total={} parquet_peak_inflight_row_groups={} raw_temp_write_ms={:.3} train_finish_ms={:.3} raw_temp_reread_ms={:.3} index_add_ms={:.3} serialize_upload_ms={:.3} commit_ms={:.3} sample_read_ms=0.000 full_scan_add_ms=0.000 pipeline_blocked_ms=0.000 producer_blocked_ms=0.000 consumer_add_ms=0.000 data_file_count={} data_file_read_concurrency=1 peak_ready_batches=0 total_ms={:.3} unattributed_ms={:.3}",
             index_type,
             self.file_name,
             self.rows,
@@ -102,6 +105,9 @@ impl VectorIndexBuildTiming {
             self.source_batch_wait.as_secs_f64() * 1000.0,
             self.oss_read.as_secs_f64() * 1000.0,
             self.parquet_decode.as_secs_f64() * 1000.0,
+            self.file_schema_open.as_secs_f64() * 1000.0,
+            self.first_batch_wait.as_secs_f64() * 1000.0,
+            self.remaining_batch_wait.as_secs_f64() * 1000.0,
             self.parquet_row_group_count,
             self.parquet_projected_bytes_min,
             self.parquet_projected_bytes_max,
@@ -656,6 +662,11 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             .map_or((Duration::ZERO, Duration::ZERO), |timing| {
                 (timing.file_read(), timing.parquet_decode())
             });
+        let (file_schema_open, first_batch_wait, remaining_batch_wait) = read_timing
+            .as_ref()
+            .map_or((Duration::ZERO, Duration::ZERO, Duration::ZERO), |timing| {
+                timing.file_waits()
+            });
         let parquet_diagnostics = parquet_read_budget
             .as_ref()
             .map_or_else(Default::default, |budget| budget.diagnostics());
@@ -664,6 +675,9 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             source_batch_wait,
             oss_read,
             parquet_decode,
+            file_schema_open,
+            first_batch_wait,
+            remaining_batch_wait,
             parquet_row_group_count: parquet_diagnostics.row_group_count,
             parquet_projected_bytes_min: parquet_diagnostics.projected_bytes_min,
             parquet_projected_bytes_max: parquet_diagnostics.projected_bytes_max,

From b445750ba61a8cf31aae0b5b7b576975d42df9d7 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Thu, 20 Aug 2026 12:53:24 +0800
Subject: [PATCH 07/16] perf(vindex): finalize read-path optimization

---
 crates/paimon/src/table/data_file_reader.rs   | 44 ++++++++++++++++++-
 .../src/table/vindex_index_build_builder.rs   | 43 ++----------------
 2 files changed, 46 insertions(+), 41 deletions(-)

diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs
index bf261a2a..1306bba9 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -924,7 +924,11 @@ fn merge_row_selection(
     }
 
     if !has_dv {
-        return row_ranges.map(|r| r.to_vec());
+        return match row_ranges {
+            Some(ranges) if ranges_cover_all_rows(ranges, row_count) => None,
+            Some(ranges) => Some(ranges.to_vec()),
+            None => None,
+        };
     }
 
     let dv_ranges = dv_to_non_deleted_ranges(dv.unwrap(), row_count);
@@ -935,6 +939,20 @@ fn merge_row_selection(
     }
 }
 
+fn ranges_cover_all_rows(ranges: &[RowRange], row_count: i64) -> bool {
+    if row_count <= 0 || ranges.is_empty() || ranges[0].from() > 0 {
+        return false;
+    }
+    let mut covered_to = ranges[0].to();
+    for range in &ranges[1..] {
+        if range.from() > covered_to.saturating_add(1) {
+            return false;
+        }
+        covered_to = covered_to.max(range.to());
+    }
+    covered_to >= row_count - 1
+}
+
 /// Convert a DeletionVector into sorted non-deleted inclusive RowRanges.
 fn dv_to_non_deleted_ranges(dv: &DeletionVector, row_count: i64) -> Vec<RowRange> {
     let mut result = Vec::new();
@@ -1474,6 +1492,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn merge_row_selection_skips_only_unfiltered_full_coverage() {
+        let full = [RowRange::new(0, 9)];
+        let joined = [RowRange::new(0, 3), RowRange::new(4, 9)];
+        let partial = [RowRange::new(1, 9)];
+        let empty = [];
+
+        assert_eq!(merge_row_selection(10, None, Some(&full)), None);
+        assert_eq!(merge_row_selection(10, None, Some(&joined)), None);
+        assert_eq!(
+            merge_row_selection(10, None, Some(&partial)),
+            Some(partial.to_vec())
+        );
+        assert_eq!(merge_row_selection(10, None, Some(&empty)), Some(vec![]));
+
+        let mut deleted = RoaringBitmap::new();
+        deleted.insert(3);
+        let dv = DeletionVector::from_bitmap(deleted);
+        assert_eq!(
+            merge_row_selection(10, Some(&dv), Some(&full)),
+            Some(vec![RowRange::new(0, 2), RowRange::new(4, 9)])
+        );
+    }
+
     #[test]
     fn test_accessors_expose_read_type_and_row_filtering_predicate() {
         use crate::spec::{DataField, DataType, IntType};
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index 5b34d03c..5dfd767f 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -42,7 +42,6 @@ use tokio_util::io::SyncIoBridge;
 
 const INDEX_DIR: &str = "index";
 const VECTOR_BUFFER_BYTES: usize = 8 * 1024 * 1024;
-const INDEX_ADD_BATCH_ROWS: usize = 32 * 1024;
 const VECTOR_INDEX_BUILD_TIMING_ENV: &str = "PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING";
 
 fn vector_index_build_timing_enabled() -> bool {
@@ -533,14 +532,13 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                 if let Some(start) = reread_start {
                     raw_temp_reread = raw_temp_reread.saturating_add(start.elapsed());
                 }
-                let batch_rows = INDEX_ADD_BATCH_ROWS.min(row_count_usize);
+                let batch_rows = training_buffer_rows.min(row_count_usize);
                 let batch_bytes = checked_std_vector_bytes(batch_rows, dimension_usize)?;
                 let mut buffer = MutableBuffer::new(batch_bytes);
                 let mut ids = Vec::with_capacity(batch_rows);
                 let mut rows_added = 0usize;
                 while rows_added < row_count_usize {
                     let rows = batch_rows.min(row_count_usize - rows_added);
-                    let batch_end = checked_index_add_batch_end(rows_added, rows)?;
                     buffer.resize(checked_std_vector_bytes(rows, dimension_usize)?, 0);
                     let reread_start = timing_enabled.then(Instant::now);
                     raw_file.read_exact(buffer.as_slice_mut())?;
@@ -548,7 +546,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                         raw_temp_reread = raw_temp_reread.saturating_add(start.elapsed());
                     }
                     ids.clear();
-                    for row in rows_added..batch_end {
+                    for row in rows_added..rows_added + rows {
                         ids.push(i64::try_from(row).map_err(|_| {
                             std::io::Error::new(
                                 std::io::ErrorKind::InvalidData,
@@ -561,7 +559,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
                     if let Some(start) = add_start {
                         index_add = index_add.saturating_add(start.elapsed());
                     }
-                    rows_added = batch_end;
+                    rows_added += rows;
                 }
                 let mut trailing = [0u8; 1];
                 let reread_start = timing_enabled.then(Instant::now);
@@ -1165,15 +1163,6 @@ fn checked_std_vector_bytes(row_count: usize, dimension: usize) -> std::io::Resu
         })
 }
 
-fn checked_index_add_batch_end(start: usize, rows: usize) -> std::io::Result<usize> {
-    start.checked_add(rows).ok_or_else(|| {
-        std::io::Error::new(
-            std::io::ErrorKind::InvalidInput,
-            "vindex index-add batch end overflows usize",
-        )
-    })
-}
-
 fn checked_training_vector_count(row_count: usize, ratio: f64) -> Result<usize> {
     if row_count == 0 || !(ratio > 0.0 && ratio <= 1.0) {
         return Err(Error::DataInvalid {
@@ -1537,32 +1526,6 @@ mod tests {
         assert!(checked_training_sample_index(usize::MAX, usize::MAX, 1).is_err());
     }
 
-    #[test]
-    fn test_index_add_batch_ranges() {
-        let ranges = |row_count| {
-            let mut result = Vec::new();
-            let mut start = 0;
-            while start < row_count {
-                let rows = INDEX_ADD_BATCH_ROWS.min(row_count - start);
-                let end = checked_index_add_batch_end(start, rows).unwrap();
-                result.push(start..end);
-                start = end;
-            }
-            result
-        };
-
-        assert_eq!(ranges(10), vec![0..10]);
-        assert_eq!(ranges(INDEX_ADD_BATCH_ROWS), vec![0..INDEX_ADD_BATCH_ROWS]);
-        assert_eq!(
-            ranges(INDEX_ADD_BATCH_ROWS + 7),
-            vec![
-                0..INDEX_ADD_BATCH_ROWS,
-                INDEX_ADD_BATCH_ROWS..INDEX_ADD_BATCH_ROWS + 7
-            ]
-        );
-        assert!(checked_index_add_batch_end(usize::MAX, 1).is_err());
-    }
-
     fn test_table_with_io(file_io: FileIO, table_path: &str, schema: Schema) -> Table {
         Table::new(
             file_io,

From d6ec9f77a2ae3d1366c6275b3fb35d80b237eeeb Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Thu, 20 Aug 2026 18:07:23 +0800
Subject: [PATCH 08/16] perf(vindex): upload index parts concurrently

The vector index (~2 GB) is serialized as a sequential stream into an
opendal writer that uploads its 8 MiB multipart chunks strictly one at a
time -- ~244 serial round trips per index on object storage. Add
async_writer_with_concurrency and let the index upload keep 4 parts in
flight (32 MiB buffer), overlapping serialization with uploads. Parquet
and other async_writer users keep the previous serial behavior.
---
 crates/paimon/src/io/file_io.rs                     | 13 +++++++++++++
 .../paimon/src/table/vindex_index_build_builder.rs  |  8 +++++++-
 2 files changed, 20 insertions(+), 1 deletion(-)

diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index 63735486..ddadc9c6 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -768,10 +768,23 @@ impl OutputFile {
 
     /// Get an async streaming writer for format-level writes (e.g. parquet).
     pub(crate) async fn async_writer(&self) -> crate::Result<Box<dyn AsyncFileWrite>> {
+        self.async_writer_with_concurrency(1).await
+    }
+
+    /// Like [`Self::async_writer`], but uploads up to `concurrent` multipart
+    /// chunks in flight. The default writer uploads its 8 MiB parts strictly
+    /// one at a time, so a large sequentially-produced file (e.g. a vector
+    /// index) pays one round trip per part; a small concurrency overlaps the
+    /// producer with the uploads at a cost of `concurrent * 8 MiB` of buffer.
+    pub(crate) async fn async_writer_with_concurrency(
+        &self,
+        concurrent: usize,
+    ) -> crate::Result<Box<dyn AsyncFileWrite>> {
         let (op, relative_path, cache_path) = self.resolve().await?;
         let writer: Box<dyn AsyncFileWrite> = Box::new(
             op.writer_with(&relative_path)
                 .chunk(8 * 1024 * 1024)
+                .concurrent(concurrent.max(1))
                 .await?
                 .into_futures_async_write()
                 .compat_write(),
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs b/crates/paimon/src/table/vindex_index_build_builder.rs
index 5dfd767f..40b9cab7 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -42,6 +42,9 @@ use tokio_util::io::SyncIoBridge;
 
 const INDEX_DIR: &str = "index";
 const VECTOR_BUFFER_BYTES: usize = 8 * 1024 * 1024;
+/// In-flight multipart chunks while uploading the serialized index
+/// (4 x 8 MiB = 32 MiB of upload buffer).
+const INDEX_UPLOAD_CONCURRENCY: usize = 4;
 const VECTOR_INDEX_BUILD_TIMING_ENV: &str = "PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING";
 
 fn vector_index_build_timing_enabled() -> bool {
@@ -604,11 +607,14 @@ impl<'a> VindexIndexBuildBuilder<'a> {
             file_name
         );
         let write_result = async {
+            // Overlap index serialization with multipart uploads: the index is
+            // ~2 GB produced sequentially, and the default writer uploads its
+            // 8 MiB parts one at a time (one round trip per part).
             let async_writer = self
                 .table
                 .file_io()
                 .new_output(&index_path)?
-                .async_writer()
+                .async_writer_with_concurrency(INDEX_UPLOAD_CONCURRENCY)
                 .await?;
             let mut output = SyncIoBridge::new(async_writer);
             tokio::task::spawn_blocking(move || -> std::io::Result<()> {

From d55363376a58325085ca9dd632c782e5443f3dd8 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 10:24:56 +0800
Subject: [PATCH 09/16] fix

---
 bindings/c/Cargo.toml    | 2 +-
 crates/paimon/Cargo.toml | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml
index 51e3731c..af0a5121 100644
--- a/bindings/c/Cargo.toml
+++ b/bindings/c/Cargo.toml
@@ -43,4 +43,4 @@ serde_json = "1.0.120"
 # Test-only: the vector-search integration tests build a real primary-key vindex
 # IVF-flat ANN segment fixture in-process. Versions match crates/paimon.
 bytes = "1.7.1"
-paimon-vindex-core = "0.4.0"
+paimon-vindex-core = "0.3.0"
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index ae96e242..5283ea27 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -132,7 +132,7 @@ urlencoding = "2.1"
 paimon-mosaic-core = "0.2.0"
 paimon-ftindex-core = { version = "0.1.0", optional = true }
 tempfile = "3"
-paimon-vindex-core = "0.4.0"
+paimon-vindex-core = "0.3.0"
 vortex = { version = "0.75.0", features = ["tokio"], optional = true }
 libloading = "0.9"
 log = "0.4"

From bad3f34c7440048962b971406c98f4b93e39fd57 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 10:32:44 +0800
Subject: [PATCH 10/16] fix

---
 Cargo.lock | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/Cargo.lock b/Cargo.lock
index 4ecdb74e..627ac1d4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4751,7 +4751,9 @@ dependencies = [
 
 [[package]]
 name = "paimon-vindex-core"
-version = "0.4.0"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2c67c916596e578ed09f78ace8b4c54fc45d262933c658e1ca7fdb61b722635"
 dependencies = [
  "half",
  "matrixmultiply",

From c07b87f2e8c10b953435d4c16d4dcd1bdc625692 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 11:48:08 +0800
Subject: [PATCH 11/16] perf: cap single row-group budget accounting to a fair
 share

A row group whose projected bytes exceed the whole read budget previously
clamped to every byte permit, so one oversized row group serialized the
scan: wide vector columns project ~294 MiB per row group against the
256 MiB default budget, and parallel row-group reads silently degraded
to 1 in flight unless the user hand-tuned max-inflight-bytes.

Cap a single acquisition at budget / min(parallelism, 4) instead. Row
groups at or below their fair share keep exact accounting (no behavior
change for ordinary layouts); oversized row groups admit up to four
concurrent reads, matching what the 768 MiB hand-tuned budget achieved
(source wait -43.7% on a 10M-row 768-dim build) without configuration.

The byte budget thereby becomes a fair-admission mechanism for large
row groups rather than a strict projected-byte ceiling; the share
divisor is capped at 4 until wider RSS measurements justify more.
---
 .../paimon/src/arrow/parquet_read_budget.rs   | 111 ++++++++++++++++++
 1 file changed, 111 insertions(+)

diff --git a/crates/paimon/src/arrow/parquet_read_budget.rs b/crates/paimon/src/arrow/parquet_read_budget.rs
index 3e60129e..eee4b732 100644
--- a/crates/paimon/src/arrow/parquet_read_budget.rs
+++ b/crates/paimon/src/arrow/parquet_read_budget.rs
@@ -23,6 +23,16 @@ use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 const BYTE_PERMIT_UNIT: u64 = 1024 * 1024;
 const DEFAULT_PARALLELISM: usize = 8;
 const DEFAULT_MAX_INFLIGHT_BYTES: u64 = 256 * 1024 * 1024;
+/// A single row group's byte accounting is capped to the budget divided by
+/// this share, so one oversized row group (e.g. a ~294 MiB projected wide
+/// vector column under the 256 MiB default budget) cannot consume every
+/// byte permit and silently serialize the scan. This makes the byte budget
+/// a fair-admission mechanism for large row groups rather than a strict
+/// projected-byte ceiling: up to `min(parallelism, MAX_BUDGET_SHARES)` such
+/// row groups may be in flight, each accounted at `budget / shares` even
+/// though its projected size is larger. Capped at 4 shares until wider RSS
+/// measurements justify the full parallelism.
+const MAX_BUDGET_SHARES: usize = 4;
 
 /// Shared resource budget for concurrent Parquet row-group reads.
 #[derive(Debug)]
@@ -166,9 +176,17 @@ impl ParquetReadBudget {
                 message: "Parquet row-group read budget was closed".to_string(),
                 source: Some(Box::new(error)),
             })?;
+        // Cap a single row group's accounting to a fair share of the budget
+        // (see MAX_BUDGET_SHARES): an oversized row group must not take every
+        // permit and serialize the scan. `parallelism >= 1` is validated at
+        // construction, and `max(1)` keeps tiny budgets sound — the semaphore
+        // still bounds total in-flight permits.
+        let shares = self.parallelism.min(MAX_BUDGET_SHARES) as u64;
+        let share_cap = u64::from(self.byte_permits).div_ceil(shares).max(1);
         let requested = projected_uncompressed_bytes
             .max(1)
             .div_ceil(BYTE_PERMIT_UNIT)
+            .min(share_cap)
             .min(u64::from(self.byte_permits)) as u32;
         let bytes = Arc::clone(&self.bytes)
             .acquire_many_owned(requested)
@@ -277,4 +295,97 @@ mod tests {
                 .is_err()
         );
     }
+
+    /// A row group larger than the whole budget must not serialize the scan:
+    /// its accounting is capped to budget / min(parallelism, MAX_BUDGET_SHARES),
+    /// so min(parallelism, MAX_BUDGET_SHARES) oversized row groups run
+    /// concurrently. This is the wide-vector-column case (a ~294 MiB projected
+    /// row group under the 256 MiB default budget).
+    #[tokio::test]
+    async fn oversized_row_groups_share_the_budget() {
+        // parallelism 8 > MAX_BUDGET_SHARES: shares = 4.
+        let budget = Arc::new(ParquetReadBudget::new(8, 8 * BYTE_PERMIT_UNIT).unwrap());
+        budget.enable_diagnostics();
+        let oversized = 100 * BYTE_PERMIT_UNIT; // far above the whole budget
+
+        // 4 oversized acquisitions must all succeed (each accounted at 2 permits).
+        let mut permits = Vec::new();
+        for _ in 0..MAX_BUDGET_SHARES {
+            permits.push(
+                tokio::time::timeout(Duration::from_secs(1), budget.acquire(oversized))
+                    .await
+                    .expect("an oversized row group must only take a fair share")
+                    .unwrap(),
+            );
+        }
+        assert_eq!(budget.diagnostics().peak_inflight, MAX_BUDGET_SHARES);
+
+        // The 5th oversized request exhausts the byte semaphore and must wait.
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), budget.acquire(oversized))
+                .await
+                .is_err(),
+            "the byte budget still bounds total in-flight accounting"
+        );
+
+        drop(permits);
+        tokio::time::timeout(Duration::from_secs(1), budget.acquire(oversized))
+            .await
+            .expect("released shares must become available again")
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn non_divisible_budget_does_not_exceed_max_shares() {
+        let budget = Arc::new(ParquetReadBudget::new(8, 10 * BYTE_PERMIT_UNIT).unwrap());
+        let oversized = 100 * BYTE_PERMIT_UNIT;
+
+        // ceil(10 / 4) = 3 permits, so only 3 oversized reads fit. Rounding
+        // down to 2 would incorrectly admit 5 reads and exceed the cap.
+        let permits = futures::future::try_join_all((0..3).map(|_| budget.acquire(oversized)))
+            .await
+            .unwrap();
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), budget.acquire(oversized))
+                .await
+                .is_err()
+        );
+        drop(permits);
+    }
+
+    /// Row groups at or below budget / shares keep their exact accounting —
+    /// the share cap must not change behavior for ordinary layouts.
+    #[tokio::test]
+    async fn small_row_groups_keep_exact_accounting() {
+        let budget = Arc::new(ParquetReadBudget::new(4, 4 * BYTE_PERMIT_UNIT).unwrap());
+        // share_cap = 4 / min(4,4) = 1 permit; a 1 MiB row group requests
+        // exactly 1 permit, so 4 fit and the 5th blocks — identical to the
+        // pre-cap behavior for small row groups.
+        let mut permits = Vec::new();
+        for _ in 0..4 {
+            permits.push(budget.acquire(BYTE_PERMIT_UNIT).await.unwrap());
+        }
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), budget.acquire(BYTE_PERMIT_UNIT))
+                .await
+                .is_err()
+        );
+    }
+
+    /// Tiny budgets stay sound: the cap never rounds a request down to zero
+    /// permits, and a budget smaller than the share divisor still admits one
+    /// row group at a time.
+    #[tokio::test]
+    async fn tiny_budget_still_admits_one_at_a_time() {
+        let budget = Arc::new(ParquetReadBudget::new(8, BYTE_PERMIT_UNIT).unwrap());
+        let first = budget.acquire(100 * BYTE_PERMIT_UNIT).await.unwrap();
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), budget.acquire(1))
+                .await
+                .is_err(),
+            "a single-permit budget admits exactly one read"
+        );
+        drop(first);
+        budget.acquire(1).await.unwrap();
+    }
 }

From 249158057907a5162c09ea1177b83a45a394bad3 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 13:35:05 +0800
Subject: [PATCH 12/16] bench(vindex): add IVF-PQ build benchmark

---
 .../paimon/examples/ivfpq_build_benchmark.rs  | 93 +++++++++++++++++++
 1 file changed, 93 insertions(+)
 create mode 100644 crates/paimon/examples/ivfpq_build_benchmark.rs

diff --git a/crates/paimon/examples/ivfpq_build_benchmark.rs b/crates/paimon/examples/ivfpq_build_benchmark.rs
new file mode 100644
index 00000000..cbfe1f3f
--- /dev/null
+++ b/crates/paimon/examples/ivfpq_build_benchmark.rs
@@ -0,0 +1,93 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Build an IVF-PQ index through the production Paimon path.
+//!
+//! ```text
+//! PAIMON_CATALOG_OPTIONS='{"metastore":"filesystem","warehouse":"/tmp/warehouse"}' \
+//! PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING=1 \
+//! cargo run --release -p paimon --example ivfpq_build_benchmark -- \
+//!   <database> <table> <vector-column> [--drop-existing]
+//! ```
+
+use std::collections::HashMap;
+use std::error::Error;
+use std::time::Instant;
+
+use paimon::catalog::Identifier;
+use paimon::{CatalogFactory, Options};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let mut args = std::env::args().skip(1);
+    let database = required_arg(&mut args, "database")?;
+    let table_name = required_arg(&mut args, "table")?;
+    let column = required_arg(&mut args, "vector-column")?;
+    let drop_existing = args.any(|arg| arg == "--drop-existing");
+
+    let catalog_options = std::env::var("PAIMON_CATALOG_OPTIONS")?;
+    let catalog =
+        CatalogFactory::create(Options::from_map(serde_json::from_str(&catalog_options)?)).await?;
+    let table = catalog
+        .get_table(&Identifier::new(&database, &table_name))
+        .await?;
+
+    let dropped_index_files = if drop_existing {
+        let mut builder = table.new_global_index_drop_builder();
+        builder.with_index_column(&column).with_index_type("ivf-pq");
+        builder.execute().await?
+    } else {
+        0
+    };
+
+    let options = HashMap::from([
+        ("dimension".to_string(), "768".to_string()),
+        ("metric".to_string(), "cosine".to_string()),
+        ("nlist".to_string(), "4096".to_string()),
+        ("pq.m".to_string(), "192".to_string()),
+    ]);
+    let started = Instant::now();
+    let built_shards = table
+        .new_vindex_index_build_builder("ivf-pq")
+        .with_index_column(&column)
+        .with_options(options.clone())
+        .execute()
+        .await?;
+
+    println!(
+        "{}",
+        serde_json::to_string_pretty(&serde_json::json!({
+            "database": database,
+            "table": table_name,
+            "column": column,
+            "index_type": "ivf-pq",
+            "build_options": options,
+            "dropped_index_files": dropped_index_files,
+            "built_shards": built_shards,
+            "duration_seconds": started.elapsed().as_secs_f64(),
+        }))?
+    );
+    Ok(())
+}
+
+fn required_arg(
+    args: &mut impl Iterator<Item = String>,
+    name: &str,
+) -> Result<String, Box<dyn Error>> {
+    args.next()
+        .ok_or_else(|| format!("missing <{name}> argument").into())
+}

From ccc3b8d4dd861896242cc35a0867b69ea9b25ed3 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 15:58:11 +0800
Subject: [PATCH 13/16] fix(parquet): preserve strict read budget accounting

---
 .../paimon/src/arrow/parquet_read_budget.rs   | 81 ++-----------------
 1 file changed, 6 insertions(+), 75 deletions(-)

diff --git a/crates/paimon/src/arrow/parquet_read_budget.rs b/crates/paimon/src/arrow/parquet_read_budget.rs
index eee4b732..2d773746 100644
--- a/crates/paimon/src/arrow/parquet_read_budget.rs
+++ b/crates/paimon/src/arrow/parquet_read_budget.rs
@@ -23,16 +23,6 @@ use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 const BYTE_PERMIT_UNIT: u64 = 1024 * 1024;
 const DEFAULT_PARALLELISM: usize = 8;
 const DEFAULT_MAX_INFLIGHT_BYTES: u64 = 256 * 1024 * 1024;
-/// A single row group's byte accounting is capped to the budget divided by
-/// this share, so one oversized row group (e.g. a ~294 MiB projected wide
-/// vector column under the 256 MiB default budget) cannot consume every
-/// byte permit and silently serialize the scan. This makes the byte budget
-/// a fair-admission mechanism for large row groups rather than a strict
-/// projected-byte ceiling: up to `min(parallelism, MAX_BUDGET_SHARES)` such
-/// row groups may be in flight, each accounted at `budget / shares` even
-/// though its projected size is larger. Capped at 4 shares until wider RSS
-/// measurements justify the full parallelism.
-const MAX_BUDGET_SHARES: usize = 4;
 
 /// Shared resource budget for concurrent Parquet row-group reads.
 #[derive(Debug)]
@@ -176,17 +166,9 @@ impl ParquetReadBudget {
                 message: "Parquet row-group read budget was closed".to_string(),
                 source: Some(Box::new(error)),
             })?;
-        // Cap a single row group's accounting to a fair share of the budget
-        // (see MAX_BUDGET_SHARES): an oversized row group must not take every
-        // permit and serialize the scan. `parallelism >= 1` is validated at
-        // construction, and `max(1)` keeps tiny budgets sound — the semaphore
-        // still bounds total in-flight permits.
-        let shares = self.parallelism.min(MAX_BUDGET_SHARES) as u64;
-        let share_cap = u64::from(self.byte_permits).div_ceil(shares).max(1);
         let requested = projected_uncompressed_bytes
             .max(1)
             .div_ceil(BYTE_PERMIT_UNIT)
-            .min(share_cap)
             .min(u64::from(self.byte_permits)) as u32;
         let bytes = Arc::clone(&self.bytes)
             .acquire_many_owned(requested)
@@ -296,71 +278,23 @@ mod tests {
         );
     }
 
-    /// A row group larger than the whole budget must not serialize the scan:
-    /// its accounting is capped to budget / min(parallelism, MAX_BUDGET_SHARES),
-    /// so min(parallelism, MAX_BUDGET_SHARES) oversized row groups run
-    /// concurrently. This is the wide-vector-column case (a ~294 MiB projected
-    /// row group under the 256 MiB default budget).
     #[tokio::test]
-    async fn oversized_row_groups_share_the_budget() {
-        // parallelism 8 > MAX_BUDGET_SHARES: shares = 4.
+    async fn oversized_row_group_consumes_the_budget() {
         let budget = Arc::new(ParquetReadBudget::new(8, 8 * BYTE_PERMIT_UNIT).unwrap());
-        budget.enable_diagnostics();
-        let oversized = 100 * BYTE_PERMIT_UNIT; // far above the whole budget
-
-        // 4 oversized acquisitions must all succeed (each accounted at 2 permits).
-        let mut permits = Vec::new();
-        for _ in 0..MAX_BUDGET_SHARES {
-            permits.push(
-                tokio::time::timeout(Duration::from_secs(1), budget.acquire(oversized))
-                    .await
-                    .expect("an oversized row group must only take a fair share")
-                    .unwrap(),
-            );
-        }
-        assert_eq!(budget.diagnostics().peak_inflight, MAX_BUDGET_SHARES);
-
-        // The 5th oversized request exhausts the byte semaphore and must wait.
+        let first = budget.acquire(100 * BYTE_PERMIT_UNIT).await.unwrap();
         assert!(
-            tokio::time::timeout(Duration::from_millis(20), budget.acquire(oversized))
+            tokio::time::timeout(Duration::from_millis(20), budget.acquire(1))
                 .await
                 .is_err(),
-            "the byte budget still bounds total in-flight accounting"
-        );
-
-        drop(permits);
-        tokio::time::timeout(Duration::from_secs(1), budget.acquire(oversized))
-            .await
-            .expect("released shares must become available again")
-            .unwrap();
-    }
-
-    #[tokio::test]
-    async fn non_divisible_budget_does_not_exceed_max_shares() {
-        let budget = Arc::new(ParquetReadBudget::new(8, 10 * BYTE_PERMIT_UNIT).unwrap());
-        let oversized = 100 * BYTE_PERMIT_UNIT;
-
-        // ceil(10 / 4) = 3 permits, so only 3 oversized reads fit. Rounding
-        // down to 2 would incorrectly admit 5 reads and exceed the cap.
-        let permits = futures::future::try_join_all((0..3).map(|_| budget.acquire(oversized)))
-            .await
-            .unwrap();
-        assert!(
-            tokio::time::timeout(Duration::from_millis(20), budget.acquire(oversized))
-                .await
-                .is_err()
+            "an oversized row group must consume the whole byte budget"
         );
-        drop(permits);
+        drop(first);
+        budget.acquire(1).await.unwrap();
     }
 
-    /// Row groups at or below budget / shares keep their exact accounting —
-    /// the share cap must not change behavior for ordinary layouts.
     #[tokio::test]
     async fn small_row_groups_keep_exact_accounting() {
         let budget = Arc::new(ParquetReadBudget::new(4, 4 * BYTE_PERMIT_UNIT).unwrap());
-        // share_cap = 4 / min(4,4) = 1 permit; a 1 MiB row group requests
-        // exactly 1 permit, so 4 fit and the 5th blocks — identical to the
-        // pre-cap behavior for small row groups.
         let mut permits = Vec::new();
         for _ in 0..4 {
             permits.push(budget.acquire(BYTE_PERMIT_UNIT).await.unwrap());
@@ -372,9 +306,6 @@ mod tests {
         );
     }
 
-    /// Tiny budgets stay sound: the cap never rounds a request down to zero
-    /// permits, and a budget smaller than the share divisor still admits one
-    /// row group at a time.
     #[tokio::test]
     async fn tiny_budget_still_admits_one_at_a_time() {
         let budget = Arc::new(ParquetReadBudget::new(8, BYTE_PERMIT_UNIT).unwrap());

From 267b6d8282c0bde7709f3649f823afd24c6d5bf7 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 15:58:56 +0800
Subject: [PATCH 14/16] perf(vindex): enable approximate IVF-PQ assignment

---
 .../paimon/examples/ivfpq_build_benchmark.rs  |  1 +
 crates/paimon/src/vindex/mod.rs               | 23 +++++++++++++++++++
 2 files changed, 24 insertions(+)

diff --git a/crates/paimon/examples/ivfpq_build_benchmark.rs b/crates/paimon/examples/ivfpq_build_benchmark.rs
index cbfe1f3f..bbbf5d88 100644
--- a/crates/paimon/examples/ivfpq_build_benchmark.rs
+++ b/crates/paimon/examples/ivfpq_build_benchmark.rs
@@ -59,6 +59,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         ("metric".to_string(), "cosine".to_string()),
         ("nlist".to_string(), "4096".to_string()),
         ("pq.m".to_string(), "192".to_string()),
+        ("approximate-assignment".to_string(), "true".to_string()),
     ]);
     let started = Instant::now();
     let built_shards = table
diff --git a/crates/paimon/src/vindex/mod.rs b/crates/paimon/src/vindex/mod.rs
index 8ec93a13..b74d85c8 100644
--- a/crates/paimon/src/vindex/mod.rs
+++ b/crates/paimon/src/vindex/mod.rs
@@ -190,6 +190,16 @@ impl VindexVectorIndexOptions {
                     DEFAULT_PQ_USE_OPQ,
                 ),
             );
+            if let Some(value) = optional_value(
+                table_options,
+                user_options,
+                field.name(),
+                index_type,
+                "approximate-assignment",
+                "approximate-assignment",
+            ) {
+                native_options.insert("approximate-assignment".to_string(), value);
+            }
         }
         if index_type == IVF_RQ_IDENTIFIER {
             for key in ["rq.bits", "max-bytes-per-vector"] {
@@ -312,6 +322,7 @@ fn is_allowed_native_key(key: &str, index_type: &str) -> bool {
         "dimension" | "metric" => true,
         "nlist" => index_type != DISKANN_IDENTIFIER,
         "use-opq" => index_type == IVF_PQ_IDENTIFIER,
+        "approximate-assignment" => index_type == IVF_PQ_IDENTIFIER,
         "rq.bits" => index_type == IVF_RQ_IDENTIFIER,
         "max-bytes-per-vector" => {
             matches!(index_type, IVF_RQ_IDENTIFIER | DISKANN_IDENTIFIER)
@@ -332,6 +343,7 @@ fn is_allowed_paimon_suffix(suffix: &str, index_type: &str) -> bool {
         "nlist" => index_type != DISKANN_IDENTIFIER,
         "train.sample-ratio" => true,
         "pq.use-opq" => index_type == IVF_PQ_IDENTIFIER,
+        "approximate-assignment" => index_type == IVF_PQ_IDENTIFIER,
         "rq.bits" => index_type == IVF_RQ_IDENTIFIER,
         "max-bytes-per-vector" => {
             matches!(index_type, IVF_RQ_IDENTIFIER | DISKANN_IDENTIFIER)
@@ -528,6 +540,10 @@ mod tests {
             ("ivf-pq.distance.metric".to_string(), "cosine".to_string()),
             ("ivf-pq.pq.m".to_string(), "2".to_string()),
             ("ivf-pq.pq.use-opq".to_string(), "true".to_string()),
+            (
+                "ivf-pq.approximate-assignment".to_string(),
+                "true".to_string(),
+            ),
         ]);
 
         let options = VindexVectorIndexOptions::new(
@@ -555,6 +571,13 @@ mod tests {
             options.native_options.get("use-opq").map(String::as_str),
             Some("true")
         );
+        assert_eq!(
+            options
+                .native_options
+                .get("approximate-assignment")
+                .map(String::as_str),
+            Some("true")
+        );
     }
 
     #[test]

From bbf6399767f5da6af5c765678f7babcb841d5834 Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 16:21:18 +0800
Subject: [PATCH 15/16] fix(vindex): keep options compatible with core 0.3

---
 .../paimon/examples/ivfpq_build_benchmark.rs  |  1 -
 crates/paimon/src/vindex/mod.rs               | 23 -------------------
 2 files changed, 24 deletions(-)

diff --git a/crates/paimon/examples/ivfpq_build_benchmark.rs b/crates/paimon/examples/ivfpq_build_benchmark.rs
index bbbf5d88..cbfe1f3f 100644
--- a/crates/paimon/examples/ivfpq_build_benchmark.rs
+++ b/crates/paimon/examples/ivfpq_build_benchmark.rs
@@ -59,7 +59,6 @@ async fn main() -> Result<(), Box<dyn Error>> {
         ("metric".to_string(), "cosine".to_string()),
         ("nlist".to_string(), "4096".to_string()),
         ("pq.m".to_string(), "192".to_string()),
-        ("approximate-assignment".to_string(), "true".to_string()),
     ]);
     let started = Instant::now();
     let built_shards = table
diff --git a/crates/paimon/src/vindex/mod.rs b/crates/paimon/src/vindex/mod.rs
index b74d85c8..8ec93a13 100644
--- a/crates/paimon/src/vindex/mod.rs
+++ b/crates/paimon/src/vindex/mod.rs
@@ -190,16 +190,6 @@ impl VindexVectorIndexOptions {
                     DEFAULT_PQ_USE_OPQ,
                 ),
             );
-            if let Some(value) = optional_value(
-                table_options,
-                user_options,
-                field.name(),
-                index_type,
-                "approximate-assignment",
-                "approximate-assignment",
-            ) {
-                native_options.insert("approximate-assignment".to_string(), value);
-            }
         }
         if index_type == IVF_RQ_IDENTIFIER {
             for key in ["rq.bits", "max-bytes-per-vector"] {
@@ -322,7 +312,6 @@ fn is_allowed_native_key(key: &str, index_type: &str) -> bool {
         "dimension" | "metric" => true,
         "nlist" => index_type != DISKANN_IDENTIFIER,
         "use-opq" => index_type == IVF_PQ_IDENTIFIER,
-        "approximate-assignment" => index_type == IVF_PQ_IDENTIFIER,
         "rq.bits" => index_type == IVF_RQ_IDENTIFIER,
         "max-bytes-per-vector" => {
             matches!(index_type, IVF_RQ_IDENTIFIER | DISKANN_IDENTIFIER)
@@ -343,7 +332,6 @@ fn is_allowed_paimon_suffix(suffix: &str, index_type: &str) -> bool {
         "nlist" => index_type != DISKANN_IDENTIFIER,
         "train.sample-ratio" => true,
         "pq.use-opq" => index_type == IVF_PQ_IDENTIFIER,
-        "approximate-assignment" => index_type == IVF_PQ_IDENTIFIER,
         "rq.bits" => index_type == IVF_RQ_IDENTIFIER,
         "max-bytes-per-vector" => {
             matches!(index_type, IVF_RQ_IDENTIFIER | DISKANN_IDENTIFIER)
@@ -540,10 +528,6 @@ mod tests {
             ("ivf-pq.distance.metric".to_string(), "cosine".to_string()),
             ("ivf-pq.pq.m".to_string(), "2".to_string()),
             ("ivf-pq.pq.use-opq".to_string(), "true".to_string()),
-            (
-                "ivf-pq.approximate-assignment".to_string(),
-                "true".to_string(),
-            ),
         ]);
 
         let options = VindexVectorIndexOptions::new(
@@ -571,13 +555,6 @@ mod tests {
             options.native_options.get("use-opq").map(String::as_str),
             Some("true")
         );
-        assert_eq!(
-            options
-                .native_options
-                .get("approximate-assignment")
-                .map(String::as_str),
-            Some("true")
-        );
     }
 
     #[test]

From 46d22b73484cea4e48d458d65a79c9d9a0f1a06b Mon Sep 17 00:00:00 2001
From: yantian <lining.jln@alibaba-inc.com>
Date: Fri, 21 Aug 2026 16:50:33 +0800
Subject: [PATCH 16/16] Warn when Parquet row groups exceed read budget

---
 crates/paimon/src/arrow/parquet_read_budget.rs | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/crates/paimon/src/arrow/parquet_read_budget.rs b/crates/paimon/src/arrow/parquet_read_budget.rs
index 2d773746..b3f2f924 100644
--- a/crates/paimon/src/arrow/parquet_read_budget.rs
+++ b/crates/paimon/src/arrow/parquet_read_budget.rs
@@ -31,6 +31,7 @@ pub struct ParquetReadBudget {
     row_groups: Arc<Semaphore>,
     bytes: Arc<Semaphore>,
     byte_permits: u32,
+    oversized_warning_logged: AtomicBool,
     diagnostics: Arc<ParquetReadDiagnostics>,
 }
 
@@ -96,6 +97,7 @@ impl ParquetReadBudget {
             row_groups: Arc::new(Semaphore::new(parallelism)),
             bytes: Arc::new(Semaphore::new(byte_permits as usize)),
             byte_permits,
+            oversized_warning_logged: AtomicBool::new(false),
             diagnostics: Arc::new(ParquetReadDiagnostics::default()),
         })
     }
@@ -170,6 +172,17 @@ impl ParquetReadBudget {
             .max(1)
             .div_ceil(BYTE_PERMIT_UNIT)
             .min(u64::from(self.byte_permits)) as u32;
+        if projected_uncompressed_bytes > u64::from(self.byte_permits) * BYTE_PERMIT_UNIT
+            && !self.oversized_warning_logged.swap(true, Ordering::Relaxed)
+        {
+            log::warn!(
+                "Parquet row group projected size ({projected_uncompressed_bytes} bytes) exceeds \
+                 read.parquet.row-group.max-inflight-bytes ({} bytes); it will consume the entire \
+                 byte budget and may reduce row-group read parallelism; increase the option if \
+                 memory allows",
+                u64::from(self.byte_permits) * BYTE_PERMIT_UNIT
+            );
+        }
         let bytes = Arc::clone(&self.bytes)
             .acquire_many_owned(requested)
             .await
@@ -282,6 +295,7 @@ mod tests {
     async fn oversized_row_group_consumes_the_budget() {
         let budget = Arc::new(ParquetReadBudget::new(8, 8 * BYTE_PERMIT_UNIT).unwrap());
         let first = budget.acquire(100 * BYTE_PERMIT_UNIT).await.unwrap();
+        assert!(budget.oversized_warning_logged.load(Ordering::Relaxed));
         assert!(
             tokio::time::timeout(Duration::from_millis(20), budget.acquire(1))
                 .await