[VL] Push Velox dynamic filters to shuffle reader

[zhztheplayer]

As Velox's dynamic filter API is extensible, it's worth investigating whether we can push the filters down to Gluten's Velox shuffle reader to reduce the amount of data materialized from shuffle.

[acvictor]

I think it is feasible by making Gluten's ValueStream operator in RowVectorStream.h override canAddDynamicFilter() to return true, addDynamicFilterLocked() to receive filters, then apply them in getOutput() using Filter::testX() + Velox zero-copy wrap(). I think the changes can be contained to RowVectorStream.h. I can try out a PoC!

[zhztheplayer]

@acvictor thanks a lot for helping out here.

I think it is feasible by making Gluten's ValueStream operator in RowVectorStream.h

I think we don't have to expect much performance gain by filtering from ValueStream solely, since the filter operation is not guaranteed to be cheaper than hash probing. We might benefit from better cache locality as filters can be smaller than hash tables, but I am not sure how much. Worth a try anyway because it's a prerequisite for deeper pruning, e.g., from shuffle reader, since we need to obtain the pushed filters from Gluten side.

[zhztheplayer]

#11657

[acvictor]

@acvictor thanks a lot for helping out here.

I think it is feasible by making Gluten's ValueStream operator in RowVectorStream.h

I think we don't have to expect much performance gain by filtering from ValueStream solely, since the filter operation is not guaranteed to be cheaper than hash probing. We might benefit from better cache locality as filters can be smaller than hash tables, but I am not sure how much. Worth a try anyway because it's a prerequisite for deeper pruning, e.g., from shuffle reader, since we need to obtain the pushed filters from Gluten side.

I see what you mean. Should this filter push down be under a feature flag?

[zhztheplayer]

@acvictor yes. If the benchmark doesn't show much benefit, we might consider using a flag to disable it by default.

[zhztheplayer]

For the shuffle-reader pruning, I think we haven't had page-level metadata in shuffle blocks to support that yet. So maybe we need to start from writing metadata from the shuffle writer side - perhaps @marin-ma can help validate the idea later.

[acvictor]

For the shuffle-reader pruning, I think we haven't had page-level metadata in shuffle blocks to support that yet. So maybe we need to start from writing metadata from the shuffle writer side - perhaps @marin-ma can help validate the idea later.

@zhztheplayer @marin-ma we can now push dynamic filters down to ValueStreamDataSource with #11657, but filtering still happens after full deserialization of every shuffle block. I think the next step is block-level pruning so we can skip shuffle blocks without deserializing them just like Parquet row group pruning with statistics. I think we will need the following changes

Writer side:

• Compute per-column min/max/hasNull stats in VeloxHashShuffleWriter::evictBuffers() during buffer assembly (data is already in cache at that point)
• Serialize a compact stats header before each BlockPayload, e.g. via a new BlockType value

Reader side:

• Read the stats header in VeloxHashShuffleReaderDeserializer::resolveNextBlockType()
• Test pushed filters against block stats using Velox's existing Filter::testInt64Range() / testBytesRange() / testDoubleRange() etc. (already used by Parquet/ORC readers in velox/dwio/common/ScanSpec.cpp)
• If no rows can match → skip the block entirely (seek past it without decompressing)
• Propagate dynamic filters from ValueStreamDataSource down through RowVectorStream to the reader/deserializer

What do you think?