[GLUTEN-2051] Make ColumnarBatchSerializer supports relocation so that continuous shuffle block fetching can be enabled

[WangGuangxin]

What changes were proposed in this pull request?

Spark supports fetch the contiguous shuffle blocks in batch, which is enabled by default (by conf spark.sql.adaptive.fetchShuffleBlocksInBatch). This feature has a big performance improvement in our production.

However, currently, since ColumnarBatchSerializer's supportsRelocationOfSerializedObjects return false, so that this feature cann't take effect.
In fact, the arrow serialization does support reloation if we don't write schema (which is default to true) and don't write EOS (which is an optional in arrow rpc serialization format)

https://wesm.github.io/arrow-site-test/format/IPC.html#streaming-format

(Fixes: #2051)

How was this patch tested?

Manually test using our internal query

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

#2051

[github-actions]

Run Gluten Clickhouse CI

[WangGuangxin]

@zhztheplayer @zhouyuan Can you please help review this?

[zhouyuan]

@WangGuangxin good finding! Curious how the performance gain look like? The marker is 4 bytes only if I i understand this correctly

-yuan

[WangGuangxin]

spark.sql.adaptive.fetchShuffleBlocksInBatch

@zhouyuan Hi, the benefits are not comes from the 4bytes savings, but make the Spark AQE feature "batch fetch shuffle blocks" take effect.
You can refer the feature by the initial commit apache/spark#26040

In short, when Spark AQE do coalesce partitions, it can fetch continuous blocks in batch, instead of fetch blocks one by one. But it has some preconditions, one is the serialization of shuffle must support the so called relocation https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala#L48C48-L48C85.

Currently, the ColumnarBatchSerializer doesn't support relocation. In fact, We can make some simple modifications to make it support relocation by not writing EOS after each shuffle blocks.

[zhouyuan]

spark.sql.adaptive.fetchShuffleBlocksInBatch

@zhouyuan Hi, the benefits are not comes from the 4bytes savings, but make the Spark AQE feature "batch fetch shuffle blocks" take effect. You can refer the feature by the initial commit apache/spark#26040

In short, when Spark AQE do coalesce partitions, it can fetch continuous blocks in batch, instead of fetch blocks one by one. But it has some preconditions, one is the serialization of shuffle must support the so called relocation https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala#L48C48-L48C85.

Currently, the ColumnarBatchSerializer doesn't support relocation. In fact, We can make some simple modifications to make it support relocation by not writing EOS after each shuffle blocks.

@WangGuangxin thanks for detailed explanation, got the idea now

[zhouyuan]

👍

[CLTFOREVER]

@WangGuangxin @zhouyuan we have meet the same problem，if the pr is ready，can we merge it？

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

Run Gluten Clickhouse CI

[github-actions]

Run Gluten Clickhouse CI

[zhouyuan]

👍 works for me