You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by jo...@apache.org on 2024/04/15 20:40:05 UTC
(hudi) branch master updated: [HUDI-7566] Add schema evolution to spark file readers (#10956)
This is an automated email from the ASF dual-hosted git repository.
jonvex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c71ac326b79 [HUDI-7566] Add schema evolution to spark file readers (#10956)
c71ac326b79 is described below
commit c71ac326b791fafe30ca0e46b64d85ddedba396a
Author: Jon Vexler <jb...@gmail.com>
AuthorDate: Mon Apr 15 16:39:58 2024 -0400
[HUDI-7566] Add schema evolution to spark file readers (#10956)
* add spark 3.3 reader
* add spark3.4
* add spark 3.5
* add spark 3.2
* add spark 3.1
* add spark 3.0
* add spark 2.4
* spark 3.3 use properties class
* spark 3.2 add props class
* spark 3.4 add properties
* add spark 3.5 properties
* add properties spark 3.1
* add props spark 3.0
* add properties spark 2.4
* fix 3.0
* refactor to get rid of properties, spark 3.1
* remove props spark 3.0
* use class model for spark 3.3
* remove props spark 3.3
* remove props spark 3.4
* remove props spark 3.5
* remove props spark 2.4
* remove change
* remove bad import
* add spark 3.3
* add spark 3.4
* add spark 3.5
* add spark 3.2
* add spark 3.1
* add spark 3.0
* add spark 2.4
* fix 2.4
* create a copy of the conf when reading
* make conf copy during read
* add test
* allow vectorized read and comment better
* address review comments 3.5
* rename spark 3.4
* rename for spark3.3
* rename for spark 3.2
* rename spark 3.1
* rename spark 30
* rename for spark 2
* remove empty line
* address hidden review comments
* add missing import
* address comments and add changes to legacy 3.5
* spark 3.4 update legacy
* make changes to spark 3.3 and restore legacy for 3.4 and 3.5
* update spark 3.2
* update spark 3.1
* update spark 3.0
---------
Co-authored-by: Jonathan Vexler <=>
---
.../datasources/parquet/Spark24ParquetReader.scala | 64 +++++--
.../Spark3ParquetSchemaEvolutionUtils.scala | 196 +++++++++++++++++++++
.../datasources/parquet/Spark30ParquetReader.scala | 26 ++-
.../Spark30ParquetSchemaEvolutionUtils.scala | 54 ++++++
.../datasources/parquet/Spark31ParquetReader.scala | 49 ++++--
.../Spark31ParquetSchemaEvolutionUtils.scala | 57 ++++++
.../datasources/parquet/Spark32ParquetReader.scala | 112 +++++++++---
.../Spark32PlusParquetSchemaEvolutionUtils.scala | 64 +++++++
.../datasources/parquet/Spark33ParquetReader.scala | 18 +-
.../datasources/parquet/Spark34ParquetReader.scala | 10 +-
.../datasources/parquet/Spark35ParquetReader.scala | 19 +-
11 files changed, 590 insertions(+), 79 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark24ParquetReader.scala b/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark24ParquetReader.scala
index 7fa30a36222..42808f337b7 100644
--- a/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark24ParquetReader.scala
+++ b/hudi-spark-datasource/hudi-spark2/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark24ParquetReader.scala
@@ -30,12 +30,12 @@ import org.apache.parquet.hadoop.{ParquetFileReader, ParquetInputFormat, Parquet
import org.apache.spark.TaskContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
-import org.apache.spark.sql.catalyst.expressions.{JoinedRow, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.{Cast, JoinedRow, UnsafeRow}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources.{PartitionedFile, RecordReaderIterator}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.Filter
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{StructField, StructType}
import java.net.URI
@@ -131,8 +131,17 @@ class Spark24ParquetReader(enableVectorizedReader: Boolean,
}
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
+
+ // Clone new conf
+ val hadoopAttemptConf = new Configuration(sharedConf)
+ val (implicitTypeChangeInfos, sparkRequestSchema) = HoodieParquetFileFormatHelper.buildImplicitSchemaChangeInfo(hadoopAttemptConf, footerFileMetaData, requiredSchema)
+
+ if (!implicitTypeChangeInfos.isEmpty) {
+ hadoopAttemptConf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA, sparkRequestSchema.json)
+ }
+
val hadoopAttemptContext =
- new TaskAttemptContextImpl(sharedConf, attemptId)
+ new TaskAttemptContextImpl(hadoopAttemptConf, attemptId)
// Try to push down filters when filter push-down is enabled.
// Notice: This push-down is RowGroups level, not individual records.
@@ -141,8 +150,20 @@ class Spark24ParquetReader(enableVectorizedReader: Boolean,
}
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
- val vectorizedReader = new VectorizedParquetRecordReader(
- convertTz.orNull, enableOffHeapColumnVector && taskContext.isDefined, capacity)
+ val vectorizedReader = if (!implicitTypeChangeInfos.isEmpty) {
+ new Spark24HoodieVectorizedParquetRecordReader(
+ convertTz.orNull,
+ enableOffHeapColumnVector && taskContext.isDefined,
+ capacity,
+ implicitTypeChangeInfos
+ )
+ } else {
+ new VectorizedParquetRecordReader(
+ convertTz.orNull,
+ enableOffHeapColumnVector && taskContext.isDefined,
+ capacity)
+ }
+
val iter = new RecordReaderIterator(vectorizedReader)
// SPARK-23457 Register a task completion lister before `initialization`.
taskContext.foreach(_.addTaskCompletionListener[Unit](_ => iter.close()))
@@ -156,11 +177,12 @@ class Spark24ParquetReader(enableVectorizedReader: Boolean,
iter.asInstanceOf[Iterator[InternalRow]]
} else {
// ParquetRecordReader returns UnsafeRow
+ val readSupport = new ParquetReadSupport(convertTz)
val reader = if (pushed.isDefined && enableRecordFilter) {
val parquetFilter = FilterCompat.get(pushed.get, null)
- new ParquetRecordReader[UnsafeRow](new ParquetReadSupport(convertTz), parquetFilter)
+ new ParquetRecordReader[UnsafeRow](readSupport, parquetFilter)
} else {
- new ParquetRecordReader[UnsafeRow](new ParquetReadSupport(convertTz))
+ new ParquetRecordReader[UnsafeRow](readSupport)
}
val iter = new RecordReaderIterator(reader)
// SPARK-23457 Register a task completion lister before `initialization`.
@@ -168,18 +190,38 @@ class Spark24ParquetReader(enableVectorizedReader: Boolean,
reader.initialize(split, hadoopAttemptContext)
val fullSchema = requiredSchema.toAttributes ++ partitionSchema.toAttributes
- val joinedRow = new JoinedRow()
- val appendPartitionColumns = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ val unsafeProjection = if (implicitTypeChangeInfos.isEmpty) {
+ GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ } else {
+ val newFullSchema = new StructType(requiredSchema.fields.zipWithIndex.map { case (f, i) =>
+ if (implicitTypeChangeInfos.containsKey(i)) {
+ StructField(f.name, implicitTypeChangeInfos.get(i).getRight, f.nullable, f.metadata)
+ } else f
+ }).toAttributes ++ partitionSchema.toAttributes
+ val castSchema = newFullSchema.zipWithIndex.map { case (attr, i) =>
+ if (implicitTypeChangeInfos.containsKey(i)) {
+ val srcType = implicitTypeChangeInfos.get(i).getRight
+ val dstType = implicitTypeChangeInfos.get(i).getLeft
+ val needTimeZone = Cast.needsTimeZone(srcType, dstType)
+ Cast(attr, dstType, if (needTimeZone) timeZoneId else None)
+ } else attr
+ }
+ GenerateUnsafeProjection.generate(castSchema, newFullSchema)
+ }
// This is a horrible erasure hack... if we type the iterator above, then it actually check
// the type in next() and we get a class cast exception. If we make that function return
// Object, then we can defer the cast until later!
+ //
+ // NOTE: We're making appending of the partitioned values to the rows read from the
+ // data file configurable
if (partitionSchema.length == 0) {
// There is no partition columns
- iter.asInstanceOf[Iterator[InternalRow]]
+ iter.asInstanceOf[Iterator[InternalRow]].map(unsafeProjection)
} else {
+ val joinedRow = new JoinedRow()
iter.asInstanceOf[Iterator[InternalRow]]
- .map(d => appendPartitionColumns(joinedRow(d, file.partitionValues)))
+ .map(d => unsafeProjection(joinedRow(d, file.partitionValues)))
}
}
}
diff --git a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/Spark3ParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/Spark3ParquetSchemaEvolutionUtils.scala
new file mode 100644
index 00000000000..4f83ef5a569
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/Spark3ParquetSchemaEvolutionUtils.scala
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hudi.SparkAdapterSupport
+import org.apache.hudi.client.utils.SparkInternalSchemaConverter
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.util
+import org.apache.hudi.common.util.InternalSchemaCache
+import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
+import org.apache.hudi.common.util.collection.Pair
+import org.apache.hudi.internal.schema.InternalSchema
+import org.apache.hudi.internal.schema.action.InternalSchemaMerger
+import org.apache.hudi.internal.schema.utils.{InternalSchemaUtils, SerDeHelper}
+import org.apache.parquet.hadoop.metadata.FileMetaData
+import org.apache.spark.sql.HoodieSchemaUtils
+import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, UnsafeProjection}
+import org.apache.spark.sql.execution.datasources.Spark3ParquetSchemaEvolutionUtils.pruneInternalSchema
+import org.apache.spark.sql.execution.datasources.parquet.{HoodieParquetFileFormatHelper, ParquetReadSupport}
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
+
+import scala.collection.convert.ImplicitConversions.`collection AsScalaIterable`
+
+abstract class Spark3ParquetSchemaEvolutionUtils(sharedConf: Configuration,
+ filePath: Path,
+ requiredSchema: StructType,
+ partitionSchema: StructType) extends SparkAdapterSupport{
+ // Fetch internal schema
+ private lazy val internalSchemaStr: String = sharedConf.get(SparkInternalSchemaConverter.HOODIE_QUERY_SCHEMA)
+
+ private lazy val querySchemaOption: util.Option[InternalSchema] = pruneInternalSchema(internalSchemaStr, requiredSchema)
+
+ var shouldUseInternalSchema: Boolean = !isNullOrEmpty(internalSchemaStr) && querySchemaOption.isPresent
+
+ private lazy val schemaUtils: HoodieSchemaUtils = sparkAdapter.getSchemaUtils
+
+ private lazy val tablePath: String = sharedConf.get(SparkInternalSchemaConverter.HOODIE_TABLE_PATH)
+ private lazy val fileSchema: InternalSchema = if (shouldUseInternalSchema) {
+ val commitInstantTime = FSUtils.getCommitTime(filePath.getName).toLong;
+ val validCommits = sharedConf.get(SparkInternalSchemaConverter.HOODIE_VALID_COMMITS_LIST)
+ InternalSchemaCache.getInternalSchemaByVersionId(commitInstantTime, tablePath, sharedConf, if (validCommits == null) "" else validCommits)
+ } else {
+ null
+ }
+
+ def rebuildFilterFromParquet(filter: Filter): Filter = {
+ rebuildFilterFromParquetHelper(filter, fileSchema, querySchemaOption.orElse(null))
+ }
+
+ private def rebuildFilterFromParquetHelper(oldFilter: Filter, fileSchema: InternalSchema, querySchema: InternalSchema): Filter = {
+ if (fileSchema == null || querySchema == null) {
+ oldFilter
+ } else {
+ oldFilter match {
+ case eq: EqualTo =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute = newAttribute)
+ case eqs: EqualNullSafe =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute = newAttribute)
+ case gt: GreaterThan =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute = newAttribute)
+ case gtr: GreaterThanOrEqual =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute = newAttribute)
+ case lt: LessThan =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute = newAttribute)
+ case lte: LessThanOrEqual =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute = newAttribute)
+ case i: In =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute = newAttribute)
+ case isn: IsNull =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute = newAttribute)
+ case isnn: IsNotNull =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute = newAttribute)
+ case And(left, right) =>
+ And(rebuildFilterFromParquetHelper(left, fileSchema, querySchema), rebuildFilterFromParquetHelper(right, fileSchema, querySchema))
+ case Or(left, right) =>
+ Or(rebuildFilterFromParquetHelper(left, fileSchema, querySchema), rebuildFilterFromParquetHelper(right, fileSchema, querySchema))
+ case Not(child) =>
+ Not(rebuildFilterFromParquetHelper(child, fileSchema, querySchema))
+ case ssw: StringStartsWith =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute = newAttribute)
+ case ses: StringEndsWith =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute = newAttribute)
+ case sc: StringContains =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute = newAttribute)
+ case AlwaysTrue =>
+ AlwaysTrue
+ case AlwaysFalse =>
+ AlwaysFalse
+ case _ =>
+ AlwaysTrue
+ }
+ }
+ }
+
+ protected var typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = null
+
+ def getHadoopConfClone(footerFileMetaData: FileMetaData, enableVectorizedReader: Boolean): Configuration = {
+ // Clone new conf
+ val hadoopAttemptConf = new Configuration(sharedConf)
+ typeChangeInfos = if (shouldUseInternalSchema) {
+ val mergedInternalSchema = new InternalSchemaMerger(fileSchema, querySchemaOption.get(), true, true).mergeSchema()
+ val mergedSchema = SparkInternalSchemaConverter.constructSparkSchemaFromInternalSchema(mergedInternalSchema)
+
+ hadoopAttemptConf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA, mergedSchema.json)
+
+ SparkInternalSchemaConverter.collectTypeChangedCols(querySchemaOption.get(), mergedInternalSchema)
+ } else {
+ val (implicitTypeChangeInfo, sparkRequestSchema) = HoodieParquetFileFormatHelper.buildImplicitSchemaChangeInfo(hadoopAttemptConf, footerFileMetaData, requiredSchema)
+ if (!implicitTypeChangeInfo.isEmpty) {
+ shouldUseInternalSchema = true
+ hadoopAttemptConf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA, sparkRequestSchema.json)
+ }
+ implicitTypeChangeInfo
+ }
+
+ if (enableVectorizedReader && shouldUseInternalSchema &&
+ !typeChangeInfos.values().forall(_.getLeft.isInstanceOf[AtomicType])) {
+ throw new IllegalArgumentException(
+ "Nested types with type changes(implicit or explicit) cannot be read in vectorized mode. " +
+ "To workaround this issue, set spark.sql.parquet.enableVectorizedReader=false.")
+ }
+
+ hadoopAttemptConf
+ }
+
+ def generateUnsafeProjection(fullSchema: Seq[AttributeReference], timeZoneId: Option[String]): UnsafeProjection = {
+
+ if (typeChangeInfos.isEmpty) {
+ GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ } else {
+ // find type changed.
+ val newSchema = new StructType(requiredSchema.fields.zipWithIndex.map { case (f, i) =>
+ if (typeChangeInfos.containsKey(i)) {
+ StructField(f.name, typeChangeInfos.get(i).getRight, f.nullable, f.metadata)
+ } else f
+ })
+ val newFullSchema = schemaUtils.toAttributes(newSchema) ++ schemaUtils.toAttributes(partitionSchema)
+ val castSchema = newFullSchema.zipWithIndex.map { case (attr, i) =>
+ if (typeChangeInfos.containsKey(i)) {
+ val srcType = typeChangeInfos.get(i).getRight
+ val dstType = typeChangeInfos.get(i).getLeft
+ val needTimeZone = Cast.needsTimeZone(srcType, dstType)
+ Cast(attr, dstType, if (needTimeZone) timeZoneId else None)
+ } else attr
+ }
+ GenerateUnsafeProjection.generate(castSchema, newFullSchema)
+ }
+ }
+}
+
+object Spark3ParquetSchemaEvolutionUtils {
+ def pruneInternalSchema(internalSchemaStr: String, requiredSchema: StructType): util.Option[InternalSchema] = {
+ if (!isNullOrEmpty(internalSchemaStr) ) {
+ val querySchemaOption = SerDeHelper.fromJson(internalSchemaStr)
+ if (querySchemaOption.isPresent && requiredSchema.nonEmpty) {
+ util.Option.of(SparkInternalSchemaConverter.convertAndPruneStructTypeToInternalSchema(requiredSchema, querySchemaOption.get()))
+ } else {
+ querySchemaOption
+ }
+ } else {
+ util.Option.empty()
+ }
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30ParquetReader.scala
index 9088676c335..a6e9a3c5fe0 100644
--- a/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30ParquetReader.scala
+++ b/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30ParquetReader.scala
@@ -30,7 +30,6 @@ import org.apache.parquet.hadoop._
import org.apache.spark.TaskContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.JoinedRow
-import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SQLConf
@@ -95,14 +94,22 @@ class Spark30ParquetReader(enableVectorizedReader: Boolean,
Array.empty,
null)
+ val schemaEvolutionUtils = new Spark30ParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema)
+
lazy val footerFileMetaData =
ParquetFileReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
// Try to push down filters when filter push-down is enabled.
val pushed = if (enableParquetFilterPushDown) {
val parquetSchema = footerFileMetaData.getSchema
- val parquetFilters = new ParquetFilters(parquetSchema, pushDownDate, pushDownTimestamp,
- pushDownDecimal, pushDownStringStartWith, pushDownInFilterThreshold, isCaseSensitive)
- filters
+ val parquetFilters = createParquetFilters(
+ parquetSchema,
+ pushDownDate,
+ pushDownTimestamp,
+ pushDownDecimal,
+ pushDownStringStartWith,
+ pushDownInFilterThreshold,
+ isCaseSensitive)
+ filters.map(schemaEvolutionUtils.rebuildFilterFromParquet)
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -133,7 +140,7 @@ class Spark30ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val hadoopAttemptContext =
- new TaskAttemptContextImpl(sharedConf, attemptId)
+ new TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData, enableVectorizedReader), attemptId)
// Try to push down filters when filter push-down is enabled.
// Notice: This push-down is RowGroups level, not individual records.
@@ -142,7 +149,7 @@ class Spark30ParquetReader(enableVectorizedReader: Boolean,
}
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
- val vectorizedReader = new VectorizedParquetRecordReader(
+ val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
convertTz.orNull,
datetimeRebaseMode.toString,
enableOffHeapColumnVector && taskContext.isDefined,
@@ -174,7 +181,7 @@ class Spark30ParquetReader(enableVectorizedReader: Boolean,
reader.initialize(split, hadoopAttemptContext)
val fullSchema = requiredSchema.toAttributes ++ partitionSchema.toAttributes
- val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ val unsafeProjection = schemaEvolutionUtils.generateUnsafeProjection(fullSchema, timeZoneId)
if (partitionSchema.length == 0) {
// There is no partition columns
@@ -186,6 +193,11 @@ class Spark30ParquetReader(enableVectorizedReader: Boolean,
}
}
+ private def createParquetFilters(args: Any*): ParquetFilters = {
+ val ctor = classOf[ParquetFilters].getConstructors.head
+ ctor.newInstance(args.map(_.asInstanceOf[AnyRef]): _*)
+ .asInstanceOf[ParquetFilters]
+ }
}
object Spark30ParquetReader extends SparkParquetReaderBuilder {
diff --git a/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30ParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30ParquetSchemaEvolutionUtils.scala
new file mode 100644
index 00000000000..28acb6e0474
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark3.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark30ParquetSchemaEvolutionUtils.scala
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.execution.datasources.Spark3ParquetSchemaEvolutionUtils
+import org.apache.spark.sql.types.StructType
+
+import java.time.ZoneId
+
+class Spark30ParquetSchemaEvolutionUtils(sharedConf: Configuration,
+ filePath: Path,
+ requiredSchema: StructType,
+ partitionSchema: StructType) extends
+ Spark3ParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema) {
+
+ def buildVectorizedReader(convertTz: ZoneId,
+ datetimeRebaseMode: String,
+ useOffHeap: Boolean,
+ capacity: Int): VectorizedParquetRecordReader = {
+ if (shouldUseInternalSchema) {
+ new Spark30HoodieVectorizedParquetRecordReader(
+ convertTz,
+ datetimeRebaseMode,
+ useOffHeap,
+ capacity,
+ typeChangeInfos)
+ } else {
+ new VectorizedParquetRecordReader(
+ convertTz,
+ datetimeRebaseMode,
+ useOffHeap,
+ capacity)
+ }
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31ParquetReader.scala
index 94a5efaee61..a3e179fa184 100644
--- a/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31ParquetReader.scala
+++ b/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31ParquetReader.scala
@@ -23,6 +23,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.mapreduce._
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
+import org.apache.hudi.HoodieSparkUtils
import org.apache.parquet.filter2.compat.FilterCompat
import org.apache.parquet.filter2.predicate.FilterApi
import org.apache.parquet.format.converter.ParquetMetadataConverter.SKIP_ROW_GROUPS
@@ -30,7 +31,6 @@ import org.apache.parquet.hadoop._
import org.apache.spark.TaskContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.JoinedRow
-import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SQLConf
@@ -95,6 +95,8 @@ class Spark31ParquetReader(enableVectorizedReader: Boolean,
Array.empty,
null)
+ val schemaEvolutionUtils = new Spark31ParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema)
+
lazy val footerFileMetaData =
ParquetFileReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
val datetimeRebaseMode = DataSourceUtils.datetimeRebaseMode(
@@ -103,16 +105,27 @@ class Spark31ParquetReader(enableVectorizedReader: Boolean,
// Try to push down filters when filter push-down is enabled.
val pushed = if (enableParquetFilterPushDown) {
val parquetSchema = footerFileMetaData.getSchema
- val parquetFilters = new ParquetFilters(
- parquetSchema,
- pushDownDate,
- pushDownTimestamp,
- pushDownDecimal,
- pushDownStringStartWith,
- pushDownInFilterThreshold,
- isCaseSensitive,
- datetimeRebaseMode)
- filters
+ val parquetFilters = if (HoodieSparkUtils.gteqSpark3_1_3) {
+ createParquetFilters(
+ parquetSchema,
+ pushDownDate,
+ pushDownTimestamp,
+ pushDownDecimal,
+ pushDownStringStartWith,
+ pushDownInFilterThreshold,
+ isCaseSensitive,
+ datetimeRebaseMode)
+ } else {
+ createParquetFilters(
+ parquetSchema,
+ pushDownDate,
+ pushDownTimestamp,
+ pushDownDecimal,
+ pushDownStringStartWith,
+ pushDownInFilterThreshold,
+ isCaseSensitive)
+ }
+ filters.map(schemaEvolutionUtils.rebuildFilterFromParquet)
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -143,7 +156,7 @@ class Spark31ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val hadoopAttemptContext =
- new TaskAttemptContextImpl(sharedConf, attemptId)
+ new TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData, enableVectorizedReader), attemptId)
// Try to push down filters when filter push-down is enabled.
// Notice: This push-down is RowGroups level, not individual records.
@@ -152,7 +165,7 @@ class Spark31ParquetReader(enableVectorizedReader: Boolean,
}
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
- val vectorizedReader = new VectorizedParquetRecordReader(
+ val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
convertTz.orNull,
datetimeRebaseMode.toString,
int96RebaseMode.toString,
@@ -188,8 +201,7 @@ class Spark31ParquetReader(enableVectorizedReader: Boolean,
reader.initialize(split, hadoopAttemptContext)
val fullSchema = requiredSchema.toAttributes ++ partitionSchema.toAttributes
- val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
-
+ val unsafeProjection = schemaEvolutionUtils.generateUnsafeProjection(fullSchema, timeZoneId)
if (partitionSchema.length == 0) {
// There is no partition columns
iter.map(unsafeProjection)
@@ -199,6 +211,13 @@ class Spark31ParquetReader(enableVectorizedReader: Boolean,
}
}
}
+
+ private def createParquetFilters(args: Any*): ParquetFilters = {
+ // ParquetFilters bears a single ctor (in Spark 3.1)
+ val ctor = classOf[ParquetFilters].getConstructors.head
+ ctor.newInstance(args.map(_.asInstanceOf[AnyRef]): _*)
+ .asInstanceOf[ParquetFilters]
+ }
}
object Spark31ParquetReader extends SparkParquetReaderBuilder {
diff --git a/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31ParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31ParquetSchemaEvolutionUtils.scala
new file mode 100644
index 00000000000..d31efde4d29
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark3.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark31ParquetSchemaEvolutionUtils.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.execution.datasources.Spark3ParquetSchemaEvolutionUtils
+import org.apache.spark.sql.types.StructType
+
+import java.time.ZoneId
+
+class Spark31ParquetSchemaEvolutionUtils(sharedConf: Configuration,
+ filePath: Path,
+ requiredSchema: StructType,
+ partitionSchema: StructType) extends
+ Spark3ParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema) {
+
+ def buildVectorizedReader(convertTz: ZoneId,
+ datetimeRebaseMode: String,
+ int96RebaseMode: String,
+ useOffHeap: Boolean,
+ capacity: Int): VectorizedParquetRecordReader = {
+ if (shouldUseInternalSchema) {
+ new Spark31HoodieVectorizedParquetRecordReader(
+ convertTz,
+ datetimeRebaseMode,
+ int96RebaseMode,
+ useOffHeap,
+ capacity,
+ typeChangeInfos)
+ } else {
+ new VectorizedParquetRecordReader(
+ convertTz,
+ datetimeRebaseMode,
+ int96RebaseMode,
+ useOffHeap,
+ capacity)
+ }
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32ParquetReader.scala
index 5437a18cd4b..e2189e5cbdf 100644
--- a/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32ParquetReader.scala
+++ b/hudi-spark-datasource/hudi-spark3.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32ParquetReader.scala
@@ -24,6 +24,7 @@ import org.apache.hadoop.fs.Path
import org.apache.hadoop.mapred.FileSplit
import org.apache.hadoop.mapreduce._
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
+import org.apache.hudi.HoodieSparkUtils
import org.apache.parquet.filter2.compat.FilterCompat
import org.apache.parquet.filter2.predicate.FilterApi
import org.apache.parquet.format.converter.ParquetMetadataConverter.SKIP_ROW_GROUPS
@@ -31,7 +32,6 @@ import org.apache.parquet.hadoop._
import org.apache.spark.TaskContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.JoinedRow
-import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SQLConf
@@ -82,15 +82,17 @@ class Spark32ParquetReader(enableVectorizedReader: Boolean,
* @return iterator of rows read from the file output type says [[InternalRow]] but could be [[ColumnarBatch]]
*/
protected def doRead(file: PartitionedFile,
- requiredSchema: StructType,
- partitionSchema: StructType,
- filters: Seq[Filter],
- sharedConf: Configuration): Iterator[InternalRow] = {
+ requiredSchema: StructType,
+ partitionSchema: StructType,
+ filters: Seq[Filter],
+ sharedConf: Configuration): Iterator[InternalRow] = {
assert(file.partitionValues.numFields == partitionSchema.size)
val filePath = new Path(new URI(file.filePath))
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
+ val schemaEvolutionUtils = new Spark32PlusParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema)
+
lazy val footerFileMetaData =
ParquetFooterReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec(
@@ -99,16 +101,36 @@ class Spark32ParquetReader(enableVectorizedReader: Boolean,
// Try to push down filters when filter push-down is enabled.
val pushed = if (enableParquetFilterPushDown) {
val parquetSchema = footerFileMetaData.getSchema
- val parquetFilters = new ParquetFilters(
- parquetSchema,
- pushDownDate,
- pushDownTimestamp,
- pushDownDecimal,
- pushDownStringStartWith,
- pushDownInFilterThreshold,
- isCaseSensitive,
- datetimeRebaseSpec)
- filters
+ val parquetFilters = if (HoodieSparkUtils.gteqSpark3_2_1) {
+ // NOTE: Below code could only be compiled against >= Spark 3.2.1,
+ // and unfortunately won't compile against Spark 3.2.0
+ // However this code is runtime-compatible w/ both Spark 3.2.0 and >= Spark 3.2.1
+ val datetimeRebaseSpec =
+ DataSourceUtils.datetimeRebaseSpec(footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead)
+ new ParquetFilters(
+ parquetSchema,
+ pushDownDate,
+ pushDownTimestamp,
+ pushDownDecimal,
+ pushDownStringStartWith,
+ pushDownInFilterThreshold,
+ isCaseSensitive,
+ datetimeRebaseSpec)
+ } else {
+ // Spark 3.2.0
+ val datetimeRebaseMode =
+ Spark32DataSourceUtils.datetimeRebaseMode(footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead)
+ createParquetFilters(
+ parquetSchema,
+ pushDownDate,
+ pushDownTimestamp,
+ pushDownDecimal,
+ pushDownStringStartWith,
+ pushDownInFilterThreshold,
+ isCaseSensitive,
+ datetimeRebaseMode)
+ }
+ filters.map(schemaEvolutionUtils.rebuildFilterFromParquet)
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -139,7 +161,7 @@ class Spark32ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val hadoopAttemptContext =
- new TaskAttemptContextImpl(sharedConf, attemptId)
+ new TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData, enableVectorizedReader), attemptId)
// Try to push down filters when filter push-down is enabled.
// Notice: This push-down is RowGroups level, not individual records.
@@ -148,7 +170,7 @@ class Spark32ParquetReader(enableVectorizedReader: Boolean,
}
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
- val vectorizedReader = new VectorizedParquetRecordReader(
+ val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
convertTz.orNull,
datetimeRebaseSpec.mode.toString,
datetimeRebaseSpec.timeZone,
@@ -182,11 +204,31 @@ class Spark32ParquetReader(enableVectorizedReader: Boolean,
}
} else {
// ParquetRecordReader returns InternalRow
- val readSupport = new ParquetReadSupport(
- convertTz,
- enableVectorizedReader = false,
- datetimeRebaseSpec,
- int96RebaseSpec)
+ val readSupport = if (HoodieSparkUtils.gteqSpark3_2_1) {
+ // ParquetRecordReader returns InternalRow
+ // NOTE: Below code could only be compiled against >= Spark 3.2.1,
+ // and unfortunately won't compile against Spark 3.2.0
+ // However this code is runtime-compatible w/ both Spark 3.2.0 and >= Spark 3.2.1
+ val int96RebaseSpec =
+ DataSourceUtils.int96RebaseSpec(footerFileMetaData.getKeyValueMetaData.get, int96RebaseModeInRead)
+ val datetimeRebaseSpec =
+ DataSourceUtils.datetimeRebaseSpec(footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead)
+ new ParquetReadSupport(
+ convertTz,
+ enableVectorizedReader = false,
+ datetimeRebaseSpec,
+ int96RebaseSpec)
+ } else {
+ val datetimeRebaseMode =
+ Spark32DataSourceUtils.datetimeRebaseMode(footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead)
+ val int96RebaseMode =
+ Spark32DataSourceUtils.int96RebaseMode(footerFileMetaData.getKeyValueMetaData.get, int96RebaseModeInRead)
+ createParquetReadSupport(
+ convertTz,
+ /* enableVectorizedReader = */ false,
+ datetimeRebaseMode,
+ int96RebaseMode)
+ }
val reader = if (pushed.isDefined && enableRecordFilter) {
val parquetFilter = FilterCompat.get(pushed.get, null)
new ParquetRecordReader[InternalRow](readSupport, parquetFilter)
@@ -198,7 +240,7 @@ class Spark32ParquetReader(enableVectorizedReader: Boolean,
reader.initialize(split, hadoopAttemptContext)
val fullSchema = requiredSchema.toAttributes ++ partitionSchema.toAttributes
- val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ val unsafeProjection = schemaEvolutionUtils.generateUnsafeProjection(fullSchema, timeZoneId)
if (partitionSchema.length == 0) {
// There is no partition columns
@@ -216,6 +258,30 @@ class Spark32ParquetReader(enableVectorizedReader: Boolean,
}
}
}
+
+ /**
+ * NOTE: This method is specific to Spark 3.2.0
+ */
+ private def createParquetFilters(args: Any*): ParquetFilters = {
+ // NOTE: ParquetFilters ctor args contain Scala enum, therefore we can't look it
+ // up by arg types, and have to instead rely on the number of args based on individual class;
+ // the ctor order is not guaranteed
+ val ctor = classOf[ParquetFilters].getConstructors.maxBy(_.getParameterCount)
+ ctor.newInstance(args.map(_.asInstanceOf[AnyRef]): _*)
+ .asInstanceOf[ParquetFilters]
+ }
+
+ /**
+ * NOTE: This method is specific to Spark 3.2.0
+ */
+ private def createParquetReadSupport(args: Any*): ParquetReadSupport = {
+ // NOTE: ParquetReadSupport ctor args contain Scala enum, therefore we can't look it
+ // up by arg types, and have to instead rely on the number of args based on individual class;
+ // the ctor order is not guaranteed
+ val ctor = classOf[ParquetReadSupport].getConstructors.maxBy(_.getParameterCount)
+ ctor.newInstance(args.map(_.asInstanceOf[AnyRef]): _*)
+ .asInstanceOf[ParquetReadSupport]
+ }
}
object Spark32ParquetReader extends SparkParquetReaderBuilder {
diff --git a/hudi-spark-datasource/hudi-spark3.2plus-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32PlusParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark3.2plus-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32PlusParquetSchemaEvolutionUtils.scala
new file mode 100644
index 00000000000..05523bcd9f0
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark3.2plus-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark32PlusParquetSchemaEvolutionUtils.scala
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.spark.TaskContext
+import org.apache.spark.sql.catalyst.util.RebaseDateTime
+import org.apache.spark.sql.execution.datasources.Spark3ParquetSchemaEvolutionUtils
+import org.apache.spark.sql.types.StructType
+
+import java.time.ZoneId
+
+class Spark32PlusParquetSchemaEvolutionUtils(sharedConf: Configuration,
+ filePath: Path,
+ requiredSchema: StructType,
+ partitionSchema: StructType) extends
+ Spark3ParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema) {
+
+ def buildVectorizedReader(convertTz: ZoneId,
+ datetimeRebaseMode: String,
+ datetimeRebaseTz: String,
+ int96RebaseMode: String,
+ int96RebaseTz: String,
+ useOffHeap: Boolean,
+ capacity: Int): VectorizedParquetRecordReader = {
+ if (shouldUseInternalSchema) {
+ new Spark32PlusHoodieVectorizedParquetRecordReader(
+ convertTz,
+ datetimeRebaseMode,
+ datetimeRebaseTz,
+ int96RebaseMode,
+ int96RebaseTz,
+ useOffHeap,
+ capacity,
+ typeChangeInfos)
+ } else {
+ new VectorizedParquetRecordReader(
+ convertTz,
+ datetimeRebaseMode,
+ datetimeRebaseTz,
+ int96RebaseMode,
+ int96RebaseTz,
+ useOffHeap,
+ capacity)
+ }
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
index 0bd8cca3599..8d882d7945d 100644
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
+++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
@@ -32,7 +32,6 @@ import org.apache.parquet.hadoop._
import org.apache.spark.TaskContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.JoinedRow
-import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SQLConf
@@ -83,15 +82,16 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean,
* @return iterator of rows read from the file output type says [[InternalRow]] but could be [[ColumnarBatch]]
*/
protected def doRead(file: PartitionedFile,
- requiredSchema: StructType,
- partitionSchema: StructType,
- filters: Seq[Filter],
- sharedConf: Configuration): Iterator[InternalRow] = {
+ requiredSchema: StructType,
+ partitionSchema: StructType,
+ filters: Seq[Filter],
+ sharedConf: Configuration): Iterator[InternalRow] = {
assert(file.partitionValues.numFields == partitionSchema.size)
val filePath = new Path(new URI(file.filePath))
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
+ val schemaEvolutionUtils = new Spark32PlusParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema)
lazy val footerFileMetaData =
ParquetFooterReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
@@ -110,7 +110,7 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean,
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
- filters
+ filters.map(schemaEvolutionUtils.rebuildFilterFromParquet)
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -141,7 +141,7 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val hadoopAttemptContext =
- new TaskAttemptContextImpl(sharedConf, attemptId)
+ new TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData, enableVectorizedReader), attemptId)
// Try to push down filters when filter push-down is enabled.
// Notice: This push-down is RowGroups level, not individual records.
@@ -150,7 +150,7 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean,
}
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
- val vectorizedReader = new VectorizedParquetRecordReader(
+ val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
convertTz.orNull,
datetimeRebaseSpec.mode.toString,
datetimeRebaseSpec.timeZone,
@@ -200,7 +200,7 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean,
reader.initialize(split, hadoopAttemptContext)
val fullSchema = requiredSchema.toAttributes ++ partitionSchema.toAttributes
- val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ val unsafeProjection = schemaEvolutionUtils.generateUnsafeProjection(fullSchema, timeZoneId)
if (partitionSchema.length == 0) {
// There is no partition columns
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
index 73db889a044..77e90b82849 100644
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
+++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
@@ -30,7 +30,6 @@ import org.apache.parquet.hadoop._
import org.apache.spark.TaskContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.JoinedRow
-import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SQLConf
@@ -88,6 +87,7 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean,
val filePath = file.toPath
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
+ val schemaEvolutionUtils = new Spark32PlusParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema)
lazy val footerFileMetaData =
ParquetFooterReader.readFooter(sharedConf, filePath, SKIP_ROW_GROUPS).getFileMetaData
@@ -106,7 +106,7 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean,
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
- filters
+ filters.map(schemaEvolutionUtils.rebuildFilterFromParquet)
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -137,7 +137,7 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val hadoopAttemptContext =
- new TaskAttemptContextImpl(sharedConf, attemptId)
+ new TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData, enableVectorizedReader), attemptId)
// Try to push down filters when filter push-down is enabled.
// Notice: This push-down is RowGroups level, not individual records.
@@ -146,7 +146,7 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean,
}
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
- val vectorizedReader = new VectorizedParquetRecordReader(
+ val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
convertTz.orNull,
datetimeRebaseSpec.mode.toString,
datetimeRebaseSpec.timeZone,
@@ -198,7 +198,7 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean,
readerWithRowIndexes.initialize(split, hadoopAttemptContext)
val fullSchema = requiredSchema.toAttributes ++ partitionSchema.toAttributes
- val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ val unsafeProjection = schemaEvolutionUtils.generateUnsafeProjection(fullSchema, timeZoneId)
if (partitionSchema.length == 0) {
// There is no partition columns
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
index f088efd07e1..39c2906c0e5 100644
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
+++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
@@ -29,7 +29,6 @@ import org.apache.parquet.hadoop.{ParquetInputFormat, ParquetRecordReader}
import org.apache.spark.TaskContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.JoinedRow
-import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources.{DataSourceUtils, FileFormat, PartitionedFile, RecordReaderIterator}
@@ -79,15 +78,17 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean,
* @return iterator of rows read from the file output type says [[InternalRow]] but could be [[ColumnarBatch]]
*/
protected def doRead(file: PartitionedFile,
- requiredSchema: StructType,
- partitionSchema: StructType,
- filters: Seq[Filter],
- sharedConf: Configuration): Iterator[InternalRow] = {
+ requiredSchema: StructType,
+ partitionSchema: StructType,
+ filters: Seq[Filter],
+ sharedConf: Configuration): Iterator[InternalRow] = {
assert(file.partitionValues.numFields == partitionSchema.size)
val filePath = file.toPath
val split = new FileSplit(filePath, file.start, file.length, Array.empty[String])
+ val schemaEvolutionUtils = new Spark32PlusParquetSchemaEvolutionUtils(sharedConf, filePath, requiredSchema, partitionSchema)
+
val fileFooter = if (enableVectorizedReader) {
// When there are vectorized reads, we can avoid reading the footer twice by reading
// all row groups in advance and filter row groups according to filters that require
@@ -116,7 +117,7 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean,
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
- filters
+ filters.map(schemaEvolutionUtils.rebuildFilterFromParquet)
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -144,7 +145,7 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean,
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val hadoopAttemptContext =
- new TaskAttemptContextImpl(sharedConf, attemptId)
+ new TaskAttemptContextImpl(schemaEvolutionUtils.getHadoopConfClone(footerFileMetaData, enableVectorizedReader), attemptId)
// Try to push down filters when filter push-down is enabled.
// Notice: This push-down is RowGroups level, not individual records.
@@ -153,7 +154,7 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean,
}
val taskContext = Option(TaskContext.get())
if (enableVectorizedReader) {
- val vectorizedReader = new VectorizedParquetRecordReader(
+ val vectorizedReader = schemaEvolutionUtils.buildVectorizedReader(
convertTz.orNull,
datetimeRebaseSpec.mode.toString,
datetimeRebaseSpec.timeZone,
@@ -205,7 +206,7 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean,
readerWithRowIndexes.initialize(split, hadoopAttemptContext)
val fullSchema = toAttributes(requiredSchema) ++ toAttributes(partitionSchema)
- val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
+ val unsafeProjection = schemaEvolutionUtils.generateUnsafeProjection(fullSchema, timeZoneId)
if (partitionSchema.length == 0) {
// There is no partition columns