This is an automated email from the ASF dual-hosted git repository.
yihua 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 2fc6486f9b29 perf(spark): ship file group reader scan state once per
executor (#20077)
2fc6486f9b29 is described below
commit 2fc6486f9b298f40a232c062da3bb461de52caa3
Author: Y Ethan Guo <[email protected]>
AuthorDate: Mon Oct 5 14:47:14 2026 -0700
perf(spark): ship file group reader scan state once per executor (#20077)
---
.../hudi/common/table/HoodieTableMetaClient.java | 4 +-
.../common/table/TestHoodieTableMetaClient.java | 41 +++
.../hudi/HoodieHadoopFsRelationFactory.scala | 2 +-
.../HoodieFileGroupReaderBasedFileFormat.scala | 310 ++---------------
.../parquet/HoodieFileGroupReaderFunction.scala | 376 +++++++++++++++++++++
.../TestReadOptionsDoNotOverrideTableConfig.java | 247 ++++++++++++++
.../TestHoodieFileGroupReaderFunction.scala | 112 ++++++
7 files changed, 812 insertions(+), 280 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
index d7b120828a94..87218cae4f40 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
@@ -167,7 +167,8 @@ public class HoodieTableMetaClient implements Serializable {
// In-memory cache for archived timeline based on the start instant time
// Only one entry should be present in this map
- private final Map<String, HoodieArchivedTimeline> archivedTimelineMap = new
HashMap<>();
+ // Not serialized, so a copy shipped to an executor does not carry the
loaded archived instants and their details
+ private transient Map<String, HoodieArchivedTimeline> archivedTimelineMap =
new HashMap<>();
protected StoragePath basePath;
protected StoragePath metaPath;
@@ -388,6 +389,7 @@ public class HoodieTableMetaClient implements Serializable {
in.defaultReadObject();
storage = null; // will be lazily initialized
+ archivedTimelineMap = new HashMap<>();
}
private void writeObject(java.io.ObjectOutputStream out) throws IOException {
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
index 2754dc35f8bd..6ee96663cd63 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.model.HoodieIndexMetadata;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.MetaFieldsMode;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
@@ -43,20 +44,27 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -106,6 +114,39 @@ class TestHoodieTableMetaClient extends
HoodieCommonTestHarness {
assertEquals("val",
metaClient.getActiveTimeline().readCommitMetadata(completedInstant).getExtraMetadata().get("key"));
}
+ /**
+ * The archived timeline is cached by the meta client that loaded it, and a
serialized copy (as shipped to
+ * executors) must not carry it, since it holds every loaded archived
instant and its details.
+ */
+ @Test
+ void testArchivedTimelineCacheIsNotSerialized() throws IOException,
ClassNotFoundException {
+ HoodieArchivedTimeline archivedTimeline = metaClient.getArchivedTimeline();
+ assertSame(archivedTimeline, metaClient.getArchivedTimeline());
+
+ Set<Class<?>> serializedClasses = new HashSet<>();
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream out = new ObjectOutputStream(bytes) {
+ {
+ enableReplaceObject(true);
+ }
+
+ @Override
+ protected Object replaceObject(Object obj) {
+ serializedClasses.add(obj.getClass());
+ return obj;
+ }
+ }) {
+ out.writeObject(metaClient);
+ }
+
assertTrue(serializedClasses.stream().noneMatch(HoodieArchivedTimeline.class::isAssignableFrom),
+ "The archived timeline was serialized with the meta client: " +
serializedClasses);
+
+ try (ObjectInputStream in = new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray()))) {
+ HoodieTableMetaClient deserializedMetaClient = (HoodieTableMetaClient)
in.readObject();
+ assertNotNull(deserializedMetaClient.getArchivedTimeline());
+ }
+ }
+
@Test
void testCommitTimeline() throws IOException {
HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
index a77329b857bb..a4e8f18bf499 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
@@ -247,7 +247,7 @@ abstract class HoodieBaseHadoopFsRelationFactory(val
sqlContext: SQLContext,
HoodieTableSchema(tableStructSchema, tableSchema, internalSchemaOpt),
tableConfig.getTableName, queryTimestamp.get, getMandatoryFields, isMOR,
isBootstrap,
isIncremental, validCommits, shouldUseRecordPosition, getRequiredFilters,
- tableConfig.isMultipleBaseFileFormatsEnabled,
tableConfig.getBaseFileFormat, getInstantRange)
+ tableConfig.isMultipleBaseFileFormatsEnabled,
tableConfig.getBaseFileFormat, getInstantRange, Some(metaClient))
}
override def buildBucketSpec(): Option[BucketSpec] = None
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index 241bdff63178..7a0be7d5b538 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -17,41 +17,32 @@
package org.apache.spark.sql.execution.datasources.parquet
-import org.apache.hudi.{HoodieFileIndex, HoodiePartitionCDCFileGroupMapping,
HoodiePartitionFileSliceMapping, HoodieSchemaConversionUtils, HoodieSparkUtils,
HoodieTableSchema, SparkAdapterSupport, SparkFileFormatInternalRowReaderContext}
-import org.apache.hudi.cdc.{CDCFileGroupIterator, HoodieCDCFileGroupSplit,
HoodieCDCFileIndex}
+import org.apache.hudi.{HoodieFileIndex, HoodieSchemaConversionUtils,
HoodieSparkUtils, HoodieTableSchema, SparkAdapterSupport,
SparkFileFormatInternalRowReaderContext}
import org.apache.hudi.client.common.HoodieSparkEngineContext
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
-import org.apache.hudi.common.config.{HoodieMemoryConfig, HoodieReaderConfig,
TypedProperties}
-import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.config.{HoodieMemoryConfig, TypedProperties}
import org.apache.hudi.common.model.HoodieFileFormat
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.HoodieSchemaRepair
import org.apache.hudi.common.schema.HoodieSchemaUtils
import org.apache.hudi.common.schema.internal.InternalSchema
-import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient,
ParquetTableSchemaResolver}
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
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.common.table.read.HoodieFileGroupReader
+import org.apache.hudi.common.util.{Option => HOption}
import org.apache.hudi.exception.HoodieNotSupportedException
import org.apache.hudi.io.MergeUtils
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.hadoop.fs.{FileStatus, Path}
import org.apache.hadoop.mapreduce.Job
-import org.apache.parquet.schema.MessageType
import org.apache.spark.api.java.JavaSparkContext
import org.apache.spark.internal.Logging
-import
org.apache.spark.sql.HoodieCatalystExpressionUtils.generateUnsafeProjection
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{JoinedRow, UnsafeProjection}
import org.apache.spark.sql.execution.datasources.{OutputWriterFactory,
PartitionedFile, SparkColumnarFileReader, SparkSchemaTransformUtils}
import org.apache.spark.sql.execution.datasources.orc.OrcUtils
import org.apache.spark.sql.execution.vectorized.{OffHeapColumnVector,
OnHeapColumnVector}
@@ -59,11 +50,8 @@ import
org.apache.spark.sql.hudi.MultipleColumnarFileFormatReader
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
-import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnarBatchUtils}
import org.apache.spark.util.SerializableConfiguration
-import java.io.Closeable
-
import scala.collection.JavaConverters.mapAsJavaMapConverter
trait HoodieFormatTrait {
@@ -94,17 +82,12 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
requiredFilters: Seq[Filter],
isMultipleBaseFileFormatsEnabled:
Boolean,
hoodieFileFormat: HoodieFileFormat,
- instantRangeOpt:
HOption[InstantRange] = HOption.empty())
+ instantRangeOpt:
HOption[InstantRange] = HOption.empty(),
+ @transient tableMetaClient:
Option[HoodieTableMetaClient] = None)
extends ParquetFileFormat with SparkAdapterSupport with HoodieFormatTrait
with Logging with Serializable {
private lazy val schema = tableSchema.schema
- private lazy val tableSchemaAsMessageType: HOption[MessageType] = {
- HOption.ofNullable(
- ParquetTableSchemaResolver.convertAvroSchemaToParquet(schema, new
Configuration())
- )
- }
-
private lazy val hasTimestampMillisFieldInTableSchema =
HoodieSchemaRepair.hasTimestampMillisField(schema)
private lazy val supportBatchWithTableSchema = HoodieSparkUtils.gteqSpark3_5
|| !hasTimestampMillisFieldInTableSchema
override def shortName(): String = "HudiFileGroup"
@@ -127,7 +110,7 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
/**
* Cached result of vector column detection keyed by schema identity.
- * Avoids re-parsing metadata on repeated supportBatch / readBaseFile calls
with the same schema.
+ * Avoids re-parsing metadata on repeated supportBatch calls with the same
schema.
*/
@transient private var cachedVectorDetection: (StructType, Map[Int,
HoodieSchema.Vector]) = _
@@ -358,97 +341,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)
val engineContext = new HoodieSparkEngineContext(new
JavaSparkContext(spark.sparkContext))
val maxMemoryPerCompaction =
MergeUtils.getMaxMemoryPerCompaction(engineContext.getTaskContextSupplier,
options.asJava)
- // Create metaclient on driver to avoid expensive operations on executors
- val metaClient: HoodieTableMetaClient = HoodieTableMetaClient
- .builder().setConf(augmentedStorageConf).setBasePath(tablePath).build
-
- (file: PartitionedFile) => {
- // executor
- val storageConf = new
HadoopStorageConfiguration(broadcastedStorageConf.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 !isCount && (requiredSchema.nonEmpty ||
fileSlice.getLogFiles.findAny().isPresent) =>
- // 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, filters, requiredFilters,
storageConf, metaClient.getTableConfig,
- sparkRequiredSchema = Some(requiredSchema), instantRangeOpt =
instantRangeOpt)
-
readerContext.enableLogicalTimestampFieldRepair(storageConf.getBoolean(ENABLE_LOGICAL_TIMESTAMP_REPAIR,
true))
- 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))
- val baseFileLength = if (fileSlice.getBaseFile.isPresent) {
- fileSlice.getBaseFile.get.getFileSize
- } else {
- 0
- }
- val reader: HoodieRecordReader[InternalRow] =
- if (LsmReaderUtils.shouldUseLsmReader(
- metaClient.getTableConfig,
- ConfigUtils.getStringWithAltKeys(props,
HoodieReaderConfig.MERGE_TYPE, true))) {
- HoodieLsmFileGroupReader.builder[InternalRow]()
- .withReaderContext(readerContext)
- .withHoodieTableMetaClient(metaClient)
- .withLatestCommitTime(queryTimestamp)
- .withBaseFileOption(fileSlice.getBaseFile)
- .withLogFiles(fileSlice.getLogFiles)
- .withPartitionPath(fileSlice.getPartitionPath)
- .withDataSchema(dataSchema)
- .withRequestedSchema(requestedSchema)
- .withInternalSchemaOpt(internalSchemaOpt)
- .withProps(props)
- .withStart(file.start)
- .withLength(baseFileLength)
- .build()
- } else {
- HoodieFileGroupReader.builder[InternalRow]()
- .withReaderContext(readerContext)
- .withHoodieTableMetaClient(metaClient)
- .withLatestCommitTime(queryTimestamp)
- .withBaseFileOption(fileSlice.getBaseFile)
- .withLogFiles(fileSlice.getLogFiles)
- .withPartitionPath(fileSlice.getPartitionPath)
- .withDataSchema(dataSchema)
- .withRequestedSchema(requestedSchema)
- .withInternalSchemaOpt(internalSchemaOpt)
- .withProps(props)
- .withStart(file.start)
- .withLength(baseFileLength)
- .withShouldUseRecordPosition(shouldUseRecordPosition)
- .build()
- }
- // Append partition values to rows and project to output schema
- appendPartitionAndProject(
- reader.getClosableIterator,
- projectionInputSchema,
- remainingPartitionSchema,
- outputSchema,
- fileSliceMapping.getPartitionValues,
- fixedPartitionIndexes)
-
- case _ =>
- readBaseFile(file, baseFileReader.value, requestedStructType,
remainingPartitionSchema, fixedPartitionIndexes,
- readRequiredSchema, partitionSchema, outputSchema, filters ++
requiredFilters, storageConf)
- }
- // CDC queries.
- case hoodiePartitionCDCFileGroupSliceMapping:
HoodiePartitionCDCFileGroupMapping =>
- buildCDCRecordIterator(hoodiePartitionCDCFileGroupSliceMapping,
fileGroupBaseFileReader.value, storageConf, fileIndexProps, requiredSchema,
metaClient)
-
- case _ =>
- readBaseFile(file, baseFileReader.value, requestedStructType,
remainingPartitionSchema, fixedPartitionIndexes,
- readRequiredSchema, partitionSchema, outputSchema, filters ++
requiredFilters, storageConf)
- }
- CloseableIteratorListener.addListener(iter)
+ // The relation's meta client carries the timeline the scan was planned
against; build one only when the
+ // format is used without a relation. The field is null rather than None
on a deserialized format.
+ val metaClient: HoodieTableMetaClient =
Option(tableMetaClient).flatten.getOrElse(HoodieTableMetaClient
+ .builder().setConf(augmentedStorageConf).setBasePath(tablePath).build)
+ val readerProps = TypedProperties.copy(metaClient.getTableConfig.getProps)
+ options.foreach(kv => readerProps.setProperty(kv._1, kv._2))
+ readerProps.put(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(),
String.valueOf(maxMemoryPerCompaction))
+
+ val (baseReadRequiredSchema, readVectorColumns) =
withVectorRewrite(readRequiredSchema)
+ val baseFileReadSchemas = if (readVectorColumns.nonEmpty) {
+ val (baseOutputSchema, outputVectorColumns) =
withVectorRewrite(outputSchema)
+ BaseFileReadSchemas(baseReadRequiredSchema, readVectorColumns,
baseOutputSchema, outputVectorColumns,
+ withVectorRewrite(requestedStructType)._1)
+ } else {
+ BaseFileReadSchemas(baseReadRequiredSchema, readVectorColumns,
outputSchema, Map.empty, requestedStructType)
}
+
+ val state = new HoodieFileGroupReadState(metaClient, tableSchema,
queryTimestamp, readerProps, cdcProps,
+ dataSchema, requestedSchema, internalSchemaOpt, instantRangeOpt,
shouldUseRecordPosition, isCount, filters,
+ requiredFilters, requiredSchema, partitionSchema,
remainingPartitionSchema, fixedPartitionIndexes, outputSchema,
+ projectionInputSchema, baseFileReadSchemas)
+ new HoodieFileGroupReaderFunction(baseFileReader, fileGroupBaseFileReader,
broadcastedStorageConf,
+ spark.sparkContext.broadcast(JavaSerializedValue(state)))
}
private def buildBaseFileReader(spark: SparkSession,
@@ -482,85 +403,6 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
}
}
- private def buildCDCRecordIterator(hoodiePartitionCDCFileGroupSliceMapping:
HoodiePartitionCDCFileGroupMapping,
- baseFileReader: SparkColumnarFileReader,
- storageConf:
StorageConfiguration[Configuration],
- props: TypedProperties,
- requiredSchema: StructType,
- metaClient: HoodieTableMetaClient):
Iterator[InternalRow] = {
- val fileSplits =
hoodiePartitionCDCFileGroupSliceMapping.getFileSplits().toArray
- val cdcFileGroupSplit: HoodieCDCFileGroupSplit =
HoodieCDCFileGroupSplit(fileSplits)
- props.setProperty(HoodieTableConfig.HOODIE_TABLE_NAME_KEY, tableName)
- val cdcSchema = HoodieCDCFileIndex.FULL_CDC_SPARK_SCHEMA
- new CDCFileGroupIterator(
- cdcFileGroupSplit,
- metaClient,
- storageConf,
- baseFileReader,
- tableSchema,
- cdcSchema,
- requiredSchema,
- props)
- }
-
- private def appendPartitionAndProject(iter: ClosableIterator[InternalRow],
- inputSchema: StructType,
- partitionSchema: StructType,
- to: StructType,
- partitionValues: InternalRow,
- fixedPartitionIndexes: Set[Int]):
Iterator[InternalRow] = {
- if (partitionSchema.isEmpty) {
- //'inputSchema' and 'to' should be the same so the projection will just
be an identity func
- projectSchema(iter, inputSchema, to)
- } else {
- val fixedPartitionValues = if (partitionSchema.length ==
partitionValues.numFields) {
- //need to append all of the partition fields
- partitionValues
- } else {
- //some partition fields read from file, some were not
- getFixedPartitionValues(partitionValues, partitionSchema,
fixedPartitionIndexes)
- }
- val unsafeProjection =
generateOutputProjection(StructType(inputSchema.fields ++
partitionSchema.fields), to)
- val joinedRow = new JoinedRow()
- makeCloseableFileGroupMappingRecordIterator(iter, d =>
unsafeProjection(joinedRow(d, fixedPartitionValues)))
- }
- }
-
- private def projectSchema(iter: ClosableIterator[InternalRow],
- from: StructType,
- to: StructType): Iterator[InternalRow] = {
- val unsafeProjection = generateOutputProjection(from, to)
- makeCloseableFileGroupMappingRecordIterator(iter, d => unsafeProjection(d))
- }
-
- /**
- * The scan's output projection. Stays the by-name top-level projection
unless a column comes out of the
- * reader with nested fields Spark did not ask for (see
`projectionInputSchema` in
- * buildReaderWithPartitionValues), in which case those are dropped by name
at every depth.
- */
- private def generateOutputProjection(from: StructType, to: StructType):
UnsafeProjection = {
- val hasWiderNestedInput = to.fields.exists { f =>
- from.getFieldIndex(f.name).exists(i =>
SparkSchemaTransformUtils.needsNestedPruning(from.fields(i).dataType,
f.dataType))
- }
- if (hasWiderNestedInput) {
- SparkSchemaTransformUtils.generateNestedPruningProjection(from, to)
- } else {
- generateUnsafeProjection(from, to)
- }
- }
-
- private def
makeCloseableFileGroupMappingRecordIterator(closeableFileGroupRecordIterator:
ClosableIterator[InternalRow],
- mappingFunction:
Function[InternalRow, InternalRow]): Iterator[InternalRow] = {
- CloseableIteratorListener.addListener(closeableFileGroupRecordIterator)
- new Iterator[InternalRow] with Closeable {
- override def hasNext: Boolean = closeableFileGroupRecordIterator.hasNext
-
- override def next(): InternalRow =
mappingFunction(closeableFileGroupRecordIterator.next())
-
- override def close(): Unit = closeableFileGroupRecordIterator.close()
- }
- }
-
private def detectVectorColumns(schema: StructType): Map[Int,
HoodieSchema.Vector] =
SparkFileFormatInternalRowReaderContext.detectVectorColumnsFromMetadata(schema)
@@ -593,78 +435,6 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
}
}
- /**
- * Wraps an iterator to convert binary VECTOR columns back to typed arrays.
- * The read schema has BinaryType for vector columns; the target schema has
ArrayType.
- */
- private def wrapWithVectorConversion(iter: Iterator[InternalRow],
- readSchema: StructType,
- targetSchema: StructType,
- vectorCols: Map[Int,
HoodieSchema.Vector]): Iterator[InternalRow] = {
- val vectorProjection = UnsafeProjection.create(targetSchema)
- val javaVectorCols: java.util.Map[Integer, HoodieSchema.Vector] =
- vectorCols.map { case (k, v) => (Integer.valueOf(k), v) }.asJava
- val mapper = VectorConversionUtils.buildRowMapper(readSchema,
javaVectorCols, vectorProjection.apply(_))
- iter.map(mapper.apply(_))
- }
-
- // executor
- private def readBaseFile(file: PartitionedFile, parquetFileReader:
SparkColumnarFileReader, requestedSchema: StructType,
- remainingPartitionSchema: StructType,
fixedPartitionIndexes: Set[Int], readRequiredSchema: StructType,
- partitionSchema: StructType, outputSchema:
StructType, filters: Seq[Filter],
- storageConf: StorageConfiguration[Configuration]):
Iterator[InternalRow] = {
- // Detect vector columns and create modified schemas with BinaryType.
- // Each schema is detected independently because ordinals are relative to
the schema being
- // modified — outputSchema and requestedSchema may have vector columns at
different positions
- // than readRequiredSchema (e.g. when partition columns are interleaved).
- val (modifiedReadRequiredSchema, vectorCols) =
withVectorRewrite(readRequiredSchema)
- val hasVectors = vectorCols.nonEmpty
- val (modifiedOutputSchema, outputVectorCols) = if (hasVectors)
withVectorRewrite(outputSchema) else (outputSchema, Map.empty[Int,
HoodieSchema.Vector])
- val (modifiedRequestedSchema, _) = if (hasVectors)
withVectorRewrite(requestedSchema) else (requestedSchema, Map.empty[Int,
HoodieSchema.Vector])
-
- val rawIter = if (remainingPartitionSchema.fields.length ==
partitionSchema.fields.length) {
- //none of partition fields are read from the file, so the reader will do
the appending for us
- val iter = parquetFileReader.read(file, modifiedReadRequiredSchema,
partitionSchema, internalSchemaOpt, filters, storageConf,
tableSchemaAsMessageType)
- projectIfNeeded(iter, StructType(modifiedReadRequiredSchema.fields ++
partitionSchema.fields), modifiedOutputSchema)
- } else if (remainingPartitionSchema.fields.length == 0) {
- //we read all of the partition fields from the file
- val pfileUtils = sparkAdapter.getSparkPartitionedFileUtils
- //we need to modify the partitioned file so that the partition values
are empty
- val modifiedFile = pfileUtils.createPartitionedFile(InternalRow.empty,
pfileUtils.getPathFromPartitionedFile(file), file.start, file.length)
- val readSchema = StructType(modifiedReadRequiredSchema.fields ++
partitionSchema.fields)
- //and we pass an empty schema for the partition schema
- val iter = parquetFileReader.read(modifiedFile, readSchema, new
StructType(), internalSchemaOpt, filters, storageConf, tableSchemaAsMessageType)
- projectIfNeeded(iter, readSchema, modifiedOutputSchema)
- } else {
- //need to do an additional projection here. The case in mind is that
partition schema is "a,b,c" mandatoryFields is "a,c",
- //then we will read (dataSchema + a + c) and append b. So the final
schema will be (data schema + a + c +b)
- //but expected output is (data schema + a + b + c)
- val pfileUtils = sparkAdapter.getSparkPartitionedFileUtils
- val partitionValues = getFixedPartitionValues(file.partitionValues,
partitionSchema, fixedPartitionIndexes)
- val modifiedFile = pfileUtils.createPartitionedFile(partitionValues,
pfileUtils.getPathFromPartitionedFile(file), file.start, file.length)
- val iter = parquetFileReader.read(modifiedFile, modifiedRequestedSchema,
remainingPartitionSchema, internalSchemaOpt, filters, storageConf,
tableSchemaAsMessageType)
- projectIter(iter, StructType(modifiedRequestedSchema.fields ++
remainingPartitionSchema.fields), modifiedOutputSchema)
- }
-
- if (hasVectors) {
- // The raw iterator has BinaryType for vector columns; convert back to
ArrayType.
- // All branches above produce rows in modifiedOutputSchema: filter-only
columns from
- // readRequiredSchema are projected away by projectIfNeeded/projectIter.
- wrapWithVectorConversion(rawIter, modifiedOutputSchema, outputSchema,
outputVectorCols)
- } else {
- rawIter
- }
- }
-
- private def projectIter(iter: Iterator[Any], from: StructType, to:
StructType): Iterator[InternalRow] = {
- val unsafeProjection = generateUnsafeProjection(from, to)
- val batchProjection = ColumnarBatchUtils.generateProjection(from, to)
- iter.map {
- case ir: InternalRow => unsafeProjection(ir)
- case cb: ColumnarBatch => batchProjection(cb)
- }.asInstanceOf[Iterator[InternalRow]]
- }
-
/**
* A partition column whose name is a nested field path (e.g.
"nested_record.level") cannot be
* read from the data file as a flat top-level column, nor converted into a
top-level Avro field
@@ -673,22 +443,6 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
*/
private def isNestedPartitionField(name: String): Boolean =
name.contains(".")
- /**
- * Projects to `to` only when the read schema was augmented with filter-only
columns;
- * otherwise returns the iterator as is, preserving columnar batches.
- */
- private def projectIfNeeded(iter: Iterator[InternalRow], from: StructType,
to: StructType): Iterator[InternalRow] = {
- if (from.fieldNames.sameElements(to.fieldNames)) {
- iter
- } else {
- projectIter(iter, from, to)
- }
- }
-
- private def getFixedPartitionValues(allPartitionValues: InternalRow,
partitionSchema: StructType, fixedPartitionIndexes: Set[Int]): InternalRow = {
-
InternalRow.fromSeq(allPartitionValues.toSeq(partitionSchema).zipWithIndex.filter(p
=> fixedPartitionIndexes.contains(p._2)).map(p => p._1))
- }
-
override def inferSchema(sparkSession: SparkSession, options: Map[String,
String], files: Seq[FileStatus]): Option[StructType] = {
if (isMultipleBaseFileFormatsEnabled || hoodieFileFormat ==
HoodieFileFormat.PARQUET) {
ParquetUtils.inferSchema(sparkSession, options, files)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala
new file mode 100644
index 000000000000..978111ac60d6
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala
@@ -0,0 +1,376 @@
+/*
+ * 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
+ * `broadcastedState` is deserialized once per executor.
+ */
+private[parquet] class HoodieFileGroupReaderFunction(baseFileReader:
Broadcast[SparkColumnarFileReader],
+ fileGroupBaseFileReader:
Broadcast[SparkColumnarFileReader],
+ storageConf:
Broadcast[SerializableConfiguration],
+ broadcastedState:
Broadcast[JavaSerializedValue[HoodieFileGroupReadState]])
+ extends (PartitionedFile => Iterator[InternalRow]) with Serializable {
+
+ import HoodieFileGroupReaderFunction._
+
+ private def sparkAdapter = SparkAdapterSupport.sparkAdapter
+
+ override def apply(file: PartitionedFile): Iterator[InternalRow] = {
+ val state = broadcastedState.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 !state.isCount &&
(state.requiredSchema.nonEmpty || fileSlice.getLogFiles.findAny().isPresent) =>
+ val tableConfig = state.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, state.filters,
state.requiredFilters, conf, tableConfig,
+ sparkRequiredSchema = Some(state.requiredSchema),
instantRangeOpt = state.instantRangeOpt)
+
readerContext.enableLogicalTimestampFieldRepair(conf.getBoolean(ENABLE_LOGICAL_TIMESTAMP_REPAIR,
true))
+ // The merger factory writes the payload class into the props it
is given, so each file gets its own copy
+ val props = TypedProperties.copy(state.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(state.metaClient)
+ .withLatestCommitTime(state.queryTimestamp)
+ .withBaseFileOption(fileSlice.getBaseFile)
+ .withLogFiles(fileSlice.getLogFiles)
+ .withPartitionPath(fileSlice.getPartitionPath)
+ .withDataSchema(state.dataSchema)
+ .withRequestedSchema(state.requestedSchema)
+ .withInternalSchemaOpt(state.internalSchemaOpt)
+ .withProps(props)
+ .withStart(file.start)
+ .withLength(baseFileLength)
+ .build()
+ } else {
+ HoodieFileGroupReader.builder[InternalRow]()
+ .withReaderContext(readerContext)
+ .withHoodieTableMetaClient(state.metaClient)
+ .withLatestCommitTime(state.queryTimestamp)
+ .withBaseFileOption(fileSlice.getBaseFile)
+ .withLogFiles(fileSlice.getLogFiles)
+ .withPartitionPath(fileSlice.getPartitionPath)
+ .withDataSchema(state.dataSchema)
+ .withRequestedSchema(state.requestedSchema)
+ .withInternalSchemaOpt(state.internalSchemaOpt)
+ .withProps(props)
+ .withStart(file.start)
+ .withLength(baseFileLength)
+ .withShouldUseRecordPosition(state.shouldUseRecordPosition)
+ .build()
+ }
+ // Append partition values to rows and project to output schema
+ appendPartitionAndProject(
+ reader.getClosableIterator,
+ state.projectionInputSchema,
+ state.remainingPartitionSchema,
+ state.outputSchema,
+ fileSliceMapping.getPartitionValues,
+ state.fixedPartitionIndexes)
+
+ case _ =>
+ readBaseFile(file, state, conf)
+ }
+ // CDC queries.
+ case cdcFileGroupMapping: HoodiePartitionCDCFileGroupMapping =>
+ // The CDC iterator hands its props to the merger factory, which
writes the payload class into them
+ new CDCFileGroupIterator(
+ HoodieCDCFileGroupSplit(cdcFileGroupMapping.getFileSplits().toArray),
+ state.metaClient,
+ conf,
+ fileGroupBaseFileReader.value,
+ state.tableSchema,
+ HoodieCDCFileIndex.FULL_CDC_SPARK_SCHEMA,
+ state.requiredSchema,
+ TypedProperties.copy(state.cdcProps))
+
+ case _ =>
+ readBaseFile(file, state, conf)
+ }
+ CloseableIteratorListener.addListener(iter)
+ }
+
+ private def readBaseFile(file: PartitionedFile,
+ state: HoodieFileGroupReadState,
+ conf: StorageConfiguration[Configuration]):
Iterator[InternalRow] = {
+ val schemas = state.baseFileReadSchemas
+ val hasVectors = schemas.readVectorColumns.nonEmpty
+ val parquetFileReader = baseFileReader.value
+ val filters = state.filters ++ state.requiredFilters
+ val partitionSchema = state.partitionSchema
+ val remainingPartitionSchema = state.remainingPartitionSchema
+
+ val rawIter = if (remainingPartitionSchema.fields.length ==
partitionSchema.fields.length) {
+ //none of partition fields are read from the file, so the reader will do
the appending for us
+ val iter = parquetFileReader.read(file, schemas.readRequiredSchema,
partitionSchema, state.internalSchemaOpt, filters, conf,
+ state.tableSchemaAsMessageType)
+ projectIfNeeded(iter, StructType(schemas.readRequiredSchema.fields ++
partitionSchema.fields), schemas.outputSchema)
+ } else if (remainingPartitionSchema.fields.length == 0) {
+ //we read all of the partition fields from the file
+ val pfileUtils = sparkAdapter.getSparkPartitionedFileUtils
+ //we need to modify the partitioned file so that the partition values
are empty
+ val modifiedFile = pfileUtils.createPartitionedFile(InternalRow.empty,
pfileUtils.getPathFromPartitionedFile(file), file.start, file.length)
+ val readSchema = StructType(schemas.readRequiredSchema.fields ++
partitionSchema.fields)
+ //and we pass an empty schema for the partition schema
+ val iter = parquetFileReader.read(modifiedFile, readSchema, new
StructType(), state.internalSchemaOpt, filters, conf,
+ state.tableSchemaAsMessageType)
+ projectIfNeeded(iter, readSchema, schemas.outputSchema)
+ } else {
+ //need to do an additional projection here. The case in mind is that
partition schema is "a,b,c" mandatoryFields is "a,c",
+ //then we will read (dataSchema + a + c) and append b. So the final
schema will be (data schema + a + c +b)
+ //but expected output is (data schema + a + b + c)
+ val pfileUtils = sparkAdapter.getSparkPartitionedFileUtils
+ val partitionValues = getFixedPartitionValues(file.partitionValues,
partitionSchema, state.fixedPartitionIndexes)
+ val modifiedFile = pfileUtils.createPartitionedFile(partitionValues,
pfileUtils.getPathFromPartitionedFile(file), file.start, file.length)
+ val iter = parquetFileReader.read(modifiedFile, schemas.requestedSchema,
remainingPartitionSchema, state.internalSchemaOpt, filters, conf,
+ state.tableSchemaAsMessageType)
+ projectIter(iter, StructType(schemas.requestedSchema.fields ++
remainingPartitionSchema.fields), schemas.outputSchema)
+ }
+
+ if (hasVectors) {
+ // The raw iterator has BinaryType for vector columns; convert back to
ArrayType.
+ // All branches above produce rows in the rewritten output schema:
filter-only columns from
+ // readRequiredSchema are projected away by projectIfNeeded/projectIter.
+ wrapWithVectorConversion(rawIter, schemas.outputSchema,
state.outputSchema, schemas.outputVectorColumns)
+ } else {
+ rawIter
+ }
+ }
+}
+
+private[parquet] object HoodieFileGroupReaderFunction {
+
+ private def appendPartitionAndProject(iter: ClosableIterator[InternalRow],
+ inputSchema: StructType,
+ partitionSchema: StructType,
+ to: StructType,
+ partitionValues: InternalRow,
+ fixedPartitionIndexes: Set[Int]):
Iterator[InternalRow] = {
+ if (partitionSchema.isEmpty) {
+ //'inputSchema' and 'to' should be the same so the projection will just
be an identity func
+ projectSchema(iter, inputSchema, to)
+ } else {
+ val fixedPartitionValues = if (partitionSchema.length ==
partitionValues.numFields) {
+ //need to append all of the partition fields
+ partitionValues
+ } else {
+ //some partition fields read from file, some were not
+ getFixedPartitionValues(partitionValues, partitionSchema,
fixedPartitionIndexes)
+ }
+ val unsafeProjection =
generateOutputProjection(StructType(inputSchema.fields ++
partitionSchema.fields), to)
+ val joinedRow = new JoinedRow()
+ makeCloseableFileGroupMappingRecordIterator(iter, d =>
unsafeProjection(joinedRow(d, fixedPartitionValues)))
+ }
+ }
+
+ private def projectSchema(iter: ClosableIterator[InternalRow],
+ from: StructType,
+ to: StructType): Iterator[InternalRow] = {
+ val unsafeProjection = generateOutputProjection(from, to)
+ makeCloseableFileGroupMappingRecordIterator(iter, d => unsafeProjection(d))
+ }
+
+ /**
+ * The scan's output projection. Stays the by-name top-level projection
unless a column comes out of the
+ * reader with nested fields Spark did not ask for (see
`projectionInputSchema` in
+ * buildReaderWithPartitionValues), in which case those are dropped by name
at every depth.
+ */
+ private def generateOutputProjection(from: StructType, to: StructType):
UnsafeProjection = {
+ val hasWiderNestedInput = to.fields.exists { f =>
+ from.getFieldIndex(f.name).exists(i =>
SparkSchemaTransformUtils.needsNestedPruning(from.fields(i).dataType,
f.dataType))
+ }
+ if (hasWiderNestedInput) {
+ SparkSchemaTransformUtils.generateNestedPruningProjection(from, to)
+ } else {
+ generateUnsafeProjection(from, to)
+ }
+ }
+
+ private def
makeCloseableFileGroupMappingRecordIterator(closeableFileGroupRecordIterator:
ClosableIterator[InternalRow],
+ mappingFunction:
Function[InternalRow, InternalRow]): Iterator[InternalRow] = {
+ CloseableIteratorListener.addListener(closeableFileGroupRecordIterator)
+ new Iterator[InternalRow] with Closeable {
+ override def hasNext: Boolean = closeableFileGroupRecordIterator.hasNext
+
+ override def next(): InternalRow =
mappingFunction(closeableFileGroupRecordIterator.next())
+
+ override def close(): Unit = closeableFileGroupRecordIterator.close()
+ }
+ }
+
+ /**
+ * Wraps an iterator to convert binary VECTOR columns back to typed arrays.
+ * The read schema has BinaryType for vector columns; the target schema has
ArrayType.
+ */
+ private def wrapWithVectorConversion(iter: Iterator[InternalRow],
+ readSchema: StructType,
+ targetSchema: StructType,
+ vectorCols: Map[Int,
HoodieSchema.Vector]): Iterator[InternalRow] = {
+ val vectorProjection = UnsafeProjection.create(targetSchema)
+ val javaVectorCols: java.util.Map[Integer, HoodieSchema.Vector] =
+ vectorCols.map { case (k, v) => (Integer.valueOf(k), v) }.asJava
+ val mapper = VectorConversionUtils.buildRowMapper(readSchema,
javaVectorCols, vectorProjection.apply(_))
+ iter.map(mapper.apply(_))
+ }
+
+ private def projectIter(iter: Iterator[Any], from: StructType, to:
StructType): Iterator[InternalRow] = {
+ val unsafeProjection = generateUnsafeProjection(from, to)
+ val batchProjection = ColumnarBatchUtils.generateProjection(from, to)
+ iter.map {
+ case ir: InternalRow => unsafeProjection(ir)
+ case cb: ColumnarBatch => batchProjection(cb)
+ }.asInstanceOf[Iterator[InternalRow]]
+ }
+
+ /**
+ * Projects to `to` only when the read schema was augmented with filter-only
columns;
+ * otherwise returns the iterator as is, preserving columnar batches.
+ */
+ private def projectIfNeeded(iter: Iterator[InternalRow], from: StructType,
to: StructType): Iterator[InternalRow] = {
+ if (from.fieldNames.sameElements(to.fieldNames)) {
+ iter
+ } else {
+ projectIter(iter, from, to)
+ }
+ }
+
+ private def getFixedPartitionValues(allPartitionValues: InternalRow,
partitionSchema: StructType, fixedPartitionIndexes: Set[Int]): InternalRow = {
+
InternalRow.fromSeq(allPartitionValues.toSeq(partitionSchema).zipWithIndex.filter(p
=> fixedPartitionIndexes.contains(p._2)).map(p => p._1))
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestReadOptionsDoNotOverrideTableConfig.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestReadOptionsDoNotOverrideTableConfig.java
new file mode 100644
index 000000000000..a0ef6fde8ddf
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestReadOptionsDoNotOverrideTableConfig.java
@@ -0,0 +1,247 @@
+/*
+ * 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.hudi.functional;
+
+import org.apache.hudi.DataSourceReadOptions;
+import org.apache.hudi.DataSourceWriteOptions;
+import org.apache.hudi.HoodieFileIndex;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.read.CustomPayloadForTesting;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation;
+import org.apache.spark.sql.execution.datasources.LogicalRelation;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import scala.collection.JavaConverters;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A snapshot read merges base and log records with the merge mode and
ordering fields persisted in
+ * the table config. Read options that happen to share a table config key must
not change that:
+ * the persisted config is authoritative, and a read must not return different
rows because an
+ * option was passed, or depending on which files a task happened to read
first.
+ *
+ * <p>The table orders by {@code ts}, through {@link
RecordMergeMode#EVENT_TIME_ORDERING} or through a
+ * persisted custom payload. Every key is first written with {@code ts = 10}
and value {@code A}, then
+ * updated with a lower {@code ts = 5} and value {@code B}, which the ordering
must discard, so every read
+ * must return {@code A}. All file groups are packed into a single task, so
that later file
+ * groups of the task are read after earlier ones have been set up. The table
config of the relation
+ * must also be left as persisted after the read.
+ */
+@Tag("functional")
+class TestReadOptionsDoNotOverrideTableConfig extends
SparkClientFunctionalTestHarness {
+
+ private static final int NUM_PARTITIONS = 4;
+ private static final int KEYS_PER_PARTITION = 5;
+
+ private static final StructType SCHEMA = DataTypes.createStructType(new
StructField[] {
+ DataTypes.createStructField("key", DataTypes.StringType, false),
+ DataTypes.createStructField("part", DataTypes.StringType, false),
+ DataTypes.createStructField("ts", DataTypes.LongType, false),
+ DataTypes.createStructField("other_ts", DataTypes.LongType, false),
+ DataTypes.createStructField("value", DataTypes.StringType, true)});
+
+ private static final String[] PACKING_CONFIGS = {
+ "spark.sql.files.minPartitionNum", "spark.sql.files.openCostInBytes"};
+
+ private final Map<String, String> savedSessionConfs = new HashMap<>();
+
+ @BeforeEach
+ void packFilesIntoOneTask() {
+ for (String conf : PACKING_CONFIGS) {
+ if (spark().conf().contains(conf)) {
+ savedSessionConfs.put(conf, spark().conf().get(conf));
+ }
+ }
+ spark().conf().set("spark.sql.files.minPartitionNum", "1");
+ spark().conf().set("spark.sql.files.openCostInBytes", "1");
+ }
+
+ @AfterEach
+ void restoreSessionConfs() {
+ for (String conf : PACKING_CONFIGS) {
+ if (savedSessionConfs.containsKey(conf)) {
+ spark().conf().set(conf, savedSessionConfs.get(conf));
+ } else {
+ spark().conf().unset(conf);
+ }
+ }
+ }
+
+ @Test
+ void testMergeModeReadOptionDoesNotOverrideTableConfig() {
+ String basePath = writeEventTimeOrderingTable();
+ assertAllRowsKeepFirstValue(read(basePath, Collections.emptyMap()), "a
read without options");
+ Dataset<Row> df = read(basePath,
Collections.singletonMap(HoodieTableConfig.RECORD_MERGE_MODE.key(),
RecordMergeMode.COMMIT_TIME_ORDERING.name()));
+ assertAllRowsKeepFirstValue(df, "a read with " +
HoodieTableConfig.RECORD_MERGE_MODE.key() + "=" +
RecordMergeMode.COMMIT_TIME_ORDERING);
+ assertEventTimeOrderingTableConfig(relationTableConfig(df));
+ }
+
+ @Test
+ void testOrderingFieldsReadOptionDoesNotOverrideTableConfig() {
+ String basePath = writeEventTimeOrderingTable();
+ // other_ts orders the update after the insert; the persisted ordering
field ts does not.
+ Dataset<Row> df = read(basePath,
Collections.singletonMap(HoodieTableConfig.ORDERING_FIELDS.key(), "other_ts"));
+ assertAllRowsKeepFirstValue(df, "a read with " +
HoodieTableConfig.ORDERING_FIELDS.key() + "=other_ts");
+ assertEventTimeOrderingTableConfig(relationTableConfig(df));
+ }
+
+ /**
+ * The persisted payload class is looked up on a different path than the
merge mode and ordering fields
+ * (see {@code ConfigUtils.getMergeProps}), so it gets its own table: a
custom merge with
+ * {@link CustomPayloadForTesting}, which keeps the record with the higher
{@code ts}.
+ */
+ @Test
+ void testPayloadClassReadOptionDoesNotOverrideTableConfig() {
+ Map<String, String> mergeOptions = new HashMap<>();
+ mergeOptions.put(HoodieWriteConfig.RECORD_MERGE_MODE.key(),
RecordMergeMode.CUSTOM.name());
+ mergeOptions.put(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key(),
CustomPayloadForTesting.class.getName());
+ String basePath = writeTable(mergeOptions);
+ assertEquals(CustomPayloadForTesting.class.getName(),
tableConfig(basePath).getPayloadClass());
+ assertAllRowsKeepFirstValue(read(basePath, Collections.emptyMap()), "a
read without options");
+ // Overwriting with the latest write would keep the ts=5 update.
+ String overwritePayload = OverwriteWithLatestAvroPayload.class.getName();
+ Dataset<Row> df = read(basePath,
Collections.singletonMap(HoodieTableConfig.PAYLOAD_CLASS_NAME.key(),
overwritePayload));
+ assertAllRowsKeepFirstValue(df, "a read with " +
HoodieTableConfig.PAYLOAD_CLASS_NAME.key() + "=" + overwritePayload);
+ assertEquals(CustomPayloadForTesting.class.getName(),
relationTableConfig(df).getPayloadClass());
+ }
+
+ private String writeEventTimeOrderingTable() {
+ String basePath =
writeTable(Collections.singletonMap(HoodieWriteConfig.RECORD_MERGE_MODE.key(),
RecordMergeMode.EVENT_TIME_ORDERING.name()));
+ assertEventTimeOrderingTableConfig(tableConfig(basePath));
+ return basePath;
+ }
+
+ private String writeTable(Map<String, String> mergeOptions) {
+ String basePath = basePath();
+ Map<String, String> options = new HashMap<>();
+ options.put(HoodieTableConfig.NAME.key(), "merge_config_from_table");
+ options.put(DataSourceWriteOptions.TABLE_TYPE().key(),
HoodieTableType.MERGE_ON_READ.name());
+ options.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "key");
+ options.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "part");
+ options.put(DataSourceWriteOptions.ORDERING_FIELDS().key(), "ts");
+ options.putAll(mergeOptions);
+ options.put("hoodie.insert.shuffle.parallelism", "2");
+ options.put("hoodie.upsert.shuffle.parallelism", "2");
+ write(rows(10L, 1L, "A"), options, basePath);
+ write(rows(5L, 2L, "B"), options, basePath);
+ assertTrue(countLogFiles(tempDir) >= NUM_PARTITIONS, "Every file group
should have a log file to merge");
+ return basePath;
+ }
+
+ private HoodieTableConfig tableConfig(String basePath) {
+ return
HoodieTableMetaClient.builder().setBasePath(basePath).setConf(storageConf()).build().getTableConfig();
+ }
+
+ private static void assertEventTimeOrderingTableConfig(HoodieTableConfig
tableConfig) {
+ assertEquals(RecordMergeMode.EVENT_TIME_ORDERING,
tableConfig.getRecordMergeMode());
+ assertEquals("ts", tableConfig.getOrderingFieldsStr().orElse(null));
+ }
+
+ private Dataset<Row> read(String basePath, Map<String, String> options) {
+ Dataset<Row> df = spark().read().format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE().key(),
DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL())
+ .options(options)
+ .load(basePath);
+ assertEquals(1, df.rdd().getNumPartitions(), "All file groups should be
read by a single task");
+ return df;
+ }
+
+ private static void assertAllRowsKeepFirstValue(Dataset<Row> df, String
description) {
+ List<Row> rows = df.select("key", "value", "ts").collectAsList();
+ assertEquals(NUM_PARTITIONS * KEYS_PER_PARTITION, rows.size(),
description);
+ List<String> overridden = rows.stream()
+ .filter(row -> !"A".equals(row.getString(1)))
+ .map(row -> row.getString(0) + "=" + row.getString(1) + "@ts" +
row.getLong(2))
+ .sorted()
+ .collect(Collectors.toList());
+ assertTrue(overridden.isEmpty(), "With the ordering on ts persisted in the
table config, "
+ + description + " must keep the ts=10 value A for every key, but these
keys returned the lower-ts update: "
+ + overridden);
+ }
+
+ /**
+ * The table config of the relation's meta client, which the reader state is
built from on the driver; the read
+ * options must not have been written into it.
+ */
+ private static HoodieTableConfig relationTableConfig(Dataset<Row> df) {
+ LogicalPlan relationPlan =
JavaConverters.seqAsJavaList(df.queryExecution().optimizedPlan().collectLeaves()).stream()
+ .filter(plan -> plan instanceof LogicalRelation)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No relation in the plan of the
read"));
+ HadoopFsRelation relation = (HadoopFsRelation) ((LogicalRelation)
relationPlan).relation();
+ return ((HoodieFileIndex)
relation.location()).metaClient().getTableConfig();
+ }
+
+ private static List<Row> rows(long ts, long otherTs, String value) {
+ return IntStream.range(0, NUM_PARTITIONS * KEYS_PER_PARTITION)
+ .mapToObj(i -> RowFactory.create(String.format("key%02d", i), "p" + (i
% NUM_PARTITIONS), ts, otherTs, value))
+ .collect(Collectors.toList());
+ }
+
+ private void write(List<Row> rows, Map<String, String> options, String
basePath) {
+ spark().createDataset(rows,
SparkAdapterSupport$.MODULE$.sparkAdapter().getCatalystExpressionUtils().getEncoder(SCHEMA))
+ .write()
+ .format("hudi")
+ .options(options)
+ .mode(SaveMode.Append)
+ .save(basePath);
+ }
+
+ private static long countLogFiles(Path tableDir) {
+ try (Stream<Path> files = Files.walk(tableDir)) {
+ return files.filter(p -> !p.toString().contains("/.hoodie/") &&
p.getFileName().toString().contains(".log.")).count();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestHoodieFileGroupReaderFunction.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestHoodieFileGroupReaderFunction.scala
new file mode 100644
index 000000000000..27e2c21eb123
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestHoodieFileGroupReaderFunction.scala
@@ -0,0 +1,112 @@
+/*
+ * 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.{DataSourceReadOptions, DataSourceWriteOptions}
+import org.apache.hudi.common.model.HoodieTableType
+import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
+import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+
+import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation,
LogicalRelation}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.EnumSource
+
+import java.io.{ByteArrayOutputStream, ObjectOutputStream, OutputStream}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+/**
+ * Spark Java-serializes the function returned by
[[HoodieFileGroupReaderBasedFileFormat.buildReaderWithPartitionValues]]
+ * into every task. It must hold only broadcast handles, not the file format,
a meta client or a Hadoop configuration.
+ */
+class TestHoodieFileGroupReaderFunction extends HoodieSparkClientTestBase {
+
+ private val maxSerializedFunctionBytes = 16 * 1024
+
+ private val classesNotSerializedWithFunction: Seq[Class[_]] = Seq(
+ classOf[HoodieTableMetaClient],
+ classOf[HoodieFileGroupReaderBasedFileFormat],
+ classOf[HadoopStorageConfiguration],
+ classOf[org.apache.hadoop.conf.Configuration])
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieTableType])
+ def testReaderFunctionHoldsNoDriverState(tableType: HoodieTableType): Unit =
{
+ val opts = Map(
+ HoodieWriteConfig.TBL_NAME.key -> "hoodie_test",
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType.name(),
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "timestamp",
+ "hoodie.insert.shuffle.parallelism" -> "2",
+ "hoodie.upsert.shuffle.parallelism" -> "2")
+ val inserts = recordsToStrings(dataGen.generateInserts("001",
20)).asScala.toSeq
+ sparkSession.read.json(sparkSession.sparkContext.parallelize(inserts, 2))
+
.write.format("hudi").options(opts).mode(SaveMode.Overwrite).save(basePath)
+ val updates = recordsToStrings(dataGen.generateUpdates("002",
10)).asScala.toSeq
+ sparkSession.read.json(sparkSession.sparkContext.parallelize(updates, 2))
+ .write.format("hudi").options(opts).mode(SaveMode.Append).save(basePath)
+
+ val df = sparkSession.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL)
+ .load(basePath)
+ assertEquals(20, df.count())
+ val relation = df.queryExecution.optimizedPlan.collectFirst {
+ case l: LogicalRelation => l.relation.asInstanceOf[HadoopFsRelation]
+ }.get
+ val format =
relation.fileFormat.asInstanceOf[HoodieFileGroupReaderBasedFileFormat]
+ // Spark's scan passes the batch decision it planned with
("returning_batch" is FileFormat.OPTION_RETURNING_BATCH).
+ val returningBatch = format.supportBatch(sparkSession, relation.schema)
+ val options = relation.options + ("returning_batch" ->
returningBatch.toString)
+ val readerFunction = format.buildReaderWithPartitionValues(sparkSession,
relation.dataSchema, relation.partitionSchema,
+ relation.dataSchema, Seq.empty, options,
sparkSession.sessionState.newHadoopConfWithOptions(options))
+ assertTrue(readerFunction.isInstanceOf[HoodieFileGroupReaderFunction],
+ s"Unexpected reader function ${readerFunction.getClass.getName}")
+
+ val bytes = new ByteArrayOutputStream()
+ val out = new ClassRecordingObjectOutputStream(bytes)
+ out.writeObject(readerFunction)
+ out.close()
+ val captured = classesNotSerializedWithFunction.flatMap(forbidden =>
+ out.classes.filter(c => forbidden.isAssignableFrom(c)).map(_.getName))
+ assertTrue(captured.isEmpty, s"The serialized reader function must not
contain $captured")
+ assertTrue(bytes.size() < maxSerializedFunctionBytes,
+ s"The serialized reader function takes ${bytes.size()} bytes, over the
budget of $maxSerializedFunctionBytes")
+ }
+
+ /**
+ * Records the class of every object written to the stream.
+ */
+ private class ClassRecordingObjectOutputStream(out: OutputStream) extends
ObjectOutputStream(out) {
+ val classes: mutable.Set[Class[_]] = mutable.LinkedHashSet[Class[_]]()
+ enableReplaceObject(true)
+
+ override def replaceObject(obj: AnyRef): AnyRef = {
+ if (obj != null) {
+ classes += obj.getClass
+ }
+ obj
+ }
+ }
+}
