yihua commented on code in PR #20077:
URL: https://github.com/apache/hudi/pull/20077#discussion_r4169418686
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -356,97 +339,35 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
}
val broadcastedStorageConf = spark.sparkContext.broadcast(new
SerializableConfiguration(augmentedStorageConf.unwrap()))
- val fileIndexProps: TypedProperties =
HoodieFileIndex.getConfigProperties(spark, options, null)
+ val cdcProps: TypedProperties = HoodieFileIndex.getConfigProperties(spark,
options, null)
+ cdcProps.setProperty(HoodieTableConfig.HOODIE_TABLE_NAME_KEY, tableName)
Review Comment:
Why is this needed now?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala:
##########
@@ -0,0 +1,374 @@
+/*
+ * 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.hudi.{HoodiePartitionCDCFileGroupMapping,
HoodiePartitionFileSliceMapping, HoodieTableSchema, SparkAdapterSupport,
SparkFileFormatInternalRowReaderContext}
+import org.apache.hudi.cdc.{CDCFileGroupIterator, HoodieCDCFileGroupSplit,
HoodieCDCFileIndex}
+import org.apache.hudi.common.config.{HoodieReaderConfig, TypedProperties}
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.schema.HoodieSchema
+import org.apache.hudi.common.schema.internal.InternalSchema
+import org.apache.hudi.common.table.{HoodieTableMetaClient,
ParquetTableSchemaResolver}
+import org.apache.hudi.common.table.log.InstantRange
+import org.apache.hudi.common.table.read.{HoodieFileGroupReader,
HoodieRecordReader}
+import org.apache.hudi.common.table.read.lsm.{HoodieLsmFileGroupReader,
LsmReaderUtils}
+import org.apache.hudi.common.util.{ConfigUtils, Option => HOption}
+import org.apache.hudi.common.util.collection.ClosableIterator
+import org.apache.hudi.data.CloseableIteratorListener
+import
org.apache.hudi.io.storage.HoodieSparkParquetReader.ENABLE_LOGICAL_TIMESTAMP_REPAIR
+import org.apache.hudi.io.storage.VectorConversionUtils
+import org.apache.hudi.storage.StorageConfiguration
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.parquet.schema.MessageType
+import org.apache.spark.SparkEnv
+import org.apache.spark.broadcast.Broadcast
+import
org.apache.spark.sql.HoodieCatalystExpressionUtils.generateUnsafeProjection
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{JoinedRow, UnsafeProjection}
+import org.apache.spark.sql.execution.datasources.{PartitionedFile,
SparkColumnarFileReader, SparkSchemaTransformUtils}
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnarBatchUtils}
+import org.apache.spark.util.{SerializableConfiguration, Utils}
+
+import java.io.Closeable
+import java.nio.ByteBuffer
+
+import scala.collection.JavaConverters.mapAsJavaMapConverter
+import scala.reflect.ClassTag
+
+/**
+ * Read-only state of one scan of [[HoodieFileGroupReaderBasedFileFormat]],
built on the driver and shared by every
+ * task of an executor through [[HoodieFileGroupReaderFunction]]. Executors
only fill thread-safe lazy caches in it;
+ * per-file state such as reader properties is copied before use.
+ */
+private[parquet] class HoodieFileGroupReadState(val metaClient:
HoodieTableMetaClient,
+ val tableSchema:
HoodieTableSchema,
+ val queryTimestamp: String,
+ val readerProps:
TypedProperties,
+ val cdcProps: TypedProperties,
+ val dataSchema: HoodieSchema,
+ val requestedSchema:
HoodieSchema,
+ val internalSchemaOpt:
HOption[InternalSchema],
+ val instantRangeOpt:
HOption[InstantRange],
+ val shouldUseRecordPosition:
Boolean,
+ val isCount: Boolean,
+ val filters: Seq[Filter],
+ val requiredFilters:
Seq[Filter],
+ val requiredSchema:
StructType,
+ val partitionSchema:
StructType,
+ val remainingPartitionSchema:
StructType,
+ val fixedPartitionIndexes:
Set[Int],
+ val outputSchema: StructType,
+ val projectionInputSchema:
StructType,
+ val baseFileReadSchemas:
BaseFileReadSchemas) extends Serializable {
+
+ /**
+ * Parquet form of the table schema for base file reads. The conversion
reads Hadoop's default resources, so it is
+ * done once per executor rather than per task or file.
+ */
+ @transient lazy val tableSchemaAsMessageType: HOption[MessageType] =
+
HOption.ofNullable(ParquetTableSchemaResolver.convertAvroSchemaToParquet(tableSchema.schema,
new Configuration()))
+}
+
+/**
+ * Base file read schemas with VECTOR columns rewritten to BinaryType, plus
the ordinals of those columns.
+ */
+private[parquet] case class BaseFileReadSchemas(readRequiredSchema: StructType,
+ readVectorColumns: Map[Int,
HoodieSchema.Vector],
+ outputSchema: StructType,
+ outputVectorColumns: Map[Int,
HoodieSchema.Vector],
+ requestedSchema: StructType)
+
+/**
+ * Holds a value as Java-serialized bytes and deserializes it at most once per
JVM instance of the holder.
+ *
+ * <p>Broadcasting this holder instead of the value keeps the broadcast
independent of `spark.serializer`: Kryo
+ * would otherwise serialize the value field by field and ignore the custom
Java serialization of types such as
+ * [[HoodieSchema]] and [[HadoopStorageConfiguration]].
+ */
+private[parquet] class JavaSerializedValue[T: ClassTag] private(bytes:
Array[Byte]) extends Serializable {
+
+ @transient lazy val value: T = SparkEnv.get.closureSerializer.newInstance()
+ .deserialize[T](ByteBuffer.wrap(bytes), Utils.getContextOrSparkClassLoader)
+}
+
+private[parquet] object JavaSerializedValue {
+
+ def apply[T: ClassTag](value: T): JavaSerializedValue[T] = {
+ val buffer = SparkEnv.get.closureSerializer.newInstance().serialize(value)
+ val bytes = new Array[Byte](buffer.remaining())
+ buffer.get(bytes)
+ new JavaSerializedValue[T](bytes)
+ }
+}
+
+/**
+ * The per-file read function returned by
[[HoodieFileGroupReaderBasedFileFormat.buildReaderWithPartitionValues]].
+ *
+ * <p>Spark deserializes this function once per task, so it holds only
broadcast handles. The scan state behind
+ * `state` is deserialized once per executor.
+ */
+private[parquet] class HoodieFileGroupReaderFunction(baseFileReader:
Broadcast[SparkColumnarFileReader],
+ fileGroupBaseFileReader:
Broadcast[SparkColumnarFileReader],
+ storageConf:
Broadcast[SerializableConfiguration],
+ state:
Broadcast[JavaSerializedValue[HoodieFileGroupReadState]])
+ extends (PartitionedFile => Iterator[InternalRow]) with Serializable {
+
+ import HoodieFileGroupReaderFunction._
+
+ private def sparkAdapter = SparkAdapterSupport.sparkAdapter
+
+ override def apply(file: PartitionedFile): Iterator[InternalRow] = {
+ val s = state.value.value
+ val conf = new HadoopStorageConfiguration(storageConf.value.value)
+ val iter = file.partitionValues match {
+ // Snapshot or incremental queries.
+ case fileSliceMapping: HoodiePartitionFileSliceMapping =>
+ val fileGroupName = FSUtils.getFileIdFromFilePath(sparkAdapter
+ .getSparkPartitionedFileUtils.getPathFromPartitionedFile(file))
+ fileSliceMapping.getSlice(fileGroupName) match {
+ case Some(fileSlice) if !s.isCount && (s.requiredSchema.nonEmpty ||
fileSlice.getLogFiles.findAny().isPresent) =>
+ val tableConfig = s.metaClient.getTableConfig
+ // requiredFilters preserve Spark's row-level filtering semantics,
while instantRangeOpt
+ // keeps out-of-range records from participating in the file-group
merge itself.
+ val readerContext = new SparkFileFormatInternalRowReaderContext(
+ fileGroupBaseFileReader.value, s.filters, s.requiredFilters,
conf, tableConfig,
+ sparkRequiredSchema = Some(s.requiredSchema), instantRangeOpt =
s.instantRangeOpt)
+
readerContext.enableLogicalTimestampFieldRepair(conf.getBoolean(ENABLE_LOGICAL_TIMESTAMP_REPAIR,
true))
+ val props = TypedProperties.copy(s.readerProps)
+ val baseFileLength = if (fileSlice.getBaseFile.isPresent) {
+ fileSlice.getBaseFile.get.getFileSize
+ } else {
+ 0
+ }
+ val reader: HoodieRecordReader[InternalRow] =
+ if (LsmReaderUtils.shouldUseLsmReader(
+ tableConfig,
+ ConfigUtils.getStringWithAltKeys(props,
HoodieReaderConfig.MERGE_TYPE, true))) {
+ HoodieLsmFileGroupReader.builder[InternalRow]()
+ .withReaderContext(readerContext)
+ .withHoodieTableMetaClient(s.metaClient)
+ .withLatestCommitTime(s.queryTimestamp)
+ .withBaseFileOption(fileSlice.getBaseFile)
+ .withLogFiles(fileSlice.getLogFiles)
+ .withPartitionPath(fileSlice.getPartitionPath)
+ .withDataSchema(s.dataSchema)
+ .withRequestedSchema(s.requestedSchema)
+ .withInternalSchemaOpt(s.internalSchemaOpt)
+ .withProps(props)
+ .withStart(file.start)
+ .withLength(baseFileLength)
+ .build()
+ } else {
+ HoodieFileGroupReader.builder[InternalRow]()
+ .withReaderContext(readerContext)
+ .withHoodieTableMetaClient(s.metaClient)
+ .withLatestCommitTime(s.queryTimestamp)
+ .withBaseFileOption(fileSlice.getBaseFile)
+ .withLogFiles(fileSlice.getLogFiles)
+ .withPartitionPath(fileSlice.getPartitionPath)
+ .withDataSchema(s.dataSchema)
+ .withRequestedSchema(s.requestedSchema)
+ .withInternalSchemaOpt(s.internalSchemaOpt)
+ .withProps(props)
+ .withStart(file.start)
+ .withLength(baseFileLength)
+ .withShouldUseRecordPosition(s.shouldUseRecordPosition)
+ .build()
+ }
+ // Append partition values to rows and project to output schema
+ appendPartitionAndProject(
+ reader.getClosableIterator,
+ s.projectionInputSchema,
+ s.remainingPartitionSchema,
+ s.outputSchema,
+ fileSliceMapping.getPartitionValues,
+ s.fixedPartitionIndexes)
+
+ case _ =>
+ readBaseFile(file, s, conf)
+ }
+ // CDC queries.
+ case cdcFileGroupMapping: HoodiePartitionCDCFileGroupMapping =>
+ new CDCFileGroupIterator(
+ HoodieCDCFileGroupSplit(cdcFileGroupMapping.getFileSplits().toArray),
+ s.metaClient,
+ conf,
+ fileGroupBaseFileReader.value,
+ s.tableSchema,
+ HoodieCDCFileIndex.FULL_CDC_SPARK_SCHEMA,
+ s.requiredSchema,
+ TypedProperties.copy(s.cdcProps))
Review Comment:
Is the copy here still needed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
