[SPARK-59056][SQL] Pre-size HashSets in ParquetFilters IN/InSet pushdown - #58349
[SPARK-59056][SQL] Pre-size HashSets in ParquetFilters IN/InSet pushdown#58349david-mollitor-db wants to merge 1 commit into
Conversation
|
+1, LGTM |
|
@uros-b Is this good to merge? |
|
|
||
| import scala.jdk.CollectionConverters._ | ||
|
|
||
| import com.google.common.collect.Sets |
There was a problem hiding this comment.
Oh, I don't think we should introduce Guava Sets here... Please see that Spark already banned com.google.common.collect.Sets in Java checkstyle (SPARK-53231 / SPARK-53217).
There was a problem hiding this comment.
The cited precedent Maps.newHashMapWithExpectedSize is a different class and is not banned. Scala is not covered by that checkstyle rule, which is why CI still passed, but it is still the wrong direction.
|
Requesting changes in order to prevent accidental merge. |
| case ParquetByteType | ParquetShortType | ParquetIntegerType => | ||
| (n: Array[String], values: Array[Any]) => | ||
| val set = new HashSet[Integer]() | ||
| val set = Sets.newHashSetWithExpectedSize[Integer](values.length) |
There was a problem hiding this comment.
On another note, there is a completeness gap. TypedParquetFilterOps.makeIn is still unsized. Current master extracted Time / timestamp-nanos IN pushdown into TypedParquetFilterOps.makeIn:
override def makeIn(columnPath: Array[String], values: Array[Any]): FilterPredicate = {
val set = new HashSet[T]()
values.foreach(v => set.add(toPhysicalOrNull(v)))
FilterApi.in(column(columnPath), set)
}
uros-b
left a comment
uros-b
left a comment
There was a problem hiding this comment.
I think we should consider other approaches here.
Not sure what is the best path forward, but we can consider something like just keeping java.util.HashSet:
new HashSet(values.length)
which would already remove almost all resizes (Java 8+ HashMap rounds capacity to a power of two).
Alternatively, we can consider to size it with the load factor:
new HashSet[Integer]((values.length / 0.75f).toInt + 1)
which is the same formula Guava’s helper uses for expectedSize >= 3. The 0.75 formula only avoids at most one extra rehash.
|
In any case, impact is real but fairly small... |
Parquet IN/InSet pushdown builds a `java.util.HashSet` for each pushed-down IN predicate and then adds exactly `values.length` elements. The sets start at the default capacity (16) and rehash as they grow, reallocating the internal bucket array several times for large IN lists. Pre-size each set from the known element count so it holds all IN values without rehashing. A `java.util.HashSet` resizes once it is more than 75% full, and its int constructor treats the argument as bucket capacity rather than expected size, so `new HashSet(values.length)` would still rehash. Add a JDK-only `SparkCollectionUtils.newHashSetWithExpectedSize` helper that derives the initial capacity as `expectedSize / 0.75 + 1`, and use it at the 12 sites in `ParquetFilters.makeInPredicate` and in `TypedParquetFilterOps.makeIn` (the Time / timestamp-nanos IN pushdown), so every IN-pushdown set is sized up front. The change is capacity-only and preserves the pushdown semantics. Generated-by: Claude Opus 4.8
7e17974 to
2c8829e
Compare
2 participants
What changes were proposed in this pull request?
Parquet
IN/InSetpushdown builds ajava.util.HashSetfor each pushed-downINpredicate, then adds exactlyvalues.lengthelements to it. The sets arecreated without an initial capacity, so they start at the default capacity (16)
and rehash as they grow, reallocating the internal bucket array several times
for large
INlists.This PR pre-sizes those sets from the known element count. A
java.util.HashSetresizes once it is more than 75% full, and its
intconstructor treats theargument as bucket capacity rather than expected size, so
new HashSet(values.length)would still rehash. To size correctly using only the JDK, a small helper
SparkCollectionUtils.newHashSetWithExpectedSizeis added that derives theinitial capacity as
expectedSize / 0.75 + 1, guaranteeing the set holds allelements without a resize.
The helper is applied at:
ParquetFilters.makeInPredicate, andTypedParquetFilterOps.makeIn, theINpushdown for the Types-FrameworkTime / timestamp-nanos types.
Why are the changes needed?
makeInPredicateis only reached when theINlist size exceedsspark.sql.parquet.pushdown.inFilterThreshold(default 10); smaller lists takea per-element equality OR-chain and never build a
HashSet. So by constructionthe set always holds more than the threshold's worth of values, and for larger
INlists (dozens to thousands) the default-capacity set rehashes several timesper pushed-down predicate, allocating and discarding bucket arrays, once per
Parquet file split scanned. Pre-sizing removes that avoidable work. The change
is capacity-only and behavior-preserving.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Existing
ParquetV1FilterSuite/ParquetV2FilterSuite, which coverIN/InSetpredicate pushdown (including the Time / timestamp-nanos types that exercise
TypedParquetFilterOps.makeIn), pass. The change is capacity-only, so predicatesemantics are unchanged.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8