shyjsarah commented on code in PR #723:
URL: https://github.com/apache/paimon-rust/pull/723#discussion_r3802837780
##########
crates/paimon/src/table/vindex_index_build_builder.rs:
##########
@@ -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);
Review Comment:
This applies the single operation-level commit duration to every shard-level
timing event. For a multi-shard build, summing events counts commit N times; an
early shard's `total_ms` also excludes time spent building later shards, so it
is neither shard wall time nor operation wall time. Suggest keeping shard
events limited to shard phases and emitting one operation-level event for
all-shard build time, the single commit, and end-to-end total (ideally linked
by a build ID).
##########
crates/paimon/src/table/data_file_reader.rs:
##########
@@ -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());
Review Comment:
`file_read_nanos` sums the elapsed time of every `FileRead::read`, but
Parquet range fetching can run multiple reads concurrently (up to
`RANGE_FETCH_CONCURRENCY`). Overlapping waits are therefore double-counted, so
the emitted `oss_read_ms` is cumulative request wait rather than storage-I/O
wall time and may exceed the enclosing phase/total. Could we either measure the
wall-clock I/O span, or rename this to something like `oss_read_wait_sum_ms`
and report the actual concurrency? The current hard-coded
`data_file_read_concurrency=1` is misleading for the same reason.
##########
crates/paimon/src/table/vindex_index_build_builder.rs:
##########
@@ -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);
Review Comment:
This reserves one `VectorIndexBuildTiming` slot per shard even when
`PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING` is disabled, leaving the default-off
path with O(shard_count) diagnostic memory. Could we read the enable flag once
and allocate/push only when enabled (or aggregate bounded operation-level stats
instead)?
##########
crates/paimon/src/table/vindex_index_build_builder.rs:
##########
@@ -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!(
Review Comment:
Could this use the logging facade instead of `eprintln!`? Direct stderr
bypasses the embedding application's filtering, targets, sinks, structured
handlers, and test logger, and performs synchronous output in this async path.
Nearby vector diagnostics already use `log::debug!`; a stable target such as
`paimon::vector_index_build` would keep this routable while the env flag can
still avoid timing overhead when disabled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
