Skip to content

[SPARK-30414][SQL] ParquetRowConverter optimizations: arrays, maps, plus misc. constant factors #27089

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
JoshRosen wants to merge 6 commits into from
Closed

[SPARK-30414][SQL] ParquetRowConverter optimizations: arrays, maps, plus misc. constant factors #27089

Changes from 5 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
90cebf0
Add private[this] to vals
JoshRosen Jan 3, 2020
7318785
Save a .updater() call for each field
JoshRosen Jan 3, 2020
e05de15
Only call currentRow.numFields once per loop
JoshRosen Jan 3, 2020
c7d1534
Replace Scala ArrayBuffer with Java ArrayList; clear buffer and re-us…
JoshRosen Jan 3, 2020
4456f91
Avoid GenericArrayData constructor perf. problems (see SPARK-30413)
JoshRosen Jan 3, 2020
6d16f59
Roll back to using Scala ArrayBuffer, but continue using clear()
joshrosen-stripe Jan 7, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 46 additions & 40 deletions ...c/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import java.nio.ByteOrder
import java.util.TimeZone

import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer

import org.apache.parquet.column.Dictionary
import org.apache.parquet.io.api.{Binary, Converter, GroupConverter, PrimitiveConverter}
Expand All @@ -42,7 +41,7 @@ import org.apache.spark.unsafe.types.UTF8String
* A [[ParentContainerUpdater]] is used by a Parquet converter to set converted values to some
* corresponding parent container. For example, a converter for a `StructType` field may set
* converted values to a [[InternalRow]]; or a converter for array elements may append converted
* values to an [[ArrayBuffer]].
* values to a [[java.util.ArrayList]].
*/
private[parquet] trait ParentContainerUpdater {
/** Called before a record field is being converted */
Expand Down Expand Up @@ -154,7 +153,7 @@ private[parquet] class ParquetRowConverter(
|${catalystType.prettyJson}
""".stripMargin)

private val UTC = DateTimeUtils.TimeZoneUTC
private[this] val UTC = DateTimeUtils.TimeZoneUTC

/**
* Updater used together with field converters within a [[ParquetRowConverter]]. It propagates
Expand All @@ -171,15 +170,15 @@ private[parquet] class ParquetRowConverter(
override def setFloat(value: Float): Unit = row.setFloat(ordinal, value)
}

private val currentRow = new SpecificInternalRow(catalystType.map(_.dataType))
private[this] val currentRow = new SpecificInternalRow(catalystType.map(_.dataType))

/**
* The [[InternalRow]] converted from an entire Parquet record.
*/
def currentRecord: InternalRow = currentRow

// Converters for each field.
private val fieldConverters: Array[Converter with HasParentContainerUpdater] = {
private[this] val fieldConverters: Array[Converter with HasParentContainerUpdater] = {
parquetType.getFields.asScala.map { parquetField =>
val fieldIndex = catalystType.fieldIndex(parquetField.getName)
val catalystField = catalystType(fieldIndex)
Expand All @@ -188,26 +187,30 @@ private[parquet] class ParquetRowConverter(
}.toArray
}

// Updaters for each field.
private[this] val fieldUpdaters: Array[ParentContainerUpdater] = fieldConverters.map(_.updater)

override def getConverter(fieldIndex: Int): Converter = fieldConverters(fieldIndex)

override def end(): Unit = {
var i = 0
while (i < fieldConverters.length) {
fieldConverters(i).updater.end()
while (i < fieldUpdaters.length) {
fieldUpdaters(i).end()
i += 1
}
updater.set(currentRow)
}

override def start(): Unit = {
var i = 0
while (i < currentRow.numFields) {
val numFields = currentRow.numFields
while (i < numFields) {
currentRow.setNullAt(i)
i += 1
}
i = 0
while (i < fieldConverters.length) {
fieldConverters(i).updater.start()
while (i < fieldUpdaters.length) {
fieldUpdaters(i).start()
i += 1
}
}
Expand Down Expand Up @@ -464,9 +467,9 @@ private[parquet] class ParquetRowConverter(
updater: ParentContainerUpdater)
extends ParquetGroupConverter(updater) {

private var currentArray: ArrayBuffer[Any] = _
private[this] val currentArray = new java.util.ArrayList[Any]()

private val elementConverter: Converter = {
private[this] val elementConverter: Converter = {
val repeatedType = parquetSchema.getType(0)
val elementType = catalystSchema.elementType

Expand Down Expand Up @@ -503,7 +506,7 @@ private[parquet] class ParquetRowConverter(
// If the repeated field corresponds to the element type, creates a new converter using the
// type of the repeated field.
newConverter(repeatedType, elementType, new ParentContainerUpdater {
override def set(value: Any): Unit = currentArray += value
override def set(value: Any): Unit = currentArray.add(value)
})
} else {
// If the repeated field corresponds to the syntactic group in the standard 3-level Parquet
Expand All @@ -515,26 +518,24 @@ private[parquet] class ParquetRowConverter(

override def getConverter(fieldIndex: Int): Converter = elementConverter

override def end(): Unit = updater.set(new GenericArrayData(currentArray.toArray))
override def end(): Unit = updater.set(ParquetRowConverter.arrayListToArrayData(currentArray))

// NOTE: We can't reuse the mutable `ArrayBuffer` here and must instantiate a new buffer for the
// next value. `Row.copy()` only copies row cells, it doesn't do deep copy to objects stored
// in row cells.
override def start(): Unit = currentArray = ArrayBuffer.empty[Any]
Comment on lines -520 to -523

@viirya viirya Jan 7, 2020

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.

I think it depends on currentArray.toArray copies the elements or not?

@joshrosen-stripe joshrosen-stripe Jan 7, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

ArrayBuffer.toArray should always return a fresh unshared array object (internally, it allocates a new array and then calls copyToArray).

It doesn't do copying / cloning of the array elements themselves, but that shouldn't be a problem: by design, the objects that are inserted into this array are unshared / immutable: the map and array converters always return unshared objects and we always .copy() rows when inserting them into a map or array parent container (this is still true after the changes in #26993).

I did a bit of archaeology and tracked down the source of the // NOTE comment here: it was added in #7231 and at that time it looks like we were actually passing the mutable.ArrayBuffer itself to updater: https://github.com/apache/spark/blame/360fe18a61538b03cac05da1c6d258e124df6feb/sql/core/src/main/scala/org/apache/spark/sql/parquet/CatalystRowConverter.scala#L321. The comment makes sense in that context: with that older code, we would wind up with Row() objects that contained mutable.ArrayBuffers.

Later, in #7724 this was changed to pass a new GenericArrayData(currentArray.toArray) to the parent updater: c0cc0ea#diff-1d6c363c04155a9328fe1f5bd08a2f90. At that point I think we could have safely made the change to begin reusing the mutable.ArrayBuffer since it no longer escaped its converter.

override def start(): Unit = currentArray.clear()

/** Array element converter */
private final class ElementConverter(parquetType: Type, catalystType: DataType)
extends GroupConverter {

private var currentElement: Any = _

private val converter = newConverter(parquetType, catalystType, new ParentContainerUpdater {
override def set(value: Any): Unit = currentElement = value
})
private[this] val converter =
newConverter(parquetType, catalystType, new ParentContainerUpdater {
override def set(value: Any): Unit = currentElement = value
})

override def getConverter(fieldIndex: Int): Converter = converter

override def end(): Unit = currentArray += currentElement
override def end(): Unit = currentArray.add(currentElement)

override def start(): Unit = currentElement = null
}
Expand All @@ -547,10 +548,10 @@ private[parquet] class ParquetRowConverter(
updater: ParentContainerUpdater)
extends ParquetGroupConverter(updater) {

private var currentKeys: ArrayBuffer[Any] = _
private var currentValues: ArrayBuffer[Any] = _
private[this] val currentKeys = new java.util.ArrayList[Any]()
private[this] val currentValues = new java.util.ArrayList[Any]()

private val keyValueConverter = {
private[this] val keyValueConverter = {
val repeatedType = parquetType.getType(0).asGroupType()
new KeyValueConverter(
repeatedType.getType(0),
Expand All @@ -565,15 +566,15 @@ private[parquet] class ParquetRowConverter(
// The parquet map may contains null or duplicated map keys. When it happens, the behavior is
// undefined.
// TODO (SPARK-26174): disallow it with a config.
updater.set(ArrayBasedMapData(currentKeys.toArray, currentValues.toArray))
updater.set(
new ArrayBasedMapData(
ParquetRowConverter.arrayListToArrayData(currentKeys),
ParquetRowConverter.arrayListToArrayData(currentValues)))
}

// NOTE: We can't reuse the mutable Map here and must instantiate a new `Map` for the next
// value. `Row.copy()` only copies row cells, it doesn't do deep copy to objects stored in row
// cells.
override def start(): Unit = {
currentKeys = ArrayBuffer.empty[Any]
currentValues = ArrayBuffer.empty[Any]
currentKeys.clear()
currentValues.clear()
}

/** Parquet converter for key-value pairs within the map. */
Expand All @@ -588,7 +589,7 @@ private[parquet] class ParquetRowConverter(

private var currentValue: Any = _

private val converters = Array(
private[this] val converters = Array(
// Converter for keys
newConverter(parquetKeyType, catalystKeyType, new ParentContainerUpdater {
override def set(value: Any): Unit = currentKey = value
Expand All @@ -602,8 +603,8 @@ private[parquet] class ParquetRowConverter(
override def getConverter(fieldIndex: Int): Converter = converters(fieldIndex)

override def end(): Unit = {
currentKeys += currentKey
currentValues += currentValue
currentKeys.add(currentKey)
currentValues.add(currentValue)
}

override def start(): Unit = {
Expand All @@ -614,12 +615,12 @@ private[parquet] class ParquetRowConverter(
}

private trait RepeatedConverter {
private var currentArray: ArrayBuffer[Any] = _
private[this] val currentArray = new java.util.ArrayList[Any]()

@HyukjinKwon HyukjinKwon Jan 7, 2020

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.

@JoshRosen, sorry if I'm ignorant about this but why do we need to change ArrayBuffer -> ArrayList? Seems ArrayBuffer itself is mutable and can clear() too.

@JoshRosen JoshRosen Jan 7, 2020

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.

From prior experience I've found ArrayList to be marginally faster; I ran some quick-and-dirty non-Spark microbenchmarks and this is indeed still the case, but the gain is pretty marginal compared to other factors.

In the interests of code simplicity and clarity, I've backed out that part of the change: the code now uses and clear()s a mutable.ArrayBuffer: 6d16f59


protected def newArrayUpdater(updater: ParentContainerUpdater) = new ParentContainerUpdater {
override def start(): Unit = currentArray = ArrayBuffer.empty[Any]
override def end(): Unit = updater.set(new GenericArrayData(currentArray.toArray))
override def set(value: Any): Unit = currentArray += value
override def start(): Unit = currentArray.clear()
override def end(): Unit = updater.set(ParquetRowConverter.arrayListToArrayData(currentArray))
override def set(value: Any): Unit = currentArray.add(value)
}
}

Expand All @@ -635,7 +636,7 @@ private[parquet] class ParquetRowConverter(

val updater: ParentContainerUpdater = newArrayUpdater(parentUpdater)

private val elementConverter: PrimitiveConverter =
private[this] val elementConverter: PrimitiveConverter =
newConverter(parquetType, catalystType, updater).asPrimitiveConverter()

override def addBoolean(value: Boolean): Unit = elementConverter.addBoolean(value)
Expand All @@ -662,7 +663,7 @@ private[parquet] class ParquetRowConverter(

val updater: ParentContainerUpdater = newArrayUpdater(parentUpdater)

private val elementConverter: GroupConverter =
private[this] val elementConverter: GroupConverter =
newConverter(parquetType, catalystType, updater).asGroupConverter()

override def getConverter(field: Int): Converter = elementConverter.getConverter(field)
Expand Down Expand Up @@ -702,4 +703,9 @@ private[parquet] object ParquetRowConverter {
val julianDay = buffer.getInt
DateTimeUtils.fromJulianDay(julianDay, timeOfDayNanos)
}

def arrayListToArrayData(arrayList: java.util.ArrayList[Any]): GenericArrayData = {
// Cast to force use of primary constructor; see SPARK-30413
new GenericArrayData(arrayList.toArray.asInstanceOf[Array[Any]])
}
}