add file listing cache

hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/FileListingCacheManager.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.hive;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.sync.common.HoodieSyncClient;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+
+public class FileListingCacheManager {
+    private static final Logger LOG = LogManager.getLogger(FileListingCacheManager.class);
+    protected HoodieSyncClient syncClient;
+    protected String cacheFilePath;
+    public FileListingCacheManager(HoodieSyncClient syncClient, String cacheFilePath) {
+        this.syncClient = syncClient;
+        this.cacheFilePath = cacheFilePath;
+    }
+
+    List<String> getAllPartitions() {
+        // first, check cacheDir for any cached partitions
+        Path path = Paths.get(this.cacheFilePath);
+        if (Files.exists(path)) {
+            try {
+                LOG.info("Found cache file at " + cacheFilePath + ". Reading partitions from file.");
+                List<String> partitions = Files.readAllLines(path, StandardCharsets.UTF_8);
+                LOG.info("Found " + partitions.size() + " partitions in local cache.");
+                return partitions;
+            } catch (IOException e) {
+                LOG.warn("Error reading partitions from cache. Continuing to list from storage", e);
+            }
+        }
+        // fetch all partitions and cache them
+        LOG.info("Fetching partitions from storage");
+        List<String> partitions = syncClient.getWrittenPartitionsSince(Option.empty());
+        try {
+            LOG.info("Caching " + partitions.size() + "partitions to local storage at path " + cacheFilePath);
+            Files.write(path, partitions, StandardCharsets.UTF_8);
+        } catch (IOException e) {
+            LOG.warn("Unable to cache partitions.", e);
+        }
+        return partitions;
+    }
+}

hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfig.java

@@ -74,6 +74,14 @@ public class HiveSyncConfig extends HoodieSyncConfig {
   public static final ConfigProperty<String> HIVE_SYNC_COMMENT = HiveSyncConfigHolder.HIVE_SYNC_COMMENT;
   public static final ConfigProperty<String> HIVE_SYNC_TABLE_STRATEGY = HiveSyncConfigHolder.HIVE_SYNC_TABLE_STRATEGY;
 
+  public static final ConfigProperty<String> PARTITION_CACHE_PATH = ConfigProperty
+          .key("hoodie.datasource.hive_sync.partition_cache_path")
+          .noDefaultValue();
+
+  public static final ConfigProperty<String> PARTITION_FETCH_FILTER = ConfigProperty
+          .key("hoodie.datasource.hive_sync.partition_fetch_filter")
+          .noDefaultValue();
+
   public static final ConfigProperty<Boolean> HIVE_SYNC_FILTER_PUSHDOWN_ENABLED = ConfigProperty
       .key("hoodie.datasource.hive_sync.filter_pushdown_enabled")
       .defaultValue(false)
@@ -169,6 +177,19 @@ public static class HiveSyncConfigParams {
     @Parameter(names = {"--sync-strategy"}, description = "Hive table synchronization strategy. Available option: RO, RT, ALL")
     public String syncStrategy;
 
+    @Parameter(names = {"--cache-file-path"}, description = "Path to file on local file system to cache partitions")
+    public String cacheFilePath;
+
+    @Parameter(names = {"--enable-filter-pushdown"}, description = "")
+    public Boolean enableFilterPushdown;
+
+    @Parameter(names = {"--filter-pushdown-max-size"}, description = "")
+    public Integer filterPushdownMaxSize;
+
+    @Parameter(names = {"--partition-fetch-filter"}, description = "Filter to use for fetching partitions from metastore")
+    public String partitionFetchFilter;
+
+
     public boolean isHelp() {
       return hoodieSyncConfigParams.isHelp();
     }
@@ -197,11 +218,17 @@ public TypedProperties toProps() {
       props.setPropertyIfNonNull(HIVE_SYNC_BUCKET_SYNC_SPEC.key(), bucketSpec);
       props.setPropertyIfNonNull(HIVE_SYNC_COMMENT.key(), syncComment);
       props.setPropertyIfNonNull(HIVE_SYNC_TABLE_STRATEGY.key(), syncStrategy);
+      props.setPropertyIfNonNull(HIVE_SYNC_FILTER_PUSHDOWN_ENABLED.key(), enableFilterPushdown);
+      props.setPropertyIfNonNull(HIVE_SYNC_FILTER_PUSHDOWN_MAX_SIZE.key(), filterPushdownMaxSize);
+      props.setPropertyIfNonNull(PARTITION_FETCH_FILTER.key(), partitionFetchFilter);
+
+      props.setPropertyIfNonNull(PARTITION_CACHE_PATH.key(), cacheFilePath);
       return props;
     }
   }
 
   public void validateParameters() {
     ValidationUtils.checkArgument(getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM) > 0, "batch-sync-num for sync hive table must be greater than 0, pls check your parameter");
+    ValidationUtils.checkArgument(getString(PARTITION_CACHE_PATH) != null, "cache-file-path must be provided for initial syncs");
   }
 }

hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java

@@ -52,6 +52,7 @@
 import static org.apache.hudi.common.util.StringUtils.isNullOrEmpty;
 import static org.apache.hudi.common.util.StringUtils.nonEmpty;
 import static org.apache.hudi.hive.HiveSyncConfig.HIVE_SYNC_FILTER_PUSHDOWN_ENABLED;
+import static org.apache.hudi.hive.HiveSyncConfig.PARTITION_FETCH_FILTER;
 import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_AUTO_CREATE_DATABASE;
 import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_IGNORE_EXCEPTIONS;
 import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_SKIP_RO_SUFFIX_FOR_READ_OPTIMIZED_TABLE;
@@ -97,6 +98,8 @@ public class HiveSyncTool extends HoodieSyncTool implements AutoCloseable {
 
   protected String hiveSyncTableStrategy;
 
+  protected FileListingCacheManager fileListingCacheManager;
+
   public HiveSyncTool(Properties props, Configuration hadoopConf) {
     super(props, hadoopConf);
     String metastoreUris = props.getProperty(METASTORE_URIS.key());
@@ -111,6 +114,10 @@ public HiveSyncTool(Properties props, Configuration hadoopConf) {
     this.databaseName = config.getStringOrDefault(META_SYNC_DATABASE_NAME);
     this.tableName = config.getStringOrDefault(META_SYNC_TABLE_NAME);
     initSyncClient(config);
+    this.fileListingCacheManager = new FileListingCacheManager(
+      this.syncClient,
+      props.getProperty(HiveSyncConfig.PARTITION_CACHE_PATH.key())
+    );
     initTableNameVars(config);
   }
 
@@ -254,13 +261,15 @@ protected void syncHoodieTable(String tableName, boolean useRealtimeInputFormat,
       lastCommitTimeSynced = syncClient.getLastCommitTimeSynced(tableName);
     }
     LOG.info("Last commit time synced was found to be " + lastCommitTimeSynced.orElse("null"));
-    List<String> writtenPartitionsSince = syncClient.getWrittenPartitionsSince(lastCommitTimeSynced);
-    LOG.info("Storage partitions scan complete. Found " + writtenPartitionsSince.size());
+
+    // For initial syncs, use the file listing cache to either read partitions from local cache, or fetch and populate
+    // the cache.
+    List<String> allPartitions = fileListingCacheManager.getAllPartitions();
 
     // Sync the partitions if needed
     // find dropped partitions, if any, in the latest commit
     Set<String> droppedPartitions = syncClient.getDroppedPartitionsSince(lastCommitTimeSynced);
-    boolean partitionsChanged = syncPartitions(tableName, writtenPartitionsSince, droppedPartitions);
+    boolean partitionsChanged = syncPartitions(tableName, allPartitions, droppedPartitions);
     boolean meetSyncConditions = schemaChanged || partitionsChanged;
     if (!config.getBoolean(META_SYNC_CONDITIONAL_SYNC) || meetSyncConditions) {
       syncClient.updateLastCommitTimeSynced(tableName);
@@ -346,6 +355,9 @@ private boolean syncSchema(String tableName, boolean tableExists, boolean useRea
    */
   private List<Partition> getTablePartitions(String tableName, List<String> writtenPartitions) {
     if (!config.getBooleanOrDefault(HIVE_SYNC_FILTER_PUSHDOWN_ENABLED)) {
+      if (config.getString(PARTITION_FETCH_FILTER) != null && !config.getString(PARTITION_FETCH_FILTER).isEmpty()) {
+        return syncClient.getPartitionsByFilter(tableName, config.getString(PARTITION_FETCH_FILTER));
+      }
       return syncClient.getAllPartitions(tableName);
     }
 
@@ -376,6 +388,7 @@ private boolean syncPartitions(String tableName, List<String> writtenPartitionsS
       }
 
       List<Partition> hivePartitions = getTablePartitions(tableName, writtenPartitionsSince);
+      LOG.info("Partitions fetched from metastore: " + hivePartitions.size());
       List<PartitionEvent> partitionEvents =
           syncClient.getPartitionEvents(hivePartitions, writtenPartitionsSince, droppedPartitions);