linliu-code commented on code in PR #13664:
URL: https://github.com/apache/hudi/pull/13664#discussion_r2261232496
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndex.scala:
##########
@@ -395,50 +397,121 @@ class TestPartitionStatsIndex extends
PartitionStatsIndexTestBase {
}
/**
- * 1. Enable column_stats, partition_stats and record_index (files already
enabled by default).
- * 2. Do an insert and validate the partition stats index initialization.
- * 3. Do an update and validate the partition stats index.
- * 4. Do a savepoint and restore, and validate partition_stats and
column_stats are deleted.
- * 5. Do an update and validate the partition stats index.
+ * 1. Enable column_stats, partition_stats and record_index (files/RLI
already enabled by default).
+ * 2. Do two inserts and validate index initialization.
+ * 3. Do a savepoint on the second commits.
+ * 4. Add three more commits to trigger clean, which cleans the files from
the first commit.
+ * 5. Restore, and validate partition_stats is deleted, but column_stats
partition exists.
+ * 6. Validate that column_stats does not contain records with file names
from first commit.
*/
@Test
def testPartitionStatsWithRestore(): Unit = {
val hudiOpts = commonOpts ++ Map(
- DataSourceWriteOptions.TABLE_TYPE.key ->
HoodieTableType.MERGE_ON_READ.name(),
- HoodieMetadataConfig.ENABLE.key() -> "true",
- HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key() -> "true",
- HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key() ->
"true",
- HoodieMetadataConfig.RECORD_INDEX_ENABLE_PROP.key() -> "true")
+ DataSourceWriteOptions.TABLE_TYPE.key ->
HoodieTableType.COPY_ON_WRITE.name(),
Review Comment:
Done.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndex.scala:
##########
@@ -395,50 +397,121 @@ class TestPartitionStatsIndex extends
PartitionStatsIndexTestBase {
}
/**
- * 1. Enable column_stats, partition_stats and record_index (files already
enabled by default).
- * 2. Do an insert and validate the partition stats index initialization.
- * 3. Do an update and validate the partition stats index.
- * 4. Do a savepoint and restore, and validate partition_stats and
column_stats are deleted.
- * 5. Do an update and validate the partition stats index.
+ * 1. Enable column_stats, partition_stats and record_index (files/RLI
already enabled by default).
+ * 2. Do two inserts and validate index initialization.
+ * 3. Do a savepoint on the second commits.
+ * 4. Add three more commits to trigger clean, which cleans the files from
the first commit.
+ * 5. Restore, and validate partition_stats is deleted, but column_stats
partition exists.
+ * 6. Validate that column_stats does not contain records with file names
from first commit.
*/
@Test
def testPartitionStatsWithRestore(): Unit = {
val hudiOpts = commonOpts ++ Map(
- DataSourceWriteOptions.TABLE_TYPE.key ->
HoodieTableType.MERGE_ON_READ.name(),
- HoodieMetadataConfig.ENABLE.key() -> "true",
- HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key() -> "true",
- HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key() ->
"true",
- HoodieMetadataConfig.RECORD_INDEX_ENABLE_PROP.key() -> "true")
+ DataSourceWriteOptions.TABLE_TYPE.key ->
HoodieTableType.COPY_ON_WRITE.name(),
+ HoodieMetadataConfig.ENABLE.key -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key -> "true",
+ HoodieMetadataConfig.RECORD_INDEX_ENABLE_PROP.key -> "true")
+ // First ingest.
doWriteAndValidateDataAndPartitionStats(
hudiOpts,
operation = DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
saveMode = SaveMode.Overwrite)
val firstCompletedInstant =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().lastInstant()
- doWriteAndValidateDataAndPartitionStats(hudiOpts, operation =
DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL, saveMode = SaveMode.Append)
- // validate files and record_index are present
-
assertTrue(metaClient.getTableConfig.getMetadataPartitions.contains(MetadataPartitionType.FILES.getPartitionPath))
-
assertTrue(metaClient.getTableConfig.getMetadataPartitions.contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath))
- // Do a savepoint
+ // Second ingest.
+ doWriteAndValidateDataAndPartitionStats(
+ hudiOpts,
+ operation = DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
+ saveMode = SaveMode.Append)
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+ val secondCompletedInstant = metaClient.getActiveTimeline
+ .getCommitsTimeline.filterCompletedInstants().getInstants().get(1)
+ // Validate index partitions are present
+ val initialMetadataPartitions =
metaClient.getTableConfig.getMetadataPartitions
+
assertTrue(initialMetadataPartitions.contains(MetadataPartitionType.FILES.getPartitionPath))
+
assertTrue(initialMetadataPartitions.contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath))
+
assertTrue(initialMetadataPartitions.contains(MetadataPartitionType.COLUMN_STATS.getPartitionPath))
+
assertTrue(initialMetadataPartitions.contains(MetadataPartitionType.PARTITION_STATS.getPartitionPath))
+ // Do a savepoint on the second commit.
val writeClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), getWriteConfig(hudiOpts))
- writeClient.savepoint(firstCompletedInstant.get().requestedTime,
"testUser", "savepoint to first commit")
+ writeClient.savepoint(secondCompletedInstant.requestedTime, "testUser",
"savepoint to second commit")
writeClient.close()
- val savepointTimestamp =
metaClient.reloadActiveTimeline().getSavePointTimeline.filterCompletedInstants().lastInstant().get().requestedTime
- assertEquals(firstCompletedInstant.get().requestedTime, savepointTimestamp)
+ val savepointTimestamp = metaClient.reloadActiveTimeline()
+
.getSavePointTimeline.filterCompletedInstants().lastInstant().get().requestedTime
+ assertEquals(secondCompletedInstant.requestedTime, savepointTimestamp)
+
+ // Add more ingests and trigger a clean to remove files from first
ingestion.
+ val writeOpt = hudiOpts ++ Map(
+ HoodieCleanConfig.AUTO_CLEAN.key -> "true",
+ HoodieCleanConfig.CLEAN_MAX_COMMITS.key -> "1",
+ HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key -> "2")
+ // Third ingest.
+ doWriteAndValidateDataAndPartitionStats(
Review Comment:
Adressed.
--
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]
