diff --git a/common/utils/src/main/scala/org/apache/spark/util/SparkCollectionUtils.scala b/common/utils/src/main/scala/org/apache/spark/util/SparkCollectionUtils.scala index 956d9c8bc9e34..9beb8b8a22a2e 100644 --- a/common/utils/src/main/scala/org/apache/spark/util/SparkCollectionUtils.scala +++ b/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. */ diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index 93f4000fa2760..a8d0ab0fce8b3 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/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,7 +765,7 @@ 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)) } @@ -773,7 +773,7 @@ class ParquetFilters( 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)) } @@ -781,7 +781,7 @@ class ParquetFilters( 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]) } @@ -789,7 +789,7 @@ class ParquetFilters( 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]) } @@ -797,7 +797,7 @@ class ParquetFilters( 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) } @@ -805,7 +805,7 @@ class ParquetFilters( 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,7 +814,7 @@ 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) } @@ -822,7 +822,7 @@ class ParquetFilters( 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) } @@ -830,7 +830,7 @@ class ParquetFilters( 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,7 +841,7 @@ 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) } @@ -849,7 +849,7 @@ class ParquetFilters( 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) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala index 8894dbfe66718..311498cef28c7 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/types/ops/ParquetFilterOps.scala @@ -18,7 +18,6 @@ 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} @@ -26,6 +25,8 @@ 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) }