yihua commented on code in PR #20115:
URL: https://github.com/apache/hudi/pull/20115#discussion_r4189884110
##########
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))
Review Comment:
It is dropped: `prepareFileSlices` filters out entries with no slices
(`slices.filter(_._2.nonEmpty)`) before building directories, which covers both
this caller and `HoodieIncrementalFileIndex`. A partition with only empty
slices yields no directory, same as before.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala:
##########
@@ -890,6 +891,162 @@ 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 writeOpts = 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)
+ assertEquals(partitionCount, directories.size)
+ assertEquals(partitionCount, directories.map(_.values).distinct.size)
+ directories.foreach { directory =>
+
assertFalse(directory.values.isInstanceOf[HoodiePartitionFileSliceMapping])
+ assertEquals(commitCount, directory.files.size)
+ }
+ assertSameSplitPlanning(fileIndex, directories, exactFileOrder = true)
+
+ val incrementalRelation = spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
Review Comment:
Not fully. The snapshot cases build the index directly with the settings the
COW and MOR snapshot factories use (`includeLogFiles` false/true with embedded
slices), and only the incremental check goes through `spark.read`; MOR
read-optimized is not covered, where after the update the listing should be 5
plain directories holding the 15 base files. Loading snapshot and
read-optimized through `spark.read` and checking the rows read back (`key-0-0`
at `ts-3` vs `ts-0`) would cover both, plus the factory wiring.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieFileIndex.scala:
##########
@@ -890,6 +891,162 @@ 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 writeOpts = 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)
+ assertEquals(partitionCount, directories.size)
+ assertEquals(partitionCount, directories.map(_.values).distinct.size)
+ directories.foreach { directory =>
+
assertFalse(directory.values.isInstanceOf[HoodiePartitionFileSliceMapping])
+ assertEquals(commitCount, directory.files.size)
+ }
+ assertSameSplitPlanning(fileIndex, directories, exactFileOrder = true)
+
+ val incrementalRelation = spark.read.format("hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, "000")
+ .load(basePath)
+ .queryExecution.analyzed.collectFirst { case relation: LogicalRelation
=> relation.relation.asInstanceOf[HadoopFsRelation] }.get
+ assertEquals(partitionCount, incrementalRelation.location.listFiles(Nil,
Nil).size)
+
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ // update one record in each of the first two partitions, which adds log
files to one file group of each
+ upsertRecords(writeOpts, Seq(("key-0-0", 0, "ts-3"), ("key-0-1", 1,
"ts-3")))
+ fileIndex.refresh()
+
+ val directoriesWithLogFiles = fileIndex.listFiles(Nil, Nil)
+ val (perSliceDirectories, perPartitionDirectories) =
+
directoriesWithLogFiles.partition(_.values.isInstanceOf[HoodiePartitionFileSliceMapping])
+ assertEquals(partitionCount, perPartitionDirectories.size)
+ assertEquals(partitionCount * commitCount - 2,
perPartitionDirectories.map(_.files.size).sum)
+ assertEquals(2, perSliceDirectories.size)
+ perSliceDirectories.foreach(directory =>
assertTrue(getMappedSlice(directory).hasLogFiles))
+ assertSameSplitPlanning(fileIndex, directoriesWithLogFiles,
exactFileOrder = false)
+
+ // 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 writeOpts = 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) {
+ upsertRecords(writeOpts, Seq(("key-0-0", 0, "ts-3")))
+ 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)
+ }
+ }
+
+ /**
+ * Writes `commitCount` inserts, each adding a new file group to every
partition, and returns the write options.
+ */
+ private def writeFileGroups(tableType: HoodieTableType, partitionCount: Int,
commitCount: Int,
+ partitioned: Boolean = true): Map[String,
String] = {
+ 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")
Review Comment:
It is not needed here: the default Spark index is SIMPLE, which cannot index
log files, so `SparkUpsertDeltaCommitPartitioner.getSmallFileCandidates`
returns no candidates once `PARQUET_SMALL_FILE_LIMIT` is 0, before the
candidates limit is applied, and the upserts are updates to existing keys,
which always go to the log files of their file group on MOR. With an index that
can index log files, both limits are bypassed (any slice under the max file
size is a candidate), so pinning `hoodie.index.type=SIMPLE` would be the
stricter guard.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
