[SPARK-59056][SQL] Pre-size HashSets in ParquetFilters IN/InSet pushdown

common/utils/src/main/scala/org/apache/spark/util/SparkCollectionUtils.scala

@@ -16,12 +16,34 @@
  */
 package org.apache.spark.util
 
-import java.util.Arrays
+import java.util.{Arrays, HashSet}
 
 import scala.collection.immutable
 import scala.reflect.ClassTag
 
 private[spark] trait SparkCollectionUtils {
+  /**
+   * Creates a [[java.util.HashSet]] pre-sized to hold the given number of elements without
+   * triggering a resize.
+   *
+   * A `HashSet` is backed by a `HashMap` whose bucket table is reallocated and every element
+   * rehashed once the entry count exceeds the table capacity multiplied by the default load
+   * factor of `0.75`. When the final element count is known in advance, allocating the table
+   * at the required size up front avoids those intermediate reallocations and rehashes that a
+   * set grown from the default capacity would incur. The initial capacity is derived as
+   * `expectedSize / 0.75 + 1` so that all `expectedSize` elements are accommodated below the
+   * resize threshold.
+   *
+   * @param expectedSize the number of elements the returned set is expected to hold; must be
+   *                     non-negative
+   * @tparam T the element type of the returned set
+   * @return an empty `HashSet` with capacity sufficient to hold `expectedSize` elements
+   *         without resizing
+   */
+  def newHashSetWithExpectedSize[T](expectedSize: Int): HashSet[T] = {
+    new HashSet[T]((expectedSize / 0.75f + 1.0f).toInt)
+  }
+
   /**
    * Same function as `keys.zipWithIndex.toMap`, but has perf gain.
    */

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala

@@ -23,7 +23,6 @@ import java.nio.charset.StandardCharsets.UTF_8
 import java.sql.{Date, Timestamp}
 import java.time.{Duration, Instant, LocalDate, LocalTime, Period}
 import java.time.temporal.ChronoField.MICRO_OF_DAY
-import java.util.HashSet
 import java.util.Locale
 
 import scala.jdk.CollectionConverters._
@@ -48,6 +47,7 @@ import org.apache.spark.sql.sources
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 import org.apache.spark.util.ArrayImplicits._
+import org.apache.spark.util.SparkCollectionUtils
 
 /**
  * Some utility function to convert Spark data source filters to Parquet filters.
@@ -765,47 +765,47 @@ class ParquetFilters(
 
     case ParquetByteType | ParquetShortType | ParquetIntegerType =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[Integer]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[Integer](values.length)
         for (value <- values) {
           set.add(toIntValue(value))
         }
         FilterApi.in(intColumn(n), set)
 
     case ParquetLongType =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[JLong]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[JLong](values.length)
         for (value <- values) {
           set.add(toLongValue(value))
         }
         FilterApi.in(longColumn(n), set)
 
     case ParquetFloatType =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[JFloat]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[JFloat](values.length)
         for (value <- values) {
           set.add(value.asInstanceOf[JFloat])
         }
         FilterApi.in(floatColumn(n), set)
 
     case ParquetDoubleType =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[JDouble]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[JDouble](values.length)
         for (value <- values) {
           set.add(value.asInstanceOf[JDouble])
         }
         FilterApi.in(doubleColumn(n), set)
 
     case ParquetStringType =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[Binary]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[Binary](values.length)
         for (value <- values) {
           set.add(Option(value).map(s => Binary.fromString(s.asInstanceOf[String])).orNull)
         }
         FilterApi.in(binaryColumn(n), set)
 
     case ParquetBinaryType =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[Binary]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[Binary](values.length)
         for (value <- values) {
           set.add(Option(value)
             .map(b => Binary.fromReusedByteArray(b.asInstanceOf[Array[Byte]])).orNull)
@@ -814,23 +814,23 @@ class ParquetFilters(
 
     case ParquetDateType if pushDownDate =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[Integer]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[Integer](values.length)
         for (value <- values) {
           set.add(Option(value).map(date => dateToDays(date).asInstanceOf[Integer]).orNull)
         }
         FilterApi.in(intColumn(n), set)
 
     case ParquetTimestampMicrosType if pushDownTimestamp =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[JLong]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[JLong](values.length)
         for (value <- values) {
           set.add(Option(value).map(timestampToMicros).orNull)
         }
         FilterApi.in(longColumn(n), set)
 
     case ParquetTimestampMillisType if pushDownTimestamp =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[JLong]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[JLong](values.length)
         for (value <- values) {
           set.add(Option(value).map(timestampToMillis).orNull)
         }
@@ -841,15 +841,15 @@ class ParquetFilters(
 
     case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT32, _) if pushDownDecimal =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[Integer]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[Integer](values.length)
         for (value <- values) {
           set.add(Option(value).map(d => decimalToInt32(d.asInstanceOf[JBigDecimal])).orNull)
         }
         FilterApi.in(intColumn(n), set)
 
     case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, INT64, _) if pushDownDecimal =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[JLong]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[JLong](values.length)
         for (value <- values) {
           set.add(Option(value).map(d => decimalToInt64(d.asInstanceOf[JBigDecimal])).orNull)
         }
@@ -858,7 +858,7 @@ class ParquetFilters(
     case ParquetSchemaType(_: DecimalLogicalTypeAnnotation, FIXED_LEN_BYTE_ARRAY, length)
       if pushDownDecimal =>
       (n: Array[String], values: Array[Any]) =>
-        val set = new HashSet[Binary]()
+        val set = SparkCollectionUtils.newHashSetWithExpectedSize[Binary](values.length)
         for (value <- values) {
           set.add(Option(value)
             .map(d => decimalToByteArray(d.asInstanceOf[JBigDecimal], length)).orNull)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala

@@ -18,14 +18,15 @@
 package org.apache.spark.sql.execution.datasources.parquet.types.ops
 
 import java.lang.{Long => JLong}
-import java.util.HashSet
 
 import org.apache.parquet.filter2.predicate.{FilterApi, FilterPredicate}
 import org.apache.parquet.filter2.predicate.Operators.{Column, SupportsLtGt}
 import org.apache.parquet.filter2.predicate.SparkFilterApi.longColumn
 import org.apache.parquet.schema.LogicalTypeAnnotation
 import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
 
+import org.apache.spark.util.SparkCollectionUtils
+
 /**
  * Optional Parquet filter-pushdown support for a Types Framework type.
  *
@@ -110,7 +111,7 @@ private[parquet] abstract class TypedParquetFilterOps[T <: Comparable[T]] extend
   override def makeGtEq(columnPath: Array[String], value: Any): FilterPredicate =
     FilterApi.gtEq(column(columnPath), toPhysical(value))
   override def makeIn(columnPath: Array[String], values: Array[Any]): FilterPredicate = {
-    val set = new HashSet[T]()
+    val set = SparkCollectionUtils.newHashSetWithExpectedSize[T](values.length)
     values.foreach(v => set.add(toPhysicalOrNull(v)))
     FilterApi.in(column(columnPath), set)
   }