You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2025/05/24 09:09:51 UTC
(hudi) branch master updated: [HUDI-9302] Enable vectorized reading for file slice without log file (#13127)
This is an automated email from the ASF dual-hosted git repository.
danny0405 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 d9822b8dd03 [HUDI-9302] Enable vectorized reading for file slice without log file (#13127)
d9822b8dd03 is described below
commit d9822b8dd03d173a9cae5f0cdebfd36bb732748f
Author: TheR1sing3un <ch...@apache.org>
AuthorDate: Sat May 24 17:09:45 2025 +0800
[HUDI-9302] Enable vectorized reading for file slice without log file (#13127)
* enable vectorized reading when only base file is in the file group.
---------
Signed-off-by: TheR1sing3un <ch...@apache.org>
Co-authored-by: danny0405 <yu...@gmail.com>
---
...odieFileGroupReaderBasedParquetFileFormat.scala | 47 ++++++++++++++++------
.../hudi/TestAvroSchemaResolutionSupport.scala | 9 +----
.../datasources/parquet/Spark33ParquetReader.scala | 13 +++++-
3 files changed, 48 insertions(+), 21 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedParquetFileFormat.scala
index afab0f391bc..47c5dafa872 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedParquetFileFormat.scala
@@ -82,18 +82,33 @@ class HoodieFileGroupReaderBasedParquetFileFormat(tablePath: String,
private val sanitizedTableName = AvroSchemaUtils.getAvroRecordQualifiedName(tableName)
/**
- * Support batch needs to remain consistent, even if one side of a bootstrap merge can support
- * while the other side can't
+ * Flag saying whether vectorized reading is supported.
*/
- private var supportBatchCalled = false
- private var supportBatchResult = false
+ private var supportVectorizedRead = false
+ /**
+ * Flag saying whether batch output is supported.
+ */
+ private var supportReturningBatch = false
+
+ /**
+ * Checks if the file format supports vectorized reading, please refer to SPARK-40918.
+ *
+ * NOTE: for mor read, even for file-slice with only base file, we can read parquet file with vectorized read,
+ * but the return result of the whole data-source-scan phase cannot be batch,
+ * because when there are any log file in a file slice, it needs to be read by the file group reader.
+ * Since we are currently performing merges based on rows, the result returned by merging should be based on rows,
+ * we cannot assume that all file slices have only base files.
+ * So we need to set the batch result back to false.
+ *
+ */
override def supportBatch(sparkSession: SparkSession, schema: StructType): Boolean = {
- if (!supportBatchCalled || supportBatchResult) {
- supportBatchCalled = true
- supportBatchResult = !isMOR && !isIncremental && !isBootstrap && super.supportBatch(sparkSession, schema)
- }
- supportBatchResult
+ val superSupportBatch = super.supportBatch(sparkSession, schema)
+ supportVectorizedRead = !isIncremental && !isBootstrap && superSupportBatch
+ supportReturningBatch = !isMOR && supportVectorizedRead
+ logInfo(s"supportReturningBatch: $supportReturningBatch, supportVectorizedRead: $supportVectorizedRead, isIncremental: $isIncremental, " +
+ s"isBootstrap: $isBootstrap, superSupportBatch: $superSupportBatch")
+ supportReturningBatch
}
//for partition columns that we read from the file, we don't want them to be constant column vectors so we
@@ -155,8 +170,16 @@ class HoodieFileGroupReaderBasedParquetFileFormat(tablePath: String,
val requestedSchema = StructType(requiredSchema.fields ++ partitionSchema.fields.filter(f => mandatoryFields.contains(f.name)))
val requestedAvroSchema = AvroConversionUtils.convertStructTypeToAvroSchema(requestedSchema, sanitizedTableName)
val dataAvroSchema = AvroConversionUtils.convertStructTypeToAvroSchema(dataSchema, sanitizedTableName)
- val parquetFileReader = spark.sparkContext.broadcast(sparkAdapter.createParquetFileReader(supportBatchResult,
+ val parquetFileReader = spark.sparkContext.broadcast(sparkAdapter.createParquetFileReader(supportVectorizedRead,
spark.sessionState.conf, options, augmentedStorageConf.unwrap()))
+ val fileGroupParquetFileReader = if (isMOR && supportVectorizedRead) {
+ // for file group reader to perform read, we always need to read the record without vectorized reader because our merging is based on row level.
+ // TODO: please consider to support vectorized reader in file group reader
+ spark.sparkContext.broadcast(sparkAdapter.createParquetFileReader(vectorized = false,
+ spark.sessionState.conf, options, augmentedStorageConf.unwrap()))
+ } else {
+ parquetFileReader
+ }
val broadcastedStorageConf = spark.sparkContext.broadcast(new SerializableConfiguration(augmentedStorageConf.unwrap()))
val fileIndexProps: TypedProperties = HoodieFileIndex.getConfigProperties(spark, options, null)
@@ -174,7 +197,7 @@ class HoodieFileGroupReaderBasedParquetFileFormat(tablePath: String,
case Some(fileSlice) if !isCount && (requiredSchema.nonEmpty || fileSlice.getLogFiles.findAny().isPresent) =>
val metaClient: HoodieTableMetaClient = HoodieTableMetaClient
.builder().setConf(storageConf).setBasePath(tablePath).build
- val readerContext = new SparkFileFormatInternalRowReaderContext(parquetFileReader.value, filters, requiredFilters, storageConf, metaClient.getTableConfig)
+ val readerContext = new SparkFileFormatInternalRowReaderContext(fileGroupParquetFileReader.value, filters, requiredFilters, storageConf, metaClient.getTableConfig)
val props = metaClient.getTableConfig.getProps
options.foreach(kv => props.setProperty(kv._1, kv._2))
props.put(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(), String.valueOf(maxMemoryPerCompaction))
@@ -211,7 +234,7 @@ class HoodieFileGroupReaderBasedParquetFileFormat(tablePath: String,
}
// CDC queries.
case hoodiePartitionCDCFileGroupSliceMapping: HoodiePartitionCDCFileGroupMapping =>
- buildCDCRecordIterator(hoodiePartitionCDCFileGroupSliceMapping, parquetFileReader.value, storageConf, fileIndexProps, requiredSchema)
+ buildCDCRecordIterator(hoodiePartitionCDCFileGroupSliceMapping, fileGroupParquetFileReader.value, storageConf, fileIndexProps, requiredSchema)
case _ =>
readBaseFile(file, parquetFileReader.value, requestedSchema, remainingPartitionSchema, fixedPartitionIndexes,
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestAvroSchemaResolutionSupport.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestAvroSchemaResolutionSupport.scala
index 8eb4c0aee85..2929534ffa7 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestAvroSchemaResolutionSupport.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestAvroSchemaResolutionSupport.scala
@@ -872,14 +872,7 @@ class TestAvroSchemaResolutionSupport extends HoodieClientTestBase with ScalaAss
upsertData(df2, tempRecordPath, isCow)
// after implicit type change, read the table with vectorized read enabled
- //fg reader with mor does not support vectorized currently and will auto read by row
- if (isCow || !useFileGroupReader) {
- assertThrows(classOf[SparkException]){
- withSQLConf("spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true") {
- readTable(tempRecordPath, useFileGroupReader)
- }
- }
- } else {
+ assertThrows(classOf[SparkException]){
withSQLConf("spark.sql.parquet.enableNestedColumnVectorizedReader" -> "true") {
readTable(tempRecordPath, useFileGroupReader)
}
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 a4d5adda191..1f49da7352f 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
@@ -252,6 +252,17 @@ object Spark33ParquetReader extends SparkParquetReaderBuilder {
sqlConf.getConfString("spark.sql.legacy.parquet.nanosAsLong", "false").toBoolean
)
+ // Should always be set by FileSourceScanExec while creating this.
+ // Check conf before checking the option, to allow working around an issue by changing conf.
+ val returningBatch = sqlConf.parquetVectorizedReaderEnabled &&
+ options.get(FileFormat.OPTION_RETURNING_BATCH)
+ .getOrElse {
+ throw new IllegalArgumentException(
+ "OPTION_RETURNING_BATCH should always be set for ParquetFileFormat. " +
+ "To workaround this issue, set spark.sql.parquet.enableVectorizedReader=false.")
+ }
+ .equals("true")
+
val parquetOptions = new ParquetOptions(options, sqlConf)
new Spark33ParquetReader(enableVectorizedReader = vectorized,
datetimeRebaseModeInRead = parquetOptions.datetimeRebaseModeInRead,
@@ -266,7 +277,7 @@ object Spark33ParquetReader extends SparkParquetReaderBuilder {
timestampConversion = sqlConf.isParquetINT96TimestampConversion,
enableOffHeapColumnVector = sqlConf.offHeapColumnVectorEnabled,
capacity = sqlConf.parquetVectorizedReaderBatchSize,
- returningBatch = sqlConf.parquetVectorizedReaderEnabled,
+ returningBatch = returningBatch,
enableRecordFilter = sqlConf.parquetRecordFilterEnabled,
timeZoneId = Some(sqlConf.sessionLocalTimeZone))
}