You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "EeshanBembi (via GitHub)" <gi...@apache.org> on 2025/12/21 19:53:36 UTC
[PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
EeshanBembi opened a new pull request, #19444:
URL: https://github.com/apache/datafusion/pull/19444
Add garbage collection for StringView and BinaryView arrays before spilling to disk. This prevents sliced arrays from carrying their entire original buffers when written to spill files.
Changes:
- Add gc_view_arrays() function to apply GC on view arrays
- Integrate GC into InProgressSpillFile::append_batch()
- Use simple threshold-based heuristic (100+ rows, 10KB+ buffer size)
Fixes #19414 where GROUP BY on StringView columns created 820MB spill files instead of 33MB due to sliced arrays maintaining references to original buffers.
Testing shows 80-98% reduction in spill file sizes for typical GROUP BY workloads.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "bharath-techie (via GitHub)" <gi...@apache.org>.
bharath-techie commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2767380889
##########
datafusion/physical-plan/Cargo.toml:
##########
@@ -73,6 +73,7 @@ pin-project-lite = "^0.2.7"
tokio = { workspace = true }
[dev-dependencies]
+arrow-data = { workspace = true }
Review Comment:
is this unintentional change / git diff ?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-4247640419
I opened https://github.com/apache/datafusion/pull/21633 to continue this work since I cannot push to this branch.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "EeshanBembi (via GitHub)" <gi...@apache.org>.
EeshanBembi commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2779395991
##########
datafusion/physical-plan/Cargo.toml:
##########
@@ -73,6 +73,7 @@ pin-project-lite = "^0.2.7"
tokio = { workspace = true }
[dev-dependencies]
+arrow-data = { workspace = true }
Review Comment:
This is intentional, arrow-data is added as a dev-dependency because the test helper calculate_string_view_waste_ratio uses arrow_data::MAX_INLINE_VIEW_LEN to determine whether a string value would be inlined in a StringView array.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-3900041410
Hi @EeshanBembi just pinging here, I'd love to see this across the line, just missing one final push!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2813210726
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -344,6 +344,138 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+/// 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.
+///
+/// # Why GC is needed
+///
+/// StringView and BinaryView arrays can accumulate significant memory waste when sliced.
+/// When a large array is sliced (e.g., taking first 100 rows of 1000), the view array
+/// still references the original data buffers containing all 1000 rows of data.
+///
+/// For example, in the ClickBench benchmark (issue #19414), repeated slicing of StringView
+/// arrays resulted in 820MB of spill files that could be reduced to just 33MB after GC -
+/// a 96% reduction in size.
+///
+/// # How it works
+///
+/// The GC process creates new compact buffers containing only the data referenced by the
+/// current views, eliminating unreferenced data from sliced arrays.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ // Early return optimization: Skip GC entirely if the batch contains no view arrays.
+ // This avoids unnecessary processing for batches with only primitive types.
+ let has_view_arrays = batch.columns().iter().any(|array| {
+ matches!(
+ array.data_type(),
+ arrow::datatypes::DataType::Utf8View | arrow::datatypes::DataType::BinaryView
+ )
+ });
+
+ if !has_view_arrays {
+ // RecordBatch::clone() is cheap - just Arc reference count bumps
+ return Ok(batch.clone());
+ }
+
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ // Only perform GC if the data buffers exceed our size threshold.
+ // This balances memory savings against GC overhead.
+ if should_gc_view_array(string_view.data_buffers()) {
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::DataType::BinaryView => {
+ let binary_view = array
+ .as_any()
+ .downcast_ref::<BinaryViewArray>()
+ .expect("BinaryView array should downcast to BinaryViewArray");
+ // Only perform GC if the data buffers exceed our size threshold.
+ // This balances memory savings against GC overhead.
+ if should_gc_view_array(binary_view.data_buffers()) {
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ // Non-view arrays are passed through unchanged
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ // Always return a new batch for consistency
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+}
+
+/// Determines whether a view array should be garbage collected based on its buffer usage.
+///
+/// Uses a minimum buffer size threshold to avoid unnecessary GC on small arrays.
+/// This prevents the overhead of GC for arrays with negligible memory footprint,
+/// while still capturing cases where sliced arrays carry large unreferenced buffers.
+///
+/// # Why not use get_record_batch_memory_size
+///
+/// We use manual buffer size calculation here because:
+/// - `get_record_batch_memory_size` operates on entire arrays, not just the data buffers
+/// - We need to check buffer capacity specifically to determine GC potential
+/// - The data buffers are what gets compacted during GC, so their size is the key metric
+fn should_gc_view_array(data_buffers: &[arrow::buffer::Buffer]) -> bool {
Review Comment:
Can we check if there is anything to gc (i.e. if the array is sliced)? Or is `gc()` a no-op in that case?
##########
datafusion/physical-plan/src/spill/in_progress_spill_file.rs:
##########
@@ -50,6 +50,7 @@ impl InProgressSpillFile {
}
/// Appends a `RecordBatch` to the spill file, initializing the writer if necessary.
+ /// Performs garbage collection on StringView/BinaryView arrays to reduce spill file size.
Review Comment:
@EeshanBembi could you enhance the comment here?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb closed pull request #19444: Fix massive spill files for StringView/BinaryView columns
URL: https://github.com/apache/datafusion/pull/19444
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "ebembi-crdb (via GitHub)" <gi...@apache.org>.
ebembi-crdb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2758321283
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
Review Comment:
Fixed, always returning a new batch for consistency.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -799,4 +890,426 @@ mod tests {
assert_eq!(alignment, 8);
Ok(())
}
+
+ #[test]
+ fn test_gc_string_view_before_spill() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ "short_string".to_string()
+ } else {
+ "this_is_a_much_longer_string_that_will_not_be_inlined".to_string()
+ }
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_binary_view_before_spill() -> Result<()> {
+ use arrow::array::BinaryViewArray;
+
+ let binaries: Vec<Vec<u8>> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ vec![1, 2, 3, 4]
+ } else {
+ vec![1; 50]
+ }
+ })
+ .collect();
+
+ let binary_array =
+ BinaryViewArray::from_iter(binaries.iter().map(|b| Some(b.as_slice())));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "binaries",
+ DataType::BinaryView,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(binary_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_skips_small_arrays() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..10).map(|i| format!("string_{i}")).collect();
+
+ let string_array = StringViewArray::from(strings);
+ let array_ref: ArrayRef = Arc::new(string_array);
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array_ref])?;
+
+ // GC should return the original batch for small arrays
+ let gc_batch = gc_view_arrays(&batch)?;
+
+ // The batch should be unchanged (cloned, not GC'd)
+ assert_eq!(gc_batch.num_rows(), batch.num_rows());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_with_mixed_columns() -> Result<()> {
+ use arrow::array::{Int32Array, StringViewArray};
+
+ let strings: Vec<String> = (0..200)
+ .map(|i| format!("long_string_for_gc_testing_{i}"))
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let int_array = Int32Array::from((0..200).collect::<Vec<i32>>());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("strings", DataType::Utf8View, false),
+ Field::new("ints", DataType::Int32, false),
+ ]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(string_array) as ArrayRef,
+ Arc::new(int_array) as ArrayRef,
+ ],
+ )?;
+
+ let sliced_batch = batch.slice(0, 50);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_columns(), 2);
+ assert_eq!(gc_batch.num_rows(), 50);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_verify_gc_triggers_for_sliced_arrays() -> Result<()> {
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ format!(
+ "http://example.com/very/long/path/that/exceeds/inline/threshold/{i}"
+ )
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "url",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let sliced = batch.slice(0, 100);
+
+ let sliced_array = sliced
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .unwrap();
+ let should_gc =
+ should_gc_view_array(sliced_array.len(), sliced_array.data_buffers());
+ let waste_ratio = calculate_string_view_waste_ratio(sliced_array);
+
+ assert!(
+ waste_ratio > 0.8,
+ "Waste ratio should be > 0.8 for sliced array"
+ );
+ assert!(
+ should_gc,
+ "GC should trigger for sliced array with high waste"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reproduce_issue_19414_string_view_spill_without_gc() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ let num_rows = 5000;
+ let mut strings = Vec::with_capacity(num_rows);
+
+ for i in 0..num_rows {
+ let url = match i % 5 {
+ 0 => format!(
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130/{i}"
+ ),
+ 1 => format!("http://komme%2F27.0.1453.116/very/long/path/{i}"),
+ 2 => format!("https://produkty%2Fproduct/category/item/{i}"),
+ 3 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668/{i}"
+ ),
+ 4 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapustic/product/{i}"
+ ),
+ _ => unreachable!(),
+ };
+ strings.push(url);
+ }
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "URL",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let original_batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let total_buffer_size: usize = string_array
+ .data_buffers()
+ .iter()
+ .map(|buffer| buffer.capacity())
+ .sum();
+
+ let mut sliced_batches = Vec::new();
+ let slice_size = 100;
+
+ for i in (0..num_rows).step_by(slice_size) {
+ let len = std::cmp::min(slice_size, num_rows - i);
+ let sliced = original_batch.slice(i, len);
+ sliced_batches.push(sliced);
+ }
+
+ 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")?;
+
+ for batch in &sliced_batches {
+ in_progress_file.append_batch(batch)?;
+ }
+
+ let spill_file = in_progress_file.finish()?.unwrap();
+ let file_size = fs::metadata(spill_file.path())?.len() as usize;
+
+ let theoretical_without_gc = total_buffer_size * sliced_batches.len();
+ let reduction_percent = ((theoretical_without_gc - file_size) as f64
+ / theoretical_without_gc as f64)
+ * 100.0;
+
+ assert!(
+ reduction_percent > 80.0,
+ "GC should reduce spill file size by >80%, got {reduction_percent:.1}%"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_exact_clickbench_issue_19414() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ // Test for clickbench issue: 820MB -> 33MB spill reduction
+ let unique_urls = vec![
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130",
+ "http://komme%2F27.0.1453.116",
+ "https://produkty%2Fproduct",
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668]=0&order_by=0",
+ "http://irr.ru/index.php?showalbum/login-kapustic/product_name",
+ "http://irr.ru/index.php",
+ "https://produkty%2F",
+ "http://irr.ru/index.php?showalbum/login",
+ "https://produkty/kurortmag",
+ "https://produkty%2Fpulove.ru/album/login",
+ ];
+
+ let mut urls = Vec::with_capacity(200_000);
Review Comment:
Done, reduced to 5,000 rows.
##########
datafusion/physical-plan/src/spill/in_progress_spill_file.rs:
##########
@@ -50,6 +50,7 @@ impl InProgressSpillFile {
}
/// Appends a `RecordBatch` to the spill file, initializing the writer if necessary.
+ /// Performs garbage collection on StringView/BinaryView arrays to reduce spill file size.
Review Comment:
Added doc comments explaining the why.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-4260614160
@EeshanBembi #21633 is now approved. It includes your commits so I think we can just merge that and you will get commit credit for your work. Does that sound okay?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "EeshanBembi (via GitHub)" <gi...@apache.org>.
EeshanBembi commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-4257404519
> I opened #21633 to continue this work since I cannot push to this branch.
Hey @adriangb , i didn't realize this required more work, i mistakenly thought it's stuck in review phase. I can still work on it if needed!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "EeshanBembi (via GitHub)" <gi...@apache.org>.
EeshanBembi commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-4261422491
> @EeshanBembi #21633 is now approved. It includes your commits so I think we can just merge that and you will get commit credit for your work. Does that sound okay?
Sure that works, thanks!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "ebembi-crdb (via GitHub)" <gi...@apache.org>.
ebembi-crdb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2758317641
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -799,4 +890,426 @@ mod tests {
assert_eq!(alignment, 8);
Ok(())
}
+
+ #[test]
+ fn test_gc_string_view_before_spill() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ "short_string".to_string()
+ } else {
+ "this_is_a_much_longer_string_that_will_not_be_inlined".to_string()
+ }
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_binary_view_before_spill() -> Result<()> {
+ use arrow::array::BinaryViewArray;
+
+ let binaries: Vec<Vec<u8>> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ vec![1, 2, 3, 4]
+ } else {
+ vec![1; 50]
+ }
+ })
+ .collect();
+
+ let binary_array =
+ BinaryViewArray::from_iter(binaries.iter().map(|b| Some(b.as_slice())));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "binaries",
+ DataType::BinaryView,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(binary_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_skips_small_arrays() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..10).map(|i| format!("string_{i}")).collect();
+
+ let string_array = StringViewArray::from(strings);
+ let array_ref: ArrayRef = Arc::new(string_array);
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array_ref])?;
+
+ // GC should return the original batch for small arrays
+ let gc_batch = gc_view_arrays(&batch)?;
+
+ // The batch should be unchanged (cloned, not GC'd)
+ assert_eq!(gc_batch.num_rows(), batch.num_rows());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_with_mixed_columns() -> Result<()> {
+ use arrow::array::{Int32Array, StringViewArray};
+
+ let strings: Vec<String> = (0..200)
+ .map(|i| format!("long_string_for_gc_testing_{i}"))
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let int_array = Int32Array::from((0..200).collect::<Vec<i32>>());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("strings", DataType::Utf8View, false),
+ Field::new("ints", DataType::Int32, false),
+ ]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(string_array) as ArrayRef,
+ Arc::new(int_array) as ArrayRef,
+ ],
+ )?;
+
+ let sliced_batch = batch.slice(0, 50);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_columns(), 2);
+ assert_eq!(gc_batch.num_rows(), 50);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_verify_gc_triggers_for_sliced_arrays() -> Result<()> {
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ format!(
+ "http://example.com/very/long/path/that/exceeds/inline/threshold/{i}"
+ )
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "url",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let sliced = batch.slice(0, 100);
+
+ let sliced_array = sliced
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .unwrap();
+ let should_gc =
+ should_gc_view_array(sliced_array.len(), sliced_array.data_buffers());
+ let waste_ratio = calculate_string_view_waste_ratio(sliced_array);
+
+ assert!(
+ waste_ratio > 0.8,
+ "Waste ratio should be > 0.8 for sliced array"
+ );
+ assert!(
+ should_gc,
+ "GC should trigger for sliced array with high waste"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reproduce_issue_19414_string_view_spill_without_gc() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ let num_rows = 5000;
+ let mut strings = Vec::with_capacity(num_rows);
+
+ for i in 0..num_rows {
+ let url = match i % 5 {
+ 0 => format!(
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130/{i}"
+ ),
+ 1 => format!("http://komme%2F27.0.1453.116/very/long/path/{i}"),
+ 2 => format!("https://produkty%2Fproduct/category/item/{i}"),
+ 3 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668/{i}"
+ ),
+ 4 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapustic/product/{i}"
+ ),
+ _ => unreachable!(),
+ };
+ strings.push(url);
+ }
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "URL",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let original_batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let total_buffer_size: usize = string_array
+ .data_buffers()
+ .iter()
+ .map(|buffer| buffer.capacity())
+ .sum();
+
+ let mut sliced_batches = Vec::new();
+ let slice_size = 100;
+
+ for i in (0..num_rows).step_by(slice_size) {
+ let len = std::cmp::min(slice_size, num_rows - i);
+ let sliced = original_batch.slice(i, len);
+ sliced_batches.push(sliced);
+ }
+
+ 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")?;
+
+ for batch in &sliced_batches {
+ in_progress_file.append_batch(batch)?;
+ }
+
+ let spill_file = in_progress_file.finish()?.unwrap();
+ let file_size = fs::metadata(spill_file.path())?.len() as usize;
+
+ let theoretical_without_gc = total_buffer_size * sliced_batches.len();
+ let reduction_percent = ((theoretical_without_gc - file_size) as f64
+ / theoretical_without_gc as f64)
+ * 100.0;
+
+ assert!(
+ reduction_percent > 80.0,
+ "GC should reduce spill file size by >80%, got {reduction_percent:.1}%"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_exact_clickbench_issue_19414() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ // Test for clickbench issue: 820MB -> 33MB spill reduction
+ let unique_urls = vec![
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130",
+ "http://komme%2F27.0.1453.116",
+ "https://produkty%2Fproduct",
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668]=0&order_by=0",
+ "http://irr.ru/index.php?showalbum/login-kapustic/product_name",
+ "http://irr.ru/index.php",
+ "https://produkty%2F",
+ "http://irr.ru/index.php?showalbum/login",
+ "https://produkty/kurortmag",
+ "https://produkty%2Fpulove.ru/album/login",
+ ];
+
+ let mut urls = Vec::with_capacity(200_000);
Review Comment:
Done, reduced to 5,000 rows.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
Review Comment:
Fixed, using `batch.clone()` now which is just Arc bumps.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "bharath-techie (via GitHub)" <gi...@apache.org>.
bharath-techie commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2639830983
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
Review Comment:
Maybe lets exit early and return the batch as it is if there are no view arrays in the `RecordBatch` ?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "pepijnve (via GitHub)" <gi...@apache.org>.
pepijnve commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2639921528
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
Review Comment:
There's a public constant for this `MAX_INLINE_VIEW_LEN`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "ebembi-crdb (via GitHub)" <gi...@apache.org>.
ebembi-crdb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2758319104
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
Review Comment:
Added early return before processing.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "ebembi-crdb (via GitHub)" <gi...@apache.org>.
ebembi-crdb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2758321447
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -799,4 +890,426 @@ mod tests {
assert_eq!(alignment, 8);
Ok(())
}
+
+ #[test]
+ fn test_gc_string_view_before_spill() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ "short_string".to_string()
+ } else {
+ "this_is_a_much_longer_string_that_will_not_be_inlined".to_string()
+ }
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_binary_view_before_spill() -> Result<()> {
+ use arrow::array::BinaryViewArray;
+
+ let binaries: Vec<Vec<u8>> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ vec![1, 2, 3, 4]
+ } else {
+ vec![1; 50]
+ }
+ })
+ .collect();
+
+ let binary_array =
+ BinaryViewArray::from_iter(binaries.iter().map(|b| Some(b.as_slice())));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "binaries",
+ DataType::BinaryView,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(binary_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_skips_small_arrays() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..10).map(|i| format!("string_{i}")).collect();
+
+ let string_array = StringViewArray::from(strings);
+ let array_ref: ArrayRef = Arc::new(string_array);
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array_ref])?;
+
+ // GC should return the original batch for small arrays
+ let gc_batch = gc_view_arrays(&batch)?;
+
+ // The batch should be unchanged (cloned, not GC'd)
+ assert_eq!(gc_batch.num_rows(), batch.num_rows());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_with_mixed_columns() -> Result<()> {
+ use arrow::array::{Int32Array, StringViewArray};
+
+ let strings: Vec<String> = (0..200)
+ .map(|i| format!("long_string_for_gc_testing_{i}"))
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let int_array = Int32Array::from((0..200).collect::<Vec<i32>>());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("strings", DataType::Utf8View, false),
+ Field::new("ints", DataType::Int32, false),
+ ]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(string_array) as ArrayRef,
+ Arc::new(int_array) as ArrayRef,
+ ],
+ )?;
+
+ let sliced_batch = batch.slice(0, 50);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_columns(), 2);
+ assert_eq!(gc_batch.num_rows(), 50);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_verify_gc_triggers_for_sliced_arrays() -> Result<()> {
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ format!(
+ "http://example.com/very/long/path/that/exceeds/inline/threshold/{i}"
+ )
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "url",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let sliced = batch.slice(0, 100);
+
+ let sliced_array = sliced
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .unwrap();
+ let should_gc =
+ should_gc_view_array(sliced_array.len(), sliced_array.data_buffers());
+ let waste_ratio = calculate_string_view_waste_ratio(sliced_array);
+
+ assert!(
+ waste_ratio > 0.8,
+ "Waste ratio should be > 0.8 for sliced array"
+ );
+ assert!(
+ should_gc,
+ "GC should trigger for sliced array with high waste"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reproduce_issue_19414_string_view_spill_without_gc() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ let num_rows = 5000;
+ let mut strings = Vec::with_capacity(num_rows);
+
+ for i in 0..num_rows {
+ let url = match i % 5 {
+ 0 => format!(
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130/{i}"
+ ),
+ 1 => format!("http://komme%2F27.0.1453.116/very/long/path/{i}"),
+ 2 => format!("https://produkty%2Fproduct/category/item/{i}"),
+ 3 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668/{i}"
+ ),
+ 4 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapustic/product/{i}"
+ ),
+ _ => unreachable!(),
+ };
+ strings.push(url);
+ }
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "URL",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let original_batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let total_buffer_size: usize = string_array
+ .data_buffers()
+ .iter()
+ .map(|buffer| buffer.capacity())
+ .sum();
+
+ let mut sliced_batches = Vec::new();
+ let slice_size = 100;
+
+ for i in (0..num_rows).step_by(slice_size) {
+ let len = std::cmp::min(slice_size, num_rows - i);
+ let sliced = original_batch.slice(i, len);
+ sliced_batches.push(sliced);
+ }
+
+ 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")?;
+
+ for batch in &sliced_batches {
+ in_progress_file.append_batch(batch)?;
+ }
+
+ let spill_file = in_progress_file.finish()?.unwrap();
+ let file_size = fs::metadata(spill_file.path())?.len() as usize;
+
+ let theoretical_without_gc = total_buffer_size * sliced_batches.len();
+ let reduction_percent = ((theoretical_without_gc - file_size) as f64
+ / theoretical_without_gc as f64)
+ * 100.0;
+
+ assert!(
+ reduction_percent > 80.0,
+ "GC should reduce spill file size by >80%, got {reduction_percent:.1}%"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_exact_clickbench_issue_19414() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ // Test for clickbench issue: 820MB -> 33MB spill reduction
+ let unique_urls = vec![
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130",
+ "http://komme%2F27.0.1453.116",
+ "https://produkty%2Fproduct",
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668]=0&order_by=0",
+ "http://irr.ru/index.php?showalbum/login-kapustic/product_name",
+ "http://irr.ru/index.php",
+ "https://produkty%2F",
+ "http://irr.ru/index.php?showalbum/login",
+ "https://produkty/kurortmag",
+ "https://produkty%2Fpulove.ru/album/login",
+ ];
+
+ let mut urls = Vec::with_capacity(200_000);
Review Comment:
Done, reduced to 5,000 rows.
##########
datafusion/physical-plan/src/spill/in_progress_spill_file.rs:
##########
@@ -50,6 +50,7 @@ impl InProgressSpillFile {
}
/// Appends a `RecordBatch` to the spill file, initializing the writer if necessary.
+ /// Performs garbage collection on StringView/BinaryView arrays to reduce spill file size.
Review Comment:
Added doc comments explaining the why.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "EeshanBembi (via GitHub)" <gi...@apache.org>.
EeshanBembi commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-3907996708
Hey @adriangb , could you please have a look? Thanks
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2817351274
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
+ }
+}
+
+fn should_gc_view_array(len: usize, data_buffers: &[arrow::buffer::Buffer]) -> bool {
+ if len < 10 {
+ return false;
+ }
+
+ let total_buffer_size: usize = data_buffers.iter().map(|b| b.capacity()).sum();
Review Comment:
@EeshanBembi could you address this please?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "pepijnve (via GitHub)" <gi...@apache.org>.
pepijnve commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2639868657
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
+ }
+}
+
+fn should_gc_view_array(len: usize, data_buffers: &[arrow::buffer::Buffer]) -> bool {
+ if len < 10 {
Review Comment:
Is the number of rows a useful heuristic to not GC? Even if there are few rows, the data buffer may still be large.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-4257492332
okay yes happy to continue here! I apologize if I misunderstood as well, I took it you weren't able to work on this anymore. how would you like to proceed with the PRs? I'm happy to close mine but it does have the rebase + some reviewer feedback addressed that is probably worth preserving.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "bharath-techie (via GitHub)" <gi...@apache.org>.
bharath-techie commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2639826123
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
Review Comment:
Can we just return the batch without clone ?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "pepijnve (via GitHub)" <gi...@apache.org>.
pepijnve commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2639880004
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
+ }
+}
+
+fn should_gc_view_array(len: usize, data_buffers: &[arrow::buffer::Buffer]) -> bool {
+ if len < 10 {
+ return false;
+ }
+
+ let total_buffer_size: usize = data_buffers.iter().map(|b| b.capacity()).sum();
Review Comment:
Can we use some of the existing size calculation methods like `get_buffer_memory_size` instead of duplicating calculations?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "ebembi-crdb (via GitHub)" <gi...@apache.org>.
ebembi-crdb commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-3840412279
Done, removed `organize_stringview_arrays()`, SpillManager handles it now.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "ebembi-crdb (via GitHub)" <gi...@apache.org>.
ebembi-crdb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2758317641
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -799,4 +890,426 @@ mod tests {
assert_eq!(alignment, 8);
Ok(())
}
+
+ #[test]
+ fn test_gc_string_view_before_spill() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ "short_string".to_string()
+ } else {
+ "this_is_a_much_longer_string_that_will_not_be_inlined".to_string()
+ }
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_binary_view_before_spill() -> Result<()> {
+ use arrow::array::BinaryViewArray;
+
+ let binaries: Vec<Vec<u8>> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ vec![1, 2, 3, 4]
+ } else {
+ vec![1; 50]
+ }
+ })
+ .collect();
+
+ let binary_array =
+ BinaryViewArray::from_iter(binaries.iter().map(|b| Some(b.as_slice())));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "binaries",
+ DataType::BinaryView,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(binary_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_skips_small_arrays() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..10).map(|i| format!("string_{i}")).collect();
+
+ let string_array = StringViewArray::from(strings);
+ let array_ref: ArrayRef = Arc::new(string_array);
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array_ref])?;
+
+ // GC should return the original batch for small arrays
+ let gc_batch = gc_view_arrays(&batch)?;
+
+ // The batch should be unchanged (cloned, not GC'd)
+ assert_eq!(gc_batch.num_rows(), batch.num_rows());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_with_mixed_columns() -> Result<()> {
+ use arrow::array::{Int32Array, StringViewArray};
+
+ let strings: Vec<String> = (0..200)
+ .map(|i| format!("long_string_for_gc_testing_{i}"))
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let int_array = Int32Array::from((0..200).collect::<Vec<i32>>());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("strings", DataType::Utf8View, false),
+ Field::new("ints", DataType::Int32, false),
+ ]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(string_array) as ArrayRef,
+ Arc::new(int_array) as ArrayRef,
+ ],
+ )?;
+
+ let sliced_batch = batch.slice(0, 50);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_columns(), 2);
+ assert_eq!(gc_batch.num_rows(), 50);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_verify_gc_triggers_for_sliced_arrays() -> Result<()> {
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ format!(
+ "http://example.com/very/long/path/that/exceeds/inline/threshold/{i}"
+ )
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "url",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let sliced = batch.slice(0, 100);
+
+ let sliced_array = sliced
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .unwrap();
+ let should_gc =
+ should_gc_view_array(sliced_array.len(), sliced_array.data_buffers());
+ let waste_ratio = calculate_string_view_waste_ratio(sliced_array);
+
+ assert!(
+ waste_ratio > 0.8,
+ "Waste ratio should be > 0.8 for sliced array"
+ );
+ assert!(
+ should_gc,
+ "GC should trigger for sliced array with high waste"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reproduce_issue_19414_string_view_spill_without_gc() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ let num_rows = 5000;
+ let mut strings = Vec::with_capacity(num_rows);
+
+ for i in 0..num_rows {
+ let url = match i % 5 {
+ 0 => format!(
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130/{i}"
+ ),
+ 1 => format!("http://komme%2F27.0.1453.116/very/long/path/{i}"),
+ 2 => format!("https://produkty%2Fproduct/category/item/{i}"),
+ 3 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668/{i}"
+ ),
+ 4 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapustic/product/{i}"
+ ),
+ _ => unreachable!(),
+ };
+ strings.push(url);
+ }
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "URL",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let original_batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let total_buffer_size: usize = string_array
+ .data_buffers()
+ .iter()
+ .map(|buffer| buffer.capacity())
+ .sum();
+
+ let mut sliced_batches = Vec::new();
+ let slice_size = 100;
+
+ for i in (0..num_rows).step_by(slice_size) {
+ let len = std::cmp::min(slice_size, num_rows - i);
+ let sliced = original_batch.slice(i, len);
+ sliced_batches.push(sliced);
+ }
+
+ 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")?;
+
+ for batch in &sliced_batches {
+ in_progress_file.append_batch(batch)?;
+ }
+
+ let spill_file = in_progress_file.finish()?.unwrap();
+ let file_size = fs::metadata(spill_file.path())?.len() as usize;
+
+ let theoretical_without_gc = total_buffer_size * sliced_batches.len();
+ let reduction_percent = ((theoretical_without_gc - file_size) as f64
+ / theoretical_without_gc as f64)
+ * 100.0;
+
+ assert!(
+ reduction_percent > 80.0,
+ "GC should reduce spill file size by >80%, got {reduction_percent:.1}%"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_exact_clickbench_issue_19414() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ // Test for clickbench issue: 820MB -> 33MB spill reduction
+ let unique_urls = vec![
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130",
+ "http://komme%2F27.0.1453.116",
+ "https://produkty%2Fproduct",
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668]=0&order_by=0",
+ "http://irr.ru/index.php?showalbum/login-kapustic/product_name",
+ "http://irr.ru/index.php",
+ "https://produkty%2F",
+ "http://irr.ru/index.php?showalbum/login",
+ "https://produkty/kurortmag",
+ "https://produkty%2Fpulove.ru/album/login",
+ ];
+
+ let mut urls = Vec::with_capacity(200_000);
Review Comment:
Done, reduced to 5,000 rows.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
Review Comment:
Fixed, using `batch.clone()` now which is just Arc bumps.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
Review Comment:
Added early return before processing.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
+ }
+}
+
+fn should_gc_view_array(len: usize, data_buffers: &[arrow::buffer::Buffer]) -> bool {
+ if len < 10 {
Review Comment:
Agreed, switched to 10KB buffer size threshold instead.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
+ }
+}
+
+fn should_gc_view_array(len: usize, data_buffers: &[arrow::buffer::Buffer]) -> bool {
+ if len < 10 {
+ return false;
+ }
+
+ let total_buffer_size: usize = data_buffers.iter().map(|b| b.capacity()).sum();
Review Comment:
We need buffer capacity of just the data buffers to detect waste, added a comment explaining why.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
Review Comment:
Done, importing from `arrow_data` now.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
Review Comment:
Fixed, always returning a new batch for consistency.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "bharath-techie (via GitHub)" <gi...@apache.org>.
bharath-techie commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2639817545
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -799,4 +890,426 @@ mod tests {
assert_eq!(alignment, 8);
Ok(())
}
+
+ #[test]
+ fn test_gc_string_view_before_spill() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ "short_string".to_string()
+ } else {
+ "this_is_a_much_longer_string_that_will_not_be_inlined".to_string()
+ }
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_binary_view_before_spill() -> Result<()> {
+ use arrow::array::BinaryViewArray;
+
+ let binaries: Vec<Vec<u8>> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ vec![1, 2, 3, 4]
+ } else {
+ vec![1; 50]
+ }
+ })
+ .collect();
+
+ let binary_array =
+ BinaryViewArray::from_iter(binaries.iter().map(|b| Some(b.as_slice())));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "binaries",
+ DataType::BinaryView,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(binary_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_skips_small_arrays() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..10).map(|i| format!("string_{i}")).collect();
+
+ let string_array = StringViewArray::from(strings);
+ let array_ref: ArrayRef = Arc::new(string_array);
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array_ref])?;
+
+ // GC should return the original batch for small arrays
+ let gc_batch = gc_view_arrays(&batch)?;
+
+ // The batch should be unchanged (cloned, not GC'd)
+ assert_eq!(gc_batch.num_rows(), batch.num_rows());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_with_mixed_columns() -> Result<()> {
+ use arrow::array::{Int32Array, StringViewArray};
+
+ let strings: Vec<String> = (0..200)
+ .map(|i| format!("long_string_for_gc_testing_{i}"))
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let int_array = Int32Array::from((0..200).collect::<Vec<i32>>());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("strings", DataType::Utf8View, false),
+ Field::new("ints", DataType::Int32, false),
+ ]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(string_array) as ArrayRef,
+ Arc::new(int_array) as ArrayRef,
+ ],
+ )?;
+
+ let sliced_batch = batch.slice(0, 50);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_columns(), 2);
+ assert_eq!(gc_batch.num_rows(), 50);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_verify_gc_triggers_for_sliced_arrays() -> Result<()> {
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ format!(
+ "http://example.com/very/long/path/that/exceeds/inline/threshold/{i}"
+ )
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "url",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let sliced = batch.slice(0, 100);
+
+ let sliced_array = sliced
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .unwrap();
+ let should_gc =
+ should_gc_view_array(sliced_array.len(), sliced_array.data_buffers());
+ let waste_ratio = calculate_string_view_waste_ratio(sliced_array);
+
+ assert!(
+ waste_ratio > 0.8,
+ "Waste ratio should be > 0.8 for sliced array"
+ );
+ assert!(
+ should_gc,
+ "GC should trigger for sliced array with high waste"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reproduce_issue_19414_string_view_spill_without_gc() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ let num_rows = 5000;
+ let mut strings = Vec::with_capacity(num_rows);
+
+ for i in 0..num_rows {
+ let url = match i % 5 {
+ 0 => format!(
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130/{i}"
+ ),
+ 1 => format!("http://komme%2F27.0.1453.116/very/long/path/{i}"),
+ 2 => format!("https://produkty%2Fproduct/category/item/{i}"),
+ 3 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668/{i}"
+ ),
+ 4 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapustic/product/{i}"
+ ),
+ _ => unreachable!(),
+ };
+ strings.push(url);
+ }
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "URL",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let original_batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let total_buffer_size: usize = string_array
+ .data_buffers()
+ .iter()
+ .map(|buffer| buffer.capacity())
+ .sum();
+
+ let mut sliced_batches = Vec::new();
+ let slice_size = 100;
+
+ for i in (0..num_rows).step_by(slice_size) {
+ let len = std::cmp::min(slice_size, num_rows - i);
+ let sliced = original_batch.slice(i, len);
+ sliced_batches.push(sliced);
+ }
+
+ 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")?;
+
+ for batch in &sliced_batches {
+ in_progress_file.append_batch(batch)?;
+ }
+
+ let spill_file = in_progress_file.finish()?.unwrap();
+ let file_size = fs::metadata(spill_file.path())?.len() as usize;
+
+ let theoretical_without_gc = total_buffer_size * sliced_batches.len();
+ let reduction_percent = ((theoretical_without_gc - file_size) as f64
+ / theoretical_without_gc as f64)
+ * 100.0;
+
+ assert!(
+ reduction_percent > 80.0,
+ "GC should reduce spill file size by >80%, got {reduction_percent:.1}%"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_exact_clickbench_issue_19414() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ // Test for clickbench issue: 820MB -> 33MB spill reduction
+ let unique_urls = vec![
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130",
+ "http://komme%2F27.0.1453.116",
+ "https://produkty%2Fproduct",
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668]=0&order_by=0",
+ "http://irr.ru/index.php?showalbum/login-kapustic/product_name",
+ "http://irr.ru/index.php",
+ "https://produkty%2F",
+ "http://irr.ru/index.php?showalbum/login",
+ "https://produkty/kurortmag",
+ "https://produkty%2Fpulove.ru/album/login",
+ ];
+
+ let mut urls = Vec::with_capacity(200_000);
Review Comment:
This might be quite heavy - maybe we can just keep the minimal reproducible version to verify that the changes are working as expected [ like the test above this ]
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2641790336
##########
datafusion/physical-plan/src/spill/in_progress_spill_file.rs:
##########
@@ -50,6 +50,7 @@ impl InProgressSpillFile {
}
/// Appends a `RecordBatch` to the spill file, initializing the writer if necessary.
+ /// Performs garbage collection on StringView/BinaryView arrays to reduce spill file size.
Review Comment:
I recommend to add more comments to explain the rationale for views gc, perhaps just copy and paste from https://github.com/apache/datafusion/blob/bb9a4a7ea19896fab7da6554cf5dd103e50d8e74/datafusion/physical-plan/src/sorts/sort.rs#L483
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -799,4 +890,426 @@ mod tests {
assert_eq!(alignment, 8);
Ok(())
}
+
+ #[test]
+ fn test_gc_string_view_before_spill() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ "short_string".to_string()
+ } else {
+ "this_is_a_much_longer_string_that_will_not_be_inlined".to_string()
+ }
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_binary_view_before_spill() -> Result<()> {
+ use arrow::array::BinaryViewArray;
+
+ let binaries: Vec<Vec<u8>> = (0..1000)
+ .map(|i| {
+ if i % 2 == 0 {
+ vec![1, 2, 3, 4]
+ } else {
+ vec![1; 50]
+ }
+ })
+ .collect();
+
+ let binary_array =
+ BinaryViewArray::from_iter(binaries.iter().map(|b| Some(b.as_slice())));
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "binaries",
+ DataType::BinaryView,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(binary_array) as ArrayRef],
+ )?;
+ let sliced_batch = batch.slice(0, 100);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_rows(), sliced_batch.num_rows());
+ assert_eq!(gc_batch.num_columns(), sliced_batch.num_columns());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_skips_small_arrays() -> Result<()> {
+ use arrow::array::StringViewArray;
+
+ let strings: Vec<String> = (0..10).map(|i| format!("string_{i}")).collect();
+
+ let string_array = StringViewArray::from(strings);
+ let array_ref: ArrayRef = Arc::new(string_array);
+
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "strings",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(Arc::clone(&schema), vec![array_ref])?;
+
+ // GC should return the original batch for small arrays
+ let gc_batch = gc_view_arrays(&batch)?;
+
+ // The batch should be unchanged (cloned, not GC'd)
+ assert_eq!(gc_batch.num_rows(), batch.num_rows());
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_gc_with_mixed_columns() -> Result<()> {
+ use arrow::array::{Int32Array, StringViewArray};
+
+ let strings: Vec<String> = (0..200)
+ .map(|i| format!("long_string_for_gc_testing_{i}"))
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let int_array = Int32Array::from((0..200).collect::<Vec<i32>>());
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("strings", DataType::Utf8View, false),
+ Field::new("ints", DataType::Int32, false),
+ ]));
+
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(string_array) as ArrayRef,
+ Arc::new(int_array) as ArrayRef,
+ ],
+ )?;
+
+ let sliced_batch = batch.slice(0, 50);
+ let gc_batch = gc_view_arrays(&sliced_batch)?;
+
+ assert_eq!(gc_batch.num_columns(), 2);
+ assert_eq!(gc_batch.num_rows(), 50);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_verify_gc_triggers_for_sliced_arrays() -> Result<()> {
+ let strings: Vec<String> = (0..1000)
+ .map(|i| {
+ format!(
+ "http://example.com/very/long/path/that/exceeds/inline/threshold/{i}"
+ )
+ })
+ .collect();
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "url",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let sliced = batch.slice(0, 100);
+
+ let sliced_array = sliced
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .unwrap();
+ let should_gc =
+ should_gc_view_array(sliced_array.len(), sliced_array.data_buffers());
+ let waste_ratio = calculate_string_view_waste_ratio(sliced_array);
+
+ assert!(
+ waste_ratio > 0.8,
+ "Waste ratio should be > 0.8 for sliced array"
+ );
+ assert!(
+ should_gc,
+ "GC should trigger for sliced array with high waste"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_reproduce_issue_19414_string_view_spill_without_gc() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ let num_rows = 5000;
+ let mut strings = Vec::with_capacity(num_rows);
+
+ for i in 0..num_rows {
+ let url = match i % 5 {
+ 0 => format!(
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130/{i}"
+ ),
+ 1 => format!("http://komme%2F27.0.1453.116/very/long/path/{i}"),
+ 2 => format!("https://produkty%2Fproduct/category/item/{i}"),
+ 3 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668/{i}"
+ ),
+ 4 => format!(
+ "http://irr.ru/index.php?showalbum/login-kapustic/product/{i}"
+ ),
+ _ => unreachable!(),
+ };
+ strings.push(url);
+ }
+
+ let string_array = StringViewArray::from(strings);
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "URL",
+ DataType::Utf8View,
+ false,
+ )]));
+
+ let original_batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(string_array.clone()) as ArrayRef],
+ )?;
+
+ let total_buffer_size: usize = string_array
+ .data_buffers()
+ .iter()
+ .map(|buffer| buffer.capacity())
+ .sum();
+
+ let mut sliced_batches = Vec::new();
+ let slice_size = 100;
+
+ for i in (0..num_rows).step_by(slice_size) {
+ let len = std::cmp::min(slice_size, num_rows - i);
+ let sliced = original_batch.slice(i, len);
+ sliced_batches.push(sliced);
+ }
+
+ 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")?;
+
+ for batch in &sliced_batches {
+ in_progress_file.append_batch(batch)?;
+ }
+
+ let spill_file = in_progress_file.finish()?.unwrap();
+ let file_size = fs::metadata(spill_file.path())?.len() as usize;
+
+ let theoretical_without_gc = total_buffer_size * sliced_batches.len();
+ let reduction_percent = ((theoretical_without_gc - file_size) as f64
+ / theoretical_without_gc as f64)
+ * 100.0;
+
+ assert!(
+ reduction_percent > 80.0,
+ "GC should reduce spill file size by >80%, got {reduction_percent:.1}%"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_exact_clickbench_issue_19414() -> Result<()> {
+ use arrow::array::StringViewArray;
+ use std::fs;
+
+ // Test for clickbench issue: 820MB -> 33MB spill reduction
+ let unique_urls = vec![
+ "http://irr.ru/index.php?showalbum/login-leniya7777294,938303130",
+ "http://komme%2F27.0.1453.116",
+ "https://produkty%2Fproduct",
+ "http://irr.ru/index.php?showalbum/login-kapusta-advert2668]=0&order_by=0",
+ "http://irr.ru/index.php?showalbum/login-kapustic/product_name",
+ "http://irr.ru/index.php",
+ "https://produkty%2F",
+ "http://irr.ru/index.php?showalbum/login",
+ "https://produkty/kurortmag",
+ "https://produkty%2Fpulove.ru/album/login",
+ ];
+
+ let mut urls = Vec::with_capacity(200_000);
Review Comment:
Yes, it would be great to make tests faster and use less memory.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
Review Comment:
nit: I think we can get rid of the `any_gc_performed` condition, and always go this branch to make it a little bit simpler
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "ebembi-crdb (via GitHub)" <gi...@apache.org>.
ebembi-crdb commented on code in PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#discussion_r2758320060
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
+ }
+}
+
+fn should_gc_view_array(len: usize, data_buffers: &[arrow::buffer::Buffer]) -> bool {
+ if len < 10 {
Review Comment:
Agreed, switched to 10KB buffer size threshold instead.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
+
+/// Performs garbage collection on view arrays before spilling.
+pub(crate) fn gc_view_arrays(batch: &RecordBatch) -> Result<RecordBatch> {
+ let mut new_columns: Vec<Arc<dyn Array>> = Vec::with_capacity(batch.num_columns());
+ let mut any_gc_performed = false;
+
+ for array in batch.columns() {
+ let gc_array = match array.data_type() {
+ arrow::datatypes::DataType::Utf8View => {
+ let string_view = array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .expect("Utf8View array should downcast to StringViewArray");
+ if should_gc_view_array(string_view.len(), string_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(string_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ arrow::datatypes::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.len(), binary_view.data_buffers()) {
+ any_gc_performed = true;
+ Arc::new(binary_view.gc()) as Arc<dyn Array>
+ } else {
+ Arc::clone(array)
+ }
+ }
+ _ => Arc::clone(array),
+ };
+ new_columns.push(gc_array);
+ }
+
+ if any_gc_performed {
+ Ok(RecordBatch::try_new(batch.schema(), new_columns)?)
+ } else {
+ Ok(batch.clone())
+ }
+}
+
+fn should_gc_view_array(len: usize, data_buffers: &[arrow::buffer::Buffer]) -> bool {
+ if len < 10 {
+ return false;
+ }
+
+ let total_buffer_size: usize = data_buffers.iter().map(|b| b.capacity()).sum();
Review Comment:
We need buffer capacity of just the data buffers to detect waste, added a comment explaining why.
##########
datafusion/physical-plan/src/spill/mod.rs:
##########
@@ -339,6 +339,97 @@ fn get_max_alignment_for_schema(schema: &Schema) -> usize {
max_alignment
}
+#[cfg(test)]
+const VIEW_SIZE_BYTES: usize = 16;
+#[cfg(test)]
+const INLINE_THRESHOLD: usize = 12;
Review Comment:
Done, importing from `arrow_data` now.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Fix massive spill files for StringView/BinaryView columns [datafusion]
Posted by "bharath-techie (via GitHub)" <gi...@apache.org>.
bharath-techie commented on PR #19444:
URL: https://github.com/apache/datafusion/pull/19444#issuecomment-3737256901
Hi @EeshanBembi ,
I see you've addressed bunch of comments , can you kindly address remaining comments and take this PR forward ? Let us know if any help is required.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org