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 211ccba18c0f fix(spark): avoid full partition listing in incremental
reads (#19898)
211ccba18c0f is described below
commit 211ccba18c0f4f1041bd2016e725ab4a9106accd
Author: Prashant Wason <[email protected]>
AuthorDate: Tue Oct 6 18:49:00 2026 -0700
fix(spark): avoid full partition listing in incremental reads (#19898)
---
.../apache/hudi/HoodieIncrementalFileIndex.scala | 21 ++-
.../hudi/MergeOnReadIncrementalRelationV1.scala | 6 +-
.../hudi/MergeOnReadIncrementalRelationV2.scala | 6 +-
.../apache/hudi/SparkHoodieTableFileIndex.scala | 173 +++++++++++++--------
.../TestIncrementalReadWithPartitionFilter.scala | 131 ++++++++++++++++
5 files changed, 267 insertions(+), 70 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieIncrementalFileIndex.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieIncrementalFileIndex.scala
index 5b5312643aac..ba03764aa4c1 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieIncrementalFileIndex.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieIncrementalFileIndex.scala
@@ -17,8 +17,9 @@
package org.apache.hudi
-import org.apache.hudi.common.model.HoodieLogFile
+import org.apache.hudi.common.model.{FileSlice, HoodieLogFile}
import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.core.read.BaseHoodieTableFileIndex
import org.apache.hudi.storage.StoragePathInfo
import org.apache.hudi.util.{JFunction, SparkConfigUtils}
@@ -43,6 +44,24 @@ class HoodieIncrementalFileIndex(override val spark:
SparkSession,
spark, metaClient, schemaSpec, options, fileStatusCache, includeLogFiles,
shouldEmbedFileSlices = true
) with FileIndex {
+ // Skip the Spark optimizer's partition pruning rule (e.g.
Spark33HoodiePruneFileSourcePartitions)
+ // which would trigger a full-table partition listing via the base class.
The incremental file
+ // index already selects only modified file groups, and prunes them by the
partition filters, in
+ // listFileSplits(). Overridden rather than set so that it also holds after
refresh().
+ override def hasPredicatesPushedDown: Boolean = true
+
+ // Serve file slices from the incremental relation rather than the base
class listing, in case
+ // anything still calls this.
+ override def filterFileSlices(dataFilters: Seq[Expression],
partitionFilters: Seq[Expression],
+ isPartitionPruneOnly: Boolean)
+ : Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])] = {
+ mergeOnReadIncrementalRelation.listFileSplits(partitionFilters,
dataFilters).values.flatten
+ .filter(!_.isEmpty)
+ .groupBy(_.getPartitionPath)
+ .map { case (partitionPath, fileSlices) =>
(Option(convertToPartitionPath(partitionPath)), fileSlices.toSeq) }
+ .toSeq
+ }
+
override def listFiles(partitionFilters: Seq[Expression], dataFilters:
Seq[Expression]): Seq[PartitionDirectory] = {
val fileSlices =
mergeOnReadIncrementalRelation.listFileSplits(partitionFilters,
dataFilters).toSeq.map {
case (partitionValues, fileSlices) =>
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
index fe0996e49e63..8c78bf68db54 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
@@ -130,8 +130,10 @@ case class MergeOnReadIncrementalRelationV1(override val
sqlContext: SQLContext,
val fsView = new HoodieTableFileSystemView(metaClient, timeline,
affectedFilesInCommits)
val modifiedPartitions = getWritePartitionPaths(commitsMetadata)
-
fileIndex.listMatchingPartitionPaths(HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient,
partitionFilters))
- .map(p => p.getPath).filter(p => modifiedPartitions.contains(p))
+ // Evaluate the partition predicates against the modified partitions
only, rather than listing
+ // every partition of the table.
+ fileIndex.filterPartitionPaths(modifiedPartitions.asScala.toSeq,
+ HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient,
partitionFilters))
.flatMap { relativePartitionPath =>
fsView.getLatestMergedFileSlicesBeforeOrOn(relativePartitionPath,
latestCommit).iterator().asScala
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
index 796e43ae8c61..501e619fa439 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
@@ -128,8 +128,10 @@ case class MergeOnReadIncrementalRelationV2(override val
sqlContext: SQLContext,
val fsView = new HoodieTableFileSystemView(metaClient, timeline,
affectedFilesInCommits)
val modifiedPartitions = getWritePartitionPaths(commitsMetadata)
-
fileIndex.listMatchingPartitionPaths(HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient,
partitionFilters))
- .map(p => p.getPath).filter(p => modifiedPartitions.contains(p))
+ // Evaluate the partition predicates against the modified partitions
only, rather than listing
+ // every partition of the table.
+ fileIndex.filterPartitionPaths(modifiedPartitions.asScala.toSeq,
+ HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient,
partitionFilters))
.flatMap { relativePartitionPath =>
fsView.getLatestMergedFileSlicesBeforeOrOn(relativePartitionPath,
latestCommit).iterator().asScala
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala
index 8187d5c54e86..7566bf148125 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkHoodieTableFileIndex.scala
@@ -250,44 +250,33 @@ class SparkHoodieTableFileIndex(spark: SparkSession,
}
/**
- * Prune the partition by the filter.This implementation is fork from
+ * Lists the partitions of the table matching the partition predicates. The
pruning is forked from
*
org.apache.spark.sql.execution.datasources.PartitioningAwareFileIndex#prunePartitions.
*
+ * Whether this lists the table depends on the predicates and the query:
+ * - No partition predicate: returns [[getAllQueryPartitionPaths]]. For an
incremental query whose start
+ * is inside the active timeline these are the partitions written by the
instants in range, read from
+ * the commit metadata. Otherwise (snapshot query, or an incremental
query starting from `earliest` or
+ * before the active timeline) every partition of the table is listed.
+ * - Equality predicates on all partition columns: a single existence
check, no listing.
+ * - Equality predicates on the leading partition columns: only the
partitions under that prefix are listed.
+ * - No bound leading prefix: hive-style, URL-encoded tables list from the
table root, evaluating the
+ * predicates through the metadata table, with no incremental shortcut.
Other tables fall back to
+ * [[getAllQueryPartitionPaths]] as above.
+ *
+ * Callers that already hold the candidate partitions (e.g. incremental
reads, which know the partitions
+ * modified by the commits in range) should use [[filterPartitionPaths]]
instead, which never lists the table.
+ *
* @param predicates The filter condition.
* @return The pruned partition paths.
*/
def listMatchingPartitionPaths(predicates: Seq[Expression]):
Seq[PartitionPath] = {
- val resolve = spark.sessionState.analyzer.resolver
val partitionColumnNames = getPartitionColumns
-
- // Resolves GetStructField chain to full dot-path:
GetStructField(attr("a"), _, "b") → "a.b"
- def getFieldPath(expr: Expression): Option[String] = expr match {
- case a: AttributeReference => Some(a.name)
- case GetStructField(child, _, Some(fieldName)) =>
- getFieldPath(child).map(_ + "." + fieldName)
- case _ => None
- }
-
- // True if every column reference in expr resolves to a partition column.
- // For nested columns, walks GetStructField chains to match the full
dot-path.
- // Example: partition = "nested_record.level"
- // nested_record.level = 'INFO' → GetStructField path
"nested_record.level" → true
- // nested_record.nested_int = 10 → GetStructField path
"nested_record.nested_int" → false
- // IsNotNull(nested_record) → AttributeReference "nested_record"
not in partitionColumnNames → false
- def referencesOnlyPartitionColumns(expr: Expression): Boolean = expr match
{
- case g: GetStructField =>
- getFieldPath(g).exists(path => partitionColumnNames.exists(pc =>
resolve(path, pc)))
- case a: AttributeReference =>
- partitionColumnNames.exists(pc => resolve(a.name, pc))
- case _ =>
- expr.children.forall(referencesOnlyPartitionColumns)
- }
-
val partitionPruningPredicates =
predicates.filter(referencesOnlyPartitionColumns)
if (partitionPruningPredicates.isEmpty) {
val queryPartitionPaths = getAllQueryPartitionPaths.asScala.toSeq
- logInfo(s"No partition predicates provided, listing full table
(${queryPartitionPaths.size} partitions)")
+ logInfo(s"No partition predicates provided, using all query partitions
(${queryPartitionPaths.size} partitions)")
queryPartitionPaths
} else {
// NOTE: We fallback to already cached partition-paths only in cases
when we can subsequently
@@ -307,48 +296,102 @@ class SparkHoodieTableFileIndex(spark: SparkSession,
// we might not be able to properly parse partition-values from
the listed partition-paths.
// In that case, we simply could not apply partition pruning and
will have to regress to scanning
// the whole table
- if (haveProperPartitionValues(partitionPaths.toSeq) &&
partitionSchema.nonEmpty) {
- val predicate = partitionPruningPredicates.reduce(expressions.And)
- val partitionFieldNames = partitionSchema.fieldNames
- val transformedPredicate = predicate.transform {
- case g @ GetStructField(_, _, Some(_)) =>
- getFieldPath(g).flatMap { path =>
- val idx = partitionFieldNames.indexWhere(name => resolve(path,
name))
- if (idx >= 0) Some(BoundReference(idx,
partitionSchema(idx).dataType, nullable = true))
- else None
- }.getOrElse(g)
- case a: AttributeReference =>
- val index = partitionSchema.indexWhere(sf => resolve(a.name,
sf.name))
- if (index >= 0) BoundReference(index,
partitionSchema(index).dataType, nullable = true)
- else a
- }
- val boundPredicate: BasePredicate = try {
- // Try using 1-arg constructor via reflection
- val clazz =
Class.forName("org.apache.spark.sql.catalyst.expressions.InterpretedPredicate")
- val ctor = clazz.getConstructor(classOf[Expression])
- ctor.newInstance(transformedPredicate).asInstanceOf[BasePredicate]
- } catch {
- case _: NoSuchMethodException | _: IllegalArgumentException =>
- // Fallback: Try using 2-arg constructor for certain Spark runtime
- val clazz =
Class.forName("org.apache.spark.sql.catalyst.expressions.InterpretedPredicate")
- val ctor = clazz.getConstructor(classOf[Expression],
classOf[Boolean])
- ctor.newInstance(transformedPredicate, java.lang.Boolean.FALSE)
- .asInstanceOf[BasePredicate]
- }
- val prunedPartitionPaths = partitionPaths.filter {
- partitionPath =>
boundPredicate.eval(InternalRow.fromSeq(partitionPath.getValues))
- }.toSeq
+ prunePartitionPaths(partitionPaths.toSeq, partitionPruningPredicates)
+ }
+ }
- logInfo(s"Using provided predicates to prune number of target table's
partitions scanned from" +
- s" ${partitionPaths.size} to ${prunedPartitionPaths.size}")
+ /**
+ * Prunes the given relative partition paths by the partition predicates,
without listing the table.
+ * Used by incremental reads, which already know the partitions modified by
the commits in range.
+ *
+ * @param relativePartitionPaths The candidate relative partition paths.
+ * @param predicates The filter condition.
+ * @return The relative partition paths matching the partition predicates.
+ */
+ def filterPartitionPaths(relativePartitionPaths: Seq[String], predicates:
Seq[Expression]): Seq[String] = {
+ val partitionPruningPredicates =
predicates.filter(referencesOnlyPartitionColumns)
+ if (partitionPruningPredicates.isEmpty) {
+ relativePartitionPaths
+ } else {
+ prunePartitionPaths(relativePartitionPaths.map(convertToPartitionPath),
partitionPruningPredicates).map(_.getPath)
+ }
+ }
- prunedPartitionPaths
- } else {
- logWarning(s"Unable to apply partition pruning, due to failure to
parse partition values from the" +
- s" following path(s): ${partitionPaths.find(_.getValues.length ==
0).map(e => e.getPath)}")
+ // Resolves GetStructField chain to full dot-path: GetStructField(attr("a"),
_, "b") → "a.b"
+ private def getFieldPath(expr: Expression): Option[String] = expr match {
+ case a: AttributeReference => Some(a.name)
+ case GetStructField(child, _, Some(fieldName)) =>
+ getFieldPath(child).map(_ + "." + fieldName)
+ case _ => None
+ }
+
+ // True if every column reference in expr resolves to a partition column.
+ // For nested columns, walks GetStructField chains to match the full
dot-path.
+ // Example: partition = "nested_record.level"
+ // nested_record.level = 'INFO' → GetStructField path
"nested_record.level" → true
+ // nested_record.nested_int = 10 → GetStructField path
"nested_record.nested_int" → false
+ // IsNotNull(nested_record) → AttributeReference "nested_record" not
in partitionColumnNames → false
+ private def referencesOnlyPartitionColumns(expr: Expression): Boolean = {
+ val resolve = spark.sessionState.analyzer.resolver
+ val partitionColumnNames = getPartitionColumns
+ expr match {
+ case g: GetStructField =>
+ getFieldPath(g).exists(path => partitionColumnNames.exists(pc =>
resolve(path, pc)))
+ case a: AttributeReference =>
+ partitionColumnNames.exists(pc => resolve(a.name, pc))
+ case _ =>
+ expr.children.forall(referencesOnlyPartitionColumns)
+ }
+ }
- partitionPaths.toSeq
+ /**
+ * Evaluates the (non-empty) partition pruning predicates against the
partition values of the given
+ * partition paths.
+ */
+ private def prunePartitionPaths(partitionPaths: Seq[PartitionPath],
+ partitionPruningPredicates:
Seq[Expression]): Seq[PartitionPath] = {
+ val resolve = spark.sessionState.analyzer.resolver
+ if (haveProperPartitionValues(partitionPaths.toSeq) &&
partitionSchema.nonEmpty) {
+ val predicate = partitionPruningPredicates.reduce(expressions.And)
+ val partitionFieldNames = partitionSchema.fieldNames
+ val transformedPredicate = predicate.transform {
+ case g @ GetStructField(_, _, Some(_)) =>
+ getFieldPath(g).flatMap { path =>
+ val idx = partitionFieldNames.indexWhere(name => resolve(path,
name))
+ if (idx >= 0) Some(BoundReference(idx,
partitionSchema(idx).dataType, nullable = true))
+ else None
+ }.getOrElse(g)
+ case a: AttributeReference =>
+ val index = partitionSchema.indexWhere(sf => resolve(a.name,
sf.name))
+ if (index >= 0) BoundReference(index,
partitionSchema(index).dataType, nullable = true)
+ else a
}
+ val boundPredicate: BasePredicate = try {
+ // Try using 1-arg constructor via reflection
+ val clazz =
Class.forName("org.apache.spark.sql.catalyst.expressions.InterpretedPredicate")
+ val ctor = clazz.getConstructor(classOf[Expression])
+ ctor.newInstance(transformedPredicate).asInstanceOf[BasePredicate]
+ } catch {
+ case _: NoSuchMethodException | _: IllegalArgumentException =>
+ // Fallback: Try using 2-arg constructor for certain Spark runtime
+ val clazz =
Class.forName("org.apache.spark.sql.catalyst.expressions.InterpretedPredicate")
+ val ctor = clazz.getConstructor(classOf[Expression],
classOf[Boolean])
+ ctor.newInstance(transformedPredicate, java.lang.Boolean.FALSE)
+ .asInstanceOf[BasePredicate]
+ }
+ val prunedPartitionPaths = partitionPaths.filter {
+ partitionPath =>
boundPredicate.eval(InternalRow.fromSeq(partitionPath.getValues))
+ }.toSeq
+
+ logInfo(s"Using provided predicates to prune number of target table's
partitions scanned from" +
+ s" ${partitionPaths.size} to ${prunedPartitionPaths.size}")
+
+ prunedPartitionPaths
+ } else {
+ logWarning(s"Unable to apply partition pruning, due to failure to parse
partition values from the" +
+ s" following path(s): ${partitionPaths.find(_.getValues.length ==
0).map(e => e.getPath)}")
+
+ partitionPaths.toSeq
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPartitionFilter.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPartitionFilter.scala
new file mode 100644
index 000000000000..564b49d2112f
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithPartitionFilter.scala
@@ -0,0 +1,131 @@
+/*
+ * 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, DataSourceWriteOptions}
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.model.HoodieTableType
+import org.apache.hudi.common.table.HoodieTableConfig
+import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.{DEFAULT_FIRST_PARTITION_PATH,
DEFAULT_SECOND_PARTITION_PATH}
+import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.hadoop.fs.RecordingLocalFileSystem
+import org.apache.hudi.hadoop.fs.RecordingLocalFileSystem.Call
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+import org.junit.jupiter.api.{AfterEach, BeforeEach}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.EnumSource
+
+import scala.collection.JavaConverters._
+
+/**
+ * Verifies that incremental reads apply partition filters when the commits in
range touch several
+ * partitions. Spark removes partition-only predicates from the post-scan
filters and relies on the
+ * file index to prune, so the incremental file index must evaluate them
against the modified partitions.
+ * Also verifies that resolving the file slices of an incremental read does
not list the table.
+ */
+class TestIncrementalReadWithPartitionFilter extends HoodieSparkClientTestBase
{
+
+ var spark: SparkSession = null
+
+ val commonOpts = Map(
+ "hoodie.insert.shuffle.parallelism" -> "4",
+ "hoodie.upsert.shuffle.parallelism" -> "4",
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "timestamp",
+ HoodieWriteConfig.TBL_NAME.key -> "hoodie_test"
+ )
+
+ @BeforeEach
+ override def setUp(): Unit = {
+ setTableName("hoodie_test")
+ initPath()
+ initSparkContexts()
+ spark = sqlContext.sparkSession
+ initTestDataGenerator()
+ initHoodieStorage()
+ }
+
+ @AfterEach
+ override def tearDown(): Unit = {
+ if (spark != null) {
+
RecordingLocalFileSystem.unregister(spark.sparkContext.hadoopConfiguration)
+ }
+ cleanupSparkContexts()
+ cleanupTestDataGenerator()
+ cleanupFileSystem()
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieTableType])
+ def testIncrementalReadWithPartitionFilter(tableType: HoodieTableType): Unit
= {
+ writeOneCommitPerPartition(tableType)
+ val incrementalDF = readIncrementally()
+
+ assertEquals(50, incrementalDF.count())
+ assertEquals(30, incrementalDF.where(s"partition =
'$DEFAULT_FIRST_PARTITION_PATH'").count())
+ assertEquals(20, incrementalDF.where(s"partition =
'$DEFAULT_SECOND_PARTITION_PATH'").count())
+ assertEquals(0, incrementalDF.where("partition = 'nonexistent'").count())
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieTableType])
+ def testIncrementalReadDoesNotListTable(tableType: HoodieTableType): Unit = {
+ writeOneCommitPerPartition(tableType)
+
+ // Record the file system calls of the read. With the metadata table
disabled, any partition
+ // listing shows up as a listStatus call on storage outside the .hoodie
folder.
+ RecordingLocalFileSystem.register(spark.sparkContext.hadoopConfiguration)
+ RecordingLocalFileSystem.reset()
+ assertEquals(50, readIncrementally(HoodieMetadataConfig.ENABLE.key ->
"false").count())
+
+ // The data files are opened through the recorder, so it is known to see
the calls under test.
+
assertTrue(RecordingLocalFileSystem.count(Call.operation("open").and(Call.pathEndsWith(".parquet")))
> 0)
+ val listings = Call.operation("listStatus",
"listLocatedStatus").and(Call.underMetaFolder().negate())
+ assertEquals(0, RecordingLocalFileSystem.count(listings),
RecordingLocalFileSystem.describe(listings))
+ }
+
+ // One commit per partition: 30 records in the first partition, 20 in the
second.
+ private def writeOneCommitPerPartition(tableType: HoodieTableType): Unit = {
+ Seq(("001", DEFAULT_FIRST_PARTITION_PATH, 30), ("002",
DEFAULT_SECOND_PARTITION_PATH, 20)).foreach {
+ case (instantTime, partition, numRecords) =>
+ val records =
recordsToStrings(dataGen.generateInsertsForPartition(instantTime, numRecords,
partition)).asScala.toList
+ spark.read.json(spark.sparkContext.parallelize(records,
2)).write.format("hudi")
+ .options(commonOpts)
+ .option(DataSourceWriteOptions.TABLE_TYPE.key, tableType.name())
+ .option(DataSourceWriteOptions.OPERATION.key,
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ }
+ }
+
+ // Reads from before the first commit, so every commit of the table is in
range.
+ private def readIncrementally(extraOpts: (String, String)*): DataFrame = {
+ spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, "000")
+ // Keep the incremental listing path under test rather than the full
table scan fallback.
+
.option(DataSourceReadOptions.INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN.key,
"false")
+ .options(extraOpts.toMap)
+ .load(basePath)
+ }
+}
