Skip to content

[GLUTEN-12613][VL] Align VeloxBloomFilterAggregate buffer capacity with native bloom_filter_agg#12614

Open
brijrajk wants to merge 1 commit into
apache:mainfrom
brijrajk:fix/12613-bloom-filter-capacity-mismatch
Open

[GLUTEN-12613][VL] Align VeloxBloomFilterAggregate buffer capacity with native bloom_filter_agg#12614
brijrajk wants to merge 1 commit into
apache:mainfrom
brijrajk:fix/12613-bloom-filter-capacity-mismatch

Conversation

@brijrajk

Copy link
Copy Markdown
Contributor

What changes are proposed in this pull request?

Fixes #12613.

VeloxBloomFilterAggregate (the JVM-side expression used for Velox bloom filters) and the
native bloom_filter_agg Velox function size their internal bit array differently, even when
given the identical estimatedNumItems/numBits arguments:

  • JVM VeloxBloomFilterAggregate.createAggregationBuffer() called
    VeloxBloomFilter.empty(estimatedNumItems), sizing the buffer purely from the raw item
    count and ignoring numBits entirely.
  • Native BloomFilterAggAggregate.cpp computes capacity_ = min(numBits, maxNumBits) / 16
    and ignores the raw item count once numBits is known.

For example, given estimatedNumItems=1000000, numBits=8388608 (Spark's plain defaults),
the JVM side allocated a 16,777,216-bit buffer while the native side allocated an
8,388,608-bit buffer -- exactly 2x different, deterministically, for any query using this
code path.

Why this matters

Two-phase aggregation runs the partial and final stages as separate physical operators,
which can independently land on the JVM or on native Velox (e.g. via Gluten's whole-stage
fallback policy, or via spark.gluten.sql.columnar.hashagg.enabled=false). When the partial
aggregate runs on one engine and the final aggregate (which merges partial buffers) runs on
the other, the merge combines two differently-sized bit arrays.

Velox's BloomFilter::merge (velox/common/base/BloomFilter.h) guards the size match with
VELOX_DCHECK_EQ, which is compiled out in release builds, so the mismatched merge proceeds
silently instead of throwing. The result is silent data corruption: bits inserted relative
to one array size are later queried relative to a different array size, causing bloom
filter false negatives (values that were definitely inserted are reported as absent).

The fix

Make VeloxBloomFilterAggregate's JVM-side buffer sizing use the same formula as the native
side (derive capacity from numBits, not from raw estimatedNumItems), so both engines
agree on capacity for the same input arguments regardless of which engine executes which
stage.

How was this patch tested?

Added a regression test to GlutenBloomFilterAggregateQuerySuite that runs the same
bloom_filter_agg query fully natively and fully on the JVM
(spark.gluten.sql.columnar.hashagg.enabled=false), and asserts the two produce
identically-sized serialized bytes.

Verified the test actually catches the bug: reverting only the fix (keeping the new test)
causes exactly this one test to fail, with 14 other tests in the same suite unaffected.
With the fix, all 15 tests in GlutenBloomFilterAggregateQuerySuite pass.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Sonnet 5 (Anthropic)

@brijrajk

Copy link
Copy Markdown
Contributor Author

@zhztheplayer @philo-he this is the capacity-alignment fix split out per the sequencing discussed on #12151. Once this merges I'll update #12151 to add the injectFinal registration on top.

Comment on lines +62 to +68
private lazy val capacityFromNumBits: Int = {
val numBits = numBitsExpression.eval().asInstanceOf[Number].longValue
val maxNumBits = SQLConf.get
.getConfString("spark.sql.optimizer.runtime.bloomFilter.maxNumBits", "67108864")
.toLong
Math.max(1, Math.toIntExact(Math.min(numBits, maxNumBits) / 16))
}

@zhztheplayer zhztheplayer Jul 24, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we add a comment explaining how this code is aligned with Velox?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done -- added a comment citing the native code (velox/functions/sparksql/aggregates/BloomFilterAggAggregate.cpp, computeCapacity()) this formula mirrors, and explaining why the two engines must agree here.

…th native bloom_filter_agg

VeloxBloomFilterAggregate's JVM-side buffer sizing (createAggregationBuffer)
derived capacity purely from estimatedNumItems, ignoring numBits entirely.
The native bloom_filter_agg Velox function derives capacity purely from
numBits (min(numBits, maxNumBits) / 16), ignoring the raw item count once
numBits is known. For the same input arguments, this produced deterministically
different bit-array sizes (e.g. 16,777,216 bits on JVM vs 8,388,608 bits
natively for Spark's plain defaults).

Two-phase aggregation runs partial and final stages as separate physical
operators that can independently land on the JVM or on native Velox. When
they disagree on capacity, Velox's BloomFilter::merge silently corrupts the
result instead of throwing, since its size-match guard is a VELOX_DCHECK
compiled out in release builds -- values inserted relative to one array size
are later queried relative to a different array size, producing bloom filter
false negatives.

Fix: size the JVM buffer from numBits using the same formula as the native
aggregate (velox/functions/sparksql/aggregates/BloomFilterAggAggregate.cpp,
computeCapacity()), so both engines agree on capacity for identical
arguments regardless of which engine executes which stage.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
@brijrajk
brijrajk force-pushed the fix/12613-bloom-filter-capacity-mismatch branch from fbcdd2a to 33660e1 Compare July 24, 2026 17:46
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

CORE works for Gluten Core VELOX

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[VL] VeloxBloomFilterAggregate sizes its buffer differently on JVM vs native, causing silent corruption on partial reversion

2 participants