Potential performance regression for TPCH q18

[alamb]

Describe the bug

While enabling StringView reading from Parquet in #13101 @Dandandan noticed a slight regression for TPCH 18 #13101 (comment)

here is the query

select
    c_name,
    c_custkey,
    o_orderkey,
    o_orderdate,
    o_totalprice,
    sum(l_quantity)
from
    customer,
    orders,
    lineitem
where
        o_orderkey in (
        select
            l_orderkey
        from
            lineitem
        group by
            l_orderkey having
                sum(l_quantity) > 300
    )
  and c_custkey = o_custkey
  and o_orderkey = l_orderkey
group by
    c_name,
    c_custkey,
    o_orderkey,
    o_orderdate,
    o_totalprice
order by
    o_totalprice desc,
    o_orderdate;

To Reproduce

To reproduce

Make data

# make the data and get to the correct location
cd datafusion/benchmarks
./bench.sh data tpch
cd data/tpch_sf1

Run query:

datafusion-cli -f ../../queries/q18.sql  | grep Elapsed
Elapsed 0.088 seconds.

When StringView is enabled it seems like it is slightly slower

Expected behavior

StringView should always be faster

Additional context

I took a brief look at the flamegraphs -- it seems like one difference could be BatchCoalescer::push_batch

There is a special case for StringView here:
https://github.com/apache/datafusion/blob/6034be42808b43e3f48f6e58ec38cc35fa253abb/datafusion/physical-plan/src/coalesce/mod.rs#L117-L116

Here are the explain plans for the query before and after the change

• q18-before.txt
• q18-after.txt

Here are the flamegraphs for the query before/after the change

• 
• 

[alamb]

This may be fixed by fixing #11628

[devanbenz]

take

[jayzhan211]

I will take a look on this issue too, how is your progress on this @devanbenz ?

[devanbenz]

I will take a look on this issue too, how is your progress on this @devanbenz ?

Looking at the following code path:

datafusion/datafusion/physical-plan/src/coalesce/mod.rs

Lines 246 to 263 in 9005585

	if actual_buffer_size > (ideal_buffer_size * 2) {
	// We set the block size to `ideal_buffer_size` so that the new StringViewArray only has one buffer, which accelerate later concat_batches.
	// See https://github.com/apache/arrow-rs/issues/6094 for more details.
	let mut builder = StringViewBuilder::with_capacity(s.len());
	if ideal_buffer_size > 0 {
	builder = builder.with_fixed_block_size(ideal_buffer_size as u32);
	}
	for v in s.iter() {
	builder.append_option(v);
	}
	let gc_string = builder.finish();
	debug_assert!(gc_string.data_buffers().len() <= 1); // buffer count can be 0 if the `ideal_buffer_size` is 0
	Arc::new(gc_string)
	} else {

I noticed when writing up a small example using this code path locally.

When taking a look at the append_option source code it implements a copy.

I've identified where the performance fault is--but I'm unsure what next steps to take to try and alleviate it. I was going to maybe modify the coalescer to try and "change the RecordBatch to stringview if the datatype is Utf8View" as the input occurs around here (if that makes sense):

datafusion/datafusion/physical-plan/src/coalesce_batches.rs

Lines 297 to 307 in 9005585

	match input_batch {
	Some(Ok(batch)) => match self.coalescer.push_batch(batch) {
	CoalescerState::Continue => {}
	CoalescerState::LimitReached => {
	self.inner_state = CoalesceBatchesStreamState::Exhausted;
	}
	CoalescerState::TargetReached => {
	self.inner_state =
	CoalesceBatchesStreamState::ReturnBuffer;
	}
	},

I don't have a meaningful plan yet just have been doing some exploratory work as of right now while benchmarking locally.

[alamb]

#11628 might have some ideas. I think @XiangpengHao has also been thinking of some way to do this too

Basically one thing you might be able to try is to switch from using take to directly building the output (with a StringViewBuilder or something equvialent)

But as you are hinting at this is a non trivial thing to do

[alamb]

I finally filed a ticket upstream in arrow trying to explain what I have been thinking about:

• Optimize take/filter/concat from multiple input arrays to a single large output array arrow-rs#6692

[jayzhan211]

Other than apache/arrow-rs#6692.
If we create filter version of GroupColumn that accumulate the filtered array into array builder for each column, output if the batch size reach target (i.e. 8192). We can then aggregate those small batches in filter exec and produce a single large output for next step. Does this sound a possible improvement?

I think the downside is again we need to implement builder per type like GroupColumn and the improvement is unclear

[alamb]

If we create filter version of GroupColumn that accumulate the filtered array into array builder for each column, output if the batch size reach target (i.e. 8192)

Would this be a specialization of a filter of the aggregates like for SELECT SUM(x) FROM ... HAVING SUM(x) > 5 ?

[jayzhan211]

If we create filter version of GroupColumn that accumulate the filtered array into array builder for each column, output if the batch size reach target (i.e. 8192)

Would this be a specialization of a filter of the aggregates like for SELECT SUM(x) FROM ... HAVING SUM(x) > 5 ?

Probably, but I think it has potential to reduce copied too #11628. It aggregates at the first place similar to coalescer's role and no gc required since we aggregate the value without necessary buffer copied.