More cleanup in HoodieTableMetadataUtil

Author: codope

hudi-cli/src/main/java/org/apache/hudi/cli/commands/MetadataCommand.java

@@ -139,7 +139,7 @@ public String create(
   @ShellMethod(key = "metadata delete", value = "Remove the Metadata Table")
   public String delete(@ShellOption(value = "--backup", help = "Backup the metadata table before delete", defaultValue = "true", arity = 1) final boolean backup) throws Exception {
     HoodieTableMetaClient dataMetaClient = HoodieCLI.getTableMetaClient();
-    String backupPath = HoodieTableMetadataUtil.deleteMetadataTable(dataMetaClient, new HoodieSparkEngineContext(jsc), backup);
+    String backupPath = HoodieTableMetadataUtil.deleteMetadataTable(dataMetaClient, backup);
     if (backup) {
       return "Metadata Table has been deleted and backed up to " + backupPath;
     } else {
@@ -150,8 +150,7 @@ public String delete(@ShellOption(value = "--backup", help = "Backup the metadat
   @ShellMethod(key = "metadata delete-record-index", value = "Delete the record index from Metadata Table")
   public String deleteRecordIndex(@ShellOption(value = "--backup", help = "Backup the record index before delete", defaultValue = "true", arity = 1) final boolean backup) throws Exception {
     HoodieTableMetaClient dataMetaClient = HoodieCLI.getTableMetaClient();
-    String backupPath = HoodieTableMetadataUtil.deleteMetadataTablePartition(dataMetaClient, new HoodieSparkEngineContext(jsc),
-        MetadataPartitionType.RECORD_INDEX.getPartitionPath(), backup);
+    String backupPath = HoodieTableMetadataUtil.deleteMetadataTablePartition(dataMetaClient, MetadataPartitionType.RECORD_INDEX.getPartitionPath(), backup);
     if (backup) {
       return "Record Index has been deleted from the Metadata Table and backed up to " + backupPath;
     } else {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java

@@ -99,6 +99,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static java.util.Objects.requireNonNull;
 import static org.apache.hudi.common.config.HoodieMetadataConfig.DEFAULT_METADATA_POPULATE_META_FIELDS;
 import static org.apache.hudi.common.table.HoodieTableConfig.TIMELINE_HISTORY_PATH;
 import static org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED;
@@ -309,7 +310,7 @@ private boolean metadataTableExists(HoodieTableMetaClient dataMetaClient) throws
     if (reInitialize) {
       metrics.ifPresent(m -> m.incrementMetric(HoodieMetadataMetrics.REBOOTSTRAP_STR, 1));
       LOG.info("Deleting Metadata Table directory so that it can be re-initialized");
-      HoodieTableMetadataUtil.deleteMetadataTable(dataMetaClient, engineContext, false);
+      HoodieTableMetadataUtil.deleteMetadataTable(dataMetaClient, false);
       exists = false;
     }
 
@@ -550,7 +551,7 @@ private Pair<List<String>, Pair<Integer, HoodieData<HoodieRecord>>> initializeCo
 
     // during initialization, we need stats for base and log files.
     HoodieData<HoodieRecord> records = HoodieTableMetadataUtil.convertFilesToColumnStatsRecords(
-        engineContext, Collections.emptyMap(), partitionToFilesMap, dataMetaClient, dataWriteConfig.getMetadataConfig(),
+        engineContext, Collections.emptyMap(), partitionToFilesMap, dataMetaClient,
         dataWriteConfig.getColumnStatsIndexParallelism(),
         dataWriteConfig.getMetadataConfig().getMaxReaderBufferSize(),
         columnsToIndex);
@@ -591,25 +592,27 @@ protected abstract HoodieData<HoodieRecord> getExpressionIndexRecords(List<Pair<
   private Pair<Integer, HoodieData<HoodieRecord>> initializeExpressionIndexPartition(String indexName, String instantTime) throws Exception {
     HoodieIndexDefinition indexDefinition = getIndexDefinition(indexName);
     ValidationUtils.checkState(indexDefinition != null, "Expression Index definition is not present for index " + indexName);
+    List<Pair<String, Pair<String, Long>>> partitionFilePathSizeTriplet = getPartitionFilePathSizeTriplet();
+    int fileGroupCount = dataWriteConfig.getMetadataConfig().getExpressionIndexFileGroupCount();
+    int parallelism = Math.min(partitionFilePathSizeTriplet.size(), dataWriteConfig.getMetadataConfig().getExpressionIndexParallelism());
+    Schema readerSchema = getProjectedSchemaForExpressionIndex(indexDefinition, dataMetaClient);
+    return Pair.of(fileGroupCount, getExpressionIndexRecords(partitionFilePathSizeTriplet, indexDefinition, dataMetaClient, parallelism, readerSchema, storageConf, instantTime));
+  }
+
+  private List<Pair<String, Pair<String, Long>>> getPartitionFilePathSizeTriplet() throws IOException {
     List<Pair<String, FileSlice>> partitionFileSlicePairs = getPartitionFileSlicePairs();
     List<Pair<String, Pair<String, Long>>> partitionFilePathSizeTriplet = new ArrayList<>();
     partitionFileSlicePairs.forEach(entry -> {
       if (entry.getValue().getBaseFile().isPresent()) {
         partitionFilePathSizeTriplet.add(Pair.of(entry.getKey(), Pair.of(entry.getValue().getBaseFile().get().getPath(), entry.getValue().getBaseFile().get().getFileLen())));
       }
       entry.getValue().getLogFiles().forEach(hoodieLogFile -> {
-        if (entry.getValue().getLogFiles().count() > 0) {
-          entry.getValue().getLogFiles().forEach(logfile -> {
-            partitionFilePathSizeTriplet.add(Pair.of(entry.getKey(), Pair.of(logfile.getPath().toString(), logfile.getFileSize())));
-          });
+        if (entry.getValue().getLogFiles().findAny().isPresent()) {
+          entry.getValue().getLogFiles().forEach(logfile -> partitionFilePathSizeTriplet.add(Pair.of(entry.getKey(), Pair.of(logfile.getPath().toString(), logfile.getFileSize()))));
         }
       });
     });
-
-    int fileGroupCount = dataWriteConfig.getMetadataConfig().getExpressionIndexFileGroupCount();
-    int parallelism = Math.min(partitionFilePathSizeTriplet.size(), dataWriteConfig.getMetadataConfig().getExpressionIndexParallelism());
-    Schema readerSchema = getProjectedSchemaForExpressionIndex(indexDefinition, dataMetaClient);
-    return Pair.of(fileGroupCount, getExpressionIndexRecords(partitionFilePathSizeTriplet, indexDefinition, dataMetaClient, parallelism, readerSchema, storageConf, instantTime));
+    return partitionFilePathSizeTriplet;
   }
 
   HoodieIndexDefinition getIndexDefinition(String indexName) {
@@ -661,7 +664,7 @@ private Pair<Integer, HoodieData<HoodieRecord>> initializeRecordIndexPartition()
     // Collect the list of latest base files present in each partition
     List<String> partitions = metadata.getAllPartitionPaths();
     fsView.loadAllPartitions();
-    HoodieData<HoodieRecord> records = null;
+    HoodieData<HoodieRecord> records;
     if (dataMetaClient.getTableConfig().getTableType() == HoodieTableType.COPY_ON_WRITE) {
       // for COW, we can only consider base files to initialize.
       final List<Pair<String, HoodieBaseFile>> partitionBaseFilePairs = new ArrayList<>();
@@ -670,8 +673,7 @@ private Pair<Integer, HoodieData<HoodieRecord>> initializeRecordIndexPartition()
             .map(basefile -> Pair.of(partition, basefile)).collect(Collectors.toList()));
       }
 
-      LOG.info("Initializing record index from " + partitionBaseFilePairs.size() + " base files in "
-          + partitions.size() + " partitions");
+      LOG.info("Initializing record index from {} base files in {} partitions", partitionBaseFilePairs.size(), partitions.size());
 
       // Collect record keys from the files in parallel
       records = readRecordKeysFromBaseFiles(
@@ -684,13 +686,12 @@ private Pair<Integer, HoodieData<HoodieRecord>> initializeRecordIndexPartition()
           this.getClass().getSimpleName());
     } else {
       final List<Pair<String, FileSlice>> partitionFileSlicePairs = new ArrayList<>();
-      String latestCommit = dataMetaClient.getActiveTimeline().filterCompletedAndCompactionInstants().lastInstant().map(instant -> instant.requestedTime()).orElse(SOLO_COMMIT_TIMESTAMP);
+      String latestCommit = dataMetaClient.getActiveTimeline().filterCompletedAndCompactionInstants().lastInstant().map(HoodieInstant::requestedTime).orElse(SOLO_COMMIT_TIMESTAMP);
       for (String partition : partitions) {
         fsView.getLatestMergedFileSlicesBeforeOrOn(partition, latestCommit).forEach(fs -> partitionFileSlicePairs.add(Pair.of(partition, fs)));
       }
 
-      LOG.info("Initializing record index from " + partitionFileSlicePairs.size() + " file slices in "
-          + partitions.size() + " partitions");
+      LOG.info("Initializing record index from {} file slices in {} partitions", partitionFileSlicePairs.size(), partitions.size());
       records = readRecordKeysFromFileSliceSnapshot(
           engineContext,
           partitionFileSlicePairs,
@@ -722,7 +723,7 @@ private Pair<Integer, HoodieData<HoodieRecord>> initializeRecordIndexPartition()
    * @param metaClient metaclient instance to use.
    * @param dataWriteConfig write config to use.
    * @param hoodieTable hoodie table instance of interest.
-   * @return
+   * @return {@link HoodieData} of {@link HoodieRecord} containing updates for record_index.
    */
   private static HoodieData<HoodieRecord> readRecordKeysFromFileSliceSnapshot(HoodieEngineContext engineContext,
                                                                               List<Pair<String, FileSlice>> partitionFileSlicePairs,
@@ -752,7 +753,7 @@ private static HoodieData<HoodieRecord> readRecordKeysFromFileSliceSnapshot(Hood
           Option.of(fileSlice)).getMergedRecords().stream().map(record -> {
             HoodieRecord record1 = (HoodieRecord) record;
             return HoodieMetadataPayload.createRecordIndexUpdate(record1.getRecordKey(), partition, fileId,
-            record1.getCurrentLocation().getInstantTime(), 0);
+            requireNonNull(record1.getCurrentLocation()).getInstantTime(), 0);
           }).iterator();
     });
   }
@@ -845,9 +846,10 @@ private List<DirectoryInfo> listAllPartitionsFromFilesystem(String initializatio
       // List all directories in parallel
       engineContext.setJobStatus(this.getClass().getSimpleName(), "Listing " + numDirsToList + " partitions from filesystem");
       List<DirectoryInfo> processedDirectories = engineContext.map(pathsToProcess, path -> {
-        HoodieStorage storage = new HoodieHadoopStorage(path, storageConf);
-        String relativeDirPath = FSUtils.getRelativePartitionPath(storageBasePath, path);
-        return new DirectoryInfo(relativeDirPath, storage.listDirectEntries(path), initializationTime, pendingDataInstants);
+        try (HoodieStorage storage = new HoodieHadoopStorage(path, storageConf)) {
+          String relativeDirPath = FSUtils.getRelativePartitionPath(storageBasePath, path);
+          return new DirectoryInfo(relativeDirPath, storage.listDirectEntries(path), initializationTime, pendingDataInstants);
+        }
       }, numDirsToList);
 
       // If the listing reveals a directory, add it to queue. If the listing reveals a hoodie partition, add it to
@@ -910,7 +912,7 @@ private void initializeFileGroups(HoodieTableMetaClient dataMetaClient, Metadata
     HoodieStorage storage = metadataMetaClient.getStorage();
     try {
       final List<StoragePathInfo> existingFiles = storage.listDirectEntries(partitionPath);
-      if (existingFiles.size() > 0) {
+      if (!existingFiles.isEmpty()) {
         LOG.warn("Deleting all existing files found in MDT partition {}", partitionName);
         storage.deleteDirectory(partitionPath);
         ValidationUtils.checkState(!storage.exists(partitionPath),
@@ -1202,7 +1204,7 @@ public void update(HoodieRestoreMetadata restoreMetadata, String instantTime) {
     InstantGenerator datainstantGenerator = dataMetaClient.getInstantGenerator();
     HoodieInstant restoreInstant = datainstantGenerator.createNewInstant(REQUESTED, HoodieTimeline.RESTORE_ACTION, instantTime);
     HoodieInstant requested = datainstantGenerator.getRestoreRequestedInstant(restoreInstant);
-    HoodieRestorePlan restorePlan = null;
+    HoodieRestorePlan restorePlan;
     try {
       restorePlan = TimelineMetadataUtils.deserializeAvroMetadata(
           dataMetaClient.getActiveTimeline().readRestoreInfoAsBytes(requested).get(), HoodieRestorePlan.class);
@@ -1608,7 +1610,7 @@ private void fetchOutofSyncFilesRecordsFromMetadataTable(Map<String, DirectoryIn
                                                            Map<String, List<String>> partitionFilesToDelete, List<String> partitionsToDelete) throws IOException {
 
     for (String partition : metadata.fetchAllPartitionPaths()) {
-      StoragePath partitionPath = null;
+      StoragePath partitionPath;
       if (StringUtils.isNullOrEmpty(partition) && !dataMetaClient.getTableConfig().isTablePartitioned()) {
         partitionPath = new StoragePath(dataWriteConfig.getBasePath());
       } else {
@@ -1619,7 +1621,7 @@ private void fetchOutofSyncFilesRecordsFromMetadataTable(Map<String, DirectoryIn
       if (!dirInfoMap.containsKey(partition)) {
         // Entire partition has been deleted
         partitionsToDelete.add(partitionId);
-        if (metadataFiles != null && metadataFiles.size() > 0) {
+        if (metadataFiles != null && !metadataFiles.isEmpty()) {
           partitionFilesToDelete.put(partitionId, metadataFiles.stream().map(f -> f.getPath().getName()).collect(Collectors.toList()));
         }
       } else {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseActionExecutor.java

@@ -74,7 +74,7 @@ protected final void writeTableMetadata(HoodieCommitMetadata metadata, String ac
     // Recreate MDT for insert_overwrite_table operation.
     if (table.getConfig().isMetadataTableEnabled()
         && WriteOperationType.INSERT_OVERWRITE_TABLE == metadata.getOperationType()) {
-      HoodieTableMetadataUtil.deleteMetadataTable(table.getMetaClient(), table.getContext(), false);
+      HoodieTableMetadataUtil.deleteMetadataTable(table.getMetaClient(), false);
     }
 
     // MDT should be recreated if it has been deleted for insert_overwrite_table operation.
@@ -157,7 +157,7 @@ protected final void dropIndexOnRestore() {
     for (String partitionPath : table.getMetaClient().getTableConfig().getMetadataPartitions()) {
       if (MetadataPartitionType.shouldDeletePartitionOnRestore(partitionPath)) {
         // setting backup to true as this delete is part of restore operation
-        HoodieTableMetadataUtil.deleteMetadataTablePartition(table.getMetaClient(), context, partitionPath, true);
+        HoodieTableMetadataUtil.deleteMetadataTablePartition(table.getMetaClient(), partitionPath, true);
       }
     }
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/EightToSevenDowngradeHandler.java

@@ -346,21 +346,17 @@ static void downgradeMetadataPartitions(HoodieEngineContext context,
         false);
 
     // Delete partitions.
-    List<String> validPartitionPaths = deleteMetadataPartition(context, metaClient, metadataPartitions);
+    List<String> validPartitionPaths = deleteMetadataPartition(metaClient, metadataPartitions);
 
     // Clean the configuration.
     tablePropsToAdd.put(TABLE_METADATA_PARTITIONS, String.join(",", validPartitionPaths));
   }
 
-  static List<String> deleteMetadataPartition(HoodieEngineContext context,
-                                              HoodieTableMetaClient metaClient,
+  static List<String> deleteMetadataPartition(HoodieTableMetaClient metaClient,
                                               List<String> metadataPartitions) {
     metadataPartitions.stream()
         .filter(metadataPath -> !SUPPORTED_METADATA_PARTITION_PATHS.contains(metadataPath))
-        .forEach(metadataPath ->
-            HoodieTableMetadataUtil.deleteMetadataTablePartition(
-                metaClient, context, metadataPath, true)
-        );
+        .forEach(metadataPath -> HoodieTableMetadataUtil.deleteMetadataTablePartition(metaClient, metadataPath, true));
 
     return metadataPartitions.stream()
         .filter(SUPPORTED_METADATA_PARTITION_PATHS::contains)

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestEightToSevenDowngradeHandler.java

@@ -104,19 +104,19 @@ void setUp() {
   void testDeleteMetadataPartition() {
     try (MockedStatic<HoodieTableMetadataUtil> mockedMetadataUtils = mockStatic(HoodieTableMetadataUtil.class)) {
       List<String> leftPartitionPaths =
-          EightToSevenDowngradeHandler.deleteMetadataPartition(context, metaClient, SAMPLE_METADATA_PATHS);
+          EightToSevenDowngradeHandler.deleteMetadataPartition(metaClient, SAMPLE_METADATA_PATHS);
 
       mockedMetadataUtils.verify(
           () -> HoodieTableMetadataUtil.deleteMetadataTablePartition(
-              metaClient, context, "expr_index_random", true),
+              metaClient, "expr_index_random", true),
           times(1));
       mockedMetadataUtils.verify(
           () -> HoodieTableMetadataUtil.deleteMetadataTablePartition(
-              metaClient, context, "secondary_index_random", true),
+              metaClient, "secondary_index_random", true),
           times(1));
       mockedMetadataUtils.verify(
           () -> HoodieTableMetadataUtil.deleteMetadataTablePartition(
-              metaClient, context, "partition_stats", true),
+              metaClient, "partition_stats", true),
           times(1));
 
       assertArrayEquals(new String[] {"files", "column_stats"}, leftPartitionPaths.toArray());

hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java

@@ -2270,7 +2270,7 @@ public void testBootstrapWithTableNotFound() throws Exception {
         new StoragePath(getMetadataTableBasePath(writeConfig.getBasePath()));
     assertTrue(storage.exists(metadataTablePath), "metadata table should exist.");
 
-    deleteMetadataTable(metaClient, context, false);
+    deleteMetadataTable(metaClient, false);
     assertFalse(storage.exists(metadataTablePath),
         "metadata table should not exist after being deleted.");