Memory leak reported by Java Arrow on q12 of CometTPCHQuerySuite

[viirya]

Describe the bug

#324 fixed a bug of CometShuffleExchangeExec's logical link, it changes query plans.

Due to the change, CometTPCHQuerySuite's q12 has test failure that is memory leak reported by Java Arrow:

- q12 *** FAILED *** (2 seconds, 244 milliseconds)
  java.lang.Exception: Expected "struct<[l_shipmode:string,high_line_count:bigint,low_line_count:bigint]>", but got "struct<[]>" Schema did not match
-- using default substitutions

select
	l_shipmode,
	sum(case
		when o_orderpriority = '1-URGENT'
			or o_orderpriority = '2-HIGH'
			then 1
		else 0
	end) as high_line_count,
	sum(case
		when o_orderpriority <> '1-URGENT'
			and o_orderpriority <> '2-HIGH'
			then 1
		else 0
	end) as low_line_count
from
	orders,
	lineitem
where
	o_orderkey = l_orderkey
	and l_shipmode in ('MAIL', 'SHIP')
	and l_commitdate < l_receiptdate
	and l_shipdate < l_commitdate
	and l_receiptdate >= date '1994-01-01'
	and l_receiptdate < date '1994-01-01' + interval '1' year
group by
	l_shipmode
order by
	l_shipmode
Output/Exception: java.lang.IllegalStateException
Memory was leaked by query. Memory leaked: (49152)
Allocator(ROOT) 0/49152/180352/9223372036854775807 (res/actual/peak/limit)
Error using configs:
spark.sql.autoBroadcastJoinThreshold=10485760

I spent some time on debugging it, and found it seems caused by native shuffle (CometTPCHQuerySuite uses native shuffle for now). During debugging, I found that the leak is occurred on the allocation in StreamReader. The read batch is correctly closed after being used. But there is still 49152 bytes cannot be released on the allocator.

I'm not sure if it is a bug of Java Arrow.

Steps to reproduce

No response

Expected behavior

No response

Additional context

No response

[viirya]

Duplicated to #381.