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 296f5d8283b9 fix(spark): fix incremental file index reporting
sizeInBytes=0 (#19912)
296f5d8283b9 is described below
commit 296f5d8283b9d6c576308a2bb78ddb7847fd090a
Author: Prashant Wason <[email protected]>
AuthorDate: Tue Oct 6 17:03:36 2026 -0700
fix(spark): fix incremental file index reporting sizeInBytes=0 (#19912)
---
.../scala/org/apache/hudi/DataSourceOptions.scala | 14 +++
.../apache/hudi/HoodieIncrementalFileIndex.scala | 16 ++-
.../hudi/MergeOnReadIncrementalRelationV1.scala | 5 +
.../hudi/MergeOnReadIncrementalRelationV2.scala | 6 +
.../TestIncrementalFileIndexSizeInBytes.scala | 127 +++++++++++++++++++++
5 files changed, 167 insertions(+), 1 deletion(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
index 6191331d4215..899b93a386f9 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
@@ -181,6 +181,20 @@ object DataSourceReadOptions {
.withDocumentation("Whether to skip clustering instants to avoid reading
base files of clustering operations "
+ "for streaming read to improve read performance.")
+ val INCREMENTAL_SIZE_ESTIMATION_ENABLE: ConfigProperty[Boolean] =
ConfigProperty
+ .key("hoodie.datasource.read.incr.size.estimation.enable")
+ .defaultValue(false)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("When enabled, the incremental file index reports the
total size of the files written by "
+ + "the commits in the queried range, taken from commit metadata, as its
sizeInBytes to the Spark planner, "
+ + "which may allow Spark to broadcast small incremental reads. When
disabled, it reports "
+ + "spark.sql.defaultSizeInBytes (the default of
BaseRelation.sizeInBytes, Long.MaxValue unless configured), "
+ + "matching the pre-1.0 IncrementalRelation and preventing Spark from
choosing BroadcastHashJoin for any "
+ + "subexpression that includes the incremental source. Spark's
selectivity estimates on filtered "
+ + "subexpressions of the incremental source can otherwise drop below the
broadcast threshold, causing "
+ + "broadcast timeouts on large tables.")
+
val TIME_TRAVEL_AS_OF_INSTANT: ConfigProperty[String] =
HoodieCommonConfig.TIMESTAMP_AS_OF
val ENABLE_DATA_SKIPPING: ConfigProperty[Boolean] = ConfigProperty
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 0568af6e9409..5b5312643aac 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
@@ -20,7 +20,7 @@ package org.apache.hudi
import org.apache.hudi.common.model.HoodieLogFile
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.storage.StoragePathInfo
-import org.apache.hudi.util.JFunction
+import org.apache.hudi.util.{JFunction, SparkConfigUtils}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.Expression
@@ -69,6 +69,20 @@ class HoodieIncrementalFileIndex(override val spark:
SparkSession,
}).map(fileStatus => fileStatus.getPath.toString).toArray
}
+ private val sizeEstimationEnabled: Boolean =
SparkConfigUtils.getStringWithAltKeys(
+ options,
DataSourceReadOptions.INCREMENTAL_SIZE_ESTIMATION_ENABLE).toBoolean
+
+ override def sizeInBytes: Long = {
+ lazy val defaultSizeInBytes = spark.sessionState.conf.defaultSizeInBytes
+ if (sizeEstimationEnabled) {
+ // Long.MaxValue means the relation falls back to a full table scan,
whose size is unknown here.
+ val actualSize = mergeOnReadIncrementalRelation.getIncrementalFilesSize
+ if (actualSize > 0 && actualSize < Long.MaxValue) actualSize else
defaultSizeInBytes
+ } else {
+ defaultSizeInBytes
+ }
+ }
+
def getRequiredFilters: Seq[Filter] = {
mergeOnReadIncrementalRelation.getRequiredFilters
}
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 0cc4be8df532..fe0996e49e63 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
@@ -156,6 +156,11 @@ case class MergeOnReadIncrementalRelationV1(override val
sqlContext: SQLContext,
}
}
+ override def getIncrementalFilesSize: Long = {
+ // A full table scan lists the whole table, so the affected files would
underestimate its size.
+ if (fullTableScan) Long.MaxValue else
affectedFilesInCommits.asScala.map(_.getLength).sum
+ }
+
override def shouldIncludeLogFiles(): Boolean = fullTableScan
private def filterFileSlices(fileSlices: Seq[FileSlice], pathGlobPattern:
String): Seq[FileSlice] = {
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 fb4ef6c38342..796e43ae8c61 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
@@ -47,6 +47,7 @@ trait MergeOnReadIncrementalRelation {
def listFileSplits(partitionFilters: Seq[Expression], dataFilters:
Seq[Expression]): Map[InternalRow, Seq[FileSlice]]
def getRequiredFilters: Seq[Filter]
def getInstantRange: HOption[InstantRange] = HOption.empty()
+ def getIncrementalFilesSize: Long
}
case class MergeOnReadIncrementalRelationV2(override val sqlContext:
SQLContext,
@@ -161,6 +162,11 @@ case class MergeOnReadIncrementalRelationV2(override val
sqlContext: SQLContext,
override def getInstantRange: HOption[InstantRange] =
if (fullTableScan) queryContext.getInstantRange else HOption.empty()
+ override def getIncrementalFilesSize: Long = {
+ // A full table scan lists the whole table, so the affected files would
underestimate its size.
+ if (fullTableScan) Long.MaxValue else
affectedFilesInCommits.asScala.map(_.getLength).sum
+ }
+
override def shouldIncludeLogFiles(): Boolean = fullTableScan
private def filterFileSlices(fileSlices: Seq[FileSlice], pathGlobPattern:
String): Seq[FileSlice] = {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalFileIndexSizeInBytes.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalFileIndexSizeInBytes.scala
new file mode 100644
index 000000000000..1029eb089f48
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalFileIndexSizeInBytes.scala
@@ -0,0 +1,127 @@
+/*
+ * 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,
HoodieIncrementalFileIndex}
+import org.apache.hudi.common.model.HoodieTableType
+import org.apache.hudi.common.table.HoodieTableConfig
+import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
+import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.testutils.{DataSourceTestUtils,
HoodieSparkClientTestBase}
+
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation,
LogicalRelation}
+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 the size reported to the Spark planner by
[[HoodieIncrementalFileIndex]], which must never be 0
+ * or Spark may pick a broadcast join against the incremental source.
+ */
+class TestIncrementalFileIndexSizeInBytes 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 = {
+ cleanupSparkContexts()
+ cleanupTestDataGenerator()
+ cleanupFileSystem()
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = classOf[HoodieTableType])
+ def testIncrementalFileIndexSizeInBytes(tableType: HoodieTableType): Unit = {
+ writeRecords(recordsToStrings(dataGen.generateInserts("001",
100)).asScala.toList, tableType,
+ DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ val commit1CompletionTime =
DataSourceTestUtils.latestCommitCompletionTime(storage, basePath)
+ // The upsert writes log files on MOR and new base file versions on COW.
+ writeRecords(recordsToStrings(dataGen.generateUniqueUpdates("002",
50)).asScala.toList, tableType,
+ DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
+ val commit2CompletionTime =
DataSourceTestUtils.latestCommitCompletionTime(storage, basePath)
+ val commit2Range = Map(
+ DataSourceReadOptions.START_COMMIT.key -> commit1CompletionTime,
+ DataSourceReadOptions.END_COMMIT.key -> commit2CompletionTime)
+
+ // Reports spark.sql.defaultSizeInBytes by default, matching the pre-1.0
IncrementalRelation.
+ assertEquals(spark.sessionState.conf.defaultSizeInBytes,
incrementalFileIndexOf(commit2Range).sizeInBytes)
+
+ // Reports the size of the files written by the commits in range when size
estimation is enabled.
+ val fileIndex = incrementalFileIndexOf(commit2Range ++ Map(
+ DataSourceReadOptions.INCREMENTAL_SIZE_ESTIMATION_ENABLE.key -> "true",
+ DataSourceReadOptions.INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN.key ->
"false"))
+ assertEquals(latestCommitWrittenFilesSize(), fileIndex.sizeInBytes)
+ }
+
+ private def writeRecords(records: List[String], tableType: HoodieTableType,
operation: String): Unit = {
+ spark.read.json(spark.sparkContext.parallelize(records,
2)).write.format("hudi")
+ .options(commonOpts)
+ .option(DataSourceWriteOptions.TABLE_TYPE.key, tableType.name())
+ .option(DataSourceWriteOptions.OPERATION.key, operation)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ }
+
+ private def latestCommitWrittenFilesSize(): Long = {
+ val metaClient = HoodieTestUtils.createMetaClient(storage, basePath)
+ val timeline = metaClient.getActiveTimeline
+ val commitMetadata = timeline.readCommitMetadata(
+
timeline.getCommitsTimeline.filterCompletedInstants().lastInstant().get())
+ val writeStats =
commitMetadata.getPartitionToWriteStats.values().asScala.flatMap(_.asScala)
+ assertTrue(writeStats.nonEmpty, "Expected the latest commit to write
files")
+ writeStats.map(stat => stat.getPath ->
stat.getFileSizeInBytes).toMap.values.sum
+ }
+
+ private def incrementalFileIndexOf(opts: Map[String, String]):
HoodieIncrementalFileIndex = {
+ val df: DataFrame = spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .options(opts)
+ .load(basePath)
+ df.queryExecution.analyzed.collectFirst {
+ case relation: LogicalRelation if
relation.relation.isInstanceOf[HadoopFsRelation] =>
+ relation.relation.asInstanceOf[HadoopFsRelation].location
+ } match {
+ case Some(fileIndex: HoodieIncrementalFileIndex) => fileIndex
+ case other => throw new IllegalStateException(s"Expected a
HoodieIncrementalFileIndex but got $other")
+ }
+ }
+}
