yihua commented on code in PR #20115:
URL: https://github.com/apache/hudi/pull/20115#discussion_r4189884523
##########
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]
