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 a11dae45c7dc perf(spark): enable vectorized reading and file splitting
for COW incremental reads (#19914)
a11dae45c7dc is described below
commit a11dae45c7dc2615033694d31e325d240f24c6c9
Author: Prashant Wason <[email protected]>
AuthorDate: Thu Oct 8 12:20:12 2026 -0700
perf(spark): enable vectorized reading and file splitting for COW
incremental reads (#19914)
---
.../hudi/HoodieHadoopFsRelationFactory.scala | 8 +-
.../sql/FileFormatUtilsForFileGroupReader.scala | 1 +
.../HoodieFileGroupReaderBasedFileFormat.scala | 32 ++++-
.../parquet/HoodieFileGroupReaderFunction.scala | 47 ++++++-
.../TestIncrementalReadWithFileGroupReader.scala | 156 +++++++++++++++++++--
.../spark/sql/adapter/BaseSpark3Adapter.scala | 1 +
.../spark/sql/adapter/BaseSpark4Adapter.scala | 1 +
7 files changed, 221 insertions(+), 25 deletions(-)
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 a4e8f18bf499..f6360e3e3ddd 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
@@ -241,13 +241,15 @@ abstract class HoodieBaseHadoopFsRelationFactory(val
sqlContext: SQLContext,
/** Requested-time range to apply before file-group record merging, when
required by the query. */
protected def getInstantRange: HOption[InstantRange] = HOption.empty()
+ protected def isCDC: Boolean = false
+
override def buildFileFormat(): FileFormat = {
val tableConfig = metaClient.getTableConfig
new HoodieFileGroupReaderBasedFileFormat(basePath.toString,
HoodieTableSchema(tableStructSchema, tableSchema, internalSchemaOpt),
tableConfig.getTableName, queryTimestamp.get, getMandatoryFields, isMOR,
isBootstrap,
isIncremental, validCommits, shouldUseRecordPosition, getRequiredFilters,
- tableConfig.isMultipleBaseFileFormatsEnabled,
tableConfig.getBaseFileFormat, getInstantRange, Some(metaClient))
+ tableConfig.isMultipleBaseFileFormatsEnabled,
tableConfig.getBaseFileFormat, getInstantRange, Some(metaClient), isCDC)
}
override def buildBucketSpec(): Option[BucketSpec] = None
@@ -362,6 +364,8 @@ class HoodieMergeOnReadCDCHadoopFsRelationFactory(override
val sqlContext: SQLCo
override def buildPartitionSchema(): StructType = StructType(Nil)
override protected def getRequiredFilters: Seq[Filter] = Seq.empty
+
+ override protected def isCDC: Boolean = true
}
class HoodieCopyOnWriteSnapshotHadoopFsRelationFactory(override val
sqlContext: SQLContext,
@@ -477,4 +481,6 @@ class HoodieCopyOnWriteCDCHadoopFsRelationFactory(override
val sqlContext: SQLCo
override def buildPartitionSchema(): StructType = StructType(Nil)
override protected def getRequiredFilters: Seq[Filter] = Seq.empty
+
+ override protected def isCDC: Boolean = true
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/FileFormatUtilsForFileGroupReader.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/FileFormatUtilsForFileGroupReader.scala
index ae6740b3017a..6ad50c8340a4 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/FileFormatUtilsForFileGroupReader.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/FileFormatUtilsForFileGroupReader.scala
@@ -34,6 +34,7 @@ object FileFormatUtilsForFileGroupReader extends
SparkAdapterSupport {
def applyNewFileFormatChanges(scanOperation: LogicalPlan, logicalRelation:
LogicalPlan, fs: HadoopFsRelation): LogicalPlan = {
val ff = fs.fileFormat.asInstanceOf[ParquetFileFormat with
HoodieFormatTrait]
ff.isProjected = true
+ ff.requiredFiltersAppliedInPlan = true
val tableSchema = fs.location match {
case _: HoodieCDCFileIndex => HoodieCDCFileIndex.FULL_CDC_SPARK_SCHEMA
case index: SparkHoodieTableFileIndex => index.schema
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 16f9a7554756..6849798d259e 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
@@ -60,6 +60,8 @@ trait HoodieFormatTrait {
// Used so that the planner only projects once and does not stack overflow
var isProjected: Boolean = false
+ // Set when the plan already wraps the scan in a Filter on the required
filters, so the scan need not enforce them
+ var requiredFiltersAppliedInPlan: Boolean = false
def getRequiredFilters: Seq[Filter]
}
@@ -70,6 +72,8 @@ trait HoodieFormatTrait {
* @param instantRangeOpt optional requested-time range applied before
file-group record merging;
* unlike Spark's required filters, this prevents a
later out-of-range log
* record from masking an earlier in-range version of
the same key
+ * @param isCDC whether this is a CDC query; CDC rows are produced
by the CDC iterator rather than read
+ * directly from base files, so CDC reads are neither
vectorized nor split
*/
class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
tableSchema: HoodieTableSchema,
@@ -85,7 +89,8 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String,
isMultipleBaseFileFormatsEnabled:
Boolean,
hoodieFileFormat: HoodieFileFormat,
instantRangeOpt:
HOption[InstantRange] = HOption.empty(),
- @transient tableMetaClient:
Option[HoodieTableMetaClient] = None)
+ @transient tableMetaClient:
Option[HoodieTableMetaClient] = None,
+ isCDC: Boolean = false)
extends ParquetFileFormat with SparkAdapterSupport with HoodieFormatTrait
with Logging with Serializable {
private lazy val schema = tableSchema.schema
@@ -113,6 +118,8 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
private val sanitizedTableName =
HoodieSchemaUtils.getRecordQualifiedName(tableName)
+ private def evaluateRequiredFiltersPerRow: Boolean =
requiredFilters.nonEmpty && !requiredFiltersAppliedInPlan
+
/**
* Flag saying whether vectorized reading is supported.
*/
@@ -222,10 +229,18 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
} else {
throw new HoodieNotSupportedException("Unsupported file format: " +
hoodieFileFormat)
}
- supportVectorizedRead = !isIncremental && !isBootstrap && supportBatch
- supportReturningBatch = !isMOR && supportVectorizedRead
+ // MOR incremental embeds file slices that may contain log files
requiring row-level
+ // merging, and CDC rows are built by a row-based iterator that reads
base files through
+ // the same reader, so vectorized reading must be disabled for both. All
other combinations
+ // (COW snapshot, COW incremental, MOR snapshot) either have no log
merging or handle it via
+ // a separate non-vectorized fileGroupBaseFileReader while the base file
reader stays vectorized.
+ supportVectorizedRead = !(isIncremental && (isMOR || isCDC)) &&
!isBootstrap && supportBatch
+ // Required filters (the commit-time range of an incremental query) must
hold for every row, but the
+ // vectorized reader only uses them to prune row groups. Unless the plan
already filters on them, return
+ // rows rather than batches so that the base file read can evaluate them
row by row; decoding stays vectorized.
+ supportReturningBatch = !isMOR && supportVectorizedRead &&
!evaluateRequiredFiltersPerRow
logDebug(s"supportReturningBatch: $supportReturningBatch,
supportVectorizedRead: $supportVectorizedRead, isIncremental: $isIncremental, "
+
- s"isBootstrap: $isBootstrap, superSupportBatch: $supportBatch")
+ s"isCDC: $isCDC, isBootstrap: $isBootstrap, superSupportBatch:
$supportBatch")
supportReturningBatch
}
}
@@ -277,8 +292,11 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
// For overly large single files, we can use multiple concurrent tasks to
read them, thereby reducing the overall job reading time consumption
val superSplitable = super.isSplitable(sparkSession, options, path)
val isLance = hoodieFileFormat == HoodieFileFormat.LANCE
- val splitable = !isMOR && !isIncremental && !isBootstrap && !isLance &&
superSplitable
- logDebug(s"isSplitable: $splitable, super.isSplitable: $superSplitable,
isMOR: $isMOR, isIncremental: $isIncremental, isBootstrap: $isBootstrap")
+ // COW incremental reads have no log files to merge, so file splitting is
safe.
+ // MOR, CDC and bootstrap reads need to disable splitting.
+ val splitable = !isMOR && !isCDC && !isBootstrap && !isLance &&
superSplitable
+ logDebug(s"isSplitable: $splitable, super.isSplitable: $superSplitable,
isMOR: $isMOR, isIncremental: $isIncremental, " +
+ s"isCDC: $isCDC, isBootstrap: $isBootstrap")
splitable
}
@@ -385,7 +403,7 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
val state = new HoodieFileGroupReadState(tableState, tableSchema,
queryTimestamp, readerProps, cdcProps,
dataSchema, requestedSchema, internalSchemaOpt, instantRangeOpt,
shouldUseRecordPosition, isCount, filters,
requiredFilters, requiredSchema, partitionSchema,
remainingPartitionSchema, fixedPartitionIndexes, outputSchema,
- projectionInputSchema, baseFileReadSchemas)
+ projectionInputSchema, baseFileReadSchemas,
evaluateRequiredFiltersPerRow)
new HoodieFileGroupReaderFunction(baseFileReader, fileGroupBaseFileReader,
broadcastedStorageConf,
spark.sparkContext.broadcast(JavaSerializedValue(state)))
}
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
index 370e35a5b961..6887f4848647 100644
---
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
@@ -30,6 +30,7 @@ import
org.apache.hudi.common.table.read.lsm.{HoodieLsmFileGroupReader, LsmReade
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.exception.HoodieException
import
org.apache.hudi.io.storage.HoodieSparkParquetReader.ENABLE_LOGICAL_TIMESTAMP_REPAIR
import org.apache.hudi.io.storage.VectorConversionUtils
import org.apache.hudi.storage.StorageConfiguration
@@ -39,9 +40,10 @@ 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
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.catalyst.expressions.{And, AttributeReference, In,
InSet, JoinedRow, Literal, Predicate, 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
@@ -78,7 +80,8 @@ private[parquet] class HoodieFileGroupReadState(val
tableState: TableState,
val fixedPartitionIndexes:
Set[Int],
val outputSchema: StructType,
val projectionInputSchema:
StructType,
- val baseFileReadSchemas:
BaseFileReadSchemas) extends Serializable {
+ val baseFileReadSchemas:
BaseFileReadSchemas,
+ val
evaluateRequiredFiltersPerRow: Boolean) extends Serializable {
/**
* Parquet form of the table schema for base file reads. The conversion
reads Hadoop's default resources, so it is
@@ -233,6 +236,7 @@ private[parquet] class
HoodieFileGroupReaderFunction(baseFileReader: Broadcast[S
val hasVectors = schemas.readVectorColumns.nonEmpty
val parquetFileReader = baseFileReader.value
val filters = state.filters ++ state.requiredFilters
+ val requiredFilters = if (state.evaluateRequiredFiltersPerRow)
state.requiredFilters else Seq.empty
val partitionSchema = state.partitionSchema
val remainingPartitionSchema = state.remainingPartitionSchema
@@ -240,7 +244,8 @@ private[parquet] class
HoodieFileGroupReaderFunction(baseFileReader: Broadcast[S
//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)
+ val readSchema = StructType(schemas.readRequiredSchema.fields ++
partitionSchema.fields)
+ projectIfNeeded(applyRequiredFilters(iter, readSchema, requiredFilters),
readSchema, schemas.outputSchema)
} else if (remainingPartitionSchema.fields.length == 0) {
//we read all of the partition fields from the file
val pfileUtils = sparkAdapter.getSparkPartitionedFileUtils
@@ -250,7 +255,7 @@ private[parquet] class
HoodieFileGroupReaderFunction(baseFileReader: Broadcast[S
//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)
+ projectIfNeeded(applyRequiredFilters(iter, readSchema, requiredFilters),
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)
@@ -260,7 +265,8 @@ private[parquet] class
HoodieFileGroupReaderFunction(baseFileReader: Broadcast[S
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)
+ val readSchema = StructType(schemas.requestedSchema.fields ++
remainingPartitionSchema.fields)
+ projectIter(applyRequiredFilters(iter, readSchema, requiredFilters),
readSchema, schemas.outputSchema)
}
if (hasVectors) {
@@ -349,6 +355,37 @@ private[parquet] object HoodieFileGroupReaderFunction {
iter.map(mapper.apply(_))
}
+ /**
+ * Evaluates `requiredFilters` on every row. Pushed-down filters only prune
row groups in the vectorized reader,
+ * so when the plan does not filter on them (the Hudi session extension is
not registered), the commit-time
+ * range of an incremental query is enforced here. The scan returns rows
rather than batches in that case.
+ */
+ private def applyRequiredFilters(iter: Iterator[InternalRow],
+ schema: StructType,
+ requiredFilters: Seq[Filter]):
Iterator[InternalRow] = {
+ if (requiredFilters.isEmpty) {
+ iter
+ } else {
+ val attrs =
SparkAdapterSupport.sparkAdapter.getSchemaUtils.toAttributes(schema)
+ val condition = requiredFilters.map { filter =>
+ HoodieCatalystExpressionUtils.convertToCatalystExpression(filter,
schema).getOrElse(
+ throw new HoodieException(s"Cannot evaluate required filter:
$filter"))
+ }.reduce(And).transform {
+ case a: AttributeReference => attrs(schema.fieldIndex(a.name))
+ }.transform {
+ // A set lookup rather than a chain of comparisons for the commit
times of an incremental range
+ case In(child, list) if list.forall(_.isInstanceOf[Literal]) &&
list.nonEmpty =>
+ InSet(child, list.map(_.eval()).toSet)
+ }
+ val predicate = Predicate.create(condition, attrs)
+ predicate.initialize(0)
+ iter.asInstanceOf[Iterator[Any]].filter {
+ case _: ColumnarBatch => throw new IllegalStateException("Required
filters cannot be evaluated on columnar batches")
+ case row: InternalRow => predicate.eval(row)
+ }.asInstanceOf[Iterator[InternalRow]]
+ }
+ }
+
private def projectIter(iter: Iterator[Any], from: StructType, to:
StructType): Iterator[InternalRow] = {
val unsafeProjection = generateUnsafeProjection(from, to)
val batchProjection = ColumnarBatchUtils.generateProjection(from, to)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
index 5b3c6581eede..b8c9e9652d1f 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFileGroupReader.scala
@@ -18,12 +18,13 @@
package org.apache.hudi.functional
import org.apache.hudi.{DataSourceReadOptions, DataSourceWriteOptions,
MergeOnReadIncrementalRelationV2}
-import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.config.{HoodieMetadataConfig,
HoodieStorageConfig, TimestampKeyGeneratorConfig}
import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
import
org.apache.hudi.common.table.read.IncrementalQueryAnalyzer.START_COMMIT_EARLIEST
import org.apache.hudi.common.table.timeline.HoodieInstant
import org.apache.hudi.config.{HoodieArchivalConfig, HoodieCleanConfig,
HoodieCompactionConfig, HoodieWriteConfig}
+import org.apache.hudi.keygen.{CustomKeyGenerator, TimestampBasedKeyGenerator}
import org.apache.hudi.storage.StoragePath
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
@@ -31,7 +32,9 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.sql.{DataFrame, SaveMode}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.CsvSource
+import org.junit.jupiter.params.provider.{CsvSource, ValueSource}
+
+import java.util.UUID
import scala.collection.JavaConverters._
@@ -64,18 +67,20 @@ class TestIncrementalReadWithFileGroupReader extends
SparkClientFunctionalTestHa
@ParameterizedTest
@CsvSource(value = Array(
- "COPY_ON_WRITE,6,6",
- "COPY_ON_WRITE,8,6",
- "COPY_ON_WRITE,8,8",
- "MERGE_ON_READ,6,6",
- "MERGE_ON_READ,6,8",
- "MERGE_ON_READ,8,6",
- "MERGE_ON_READ,8,8"
+ "COPY_ON_WRITE,6,6,PARQUET",
+ "COPY_ON_WRITE,8,6,PARQUET",
+ "COPY_ON_WRITE,8,8,PARQUET",
+ "COPY_ON_WRITE,6,6,ORC",
+ "COPY_ON_WRITE,8,8,ORC",
+ "MERGE_ON_READ,6,6,PARQUET",
+ "MERGE_ON_READ,6,8,PARQUET",
+ "MERGE_ON_READ,8,6,PARQUET",
+ "MERGE_ON_READ,8,8,PARQUET"
))
- def testIncrementalReadRanges(tableType: String, sourceVersion: Int,
readVersion: Int): Unit = {
+ def testIncrementalReadRanges(tableType: String, sourceVersion: Int,
readVersion: Int, baseFileFormat: String): Unit = {
batches.zipWithIndex.foreach { case ((data, operation), i) =>
val mode = if (i == 0) SaveMode.Overwrite else SaveMode.Append
- write(data, tableType, sourceVersion, operation, mode)
+ write(data, tableType, sourceVersion, operation, mode,
Map(HoodieTableConfig.BASE_FILE_FORMAT.key -> baseFileFormat))
if (i == 2) {
// small file handling must have kept a single file group with base
files only
val (baseFiles, logFiles) = listDataFiles()
@@ -99,7 +104,8 @@ class TestIncrementalReadWithFileGroupReader extends
SparkClientFunctionalTestHa
}
// records merged into the latest base file keep their original commit
times
val latestBaseFile = baseFiles.maxBy(name => FSUtils.getCommitTime(name))
- val commitTimesInBaseFile = spark.read.parquet(new Path(new Path(basePath,
"pt1"), latestBaseFile).toString)
+ val commitTimesInBaseFile = spark.read.format(baseFileFormat.toLowerCase)
+ .load(new Path(new Path(basePath, "pt1"), latestBaseFile).toString)
.select("_hoodie_commit_time").distinct().count()
assertTrue(commitTimesInBaseFile > 1,
s"Expected multiple commit times in the latest base file, got
$commitTimesInBaseFile")
@@ -126,6 +132,132 @@ class TestIncrementalReadWithFileGroupReader extends
SparkClientFunctionalTestHa
assertIncrementalRange(readVersion, instants, 6, 6, Set.empty)
}
+ /**
+ * COW incremental reads are splittable: a range read whose base file is
split into several
+ * input partitions must return the same rows as the unsplit read, with the
commit-time range
+ * enforced in every split.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("PARQUET", "ORC"))
+ def testCowIncrementalReadWithMultipleSplitsPerFile(baseFileFormat: String):
Unit = {
+ val options = Map(
+ HoodieTableConfig.BASE_FILE_FORMAT.key -> baseFileFormat,
+ HoodieStorageConfig.PARQUET_BLOCK_SIZE.key -> "65536",
+ HoodieStorageConfig.PARQUET_PAGE_SIZE.key -> "8192",
+ HoodieStorageConfig.ORC_STRIPE_SIZE.key -> "65536",
+ HoodieStorageConfig.ORC_BLOCK_SIZE.key -> "65536")
+ def rows(commit: Int, keys: Range): Seq[(Int, String, String, Double,
String)] = keys.map { i =>
+ (commit, f"k$i%05d",
UUID.nameUUIDFromBytes(s"$i-$commit".getBytes).toString * 3, commit.toDouble,
"pt1")
+ }
+ val c1Keys = 0 until 2000
+ val c2Keys = 2000 until 4000
+ val c3Keys = 4000 until 6000
+ // c1..c3 insert disjoint keys into one file group; c4 updates the keys of
c1, so every base file
+ // from c2 on carries rows of earlier commits that the range must filter
out
+ write(rows(1, c1Keys), "COPY_ON_WRITE", 8,
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL, SaveMode.Overwrite, options)
+ write(rows(2, c2Keys), "COPY_ON_WRITE", 8,
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL, SaveMode.Append, options)
+ write(rows(3, c3Keys), "COPY_ON_WRITE", 8,
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL, SaveMode.Append, options)
+ write(rows(4, c1Keys), "COPY_ON_WRITE", 8,
DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL, SaveMode.Append, options)
+ val (baseFiles, _) = listDataFiles()
+ assertEquals(1, baseFiles.map(FSUtils.getFileId).distinct.size, "Expected
a single file group")
+
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf().newInstance()).setBasePath(basePath()).build()
+ val instants =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ .getInstants.asScala.toList
+ assertEquals(4, instants.size)
+
+ def keysAndTs(df: DataFrame): Set[(String, Int)] =
+ df.select("key", "ts").collect().map(r => (r.getString(0),
r.getInt(1))).toSet
+
+ // (c1, c3]: c2 and c3 rows, without the c1 rows carried over in the same
base file
+ val c1c3 = (instants(0).getCompletionTime, instants(2).getCompletionTime)
+ val expectedC1C3 = c2Keys.map(i => (f"k$i%05d", 2)).toSet ++ c3Keys.map(i
=> (f"k$i%05d", 3)).toSet
+ // (c3, c4]: only the c4 updates of the c1 keys
+ val c3c4 = (instants(2).getCompletionTime, instants(3).getCompletionTime)
+ val expectedC3C4 = c1Keys.map(i => (f"k$i%05d", 4)).toSet
+ assertEquals(expectedC1C3, keysAndTs(readIncremental(8, c1c3._1, c1c3._2)))
+ assertEquals(expectedC3C4, keysAndTs(readIncremental(8, c3c4._1, c3c4._2)))
+
+ spark.conf.set("spark.sql.files.maxPartitionBytes", "65536")
+ try {
+ Seq((c1c3, expectedC1C3), (c3c4, expectedC3C4)).foreach { case ((start,
end), expected) =>
+ val df = readIncremental(8, start, end)
+ assertTrue(df.rdd.getNumPartitions > df.inputFiles.length,
+ s"Expected more input partitions than files, got
${df.rdd.getNumPartitions} for ${df.inputFiles.length} files")
+ assertEquals(expected, keysAndTs(df))
+ assertEquals(expected.size.toLong, readIncremental(8, start,
end).count())
+ }
+ } finally {
+ spark.conf.unset("spark.sql.files.maxPartitionBytes")
+ }
+ }
+
+ /**
+ * COW incremental reads where a partition column is read from the base file
rather than the
+ * partition path (timestamp-based partitioning), so the commit-time range
is evaluated against a
+ * read schema that carries the mandatory partition fields. TIMESTAMP reads
every partition column
+ * from the file; CUSTOM reads one from the file and appends the other from
the path.
+ */
+ @ParameterizedTest
+ @ValueSource(strings = Array("TIMESTAMP", "CUSTOM"))
+ def testCowIncrementalReadWithPartitionValuesReadFromFile(keyGenType:
String): Unit = {
+ val (keyGenClass, partitionFields) = keyGenType match {
+ case "TIMESTAMP" => (classOf[TimestampBasedKeyGenerator].getName,
"event_ts")
+ case "CUSTOM" => (classOf[CustomKeyGenerator].getName,
"pt:SIMPLE,event_ts:TIMESTAMP")
+ }
+ // all records fall on 2023-11-14 UTC so that they share one partition and
file group
+ def write(data: Seq[(Int, String, String, Long, String)], operation:
String, mode: SaveMode): Unit = {
+ spark.createDataFrame(data).toDF("ts", "key", "rider", "event_ts",
"pt").write.format("hudi")
+ .option(DataSourceWriteOptions.RECORDKEY_FIELD.key, "key")
+ .option(DataSourceWriteOptions.PARTITIONPATH_FIELD.key,
partitionFields)
+ .option(DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key,
keyGenClass)
+ .option(TimestampKeyGeneratorConfig.TIMESTAMP_TYPE_FIELD.key,
"EPOCHMILLISECONDS")
+ .option(TimestampKeyGeneratorConfig.TIMESTAMP_OUTPUT_DATE_FORMAT.key,
"yyyyMMdd")
+
.option(TimestampKeyGeneratorConfig.TIMESTAMP_OUTPUT_TIMEZONE_FORMAT.key, "UTC")
+ .option(HoodieTableConfig.ORDERING_FIELDS.key, "ts")
+ .option(DataSourceWriteOptions.TABLE_TYPE.key, "COPY_ON_WRITE")
+ .option(DataSourceWriteOptions.TABLE_NAME.key,
"test_incr_read_fgr_keygen")
+ .option(DataSourceWriteOptions.OPERATION.key, operation)
+ .option("hoodie.insert.shuffle.parallelism", "2")
+ .option("hoodie.upsert.shuffle.parallelism", "2")
+ .mode(mode)
+ .save(basePath)
+ }
+ def eventTs(ts: Int): Long = 1699920000000L + ts * 1000L
+ write(Seq((1, "k1", "rider-c1", eventTs(1), "pt1"), (1, "k2", "rider-c1",
eventTs(1), "pt1")),
+ DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL, SaveMode.Overwrite)
+ write(Seq((2, "k3", "rider-c2", eventTs(2), "pt1"), (2, "k4", "rider-c2",
eventTs(2), "pt1")),
+ DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL, SaveMode.Append)
+ // c3 rewrites the base file with k1 updated, carrying the c1 and c2 rows
over
+ write(Seq((3, "k1", "rider-c3", eventTs(1), "pt1")),
+ DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL, SaveMode.Append)
+
+ val metaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf().newInstance()).setBasePath(basePath()).build()
+ val instants =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
+ .getInstants.asScala.toList
+ assertEquals(3, instants.size)
+
+ // event_ts is compared as text: the timestamp key generator types it as a
string partition column
+ def assertRange(start: String, end: String, expected: Set[(String, Int,
Long, String)]): Unit = {
+ val rows = readIncremental(8, start, end).select("key", "ts",
"event_ts", "pt").collect()
+ .map(r => (r.getString(0), r.getInt(1), String.valueOf(r.get(2)),
r.getString(3))).toSet
+ assertEquals(expected.map(e => (e._1, e._2, e._3.toString, e._4)), rows)
+ assertEquals(expected.size.toLong, readIncremental(8, start,
end).count())
+ }
+ // (000, c2]: the rows as of c2, including the c1 version of k1 that c3
later updates
+ assertRange("000", instants(1).getCompletionTime,
+ Set(("k1", 1, eventTs(1), "pt1"), ("k2", 1, eventTs(1), "pt1"), ("k3",
2, eventTs(2), "pt1"),
+ ("k4", 2, eventTs(2), "pt1")))
+ // (c1, c2]: the c2 rows only
+ assertRange(instants(0).getCompletionTime, instants(1).getCompletionTime,
+ Set(("k3", 2, eventTs(2), "pt1"), ("k4", 2, eventTs(2), "pt1")))
+ // (c2, c3]: the c3 update of k1
+ assertRange(instants(1).getCompletionTime, instants(2).getCompletionTime,
+ Set(("k1", 3, eventTs(1), "pt1")))
+ }
+
@ParameterizedTest(name = "V2 MOR full-table scan applies instant range,
table version = {0}, legacy RDD = {1}")
@CsvSource(value = Array("6,false", "6,true", "8,false", "8,true"))
def testV2MorFullTableScanAppliesInstantRange(sourceVersion: Int,
useLegacyRdd: Boolean): Unit = {
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
index 92010be7f336..0f79c8c5ec06 100644
---
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
@@ -188,6 +188,7 @@ abstract class BaseSpark3Adapter extends SparkAdapter with
Logging {
def createStreamingDataFrame(sqlContext: SQLContext, relation:
HadoopFsRelation, requiredSchema: StructType): DataFrame = {
val logicalRelation = LogicalRelation(relation, isStreaming = true)
val resolvedSchema = logicalRelation.resolve(requiredSchema,
sqlContext.sparkSession.sessionState.analyzer.resolver)
+
relation.fileFormat.asInstanceOf[HoodieFormatTrait].requiredFiltersAppliedInPlan
= true
Dataset.ofRows(sqlContext.sparkSession,
applyFiltersToPlan(logicalRelation, requiredSchema, resolvedSchema,
relation.fileFormat.asInstanceOf[HoodieFormatTrait].getRequiredFilters))
}
diff --git
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
index 2f049cbedce6..977a1f51d976 100644
---
a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
@@ -180,6 +180,7 @@ abstract class BaseSpark4Adapter extends SparkAdapter with
Logging {
def createStreamingDataFrame(sqlContext: SQLContext, relation:
HadoopFsRelation, requiredSchema: StructType): DataFrame = {
val logicalRelation = LogicalRelation(relation, isStreaming = true)
val resolvedSchema = logicalRelation.resolve(requiredSchema,
sqlContext.sparkSession.sessionState.analyzer.resolver)
+
relation.fileFormat.asInstanceOf[HoodieFormatTrait].requiredFiltersAppliedInPlan
= true
org.apache.spark.sql.classic.Dataset.ofRows(sqlContext.sparkSession.asInstanceOf[org.apache.spark.sql.classic.SparkSession],
applyFiltersToPlan(logicalRelation, requiredSchema, resolvedSchema,
relation.fileFormat.asInstanceOf[HoodieFormatTrait].getRequiredFilters))
