(datafusion) 01/01: fix: compact sparse string views in FilterExec coalescing (#26436)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26436-4e84f71c49eb141e8353cc04f1d3ae390fe38543
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit d6fee726feadee177100ecc0e1824bb4e4ccc0de
Author: Jay Zhan <ja...@gmail.com>
AuthorDate: Tue Oct 20 11:57:10 2026 +0000

    fix: compact sparse string views in FilterExec coalescing (#26436)
    
    ## Which issue does this PR close?
    
    - Fixes a regression from #25503 (not released yet, so it would ship in
    56.0.0).
    
    ## Rationale for this change
    
    Since #25503, a selective filter over `Utf8View`/`BinaryView` columns
    keeps the whole input string buffers alive. ClickBench Q21-Q23 use about
    2-3x more memory and Q22/Q23 run 1.6-1.7x slower with the default
    `schema_force_view_types = true`.
    
    #25503 made `FilterExec` call `BatchCoalescer::push_batch_with_filter`
    instead of `filter_record_batch` + `push_batch`. When at most 1/16 of
    the rows are selected, arrow-select 60 takes a sparse path that filters
    the views but reuses the source data buffers as they are
    (`InProgressByteViewArray::copy_rows_by_filter_from` calls
    `append_views_and_update_buffer_index(..., false)`). The old path went
    through `push_batch`, which sees that the filtered array uses less than
    half of its buffers (`need_gc`) and copies the selected strings into
    compact buffers. So each output batch now pins every data page it was
    filtered from, and operators that hold batches (aggregation, sort, TopK)
    hold all of them. arrow-rs main has the same code.
    
    Repro (ClickBench `hits_partitioned`, default settings):
    
    ```sql
    SET datafusion.execution.parquet.binary_as_string = true;
    CREATE EXTERNAL TABLE hits STORED AS PARQUET LOCATION 'benchmarks/data/hits_partitioned';
    SELECT * FROM hits WHERE "URL" LIKE '%google%' ORDER BY "EventTime" LIMIT 10; -- Q23
    ```
    
    ## What changes are included in this PR?
    
    `LimitedBatchCoalescer::push_batch_with_filter` sends batches that have
    a `Utf8View` or `BinaryView` column with data buffers through
    `filter_record_batch` + `push_batch`, where arrow compacts sparse string
    views. All other batches keep the fused path from #25503.
    
    The root cause is tracked upstream in apache/arrow-rs#11635; once
    arrow-select compacts views on its sparse path, this check can be
    removed.
    
    This changes nothing for view batches when the filter is not sparse:
    arrow already materializes the filter and calls `push_batch` for them.
    So the only path that changes is the sparse one for view columns, which
    goes back to its behavior before #25503.
    
    ## What is the testing strategy for this PR?
    
    - New unit test `test_push_batch_with_filter_compacts_string_views`: a
    1/100 filter over 1600 long strings. Without the change the output keeps
    all 114,688 input string bytes; with it, less than a quarter.
    - Benchmarks, release-nonlto, same commit (5f2158269e) with and without
    this PR, plus #25503 reverted for reference; median of 3 interleaved
    runs, peak RSS:
    
    | Query | main | this PR | #25503 reverted |
    |---|---|---|---|
    | ClickBench Q21 | 0.60 s, 2.1 GB | 0.59 s, 1.1 GB | 0.62 s, 1.2 GB |
    | ClickBench Q22 | 2.14 s, 4.1 GB | 1.40 s, 1.5 GB | 1.26 s, 1.4 GB |
    | ClickBench Q23 | 7.10 s, 6.9 GB | 4.35 s, 3.4 GB | 4.33 s, 3.4 GB |
    | ClickBench total (43 queries) | 29.8 s | 25.5 s | 25.9 s |
    | TPC-H SF10 total (22 queries) | 6.89 s | 6.88 s | 6.93 s |
    
    Other ClickBench queries differ only within run-to-run noise. TPC-H
    keeps the #25503 result.
    
    ## Are there any user-facing changes?
    
    No.
---
 datafusion/physical-plan/src/coalesce/mod.rs | 63 +++++++++++++++++++++++++---
 1 file changed, 58 insertions(+), 5 deletions(-)

diff --git a/datafusion/physical-plan/src/coalesce/mod.rs b/datafusion/physical-plan/src/coalesce/mod.rs
index 70c86368c1..107696cd58 100644
--- a/datafusion/physical-plan/src/coalesce/mod.rs
+++ b/datafusion/physical-plan/src/coalesce/mod.rs
@@ -15,9 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use arrow::array::{Array, BooleanArray, RecordBatch};
-use arrow::compute::{BatchCoalescer, prep_null_mask_filter};
-use arrow::datatypes::SchemaRef;
+use arrow::array::{Array, AsArray, BooleanArray, RecordBatch};
+use arrow::compute::{BatchCoalescer, filter_record_batch, prep_null_mask_filter};
+use arrow::datatypes::{DataType, SchemaRef};
 use datafusion_common::{Result, assert_or_internal_err};
 
 /// Concatenate multiple [`RecordBatch`]es and apply a limit
@@ -124,7 +124,11 @@ impl LimitedBatchCoalescer {
     /// avoiding a separate materialization pass compared to calling
     /// [`filter_record_batch`] followed by [`Self::push_batch`].
     ///
-    /// [`filter_record_batch`]: arrow::compute::filter_record_batch
+    /// Batches with `Utf8View` or `BinaryView` columns that reference data
+    /// buffers are filtered first and then pushed with [`Self::push_batch`]:
+    /// arrow's fused path keeps the whole input data buffers alive for the few
+    /// selected strings, while [`Self::push_batch`] copies them into compact
+    /// buffers.
     pub fn push_batch_with_filter(
         &mut self,
         batch: RecordBatch,
@@ -135,6 +139,11 @@ impl LimitedBatchCoalescer {
             "LimitedBatchCoalescer: cannot push batch after finish"
         );
 
+        if has_view_data_buffers(&batch) {
+            let batch = filter_record_batch(&batch, filter)?;
+            return self.push_batch(batch);
+        }
+
         let Some(fetch) = self.fetch else {
             self.inner.push_batch_with_filter(batch, filter)?;
             return Ok(PushBatchStatus::Continue);
@@ -191,13 +200,23 @@ impl LimitedBatchCoalescer {
     }
 }
 
+/// Returns true if `batch` has a `Utf8View` or `BinaryView` column whose views
+/// reference data buffers (strings longer than 12 bytes).
+fn has_view_data_buffers(batch: &RecordBatch) -> bool {
+    batch.columns().iter().any(|array| match array.data_type() {
+        DataType::Utf8View => !array.as_string_view().data_buffers().is_empty(),
+        DataType::BinaryView => !array.as_binary_view().data_buffers().is_empty(),
+        _ => false,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use std::ops::Range;
     use std::sync::Arc;
 
-    use arrow::array::UInt32Array;
+    use arrow::array::{StringViewArray, UInt32Array};
     use arrow::compute::concat_batches;
     use arrow::datatypes::{DataType, Field, Schema};
 
@@ -360,6 +379,40 @@ mod tests {
         assert!(coalescer.next_completed_batch().is_none());
     }
 
+    /// A selective filter over long strings must not keep the whole input
+    /// data buffer alive for the few selected rows.
+    #[test]
+    fn test_push_batch_with_filter_compacts_string_views() {
+        let strings =
+            (0..1600).map(|i| format!("a string longer than twelve bytes {i:05}"));
+        let array = StringViewArray::from_iter_values(strings);
+        let input_bytes: usize = array.data_buffers().iter().map(|b| b.capacity()).sum();
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            "s",
+            DataType::Utf8View,
+            false,
+        )]));
+        let batch =
+            RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(array)]).unwrap();
+        // Select every 100th row: 16 of 1600 rows
+        let filter = BooleanArray::from_iter((0..1600).map(|i| Some(i % 100 == 0)));
+
+        let mut coalescer = LimitedBatchCoalescer::new(schema, 100, None);
+        coalescer.push_batch_with_filter(batch, &filter).unwrap();
+        coalescer.finish().unwrap();
+        let output = coalescer.next_completed_batch().unwrap();
+
+        let output = output.column(0).as_string_view();
+        assert_eq!(output.len(), 16);
+        assert_eq!(output.value(1), "a string longer than twelve bytes 00100");
+        let output_bytes: usize =
+            output.data_buffers().iter().map(|b| b.capacity()).sum();
+        assert!(
+            output_bytes * 4 < input_bytes,
+            "output keeps {output_bytes} of {input_bytes} input string bytes"
+        );
+    }
+
     /// Test for [`LimitedBatchCoalescer`]
     ///
     /// Pushes the input batches to the coalescer and verifies that the resulting


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org