jonvex commented on code in PR #13664:
URL: https://github.com/apache/hudi/pull/13664#discussion_r2255582001
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1604,14 +1604,57 @@ public void update(HoodieRestoreMetadata
restoreMetadata, String instantTime) {
// We need to choose a timestamp which would be a validInstantTime for
MDT. This is either a commit timestamp completed on the dataset
// or a new timestamp which we use for MDT clean, compaction etc.
String syncCommitTime = createRestoreInstantTime();
- processAndCommit(syncCommitTime, () ->
HoodieTableMetadataUtil.convertMissingPartitionRecords(engineContext,
+ // For Files partition.
+ Map<String, HoodieData<HoodieRecord>> partitionRecords = new HashMap<>();
+
partitionRecords.putAll(HoodieTableMetadataUtil.convertMissingPartitionRecords(engineContext,
partitionsToDelete, partitionFilesToAdd, partitionFilesToDelete,
syncCommitTime));
+ // For ColumnStats partition.
+ if
(dataMetaClient.getTableConfig().getMetadataPartitions().contains(COLUMN_STATS.getPartitionPath()))
{
+ partitionRecords.putAll(convertToColumnStatsRecord(
+ partitionFilesToAdd, partitionFilesToDelete, engineContext,
dataMetaClient,
+ dataWriteConfig.getMetadataConfig(),
Option.of(dataWriteConfig.getRecordMerger().getRecordType()),
+
dataWriteConfig.getMetadataConfig().getColumnStatsIndexParallelism()));
+ }
+ processAndCommit(syncCommitTime, () -> partitionRecords);
Review Comment:
why is this using a supplier? Should we pull everything here into a lambda
to prevent early execution?
##########
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:
maybe use a for loop here?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1604,14 +1604,57 @@ public void update(HoodieRestoreMetadata
restoreMetadata, String instantTime) {
// We need to choose a timestamp which would be a validInstantTime for
MDT. This is either a commit timestamp completed on the dataset
// or a new timestamp which we use for MDT clean, compaction etc.
String syncCommitTime = createRestoreInstantTime();
- processAndCommit(syncCommitTime, () ->
HoodieTableMetadataUtil.convertMissingPartitionRecords(engineContext,
+ // For Files partition.
+ Map<String, HoodieData<HoodieRecord>> partitionRecords = new HashMap<>();
+
partitionRecords.putAll(HoodieTableMetadataUtil.convertMissingPartitionRecords(engineContext,
partitionsToDelete, partitionFilesToAdd, partitionFilesToDelete,
syncCommitTime));
+ // For ColumnStats partition.
+ if
(dataMetaClient.getTableConfig().getMetadataPartitions().contains(COLUMN_STATS.getPartitionPath()))
{
+ partitionRecords.putAll(convertToColumnStatsRecord(
+ partitionFilesToAdd, partitionFilesToDelete, engineContext,
dataMetaClient,
+ dataWriteConfig.getMetadataConfig(),
Option.of(dataWriteConfig.getRecordMerger().getRecordType()),
+
dataWriteConfig.getMetadataConfig().getColumnStatsIndexParallelism()));
+ }
+ processAndCommit(syncCommitTime, () -> partitionRecords);
+ // Close.
Review Comment:
nit: don't need this comment
##########
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:
this seems pretty significant to change this from mor to cow?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1604,14 +1604,57 @@ public void update(HoodieRestoreMetadata
restoreMetadata, String instantTime) {
// We need to choose a timestamp which would be a validInstantTime for
MDT. This is either a commit timestamp completed on the dataset
// or a new timestamp which we use for MDT clean, compaction etc.
String syncCommitTime = createRestoreInstantTime();
- processAndCommit(syncCommitTime, () ->
HoodieTableMetadataUtil.convertMissingPartitionRecords(engineContext,
+ // For Files partition.
+ Map<String, HoodieData<HoodieRecord>> partitionRecords = new HashMap<>();
+
partitionRecords.putAll(HoodieTableMetadataUtil.convertMissingPartitionRecords(engineContext,
partitionsToDelete, partitionFilesToAdd, partitionFilesToDelete,
syncCommitTime));
+ // For ColumnStats partition.
+ if
(dataMetaClient.getTableConfig().getMetadataPartitions().contains(COLUMN_STATS.getPartitionPath()))
{
Review Comment:
verify that this is the right check. Do we also need to check if config is
enabled? It might be fine just double check
##########
hudi-common/src/main/java/org/apache/hudi/metadata/MetadataPartitionType.java:
##########
@@ -420,7 +420,9 @@ public HoodieMetadataPayload
combineMetadataPayloads(HoodieMetadataPayload older
* Check if the partition path should be deleted on restore.
*/
public static boolean shouldDeletePartitionOnRestore(String partitionPath) {
- return fromPartitionPath(partitionPath) != FILES &&
fromPartitionPath(partitionPath) != RECORD_INDEX;
+ return fromPartitionPath(partitionPath) != FILES
Review Comment:
use a variable so we don't call this 3 times
--
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]
