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 1 commit
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
Next Next commit
Add private[this] to vals
  • Loading branch information
@JoshRosen
JoshRosen committed Jan 3, 2020
commit 90cebf080a5d3857ea8cf2a89e8e060b8b5a2fbf
23 changes: 12 additions & 11 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 @@ -154,7 +154,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 +171,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 Down Expand Up @@ -466,7 +466,7 @@ private[parquet] class ParquetRowConverter(

private var currentArray: ArrayBuffer[Any] = _

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

Expand Down Expand Up @@ -528,9 +528,10 @@ private[parquet] class ParquetRowConverter(

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

Expand All @@ -550,7 +551,7 @@ private[parquet] class ParquetRowConverter(
private var currentKeys: ArrayBuffer[Any] = _
private var currentValues: ArrayBuffer[Any] = _

private val keyValueConverter = {
private[this] val keyValueConverter = {
val repeatedType = parquetType.getType(0).asGroupType()
new KeyValueConverter(
repeatedType.getType(0),
Expand Down Expand Up @@ -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 Down Expand Up @@ -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