yihua commented on code in PR #20115:
URL: https://github.com/apache/hudi/pull/20115#discussion_r4189854297
##########
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:
Is `hoodie.merge.small.file.group.candidates.limit` set to 0 so MOR only
writes log files for updates?
##########
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:
Are snapshot and read-optimized queries both tested?
--
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]
