You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2026/10/06 03:55:43 UTC
(hudi) branch master updated: fix(spark): group file slices by partition in the file index listing (#20115)
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 438c0d372569 fix(spark): group file slices by partition in the file index listing (#20115)
438c0d372569 is described below
commit 438c0d372569bb40138cc3046069ed405f75bdbe
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Mon Oct 5 20:55:37 2026 -0700
fix(spark): group file slices by partition in the file index listing (#20115)
---
.../scala/org/apache/hudi/HoodieFileIndex.scala | 35 +--
.../apache/hudi/HoodieIncrementalFileIndex.scala | 10 +-
.../apache/hudi/PartitionDirectoryConverter.scala | 55 +++--
.../org/apache/hudi/TestHoodieFileIndex.scala | 255 ++++++++++++++++++++-
.../hudi/TestPartitionDirectoryConverter.scala | 46 +++-
5 files changed, 356 insertions(+), 45 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
index cd10aeb9c1fe..003c571acf16 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
@@ -200,11 +200,10 @@ case class HoodieFileIndex(spark: SparkSession,
partitionFilters
}
- val slices = filterFileSlices(dataFilters, effectivePartitionFilters).flatMap(
- { case (partitionOpt, fileSlices) =>
- fileSlices.filter(!_.isEmpty).map(fs => (InternalRow.fromSeq(partitionOpt.get.getValues), fs))
- }
- )
+ val slices = filterFileSlices(dataFilters, effectivePartitionFilters).map {
+ case (partitionOpt, fileSlices) =>
+ (InternalRow.fromSeq(partitionOpt.get.getValues), fileSlices.filter(!_.isEmpty))
+ }
prepareFileSlices(slices)
}
@@ -213,24 +212,26 @@ case class HoodieFileIndex(spark: SparkSession,
HoodieFileIndex.extractNestedPartitionFilters(dataFilters, getPartitionColumns.toSet)
}
- protected def prepareFileSlices(slices: Seq[(InternalRow, FileSlice)]): Seq[PartitionDirectory] = {
+ protected def prepareFileSlices(slices: Seq[(InternalRow, Seq[FileSlice])]): Seq[PartitionDirectory] = {
hasPushedDownPartitionPredicates = true
- val prunedPartitionsAndFilteredFileSlices = slices.map {
- case (partitionValues, fileSlice) =>
+ val prunedPartitionsAndFilteredFileSlices = slices.filter(_._2.nonEmpty).flatMap {
+ case (partitionValues, fileSlices) =>
if (shouldEmbedFileSlices) {
- PartitionDirectoryConverter.convertFileSliceToPartitionDirectory(
+ PartitionDirectoryConverter.convertFileSlicesToPartitionDirectories(
partitionValues,
- fileSlice,
+ fileSlices,
hoodieConfig)
} else {
- val baseFileStatusOpt = getBaseFileInfo(Option.apply(fileSlice.getBaseFile.orElse(null)))
- val logPathInfoStream = fileSlice.getLogFiles.map[StoragePathInfo](JFunction.toJavaFunction[HoodieLogFile, StoragePathInfo](lf => lf.getPathInfo))
- val files = logPathInfoStream.collect(Collectors.toList[StoragePathInfo]).asScala
- baseFileStatusOpt.foreach(f => files.append(f))
- val allCandidateFiles = files.map(fileInfo => new FileStatus(fileInfo.getLength, fileInfo.isDirectory, 0, fileInfo.getBlockSize,
- fileInfo.getModificationTime, new Path(fileInfo.getPath.toUri))).toSeq
- sparkAdapter.getSparkPartitionedFileUtils.newPartitionDirectory(partitionValues, allCandidateFiles)
+ val allCandidateFiles = fileSlices.flatMap { fileSlice =>
+ val baseFileStatusOpt = getBaseFileInfo(Option.apply(fileSlice.getBaseFile.orElse(null)))
+ val logPathInfoStream = fileSlice.getLogFiles.map[StoragePathInfo](JFunction.toJavaFunction[HoodieLogFile, StoragePathInfo](lf => lf.getPathInfo))
+ val files = logPathInfoStream.collect(Collectors.toList[StoragePathInfo]).asScala
+ baseFileStatusOpt.foreach(f => files.append(f))
+ files.map(fileInfo => new FileStatus(fileInfo.getLength, fileInfo.isDirectory, 0, fileInfo.getBlockSize,
+ fileInfo.getModificationTime, new Path(fileInfo.getPath.toUri)))
+ }
+ Seq(sparkAdapter.getSparkPartitionedFileUtils.newPartitionDirectory(partitionValues, allCandidateFiles))
}
}
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 4c2091e8feda..0568af6e9409 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
@@ -44,12 +44,10 @@ class HoodieIncrementalFileIndex(override val spark: SparkSession,
) with FileIndex {
override def listFiles(partitionFilters: Seq[Expression], dataFilters: Seq[Expression]): Seq[PartitionDirectory] = {
- val fileSlices = mergeOnReadIncrementalRelation.listFileSplits(partitionFilters, dataFilters).toSeq.flatMap(
- {
- case (partitionValues, fileSlices) =>
- fileSlices.filter(!_.isEmpty).map(fs => ( partitionValues, fs.withLogFiles(includeLogFiles)))
- }
- )
+ val fileSlices = mergeOnReadIncrementalRelation.listFileSplits(partitionFilters, dataFilters).toSeq.map {
+ case (partitionValues, fileSlices) =>
+ (partitionValues, fileSlices.filter(!_.isEmpty).map(_.withLogFiles(includeLogFiles)))
+ }
prepareFileSlices(fileSlices)
}
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionDirectoryConverter.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionDirectoryConverter.scala
index 2565278ab3eb..b6518a8b3279 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionDirectoryConverter.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionDirectoryConverter.scala
@@ -26,23 +26,32 @@ import org.apache.spark.sql.execution.datasources.PartitionDirectory
object PartitionDirectoryConverter extends SparkAdapterSupport {
- def convertFileSliceToPartitionDirectory(partitionValues: InternalRow,
- fileSlice: FileSlice,
- config: HoodieConfig): PartitionDirectory = {
- // 1. Generate a delegate file for file slice, which spark uses to optimize rdd partition parallelism based on data such as file size
- // - For file slice only has base file, we directly use the base file size as delegate file size
- // - For file slice has log file, we estimate the delegate file size based on the log file size and option(base file) size
- val estimationFileSize = fileSlice.getTotalFileSizeAsParquetFormat(config)
- val fileInfo = if (fileSlice.getBaseFile.isPresent) {
- fileSlice.getBaseFile.get().getPathInfo
+ /**
+ * Converts the file slices of one partition into partition directories.
+ *
+ * File slices that are read as a whole (with log files or a bootstrap base) each get their own
+ * directory whose partition values carry only that slice, so that a task is shipped just the slice
+ * it reads. Base-file-only slices need no such mapping and share a single directory per partition,
+ * so that consumers of the file index see one entry per partition for the common copy-on-write case.
+ */
+ def convertFileSlicesToPartitionDirectories(partitionValues: InternalRow,
+ fileSlices: Seq[FileSlice],
+ config: HoodieConfig): Seq[PartitionDirectory] = {
+ val (slicesReadAsFileSlice, baseFileOnlySlices) = fileSlices.partition(shouldReadAsFileSlice)
+ val baseFileOnlyDirectory = if (baseFileOnlySlices.isEmpty) {
+ Seq.empty
} else {
- fileSlice.getLogFiles.findAny().get().getPathInfo
+ Seq(sparkAdapter.getSparkPartitionedFileUtils.newPartitionDirectory(
+ partitionValues, baseFileOnlySlices.map(createDelegateFile(_, config))))
}
- // create a delegate file status based on the file size estimation
- val delegateFile = new FileStatus(estimationFileSize, fileInfo.isDirectory, 0, fileInfo.getBlockSize, fileInfo.getModificationTime, new Path(fileInfo.getPath.toUri))
+ baseFileOnlyDirectory ++ slicesReadAsFileSlice.map(convertFileSliceToPartitionDirectory(partitionValues, _, config))
+ }
- // 2. Generate a partition directory based on the delegate file and partition values
- if (fileSlice.hasLogFiles || fileSlice.hasBootstrapBase) {
+ def convertFileSliceToPartitionDirectory(partitionValues: InternalRow,
+ fileSlice: FileSlice,
+ config: HoodieConfig): PartitionDirectory = {
+ val delegateFile = createDelegateFile(fileSlice, config)
+ if (shouldReadAsFileSlice(fileSlice)) {
// should read as file slice, so we need to create a mapping from fileId to file slice
sparkAdapter.getSparkPartitionedFileUtils.newPartitionDirectory(
sparkAdapter.createPartitionFileSliceMapping(partitionValues, Map(fileSlice.getFileId -> fileSlice)), Seq(delegateFile))
@@ -51,4 +60,22 @@ object PartitionDirectoryConverter extends SparkAdapterSupport {
}
}
+ private def shouldReadAsFileSlice(fileSlice: FileSlice): Boolean = {
+ fileSlice.hasLogFiles || fileSlice.hasBootstrapBase
+ }
+
+ /**
+ * Generates a delegate file for the file slice, which spark uses to optimize rdd partition parallelism based on data such as file size
+ * - For file slice only has base file, we directly use the base file size as delegate file size
+ * - For file slice has log file, we estimate the delegate file size based on the log file size and option(base file) size
+ */
+ private def createDelegateFile(fileSlice: FileSlice, config: HoodieConfig): FileStatus = {
+ val estimationFileSize = fileSlice.getTotalFileSizeAsParquetFormat(config)
+ val fileInfo = if (fileSlice.getBaseFile.isPresent) {
+ fileSlice.getBaseFile.get().getPathInfo
+ } else {
+ fileSlice.getLogFiles.findAny().get().getPathInfo
+ }
+ new FileStatus(estimationFileSize, fileInfo.isDirectory, 0, fileInfo.getBlockSize, fileInfo.getModificationTime, new Path(fileInfo.getPath.toUri))
+ }
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
index 3dd498564f6d..7daee1636fa4 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala
@@ -23,18 +23,18 @@ import org.apache.hudi.HoodieConversionUtils.toJavaOption
import org.apache.hudi.HoodieFileIndex.DataSkippingFailureMode
import org.apache.hudi.client.HoodieJavaWriteClient
import org.apache.hudi.client.common.HoodieJavaEngineContext
-import org.apache.hudi.common.config.{HoodieMetadataConfig, HoodieStorageConfig}
+import org.apache.hudi.common.config.{HoodieConfig, HoodieMetadataConfig, HoodieStorageConfig, TypedProperties}
import org.apache.hudi.common.config.TimestampKeyGeneratorConfig.{TIMESTAMP_INPUT_DATE_FORMAT, TIMESTAMP_OUTPUT_DATE_FORMAT, TIMESTAMP_TYPE_FIELD}
import org.apache.hudi.common.engine.EngineType
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.model.{HoodieBaseFile, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.model.{FileSlice, HoodieBaseFile, HoodieRecord, HoodieTableType}
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient}
import org.apache.hudi.common.table.view.HoodieTableFileSystemView
import org.apache.hudi.common.testutils.{HoodieTestDataGenerator, HoodieTestUtils}
import org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
import org.apache.hudi.common.util.PartitionPathEncodeUtils
import org.apache.hudi.common.util.StringUtils.isNullOrEmpty
-import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.config.{HoodieCompactionConfig, HoodieWriteConfig}
import org.apache.hudi.exception.HoodieException
import org.apache.hudi.keygen.TimestampBasedAvroKeyGenerator.TimestampType
import org.apache.hudi.keygen.constant.KeyGeneratorType
@@ -44,16 +44,17 @@ import org.apache.hudi.testutils.HoodieSparkClientTestBase
import org.apache.hudi.util.JFunction
import org.apache.spark.sql._
+import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, EqualTo, Expression, GetStructField, GreaterThanOrEqual, LessThan, Literal, Or}
-import org.apache.spark.sql.execution.datasources.{NoopCache, PartitionDirectory}
+import org.apache.spark.sql.execution.datasources.{FileIndex, HadoopFsRelation, LogicalRelation, NoopCache, PartitionDirectory, PartitionedFile}
import org.apache.spark.sql.functions.{lit, struct}
import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
import org.junit.jupiter.api.{BeforeEach, Test}
-import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue}
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.{Arguments, CsvSource, MethodSource, ValueSource}
+import org.junit.jupiter.params.provider.{Arguments, CsvSource, EnumSource, MethodSource, ValueSource}
import java.util.Properties
import java.util.function.Consumer
@@ -61,7 +62,7 @@ import java.util.function.Consumer
import scala.collection.JavaConverters._
import scala.util.Random
-class TestHoodieFileIndex extends HoodieSparkClientTestBase with ScalaAssertionSupport {
+class TestHoodieFileIndex extends HoodieSparkClientTestBase with ScalaAssertionSupport with SparkAdapterSupport {
protected def withRDDPersistenceValidation(f: => Unit): Unit = {
org.apache.hudi.testutils.SparkRDDValidationUtils.withRDDPersistenceValidation(spark, new org.apache.hudi.testutils.SparkRDDValidationUtils.ThrowingRunnable {
override def run(): Unit = f
@@ -890,6 +891,246 @@ class TestHoodieFileIndex extends HoodieSparkClientTestBase with ScalaAssertionS
assertEquals(expected, HoodieFileIndex.extractNestedPartitionFilters(filters, partitionColumns))
}
+
+ /**
+ * The file index returns one directory per partition for base-file-only slices, so that consumers
+ * of the listing see partitions rather than file slices, while slices with log files keep a directory
+ * of their own. Split planning stays the same as for one directory per file slice.
+ */
+ @ParameterizedTest
+ @EnumSource(classOf[HoodieTableType])
+ def testListFilesGroupsFileSlicesByPartition(tableType: HoodieTableType): Unit = {
+ val partitionCount = 5
+ val commitCount = 3
+ val table = writeFileGroups(tableType, partitionCount, commitCount)
+ val fileIndex = HoodieFileIndex(spark, metaClient, None, queryOpts,
+ includeLogFiles = tableType == HoodieTableType.MERGE_ON_READ, shouldEmbedFileSlices = true)
+
+ val directories = fileIndex.listFiles(Nil, Nil)
+ assertDirectoryLayout(directories, partitionCount, partitionCount * commitCount, fileSlicesWithLogFilesCount = 0)
+ directories.foreach(directory => assertEquals(commitCount, directory.files.size))
+ assertSameSplitPlanning(fileIndex, directories, exactFileOrder = true)
+
+ val snapshot = DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL
+ val readOptimized = DataSourceReadOptions.QUERY_TYPE_READ_OPTIMIZED_OPT_VAL
+ val incremental = DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL
+ Seq(snapshot, readOptimized, incremental).foreach { queryType =>
+ assertDirectoryLayout(loadFileIndex(queryType).listFiles(Nil, Nil), partitionCount, partitionCount * commitCount,
+ fileSlicesWithLogFilesCount = 0)
+ assertEquals(table.timestamps.toSeq.sorted, readTimestamps(queryType))
+ }
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ // update one record in each of the first two partitions, which appends log files to one file group of each
+ val updates = Seq(("key-0-0", 0, "ts-3"), ("key-0-1", 1, "ts-3"))
+ upsertToLogFiles(table, updates)
+ fileIndex.refresh()
+
+ val directoriesWithLogFiles = fileIndex.listFiles(Nil, Nil)
+ assertDirectoryLayout(directoriesWithLogFiles, partitionCount, partitionCount * commitCount - 2, fileSlicesWithLogFilesCount = 2)
+ assertSameSplitPlanning(fileIndex, directoriesWithLogFiles, exactFileOrder = false)
+
+ // snapshot and incremental queries merge the log files, while read-optimized queries read the base files only
+ val updatedTimestamps = table.timestamps ++ updates.map { case (key, _, timestamp) => key -> timestamp }
+ Seq(snapshot, incremental).foreach { queryType =>
+ assertDirectoryLayout(loadFileIndex(queryType).listFiles(Nil, Nil), partitionCount, partitionCount * commitCount - 2,
+ fileSlicesWithLogFilesCount = 2)
+ assertEquals(updatedTimestamps.toSeq.sorted, readTimestamps(queryType))
+ }
+ assertDirectoryLayout(loadFileIndex(readOptimized).listFiles(Nil, Nil), partitionCount, partitionCount * commitCount,
+ fileSlicesWithLogFilesCount = 0)
+ assertEquals(table.timestamps.toSeq.sorted, readTimestamps(readOptimized))
+
+ // without embedded file slices, a partition directory lists the base and log files of the partition
+ val legacyFileIndex = HoodieFileIndex(spark, metaClient, None, queryOpts, includeLogFiles = true)
+ val legacyDirectories = legacyFileIndex.listFiles(Nil, Nil)
+ assertEquals(partitionCount, legacyDirectories.size)
+ assertFalse(legacyDirectories.exists(_.values.isInstanceOf[HoodiePartitionFileSliceMapping]))
+ assertEquals(partitionCount * commitCount + 2, legacyDirectories.map(_.files.size).sum)
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(classOf[HoodieTableType])
+ def testListFilesGroupsFileSlicesOfNonPartitionedTable(tableType: HoodieTableType): Unit = {
+ val commitCount = 3
+ val table = writeFileGroups(tableType, partitionCount = 1, commitCount, partitioned = false)
+ val fileIndex = HoodieFileIndex(spark, metaClient, None, queryOpts,
+ includeLogFiles = tableType == HoodieTableType.MERGE_ON_READ, shouldEmbedFileSlices = true)
+
+ val directories = fileIndex.listFiles(Nil, Nil)
+ assertEquals(1, directories.size)
+ assertEquals(commitCount, directories.head.files.size)
+ assertFalse(directories.head.values.isInstanceOf[HoodiePartitionFileSliceMapping])
+ assertSameSplitPlanning(fileIndex, directories, exactFileOrder = true)
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ val updates = Seq(("key-0-0", 0, "ts-3"))
+ upsertToLogFiles(table, updates)
+ fileIndex.refresh()
+
+ val directoriesWithLogFiles = fileIndex.listFiles(Nil, Nil)
+ assertEquals(2, directoriesWithLogFiles.size)
+ assertFalse(directoriesWithLogFiles.head.values.isInstanceOf[HoodiePartitionFileSliceMapping])
+ assertEquals(commitCount - 1, directoriesWithLogFiles.head.files.size)
+ assertTrue(getMappedSlice(directoriesWithLogFiles.last).hasLogFiles)
+ assertEquals((table.timestamps ++ updates.map { case (key, _, timestamp) => key -> timestamp }).toSeq.sorted,
+ readTimestamps(DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL))
+ }
+ }
+
+ /**
+ * A table written by [[writeFileGroups]]: its write options, its partition paths, and the timestamp written for each record key.
+ */
+ private case class FileGroupsTable(writeOpts: Map[String, String], partitionPaths: Seq[String], timestamps: Map[String, String])
+
+ /**
+ * Writes `commitCount` inserts, each adding a new file group to every partition, and checks on storage that
+ * every partition holds `commitCount` base-file-only slices, which the small file limit of 0 is there to ensure.
+ */
+ private def writeFileGroups(tableType: HoodieTableType, partitionCount: Int, commitCount: Int,
+ partitioned: Boolean = true): FileGroupsTable = {
+ val writeOpts = commonOpts ++ Map(
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType.name,
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> (if (partitioned) "partition" else ""),
+ HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key -> "0")
+ val partitionPaths = if (partitioned) (0 until partitionCount).map(_.toString) else Seq("")
+ val _spark = spark
+ import _spark.implicits._
+ val records = (0 until commitCount).flatMap { commit =>
+ val tuples = for (i <- 0 until 20) yield (s"key-$commit-$i", i % partitionCount, s"ts-$commit")
+ tuples.toDF("_row_key", "partition", "timestamp")
+ .write.format("hudi").options(writeOpts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .mode(if (commit == 0) SaveMode.Overwrite else SaveMode.Append)
+ .save(basePath)
+ tuples
+ }
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
+ latestFileSlices(partitionPaths).foreach { case (partitionPath, slices) =>
+ assertEquals(commitCount, slices.size, s"File groups in partition '$partitionPath'")
+ slices.foreach(slice => assertTrue(slice.getBaseFile.isPresent && !slice.hasLogFiles, s"Expected a base-file-only slice: $slice"))
+ }
+ FileGroupsTable(writeOpts, partitionPaths, records.map { case (key, _, timestamp) => key -> timestamp }.toMap)
+ }
+
+ /**
+ * Upserts records with existing keys and checks on storage that each update was appended as a log file to
+ * its file group, rather than merged into a new base file or written to a new file group.
+ */
+ private def upsertToLogFiles(table: FileGroupsTable, updates: Seq[(String, Int, String)]): Unit = {
+ def baseFiles(slices: Map[String, Seq[FileSlice]]): Set[String] =
+ slices.values.flatten.map(_.getBaseFile.get.getPath).toSet
+ val baseFilesBeforeUpdate = baseFiles(latestFileSlices(table.partitionPaths))
+
+ val _spark = spark
+ import _spark.implicits._
+ updates.toDF("_row_key", "partition", "timestamp")
+ .write.format("hudi").options(table.writeOpts)
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(basePath)
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
+ val slices = latestFileSlices(table.partitionPaths)
+ assertEquals(baseFilesBeforeUpdate, baseFiles(slices))
+ val slicesWithLogFiles = slices.values.flatten.filter(_.hasLogFiles).toSeq
+ assertEquals(updates.size, slicesWithLogFiles.size)
+ slicesWithLogFiles.foreach(slice => assertEquals(1L, slice.getLogFiles.count()))
+ }
+
+ /**
+ * Lists the latest file slices of the partitions from storage, independent of the file index under test.
+ */
+ private def latestFileSlices(partitionPaths: Seq[String]): Map[String, Seq[FileSlice]] = {
+ val fileSystemView = HoodieTableFileSystemView.fileListingBasedFileSystemView(
+ new HoodieJavaEngineContext(new HadoopStorageConfiguration(false)), metaClient,
+ metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants)
+ try {
+ partitionPaths.map(partitionPath =>
+ partitionPath -> fileSystemView.getLatestFileSlices(partitionPath).iterator().asScala.toList).toMap
+ } finally {
+ fileSystemView.close()
+ }
+ }
+
+ private def readTable(queryType: String): DataFrame = {
+ val incrementalOpts = if (queryType == DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL) {
+ Map(DataSourceReadOptions.START_COMMIT.key -> "000")
+ } else {
+ Map.empty[String, String]
+ }
+ spark.read.format("hudi").option(DataSourceReadOptions.QUERY_TYPE.key, queryType).options(incrementalOpts).load(basePath)
+ }
+
+ private def loadFileIndex(queryType: String): FileIndex = {
+ readTable(queryType).queryExecution.analyzed.collectFirst {
+ case relation: LogicalRelation => relation.relation.asInstanceOf[HadoopFsRelation].location
+ }.get
+ }
+
+ private def readTimestamps(queryType: String): Seq[(String, String)] = {
+ readTable(queryType).select("_row_key", "timestamp").collect().map(row => (row.getString(0), row.getString(1))).toSeq.sorted
+ }
+
+ /**
+ * Asserts one directory per partition holding all base-file-only slices, plus one directory per file slice with log files.
+ */
+ private def assertDirectoryLayout(directories: Seq[PartitionDirectory], partitionCount: Int,
+ baseFileOnlyCount: Int, fileSlicesWithLogFilesCount: Int): Unit = {
+ val (perSliceDirectories, perPartitionDirectories) =
+ directories.partition(_.values.isInstanceOf[HoodiePartitionFileSliceMapping])
+ assertEquals(partitionCount, perPartitionDirectories.size)
+ assertEquals(partitionCount, perPartitionDirectories.map(_.values).distinct.size)
+ assertEquals(baseFileOnlyCount, perPartitionDirectories.map(_.files.size).sum)
+ assertEquals(fileSlicesWithLogFilesCount, perSliceDirectories.size)
+ perSliceDirectories.foreach(directory => assertTrue(getMappedSlice(directory).hasLogFiles))
+ }
+
+ private def getMappedSlice(directory: PartitionDirectory): FileSlice = {
+ assertEquals(1, directory.files.size)
+ val fileId = FSUtils.getFileIdFromFilePath(new StoragePath(directory.files.head.getPath.toString))
+ directory.values.asInstanceOf[HoodiePartitionFileSliceMapping].getSlice(fileId).get
+ }
+
+ /**
+ * Asserts that Spark plans the same file partitions from the listed directories as from one directory
+ * per file slice. Equal-sized files may swap places between tasks when the order is not exact.
+ */
+ private def assertSameSplitPlanning(fileIndex: HoodieFileIndex,
+ directories: Seq[PartitionDirectory],
+ exactFileOrder: Boolean): Unit = {
+ val config = new HoodieConfig(TypedProperties.fromMap(queryOpts.asJava))
+ val perSliceDirectories = fileIndex.filterFileSlices(Nil, Nil).flatMap { case (partitionOpt, slices) =>
+ slices.map(slice => PartitionDirectoryConverter.convertFileSliceToPartitionDirectory(
+ InternalRow.fromSeq(partitionOpt.get.getValues), slice, config))
+ }
+ assertTrue(perSliceDirectories.size > directories.size)
+ val maxSplitBytes = 4 * 1024L
+ def plan(dirs: Seq[PartitionDirectory]): Seq[Seq[(String, Long, Long, InternalRow)]] = {
+ val files = dirs.flatMap(dir => sparkAdapter.splitFiles(spark, dir, isSplitable = true, maxSplitBytes))
+ .sortBy(_.length)(implicitly[Ordering[Long]].reverse)
+ sparkAdapter.getFilePartitions(spark, files, maxSplitBytes).map(_.files.toSeq.map(describe))
+ }
+ val expected = plan(perSliceDirectories)
+ val actual = plan(directories)
+ assertEquals(expected.size, actual.size)
+ if (exactFileOrder) {
+ assertEquals(expected, actual)
+ } else {
+ assertEquals(expected.map(_.map(_._3)), actual.map(_.map(_._3)))
+ assertEquals(expected.flatten.sortBy(f => (f._1, f._2)), actual.flatten.sortBy(f => (f._1, f._2)))
+ }
+ }
+
+ private def describe(file: PartitionedFile): (String, Long, Long, InternalRow) = {
+ val partitionValues = file.partitionValues match {
+ case mapping: HoodiePartitionFileSliceMapping => mapping.getPartitionValues
+ case values => values
+ }
+ (sparkAdapter.getSparkPartitionedFileUtils.getPathFromPartitionedFile(file).toString, file.start, file.length, partitionValues)
+ }
}
object TestHoodieFileIndex {
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionDirectoryConverter.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionDirectoryConverter.scala
index 204b2558960f..d36660b9d7d3 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionDirectoryConverter.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionDirectoryConverter.scala
@@ -19,7 +19,7 @@ package org.apache.hudi
import org.apache.hudi.common.config.{HoodieConfig, HoodieStorageConfig, TypedProperties}
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.model.{FileSlice, HoodieBaseFile, HoodieFileGroupId, HoodieLogFile}
+import org.apache.hudi.common.model.{BaseFile, FileSlice, HoodieBaseFile, HoodieFileGroupId, HoodieLogFile}
import org.apache.hudi.storage.{StoragePath, StoragePathInfo}
import org.apache.hudi.testutils.HoodieClientTestUtils.getSparkConfForTest
@@ -28,6 +28,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.execution.PartitionedFileUtil
import org.apache.spark.sql.execution.datasources.FilePartition
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource
@@ -113,6 +114,49 @@ class TestPartitionDirectoryConverter extends SparkAdapterSupport {
assert(directory.files.head.getLen == recordCount * fixedSizePerRecordWithParquetFormat)
}
+ /**
+ * Base-file-only slices of a partition share one directory with plain partition values, while slices
+ * with log files or a bootstrap base each keep a directory whose partition values map only that slice.
+ */
+ @Test
+ def testConvertFileSlicesToPartitionDirectories(): Unit = {
+ val logFraction = 0.5
+ val config = new HoodieConfig(TypedProperties.fromMap(Map(
+ HoodieStorageConfig.LOGFILE_TO_PARQUET_COMPRESSION_RATIO_FRACTION.key() -> logFraction.toString).asJava))
+ val partitionValues = InternalRow.fromSeq(Seq("2025-01-01"))
+ val baseOnlySlices = (0 until 3).map(i => buildFileSlice(s"base-$i", 100 + i, Seq(), logFraction))
+ val baseAndLogSlice = buildFileSlice("base-and-log", 200, Seq(30, 40), logFraction)
+ val logOnlySlice = buildFileSlice("log-only", 0, Seq(50), logFraction)
+ val bootstrapSlice = new FileSlice(new HoodieFileGroupId(partitionPath, "bootstrap"), baseInstant)
+ bootstrapSlice.setBaseFile(new HoodieBaseFile(buildHoodieBaseFile("bootstrap", 300).getPathInfo,
+ new BaseFile(buildHoodieBaseFile("bootstrap-source", 300).getPathInfo)))
+ val slices = Seq(baseOnlySlices(0), baseAndLogSlice, baseOnlySlices(1), logOnlySlice, bootstrapSlice, baseOnlySlices(2))
+
+ val directories = PartitionDirectoryConverter.convertFileSlicesToPartitionDirectories(partitionValues, slices, config)
+
+ assertEquals(4, directories.size)
+ val baseOnlyDirectory = directories.head
+ assertFalse(baseOnlyDirectory.values.isInstanceOf[HoodiePartitionFileSliceMapping])
+ assertEquals(partitionValues, baseOnlyDirectory.values)
+ assertEquals(baseOnlySlices.map(_.getBaseFile.get.getPath), baseOnlyDirectory.files.map(_.getPath.toString))
+ assertEquals(baseOnlySlices.map(_.getBaseFile.get.getFileSize), baseOnlyDirectory.files.map(_.getLen))
+
+ val slicesReadAsFileSlice = Seq(baseAndLogSlice, logOnlySlice, bootstrapSlice)
+ directories.tail.zip(slicesReadAsFileSlice).foreach { case (directory, slice) =>
+ val perSliceDirectory = PartitionDirectoryConverter.convertFileSliceToPartitionDirectory(partitionValues, slice, config)
+ assertEquals(perSliceDirectory.files.map(f => (f.getPath, f.getLen)), directory.files.map(f => (f.getPath, f.getLen)))
+ assertEquals(1, directory.files.size)
+ val mapping = directory.values.asInstanceOf[HoodiePartitionFileSliceMapping]
+ assertEquals(partitionValues, mapping.getPartitionValues)
+ assertTrue(mapping.getSlice(slice.getFileId).contains(slice))
+ slices.filter(_ != slice).foreach(other => assertTrue(mapping.getSlice(other.getFileId).isEmpty))
+ }
+
+ assertEquals(1, PartitionDirectoryConverter.convertFileSlicesToPartitionDirectories(partitionValues, baseOnlySlices, config).size)
+ assertEquals(3, PartitionDirectoryConverter.convertFileSlicesToPartitionDirectories(partitionValues, slicesReadAsFileSlice, config).size)
+ assertTrue(PartitionDirectoryConverter.convertFileSlicesToPartitionDirectories(partitionValues, Seq(), config).isEmpty)
+ }
+
private def verifyBalanceByNum(tasks: Seq[FilePartition], totalRecordNum: Int, logFraction: Double): Unit = {
val expectedRecordNumPerTask = totalRecordNum / tasks.size
val expectedToleranceMin = expectedRecordNumPerTask * 0.9