JingsongLi commented on code in PR #800:
URL: https://github.com/apache/paimon-rust/pull/800#discussion_r3976835012
##########
crates/paimon/src/table/vindex_index_build_builder/writer.rs:
##########
@@ -87,245 +93,571 @@ impl<'a> VindexIndexBuildBuilder<'a> {
let expected_bytes = checked_vector_bytes(row_count_usize,
dimension_usize)?;
let training_vector_count =
checked_training_vector_count(row_count_usize,
options.train_sample_ratio)?;
- let training_buffer_rows =
- (VECTOR_BUFFER_BYTES / checked_vector_bytes(1,
dimension_usize)?).max(1);
- let training_buffer_floats = training_buffer_rows
- .checked_mul(dimension_usize)
- .ok_or_else(|| Error::DataInvalid {
- message: "vindex training buffer length overflows
usize".to_string(),
- source: None,
- })?;
+ let training_rows_retained = if self.index_type == DISKANN_IDENTIFIER {
+ 0
+ } else {
+ default_training_vector_count(training_vector_count,
options.config.nlist()).map_err(
+ |e| Error::DataInvalid {
+ message: format!("Failed to calculate IVF training vector
count: {e}"),
+ source: Some(Box::new(e)),
+ },
+ )?
+ };
+ let mut sparse_ranges =
+ if training_rows_retained > 0 && training_rows_retained <
row_count_usize {
+ Some(plan_ivf_training_ranges(shard, training_rows_retained)?)
+ } else {
+ None
+ };
+ let capability_start = (timing_enabled &&
sparse_ranges.is_some()).then(Instant::now);
+ if let Some(ranges) = sparse_ranges.as_ref() {
+ let mut checks = Vec::new();
+ let mut usable = true;
+ for file in &shard.files {
+ let path = file.data_file_path(&shard.bucket_path);
+ if !path.to_ascii_lowercase().ends_with(".parquet") {
+ usable = false;
+ break;
+ }
+ let Some((file_start, file_end)) = file.row_id_range() else {
+ usable = false;
+ break;
+ };
+ let local_ranges = ranges
+ .iter()
+ .filter_map(|range| {
+ let from = range.from().max(file_start);
+ let to = range.to().min(file_end);
+ (from <= to).then(|| RowRange::new(from - file_start,
to - file_start))
+ })
+ .collect::<Vec<_>>();
+ if local_ranges.is_empty()
+ || file
+ .write_cols
+ .as_ref()
+ .is_some_and(|columns| !columns.iter().any(|column|
column == index_column))
+ {
+ continue;
+ }
+ let file_size = u64::try_from(file.file_size).map_err(|e|
Error::DataInvalid {
+ message: format!(
+ "Invalid data file size for '{}': {}",
+ file.file_name, file.file_size
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ checks.push((path, file_size, local_ranges));
+ }
+ let found_vector_file = !checks.is_empty();
+ if usable && found_vector_file {
+ let concurrency = self
+ .table
+ .schema()
+ .core_options()
+ .parquet_row_group_parallelism()?;
+ let file_io = self.table.file_io();
+ let mut checks = futures::stream::iter(checks)
+ .map(|(path, file_size, local_ranges)| async move {
+ let input = file_io.new_input(&path)?;
+ has_usable_offset_index(
+ Box::new(input.reader().await?),
+ file_size,
+ index_column,
+ &local_ranges,
+ )
+ .await
Review Comment:
[P2] Check actual page savings before enabling the extra source pass
An offset index being present does not mean these ranges avoid reading
pages. With the default 100,000-row shard, 65,536 training rows,
128-dimensional vectors, and default Zstd/page settings, the 64 sample ranges
leave gaps too small to skip pages or survive the reader's existing 1 MiB range
coalescing. A tracking FileRead measured exactly 47,788,791 data bytes for both
the sample and the full read, excluding metadata; the offset-index check still
returned true. This fixture fits the default writer-buffer and file-size limits.
The subsequent full scan therefore doubles source data reads relative to the
previous single source scan, including remote reads when the files are on OSS.
This measures bytes, not overall wall time versus the saved local spill. Please
estimate selected page ranges after coalescing and fall back, or adapt the
sample plan, when the sample would read essentially the whole projection.
##########
crates/paimon/src/table/vindex_index_build_builder/planning.rs:
##########
@@ -45,3 +49,79 @@ pub(super) fn plan_vindex_shards(
"vindex",
)
}
+
+pub(super) fn plan_ivf_training_ranges(
+ shard: &VindexIndexShard,
+ training_rows: usize,
+) -> Result<Vec<RowRange>> {
+ let shard_rows = usize::try_from(checked_row_count(
+ shard.row_range_start,
+ shard.row_range_end,
+ )?)
+ .map_err(|error| Error::DataInvalid {
+ message: "vindex shard row count does not fit usize".to_string(),
+ source: Some(Box::new(error)),
+ })?;
+ if training_rows == 0 || training_rows > shard_rows {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Invalid IVF training row count: {training_rows}; shard
contains {shard_rows} rows"
+ ),
+ source: None,
+ });
+ }
+ if training_rows == shard_rows {
+ return Ok(Vec::new());
+ }
+
+ let range_count = training_rows.min(MAX_IVF_TRAINING_RANGES);
+ let gap_count = range_count + 1;
+ let skipped_rows = shard_rows - training_rows;
+ let seed = mix_seed(
+ (shard.snapshot_id as u64)
+ ^ (shard.row_range_start as u64).rotate_left(21)
+ ^ (shard.row_range_end as u64).rotate_left(42),
+ );
+ let range_extra_offset = seed as usize % range_count;
+ let gap_extra_offset = seed.rotate_left(17) as usize % gap_count;
+ let mut cursor = shard.row_range_start;
+ let mut ranges = Vec::with_capacity(range_count);
+
+ for gap_index in 0..range_count {
+ let gap = skipped_rows / gap_count
+ + usize::from(
+ (gap_index + gap_count - gap_extra_offset) % gap_count <
skipped_rows % gap_count,
+ );
+ cursor = checked_add_offset(cursor, gap, "training gap")?;
Review Comment:
[P2] Preserve representative training samples across the shard
These fixed gaps permanently exclude contiguous regions; the seed only
shifts the remainder allocation by approximately one row. For a 100,000-row
shard with the default sample ratio of 1 and 65,536 retained vectors, the last
sampled row is 99,468, so a newly appended distribution in the final 500 rows
is never trained.
I reproduced this using the actual planner and vindex 0.4 trainer: 90,000
vectors `[0]`, then 9,500 `[1]`, then 500 `[100]`, with IVF-SQ, L2, and
`nlist=nprobe=1`. Querying `[100]` returns 10/10 results from the final cluster
with the previous full-stream reservoir sampling, but 0/10 with these ranges.
The SQ bounds are trained only on the older distributions, so the new vectors
are clamped to the old upper bound even though every vector is subsequently
added and every list is searched. Please retain the existing reservoir path
until representative sampling is preserved, and add a recall regression test
for data clustered by append order.
--
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]
