This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new afeb84d5 perf: stream and spill scalar global-index sorting (#1244)
afeb84d5 is described below
commit afeb84d511d0760ab013328501f250ce73eabd4d
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Oct 28 16:47:36 2026 +0800
perf: stream and spill scalar global-index sorting (#1244)
---
crates/paimon/src/spec/core_options.rs | 18 ++
.../src/table/sorted_global_index_build_builder.rs | 4 +
.../extraction.rs | 232 ++++++++++----
.../sorted_global_index_build_builder/sort.rs | 340 +++++++++++++++++++++
.../sorted_global_index_build_builder/tests.rs | 35 ++-
.../sorted_global_index_build_builder/writer.rs | 33 +-
docs/src/getting-started.md | 18 ++
7 files changed, 604 insertions(+), 76 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index c9850975..6d2d2013 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -1195,6 +1195,24 @@ impl<'a> CoreOptions<'a> {
Ok(value)
}
+ /// Maximum retained serialized sort keys, excluding the active Arrow
batch,
+ /// format-writer buffers and the two keys used by a disk merge.
+ pub fn sorted_index_sort_memory(&self) -> crate::Result<usize> {
+ let raw = self
+ .options
+ .get("sorted-index.sort-memory")
+ .map(String::as_str)
+ .unwrap_or("64mb");
+ let bytes = parse_memory_size(raw)
+ .and_then(|size| usize::try_from(size).ok())
+ .filter(|size| *size > 0);
+ bytes.ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!(
+ "Option 'sorted-index.sort-memory' must be a positive memory
size, got: {raw}"
+ ),
+ })
+ }
+
pub fn btree_index_fallback_scan_max_size(&self) -> crate::Result<i64> {
self.fallback_scan_max_size(BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE_OPTION)
}
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs
b/crates/paimon/src/table/sorted_global_index_build_builder.rs
index 598f2c2e..238f02dc 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs
@@ -17,6 +17,7 @@
mod extraction;
mod planning;
+mod sort;
mod validation;
mod writer;
@@ -149,6 +150,9 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
let core_options = CoreOptions::new(&resolved_options);
super::global_index_build_common::validate_build_table(self.table)?;
let records_per_range = core_options.sorted_index_records_per_range()?;
+ if index_type != FM_GLOBAL_INDEX_TYPE {
+ core_options.sorted_index_sort_memory()?;
+ }
let write_options = if index_type == FM_GLOBAL_INDEX_TYPE {
GlobalIndexWriteOptions::FM(FMOptions::from_options(&resolved_options)?.write)
} else {
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
index a31fb165..64314fc2 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/extraction.rs
@@ -38,9 +38,10 @@ pub(super) async fn extract_index_rows(
index_fields: &[&DataField],
index_type: &str,
serialize_key: SerializeKeyFn,
-) -> Result<Vec<SortedIndexKeyRow>> {
+ budget: usize,
+ cmp: crate::btree::key_serde::KeyComparator,
+) -> Result<super::sort::SortedRows> {
let splits = build_read_splits_for_shard(shard)?;
-
let mut read_builder = table.new_read_builder();
let mut projection = index_fields
.iter()
@@ -49,55 +50,111 @@ pub(super) async fn extract_index_rows(
projection.push(ROW_ID_FIELD_NAME);
read_builder.with_projection(&projection)?;
let read = read_builder.new_read()?;
- let batches = read.to_arrow(&splits)?.try_collect::<Vec<_>>().await?;
- let expected_row_count = checked_row_count(shard.row_range_start,
shard.row_range_end)?;
- let index_field = index_fields[0];
- let index_column = index_field.name();
- if index_fields.len() > 1 {
- return extract_composite_index_rows_from_batches(
- &batches,
- index_fields,
- shard.row_range_start,
- expected_row_count,
- );
+ let mut batches = read.to_arrow(&splits)?;
+ let fields = std::sync::Arc::new(
+ index_fields
+ .iter()
+ .map(|field| (*field).clone())
+ .collect::<Vec<_>>(),
+ );
+ let mut sorter = super::sort::IndexSorter::new(budget, cmp);
+ let mut next_row_id = shard.row_range_start;
+ while let Some(batch) = batches.try_next().await? {
+ let fields = fields.clone();
+ let index_type = index_type.to_string();
+ let row_range_start = shard.row_range_start;
+ let row_count = i64::try_from(batch.num_rows()).map_err(|_|
Error::DataInvalid {
+ message: "Index batch row count overflow".into(),
+ source: None,
+ })?;
+ // Sorting and local scratch-file I/O must not block the async reader.
+ sorter = tokio::task::spawn_blocking(move || {
+ let refs = fields.iter().collect::<Vec<_>>();
+ let mut emit = |row| sorter.push(row);
+ let batches = std::slice::from_ref(&batch);
+ if refs.len() > 1 {
+ emit_composite_rows(
+ batches,
+ &refs,
+ row_range_start,
+ row_count,
+ next_row_id,
+ &mut emit,
+ )?;
+ } else if index_type == MULTIVALUE_GLOBAL_INDEX_TYPE {
+ let DataType::Array(array_type) = refs[0].data_type() else {
+ unreachable!("validated multivalue type")
+ };
+ emit_multivalue_rows(
+ batches,
+ refs[0].name(),
+ array_type.element_type(),
+ row_range_start,
+ row_count,
+ serialize_key,
+ next_row_id,
+ &mut emit,
+ )?;
+ } else {
+ emit_scalar_rows(
+ batches,
+ refs[0].name(),
+ refs[0].data_type(),
+ row_range_start,
+ row_count,
+ serialize_key,
+ next_row_id,
+ &mut emit,
+ )?;
+ }
+ Ok::<_, Error>(sorter)
+ })
+ .await
+ .map_err(sort_task_error)??;
+ next_row_id = next_row_id
+ .checked_add(row_count)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Index row ID overflow".into(),
+ source: None,
+ })?;
}
- if index_type == MULTIVALUE_GLOBAL_INDEX_TYPE {
- let DataType::Array(array_type) = index_field.data_type() else {
- unreachable!("multivalue field was validated before extraction")
- };
- extract_multivalue_index_rows_from_batches(
- &batches,
- index_column,
- array_type.element_type(),
- shard.row_range_start,
- expected_row_count,
- serialize_key,
- )
- } else {
- extract_index_rows_from_batches(
- &batches,
- index_column,
- index_field.data_type(),
- shard.row_range_start,
- expected_row_count,
- serialize_key,
- )
+ let expected = checked_row_count(shard.row_range_start,
shard.row_range_end)?;
+ if next_row_id - shard.row_range_start != expected {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Sorted index expected {expected} source rows, got {}",
+ next_row_id - shard.row_range_start
+ ),
+ source: None,
+ });
+ }
+ tokio::task::spawn_blocking(move || sorter.finish())
+ .await
+ .map_err(sort_task_error)?
+}
+
+fn sort_task_error(error: tokio::task::JoinError) -> Error {
+ Error::UnexpectedError {
+ message: format!("Sorted index sort task failed: {error}"),
+ source: None,
}
}
-fn extract_composite_index_rows_from_batches(
+#[allow(clippy::too_many_arguments)]
+fn emit_composite_rows(
batches: &[RecordBatch],
fields: &[&DataField],
row_range_start: i64,
expected_row_count: i64,
-) -> Result<Vec<SortedIndexKeyRow>> {
+ next_row_id: i64,
+ emit: &mut dyn FnMut(SortedIndexKeyRow) -> Result<()>,
+) -> Result<()> {
let fields = fields
.iter()
.map(|field| (*field).clone())
.collect::<Vec<_>>();
let codec = crate::btree::CompositeKeyCodec::new(&fields);
- let mut expected_row_id = row_range_start;
- let mut rows = Vec::new();
+ let mut expected_row_id = next_row_id;
for batch in batches {
let indices = fields
.iter()
@@ -136,10 +193,10 @@ fn extract_composite_index_rows_from_batches(
extract_datum_from_arrow(batch, row, *index,
field.data_type())
})
.collect::<Result<Vec<_>>>()?;
- rows.push((
+ emit((
Some(codec.serialize(&values)?),
expected_row_id - row_range_start,
- ));
+ ))?;
expected_row_id = expected_row_id
.checked_add(1)
.ok_or_else(|| Error::DataInvalid {
@@ -148,16 +205,16 @@ fn extract_composite_index_rows_from_batches(
})?;
}
}
- if expected_row_id - row_range_start != expected_row_count {
+ if expected_row_id - next_row_id != expected_row_count {
return Err(Error::DataInvalid {
message: format!(
"Composite index expected {expected_row_count} rows, got {}",
- rows.len()
+ expected_row_id - next_row_id
),
source: None,
});
}
- Ok(rows)
+ Ok(())
}
pub(super) fn build_read_splits_for_shard(
@@ -198,17 +255,18 @@ pub(super) fn build_read_splits_for_shard(
.collect()
}
-pub(super) fn extract_index_rows_from_batches(
+#[allow(clippy::too_many_arguments)]
+fn emit_scalar_rows(
batches: &[RecordBatch],
index_column: &str,
data_type: &DataType,
row_range_start: i64,
expected_row_count: i64,
serialize_key: SerializeKeyFn,
-) -> Result<Vec<SortedIndexKeyRow>> {
- let row_count = batches.iter().map(RecordBatch::num_rows).sum::<usize>();
- let mut rows = Vec::with_capacity(row_count);
- let mut expected_row_id = row_range_start;
+ next_row_id: i64,
+ emit: &mut dyn FnMut(SortedIndexKeyRow) -> Result<()>,
+) -> Result<()> {
+ let mut expected_row_id = next_row_id;
for batch in batches {
let value_index =
batch
@@ -256,10 +314,10 @@ pub(super) fn extract_index_rows_from_batches(
let key = extract_datum_from_arrow(batch, row, value_index,
data_type)?
.map(|datum| serialize_key(&datum, data_type));
- rows.push((key, row_id - row_range_start));
+ emit((key, row_id - row_range_start))?;
}
}
- let actual_row_count = expected_row_id - row_range_start;
+ let actual_row_count = expected_row_id - next_row_id;
if actual_row_count != expected_row_count {
return Err(Error::DataInvalid {
message: format!(
@@ -269,19 +327,21 @@ pub(super) fn extract_index_rows_from_batches(
source: None,
});
}
- Ok(rows)
+ Ok(())
}
-pub(super) fn extract_multivalue_index_rows_from_batches(
+#[allow(clippy::too_many_arguments)]
+fn emit_multivalue_rows(
batches: &[RecordBatch],
index_column: &str,
element_type: &DataType,
row_range_start: i64,
expected_row_count: i64,
serialize_key: SerializeKeyFn,
-) -> Result<Vec<SortedIndexKeyRow>> {
- let mut rows = Vec::new();
- let mut expected_row_id = row_range_start;
+ next_row_id: i64,
+ emit: &mut dyn FnMut(SortedIndexKeyRow) -> Result<()>,
+) -> Result<()> {
+ let mut expected_row_id = next_row_id;
for batch in batches {
let value_index =
batch
@@ -399,16 +459,16 @@ pub(super) fn extract_multivalue_index_rows_from_batches(
if let Some(datum) =
extract_datum_from_array(values, element_index,
value_index, element_type)?
{
- rows.push((
+ emit((
Some(serialize_key(&datum, element_type)),
row_id - row_range_start,
- ));
+ ))?;
}
}
}
}
- let actual_row_count = expected_row_id - row_range_start;
+ let actual_row_count = expected_row_id - next_row_id;
if actual_row_count != expected_row_count {
return Err(Error::DataInvalid {
message: format!(
@@ -418,7 +478,7 @@ pub(super) fn extract_multivalue_index_rows_from_batches(
source: None,
});
}
- Ok(rows)
+ Ok(())
}
/// Sort the extracted rows with the index's key comparator.
@@ -431,7 +491,7 @@ pub(super) fn sort_index_rows(
cmp: &DynKeyComparator<'_>,
) -> Result<()> {
let mut failure = None;
- rows.sort_by(|left, right| match (&left.0, &right.0) {
+ rows.sort_unstable_by(|left, right| match (&left.0, &right.0) {
(None, None) => left.1.cmp(&right.1),
(None, Some(_)) => Ordering::Less,
(Some(_), None) => Ordering::Greater,
@@ -448,3 +508,55 @@ pub(super) fn sort_index_rows(
None => Ok(()),
}
}
+
+#[cfg(test)]
+pub(super) fn extract_index_rows_from_batches(
+ batches: &[RecordBatch],
+ column: &str,
+ data_type: &DataType,
+ start: i64,
+ count: i64,
+ serialize: SerializeKeyFn,
+) -> Result<Vec<SortedIndexKeyRow>> {
+ let mut rows = Vec::new();
+ emit_scalar_rows(
+ batches,
+ column,
+ data_type,
+ start,
+ count,
+ serialize,
+ start,
+ &mut |row| {
+ rows.push(row);
+ Ok(())
+ },
+ )?;
+ Ok(rows)
+}
+
+#[cfg(test)]
+pub(super) fn extract_multivalue_index_rows_from_batches(
+ batches: &[RecordBatch],
+ column: &str,
+ data_type: &DataType,
+ start: i64,
+ count: i64,
+ serialize: SerializeKeyFn,
+) -> Result<Vec<SortedIndexKeyRow>> {
+ let mut rows = Vec::new();
+ emit_multivalue_rows(
+ batches,
+ column,
+ data_type,
+ start,
+ count,
+ serialize,
+ start,
+ &mut |row| {
+ rows.push(row);
+ Ok(())
+ },
+ )?;
+ Ok(rows)
+}
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/sort.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/sort.rs
new file mode 100644
index 00000000..68e1ecd2
--- /dev/null
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/sort.rs
@@ -0,0 +1,340 @@
+// 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.
+
+//! Bounded sort buffers and pairwise disk merges. Disk work runs on blocking
workers.
+
+use super::extraction::sort_index_rows;
+use super::SortedIndexKeyRow;
+use crate::btree::key_serde::{DynKeyComparator, KeyComparator};
+use crate::{Error, Result};
+use std::cmp::Ordering;
+use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write};
+use tempfile::NamedTempFile;
+use tokio::io::AsyncReadExt;
+
+/// One run per binary level bounds open files during merging to two inputs.
+/// Temporary files are private scratch state and are removed on failure or
cancellation.
+pub(super) struct IndexSorter {
+ rows: Vec<SortedIndexKeyRow>,
+ bytes: usize,
+ budget: usize,
+ cmp: KeyComparator,
+ runs: Vec<Option<NamedTempFile>>,
+}
+
+impl IndexSorter {
+ pub(super) fn new(budget: usize, cmp: KeyComparator) -> Self {
+ Self {
+ rows: Vec::new(),
+ bytes: 0,
+ budget,
+ cmp,
+ runs: Vec::new(),
+ }
+ }
+
+ pub(super) fn push(&mut self, row: SortedIndexKeyRow) -> Result<()> {
+ // Charge spare Vec slots as well as the actual serialized-key
allocation.
+ // The unstable sort uses no additional row-sized scratch allocation.
+ let bytes = row
+ .0
+ .as_ref()
+ .map_or(0, Vec::capacity)
+ .checked_add(2 * std::mem::size_of::<SortedIndexKeyRow>())
+ .ok_or_else(|| io_error(std::io::Error::other("index key size
overflow")))?;
+ if bytes > self.budget {
+ return Err(Error::ResourceExhausted {
+ message: format!("Sorted index key requires {bytes} bytes,
exceeding sorted-index.sort-memory={}", self.budget),
+ });
+ }
+ if bytes > self.budget - self.bytes {
+ self.spill()?;
+ }
+ self.bytes += bytes;
+ // Vec's first amortized allocation otherwise reserves four slots,
+ // exceeding the two-slots-per-row charge for a one-row buffer.
+ if self.rows.capacity() == 0 {
+ self.rows.reserve_exact(1);
+ }
+ self.rows.push(row);
+ Ok(())
+ }
+
+ fn spill(&mut self) -> Result<()> {
+ if self.rows.is_empty() {
+ return Ok(());
+ }
+ sort_index_rows(&mut self.rows, &self.cmp)?;
+ let mut run = NamedTempFile::new().map_err(io_error)?;
+ {
+ let mut output = BufWriter::new(run.as_file_mut());
+ for row in self.rows.drain(..) {
+ write_row(&mut output, &row)?;
+ }
+ output.flush().map_err(io_error)?;
+ }
+ // Release the buffer before merging so old capacity cannot escape the
budget.
+ self.rows = Vec::new();
+ self.bytes = 0;
+ let mut level = 0;
+ loop {
+ if level == self.runs.len() {
+ self.runs.push(None);
+ }
+ match self.runs[level].take() {
+ Some(previous) => run = merge(previous, run, &self.cmp)?,
+ None => {
+ self.runs[level] = Some(run);
+ return Ok(());
+ }
+ }
+ level += 1;
+ }
+ }
+
+ pub(super) fn finish(mut self) -> Result<SortedRows> {
+ if self.runs.is_empty() {
+ sort_index_rows(&mut self.rows, &self.cmp)?;
+ return Ok(SortedRows::Memory(self.rows.into_iter()));
+ }
+ self.spill()?;
+ let mut runs = self.runs.into_iter().flatten();
+ let mut run = runs
+ .next()
+ .ok_or_else(|| io_error(std::io::Error::other("missing index sort
run")))?;
+ for other in runs {
+ run = merge(run, other, &self.cmp)?;
+ }
+ run.as_file_mut()
+ .seek(SeekFrom::Start(0))
+ .map_err(io_error)?;
+ let reader =
+
tokio::io::BufReader::new(tokio::fs::File::from_std(run.reopen().map_err(io_error)?));
+ Ok(SortedRows::Disk { _run: run, reader })
+ }
+}
+
+pub(super) enum SortedRows {
+ Memory(std::vec::IntoIter<SortedIndexKeyRow>),
+ Disk {
+ _run: NamedTempFile,
+ reader: tokio::io::BufReader<tokio::fs::File>,
+ },
+}
+
+impl SortedRows {
+ pub(super) async fn next(&mut self) -> Result<Option<SortedIndexKeyRow>> {
+ match self {
+ Self::Memory(rows) => Ok(rows.next()),
+ Self::Disk { reader, .. } => {
+ let mut tag = [0];
+ if reader.read(&mut tag).await.map_err(io_error)? == 0 {
+ return Ok(None);
+ }
+ let row_id = reader.read_i64_le().await.map_err(io_error)?;
+ let key = if tag[0] == 0 {
+ None
+ } else {
+ let length = reader.read_u64_le().await.map_err(io_error)?;
+ let length = usize::try_from(length).map_err(|_| {
+ io_error(std::io::Error::other("index sort key length
overflow"))
+ })?;
+ let mut key = vec![0; length];
+ reader.read_exact(&mut key).await.map_err(io_error)?;
+ Some(key)
+ };
+ Ok(Some((key, row_id)))
+ }
+ }
+ }
+}
+
+fn io_error(error: std::io::Error) -> Error {
+ Error::UnexpectedError {
+ message: format!("Sorted index temporary file: {error}"),
+ source: Some(Box::new(error)),
+ }
+}
+
+fn write_row(output: &mut impl Write, row: &SortedIndexKeyRow) -> Result<()> {
+ output
+ .write_all(&[u8::from(row.0.is_some())])
+ .map_err(io_error)?;
+ output.write_all(&row.1.to_le_bytes()).map_err(io_error)?;
+ if let Some(key) = &row.0 {
+ output
+ .write_all(&(key.len() as u64).to_le_bytes())
+ .map_err(io_error)?;
+ output.write_all(key).map_err(io_error)?;
+ }
+ Ok(())
+}
+
+fn read_row(input: &mut impl Read) -> Result<Option<SortedIndexKeyRow>> {
+ let mut tag = [0];
+ if input.read(&mut tag).map_err(io_error)? == 0 {
+ return Ok(None);
+ }
+ let mut row_id = [0; 8];
+ input.read_exact(&mut row_id).map_err(io_error)?;
+ let key = if tag[0] == 0 {
+ None
+ } else {
+ let mut length = [0; 8];
+ input.read_exact(&mut length).map_err(io_error)?;
+ let length = usize::try_from(u64::from_le_bytes(length))
+ .map_err(|_| io_error(std::io::Error::other("index sort key length
overflow")))?;
+ let mut key = vec![0; length];
+ input.read_exact(&mut key).map_err(io_error)?;
+ Some(key)
+ };
+ Ok(Some((key, i64::from_le_bytes(row_id))))
+}
+
+pub(super) fn compare_rows(
+ left: &SortedIndexKeyRow,
+ right: &SortedIndexKeyRow,
+ cmp: &DynKeyComparator<'_>,
+) -> Result<Ordering> {
+ Ok(match (&left.0, &right.0) {
+ (None, None) => left.1.cmp(&right.1),
+ (None, Some(_)) => Ordering::Less,
+ (Some(_), None) => Ordering::Greater,
+ (Some(left_key), Some(right_key)) => {
+ cmp(left_key, right_key)?.then_with(|| left.1.cmp(&right.1))
+ }
+ })
+}
+
+fn merge(
+ mut left: NamedTempFile,
+ mut right: NamedTempFile,
+ cmp: &DynKeyComparator<'_>,
+) -> Result<NamedTempFile> {
+ left.as_file_mut()
+ .seek(SeekFrom::Start(0))
+ .map_err(io_error)?;
+ right
+ .as_file_mut()
+ .seek(SeekFrom::Start(0))
+ .map_err(io_error)?;
+ let mut output = NamedTempFile::new().map_err(io_error)?;
+ {
+ let mut output = BufWriter::new(output.as_file_mut());
+ let mut left = BufReader::new(left.as_file_mut());
+ let mut right = BufReader::new(right.as_file_mut());
+ let mut a = read_row(&mut left)?;
+ let mut b = read_row(&mut right)?;
+ while a.is_some() || b.is_some() {
+ let take_left = match (&a, &b) {
+ (Some(a), Some(b)) => compare_rows(a, b, cmp)? !=
Ordering::Greater,
+ (Some(_), None) => true,
+ _ => false,
+ };
+ if take_left {
+ write_row(&mut output, a.as_ref().expect("left row present"))?;
+ a = read_row(&mut left)?;
+ } else {
+ write_row(&mut output, b.as_ref().expect("right row
present"))?;
+ b = read_row(&mut right)?;
+ }
+ }
+ output.flush().map_err(io_error)?;
+ }
+ Ok(output)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn comparator() -> KeyComparator {
+ Box::new(|a, b| Ok(a.cmp(b)))
+ }
+
+ #[tokio::test]
+ async fn spilling_preserves_nulls_duplicates_and_row_id_ties() {
+ let mut input = (0..100)
+ .rev()
+ .map(|id| (Some(vec![(id % 7) as u8]), id))
+ .collect::<Vec<_>>();
+ input.extend([(None, 101), (Some(vec![]), 102), (None, 100)]);
+ let mut expected = input.clone();
+ sort_index_rows(&mut expected, &comparator()).unwrap();
+ let mut sorter = IndexSorter::new(256, comparator());
+ for row in input {
+ sorter.push(row).unwrap();
+ }
+ let mut rows = sorter.finish().unwrap();
+ let SortedRows::Disk { _run, .. } = &rows else {
+ panic!("test must spill")
+ };
+ let path = _run.path().to_owned();
+ let mut actual = Vec::new();
+ while let Some(row) = rows.next().await.unwrap() {
+ actual.push(row);
+ }
+ assert_eq!(actual, expected);
+ drop(rows);
+ assert!(!path.exists());
+ }
+
+ #[tokio::test]
+ async fn disk_merge_uses_numeric_comparison_instead_of_serialized_bytes() {
+ use crate::btree::{make_key_comparator, serialize_datum};
+ use crate::spec::{DataType, Datum, IntType};
+ let data_type = DataType::Int(IntType::new());
+ let values = [256, -1, 1, -256, 0, 255, i32::MAX, i32::MIN];
+ let mut sorter = IndexSorter::new(128,
make_key_comparator(&data_type));
+ for (id, value) in values.into_iter().enumerate() {
+ sorter
+ .push((
+ Some(serialize_datum(&Datum::Int(value), &data_type)),
+ id as i64,
+ ))
+ .unwrap();
+ }
+ let mut rows = sorter.finish().unwrap();
+ assert!(matches!(rows, SortedRows::Disk { .. }));
+ let mut ids = Vec::new();
+ while let Some((_, id)) = rows.next().await.unwrap() {
+ ids.push(id);
+ }
+ assert_eq!(ids, [7, 3, 1, 4, 2, 5, 0, 6]);
+ }
+
+ #[test]
+ fn oversized_keys_fail_and_dropping_sorter_removes_runs() {
+ let mut sorter = IndexSorter::new(128, comparator());
+ for id in 0..10 {
+ sorter.push((Some(vec![1; 16]), id)).unwrap();
+ }
+ let paths = sorter
+ .runs
+ .iter()
+ .flatten()
+ .map(|run| run.path().to_owned())
+ .collect::<Vec<_>>();
+ assert!(!paths.is_empty());
+ assert!(matches!(
+ sorter.push((Some(vec![0; 129]), 11)),
+ Err(Error::ResourceExhausted { .. })
+ ));
+ drop(sorter);
+ assert!(paths.iter().all(|path| !path.exists()));
+ }
+}
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
index 2b463ade..d393a321 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
@@ -833,14 +833,18 @@ async fn
scalar_build_with_deletion_vectors_preserves_physical_coverage() {
#[tokio::test]
async fn test_execute_writes_btree_index_manifest_and_file() {
for version in [1, 2] {
- assert_btree_build_version(version).await;
+ for budget in ["128b", "64mb"] {
+ assert_btree_build_version(version, budget).await;
+ }
}
}
-async fn assert_btree_build_version(version: u32) {
+async fn assert_btree_build_version(version: u32, budget: &str) {
let table_path = "memory:/test_btree_global_index_builder_e2e";
let mut options = table_options("10");
+ options.insert("read.batch-size".into(), "1".into());
options.insert("btree-index.file-version".into(), version.to_string());
+ options.insert("sorted-index.sort-memory".into(), budget.to_string());
options.insert(
"btree-index.bloom-filter.enabled".to_string(),
"true".to_string(),
@@ -1632,7 +1636,9 @@ async fn
test_detail_mode_defers_manifest_pruning_for_unindexed_ranges() {
#[tokio::test]
async fn test_execute_writes_bitmap_index_manifest_and_java_file() {
let table_path = "memory:/test_bitmap_global_index_builder_e2e";
- let table = test_table_with_path(table_path, table_options("10"));
+ let mut options = table_options("10");
+ options.insert("sorted-index.sort-memory".into(), "128b".into());
+ let table = test_table_with_path(table_path, options);
setup_dirs(&table).await;
let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
@@ -1735,7 +1741,10 @@ async fn
test_execute_writes_bitmap_index_manifest_and_java_file() {
#[tokio::test]
async fn test_execute_multivalue_index_and_array_queries_end_to_end() {
let table_path = "memory:/test_multivalue_global_index_builder_e2e";
- let table = multivalue_table(table_path);
+ let table = multivalue_table(table_path).copy_with_options(HashMap::from([
+ ("sorted-index.sort-memory".into(), "128b".into()),
+ ("read.batch-size".into(), "1".into()),
+ ]));
setup_dirs(&table).await;
let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
@@ -2797,6 +2806,7 @@ async fn composite_ranges(
async fn test_composite_btree_build_query_coverage_and_drop() {
for version in [1, 2] {
let mut options = table_options("2");
+ options.insert("sorted-index.sort-memory".into(), "128b".into());
options.insert("btree-index.file-version".into(), version.to_string());
options.insert("btree-index.bloom-filter.enabled".into(),
"true".into());
let table =
@@ -3627,3 +3637,20 @@ async fn
test_composite_btree_preserves_exact_column_names() {
.await
.is_err());
}
+
+#[tokio::test]
+async fn invalid_sort_budget_is_rejected_before_an_empty_build() {
+ for value in ["0", "-1mb", "invalid"] {
+ let mut options = table_options("10");
+ options.insert("sorted-index.sort-memory".into(), value.into());
+ let table = test_table(options);
+ let error = table
+ .new_global_index_build_builder()
+ .with_index_column("name")
+ .build()
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::ConfigInvalid { .. }));
+ assert!(error.to_string().contains("sorted-index.sort-memory"));
+ }
+}
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
index 683ccc31..1523e91d 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
@@ -17,7 +17,7 @@
//! Type-specific index-file writing and failure cleanup.
-use super::extraction::{build_read_splits_for_shard, extract_index_rows,
sort_index_rows};
+use super::extraction::{build_read_splits_for_shard, extract_index_rows};
use super::planning::SortedGlobalIndexShard;
use super::validation::{checked_i64, checked_row_count, index_key_type};
use super::{
@@ -77,13 +77,22 @@ impl SortedGlobalIndexBuildBuilder<'_> {
};
let (cmp, serialize_key) = make_index_key_codec(codec_type, key_type);
let mut rows = if index_type == FM_GLOBAL_INDEX_TYPE {
- Vec::new()
+ super::sort::SortedRows::Memory(Vec::new().into_iter())
} else {
- extract_index_rows(self.table, shard, index_fields, index_type,
serialize_key).await?
+ let mut options = self.table.schema().options().clone();
+ options.extend(self.options.clone());
+ let budget =
crate::spec::CoreOptions::new(&options).sorted_index_sort_memory()?;
+ extract_index_rows(
+ self.table,
+ shard,
+ index_fields,
+ index_type,
+ serialize_key,
+ budget,
+ make_index_key_codec(codec_type, key_type).0,
+ )
+ .await?
};
- if !rows.is_empty() {
- sort_index_rows(&mut rows, &cmp)?;
- }
let file_name = format!("{index_type}-global-index-{}.index",
uuid::Uuid::new_v4());
// Global index placement is a table property, as in Java
GlobalIndexFileFactory.
@@ -124,9 +133,9 @@ impl SortedGlobalIndexBuildBuilder<'_> {
message: "Invalid BTree file version".to_string(),
source: Some(Box::new(error)),
})?;
- for (key, local_row_id) in &rows {
+ while let Some((key, local_row_id)) = rows.next().await? {
writer
- .write(key.as_deref(), *local_row_id)
+ .write(key.as_deref(), local_row_id)
.await
.map_err(|e| Error::DataInvalid {
message: format!(
@@ -152,8 +161,8 @@ impl SortedGlobalIndexBuildBuilder<'_> {
write_options.compression_level,
cmp,
);
- for (key, local_row_id) in &rows {
- writer.write(key.as_deref(),
*local_row_id).map_err(|e| {
+ while let Some((key, local_row_id)) = rows.next().await? {
+ writer.write(key.as_deref(), local_row_id).map_err(|e|
{
Error::DataInvalid {
message: format!(
"Failed to write bitmap global index file
'{file_name}'"
@@ -182,11 +191,11 @@ impl SortedGlobalIndexBuildBuilder<'_> {
write_options.compression_level,
cmp,
);
- for (key, local_row_id) in &rows {
+ while let Some((key, local_row_id)) = rows.next().await? {
let key = key
.as_deref()
.expect("multivalue extraction skips null keys");
- writer.write_posting(key, *local_row_id).map_err(|e| {
+ writer.write_posting(key, local_row_id).map_err(|e| {
Error::DataInvalid {
message: format!(
"Failed to write multivalue global index
file '{file_name}'"
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index bd19817b..d64108bd 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -249,6 +249,24 @@ ordinary writer creation rejects supported index
configuration on these paths.
COW and partial DataEvolution rewrites do not generate indexes. Nested indexes
and additional index types are not supported.
+## Scalar Global Index Build Memory
+
+BTree, Bitmap, and Multivalue global index builds stream projected Arrow
batches
+and sort serialized keys within `sorted-index.sort-memory` (default `64mb`).
+This is a Rust build setting and does not change index files or shard
boundaries.
+Set it as a table property or pass it through the index builder's
`with_options`.
+
+When keys exceed the sort buffer, the build writes temporary runs to the system
+temporary directory and merges them pairwise. Temporary files are removed on
+success, failure, or when cancelled work finishes. Allow local disk space for
+serialized keys and merge output; a merge can temporarily retain both its
inputs
+and output. A single key larger than the sort budget returns
`ResourceExhausted`.
+
+The budget covers retained serialized sort keys and their row slots, not total
+process RSS. The active Arrow batch, reader prefetch, format-writer state, two
+merge keys and fixed I/O buffers have additional memory costs. FM indexes keep
+their existing streaming writer and do not use this sort setting.
+
## Catalog Management
Paimon supports multiple catalog types. The `CatalogFactory` provides a
unified way to create catalogs based on configuration options.
