You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/07 19:39:37 UTC
(datafusion) branch main updated: GC & Deduplicate String View on Spill (#23565)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new b350ea6eb6 GC & Deduplicate String View on Spill (#23565)
b350ea6eb6 is described below
commit b350ea6eb69b875e5f8d1e3d364bb4133e3ac912
Author: Peter L <ce...@hotmail.com>
AuthorDate: Wed Oct 7 19:25:57 2026 +0000
GC & Deduplicate String View on Spill (#23565)
## Which issue does this PR close?
- Closes https://github.com/apache/datafusion/issues/23564
## Rationale for this change
Spill files inflate string views via GC. Arrow's `gc()` copies the bytes
of every view separately, so a highly repeated string view column (for
example a dictionary-encoded Parquet column) spills one copy per row,
causing more memory and disk pressure.
## What changes are included in this PR?
The spill path now compacts `StringView` / `BinaryView` arrays by
copying each distinct value once:
- Values are deduplicated with a hash table over the views, and null
views are zeroed.
- The first 256 non-inline values are sampled. If they contain almost no
repeats, compaction falls back to plain `gc()`, so all-distinct data
does not pay for hashing.
## Are these changes tested?
Yes:
- `test_gc_copies_repeated_values_once` checks that repeated values are
written once, both when views share one buffer (as with Parquet
dictionary columns) and when they reference separate copies.
- `test_gc_distinct_values` checks the fallback to `gc()` for distinct
values.
- Existing spill tests cover the rest of the path.
Benchmarks (`spill_views`, `sort_tpch` with a memory limit) show
0.78–0.94x on repeated-value spills, and no change for all-distinct data
or `sort_tpch`. See the results in the comments.
## Are there any user-facing changes?
No, this is an internal method change.
---------
Co-authored-by: Claude Opus 5.5 <no...@anthropic.com>
Co-authored-by: Adrian Garcia Badaracco <17...@users.noreply.github.com>
---
datafusion/physical-plan/src/spill/mod.rs | 305 ++++++++++++++++++++++++++----
1 file changed, 270 insertions(+), 35 deletions(-)
diff --git a/datafusion/physical-plan/src/spill/mod.rs b/datafusion/physical-plan/src/spill/mod.rs
index 3f56d2a6e0..21e07fa23b 100644
--- a/datafusion/physical-plan/src/spill/mod.rs
+++ b/datafusion/physical-plan/src/spill/mod.rs
@@ -28,7 +28,7 @@ pub use datafusion_common::utils::memory::get_record_batch_memory_size;
#[doc(hidden)]
pub use spill_manager::SpillManager;
-use std::collections::VecDeque;
+use std::collections::{HashMap, VecDeque};
use std::pin::Pin;
use std::sync::{Arc, LazyLock};
use std::task::{Context, Poll};
@@ -37,7 +37,7 @@ use arrow::array::{
Array, ArrayRef, BinaryViewArray, BufferSpec, GenericByteViewArray, StringViewArray,
layout, make_array,
};
-use arrow::buffer::Buffer;
+use arrow::buffer::{Buffer, ScalarBuffer};
use arrow::datatypes::DataType;
use arrow::datatypes::{ByteViewType, Schema, SchemaRef};
use arrow::ipc::{
@@ -46,7 +46,7 @@ use arrow::ipc::{
writer::{IpcWriteOptions, StreamEncoder},
};
use arrow::record_batch::RecordBatch;
-use arrow_data::ArrayDataBuilder;
+use arrow_data::{ArrayDataBuilder, ByteView, MAX_INLINE_VIEW_LEN};
#[cfg(test)]
use arrow_ipc::writer::StreamWriter;
use arrow_ipc::{CompressionType, root_as_message};
@@ -746,6 +746,7 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
/// Size of a single view structure in StringView/BinaryView arrays (in bytes).
/// Each view is 16 bytes: 4 bytes length + 4 bytes prefix + 8 bytes buffer ID/offset.
+#[cfg(test)]
const VIEW_SIZE_BYTES: usize = 16;
/// Performs garbage collection on StringView and BinaryView arrays before spilling to reduce memory usage.
@@ -807,27 +808,105 @@ fn gc_array(array: &ArrayRef) -> Result<(ArrayRef, bool)> {
.as_any()
.downcast_ref::<StringViewArray>()
.expect("Utf8View array should downcast to StringViewArray");
- if should_gc_view_array(string_view) {
- Ok((Arc::new(string_view.gc()) as ArrayRef, true))
- } else {
- Ok((Arc::clone(array), false))
- }
+ Ok(match compact_view_array(string_view) {
+ Some(compacted) => (Arc::new(compacted) as ArrayRef, true),
+ None => (Arc::clone(array), false),
+ })
}
DataType::BinaryView => {
let binary_view = array
.as_any()
.downcast_ref::<BinaryViewArray>()
.expect("BinaryView array should downcast to BinaryViewArray");
- if should_gc_view_array(binary_view) {
- Ok((Arc::new(binary_view.gc()) as ArrayRef, true))
- } else {
- Ok((Arc::clone(array), false))
- }
+ Ok(match compact_view_array(binary_view) {
+ Some(compacted) => (Arc::new(compacted) as ArrayRef, true),
+ None => (Arc::clone(array), false),
+ })
}
_ => gc_array_children(array),
}
}
+/// Compacts a view array before spilling, or returns `None` to spill it as is.
+///
+/// The IPC writer writes every data buffer of an array in full, once per
+/// entry, and `gc()` copies the bytes of every view separately. Neither
+/// handles views that share bytes, as the views of a dictionary-encoded
+/// Parquet column do: after `concat` or `interleave` the dictionary is often
+/// listed once per input batch, and `gc()` copies it once per row
+/// (<https://github.com/apache/datafusion/issues/23564>).
+///
+/// Repeated buffer entries are merged first. `gc()` then runs only if it
+/// writes less than the merged buffers hold, and runs on the original array
+/// so that the views are rewritten at most once.
+fn compact_view_array<T: ByteViewType>(
+ array: &GenericByteViewArray<T>,
+) -> Option<GenericByteViewArray<T>> {
+ let buffers = array.data_buffers();
+ if buffers.is_empty() {
+ return None;
+ }
+
+ let (remap, unique) = dedup_buffer_entries(buffers);
+ let unique_size: usize = unique.iter().map(|b| b.len()).sum();
+ if should_gc_view_array(array, unique_size) {
+ return Some(array.gc());
+ }
+ if unique.len() == buffers.len() {
+ return None;
+ }
+
+ let views: Vec<u128> = array
+ .views()
+ .iter()
+ .enumerate()
+ .map(|(i, &raw)| {
+ if array.is_null(i) {
+ 0
+ } else if (raw as u32) <= MAX_INLINE_VIEW_LEN {
+ raw
+ } else {
+ let view = ByteView::from(raw);
+ ByteView {
+ buffer_index: remap[view.buffer_index as usize],
+ ..view
+ }
+ .as_u128()
+ }
+ })
+ .collect();
+
+ // SAFETY: every non-null, non-inline view points at the same bytes as
+ // before, in an entry of `unique` that holds the same buffer, and inline
+ // views are unchanged.
+ Some(unsafe {
+ GenericByteViewArray::new_unchecked(
+ ScalarBuffer::from(views),
+ unique.into(),
+ array.nulls().cloned(),
+ )
+ })
+}
+
+/// Returns, for each entry of `buffers`, its index in a list without repeated
+/// entries of the same buffer, and that list.
+fn dedup_buffer_entries(buffers: &[Buffer]) -> (Vec<u32>, Vec<Buffer>) {
+ let mut first_index: HashMap<(*const u8, usize), u32> = HashMap::new();
+ let mut unique: Vec<Buffer> = Vec::with_capacity(buffers.len());
+ let remap = buffers
+ .iter()
+ .map(|buffer| {
+ *first_index
+ .entry((buffer.as_ptr(), buffer.len()))
+ .or_insert_with(|| {
+ unique.push(buffer.clone());
+ (unique.len() - 1) as u32
+ })
+ })
+ .collect();
+ (remap, unique)
+}
+
fn gc_array_children(array: &ArrayRef) -> Result<(ArrayRef, bool)> {
let data = array.to_data();
if data.child_data().is_empty() {
@@ -858,28 +937,32 @@ fn gc_array_children(array: &ArrayRef) -> Result<(ArrayRef, bool)> {
Ok((make_array(rebuilt), true))
}
-/// Determines whether a view array should be garbage collected before spilling.
+/// Determines whether a view array should be garbage collected before spilling,
+/// given `data_size`, the bytes its data buffers hold after
+/// [`dedup_buffer_entries`].
///
-/// Arrow's `gc()` always allocates new compact buffers (it is never a no-op), so we
-/// check here to skip the allocation cost when data buffers are small. We subtract
-/// the views buffer (16 bytes × n_rows) from `get_buffer_memory_size()` so the
-/// threshold tracks non-inline string data rather than row count.
-fn should_gc_view_array<T: ByteViewType>(array: &GenericByteViewArray<T>) -> bool {
+/// Arrow's `gc()` always allocates new compact buffers (it is never a no-op), so
+/// it is skipped when the data buffers are small. It is also skipped when it
+/// would write more than the data buffers hold: `gc()` copies the bytes of
+/// every view separately, so views that share bytes each get their own copy.
+/// Both sizes come from the views and buffer lengths, without reading values.
+fn should_gc_view_array<T: ByteViewType>(
+ array: &GenericByteViewArray<T>,
+ data_size: usize,
+) -> bool {
const MIN_BUFFER_SIZE_FOR_GC: usize = 10 * 1024; // 10KB threshold
- if array.data_buffers().is_empty() {
- return false;
- }
+ // `total_buffer_bytes_used` is the size of the data buffer `gc()` writes
+ data_size > MIN_BUFFER_SIZE_FOR_GC && array.total_buffer_bytes_used() < data_size
+}
- let data_buffer_size = array
- .get_buffer_memory_size()
- .saturating_sub(array.len() * VIEW_SIZE_BYTES);
- data_buffer_size > MIN_BUFFER_SIZE_FOR_GC
+#[cfg(test)]
+fn data_buffers_size(array: &StringViewArray) -> usize {
+ array.data_buffers().iter().map(|b| b.len()).sum()
}
#[cfg(test)]
fn calculate_string_view_waste_ratio(array: &StringViewArray) -> f64 {
- use arrow_data::MAX_INLINE_VIEW_LEN;
calculate_view_waste_ratio(array.len(), array.data_buffers(), |i| {
if !array.is_null(i) {
let value = array.value(i);
@@ -2178,13 +2261,12 @@ mod tests {
let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array_ref])?;
// GC should return the original batch for small arrays
- let should_gc = should_gc_view_array(
- batch
- .column(0)
- .as_any()
- .downcast_ref::<StringViewArray>()
- .unwrap(),
- );
+ let array = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .unwrap();
+ let should_gc = should_gc_view_array(array, data_buffers_size(array));
let gc_batch = gc_view_arrays(&batch)?;
assert!(!should_gc);
@@ -2256,7 +2338,8 @@ mod tests {
.as_any()
.downcast_ref::<StringViewArray>()
.unwrap();
- let should_gc = should_gc_view_array(sliced_array);
+ let should_gc =
+ should_gc_view_array(sliced_array, data_buffers_size(sliced_array));
let waste_ratio = calculate_string_view_waste_ratio(sliced_array);
assert!(
@@ -2407,6 +2490,158 @@ mod tests {
Ok(())
}
+ /// `len` distinct 63 byte values in a single data buffer
+ fn dictionary_array(len: u32) -> Result<StringViewArray> {
+ use arrow::array::AsArray;
+
+ let values = StringArray::from_iter_values(
+ (0..len).map(|i| format!("container-id-{i:0>50}")),
+ );
+ // Casting to Utf8View keeps the values in the one existing buffer
+ let dictionary = cast(&values, &DataType::Utf8View)?;
+ let dictionary = dictionary.as_string_view().clone();
+ assert_eq!(dictionary.data_buffers().len(), 1);
+ Ok(dictionary)
+ }
+
+ /// Arrays that each hold views into one shared dictionary buffer, as the
+ /// Parquet reader produces for a dictionary-encoded column, combined the
+ /// way a sort combines its input batches. Each output row cycles through
+ /// the dictionary (<https://github.com/apache/datafusion/issues/23564>).
+ fn shared_dictionary_views(
+ dictionary_len: u32,
+ inputs: u32,
+ rows_per_input: u32,
+ ) -> Result<(StringViewArray, StringViewArray)> {
+ use arrow::array::{AsArray, UInt32Array};
+ use arrow::compute::{interleave, take};
+
+ let dictionary = dictionary_array(dictionary_len)?;
+ let arrays = (0..inputs)
+ .map(|input| {
+ let indices = UInt32Array::from_iter_values(
+ (0..rows_per_input)
+ .map(|row| (input * rows_per_input + row) % dictionary_len),
+ );
+ take(&dictionary, &indices, None)
+ })
+ .collect::<std::result::Result<Vec<_>, _>>()?;
+ let arrays: Vec<&dyn Array> = arrays.iter().map(|a| a.as_ref()).collect();
+ let indices: Vec<(usize, usize)> = (0..rows_per_input as usize)
+ .flat_map(|row| (0..inputs as usize).map(move |input| (input, row)))
+ .collect();
+ let merged = interleave(&arrays, &indices)?;
+ Ok((dictionary, merged.as_string_view().clone()))
+ }
+
+ #[test]
+ fn test_gc_keeps_shared_dictionary_buffer() -> Result<()> {
+ let (dictionary, merged) = shared_dictionary_views(1000, 8, 1024)?;
+ // `interleave` lists the dictionary once per input
+ assert_eq!(merged.data_buffers().len(), 8);
+ // `gc()` would copy 8192 values of 63 bytes, 8x the dictionary
+ assert!(merged.total_buffer_bytes_used() > data_buffers_size(&dictionary));
+
+ let compacted = compact_view_array(&merged).expect("buffer entries repeat");
+ assert_eq!(compacted.data_buffers().len(), 1);
+ assert_eq!(
+ compacted.data_buffers()[0].as_ptr(),
+ dictionary.data_buffers()[0].as_ptr()
+ );
+ assert_eq!(compacted, merged);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_small_slice_of_shared_buffer() -> Result<()> {
+ // 8 inputs of 4 rows each reference a 63 KB dictionary: `gc()` writes
+ // 32 values, far less than the dictionary
+ let (dictionary, merged) = shared_dictionary_views(1000, 8, 4)?;
+ assert_eq!(merged.data_buffers().len(), 8);
+
+ let compacted = compact_view_array(&merged).expect("gc() is smaller");
+ assert_eq!(compacted.data_buffers().len(), 1);
+ assert_eq!(data_buffers_size(&compacted), 32 * 63);
+ assert_ne!(
+ compacted.data_buffers()[0].as_ptr(),
+ dictionary.data_buffers()[0].as_ptr()
+ );
+ assert_eq!(compacted, merged);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_shared_buffer_with_nulls() -> Result<()> {
+ use arrow::array::{AsArray, BooleanArray};
+ use arrow::compute::nullif;
+
+ let (_, merged) = shared_dictionary_views(1000, 8, 1024)?;
+ let mask = BooleanArray::from_iter((0..merged.len()).map(|i| Some(i % 7 == 0)));
+ let with_nulls = nullif(&merged, &mask)?;
+ let with_nulls = with_nulls.as_string_view();
+
+ let compacted = compact_view_array(with_nulls).expect("buffer entries repeat");
+ assert_eq!(compacted.data_buffers().len(), 1);
+ assert_eq!(compacted.null_count(), with_nulls.null_count());
+ assert_eq!(&compacted, with_nulls);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_skips_array_that_shares_one_buffer() -> Result<()> {
+ use arrow::array::{AsArray, UInt32Array};
+ use arrow::compute::take;
+
+ // 10,000 views into one 63 KB dictionary: `gc()` would write 630 KB
+ let dictionary = dictionary_array(1000)?;
+ let indices = UInt32Array::from_iter_values((0..10_000).map(|i| i % 1000));
+ let shared = take(&dictionary, &indices, None)?;
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "label",
+ DataType::Utf8View,
+ false,
+ )]));
+ let batch = RecordBatch::try_new(schema, vec![Arc::clone(&shared)])?;
+
+ assert!(compact_view_array(shared.as_string_view()).is_none());
+ let gc_batch = gc_view_arrays(&batch)?;
+ assert!(Arc::ptr_eq(batch.column(0), gc_batch.column(0)));
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_spill_shared_dictionary_writes_it_once() -> Result<()> {
+ let (dictionary, merged) = shared_dictionary_views(1000, 8, 1024)?;
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "label",
+ DataType::Utf8View,
+ false,
+ )]));
+ let batch = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(merged)])?;
+
+ let env = Arc::new(RuntimeEnv::default());
+ let metrics = SpillMetrics::new(&ExecutionPlanMetricsSet::new(), 0);
+ let spill_manager = SpillManager::new(env, metrics, schema);
+ let mut in_progress_file = spill_manager.create_in_progress_file("Test GC")?;
+ in_progress_file.append_batch(&batch)?;
+ let spill_file = in_progress_file.finish()?.unwrap();
+ let file_size = std::fs::metadata(spill_file.path().unwrap())?.len() as usize;
+
+ // The views and one copy of the dictionary, not 8192 copied values
+ let views_size = batch.num_rows() * VIEW_SIZE_BYTES;
+ let dictionary_size = data_buffers_size(&dictionary);
+ assert!(
+ file_size < views_size + 2 * dictionary_size,
+ "spill file is {file_size} bytes, views {views_size}, dictionary {dictionary_size}"
+ );
+
+ Ok(())
+ }
+
#[test]
fn test_gc_recurses_into_nested_view_arrays() -> Result<()> {
use arrow::array::{DictionaryArray, Int32Array};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org