[controller] log compaction

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerClient.java

@@ -1156,6 +1156,16 @@ public MultiStoreInfoResponse getClusterStores(String clusterName) {
     return request(ControllerRoute.GET_STORES_IN_CLUSTER, params, MultiStoreInfoResponse.class);
   }
 
+  /**
+   * This method gets a list of store names that are ready for compaction.
+   * @param clusterName, the name of the cluster to query for compaction eligible stores
+   * @return The list of store names that are ready for compaction.
+   */
+  public MultiStoreInfoResponse getStoresForCompaction(String clusterName) {
+    QueryParams params = newParams().add(CLUSTER, clusterName);
+    return request(ControllerRoute.GET_STORES_FOR_COMPACTION, params, MultiStoreInfoResponse.class);
+  }
+
   public VersionResponse getStoreLargestUsedVersion(String clusterName, String storeName) {
     QueryParams params = newParams().add(CLUSTER, clusterName).add(NAME, storeName);
     return request(ControllerRoute.GET_STORE_LARGEST_USED_VERSION, params, VersionResponse.class);

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerRoute.java

@@ -266,6 +266,7 @@ public enum ControllerRoute {
       Arrays.asList(CLUSTER, SOURCE_FABRIC, FABRIC, NAME, VERSION), AMPLIFICATION_FACTOR
   ), GET_STALE_STORES_IN_CLUSTER("/get_stale_stores_in_cluster", HttpMethod.GET, Collections.singletonList(CLUSTER)),
   GET_STORES_IN_CLUSTER("/get_stores_in_cluster", HttpMethod.GET, Collections.singletonList(CLUSTER)),
+  GET_STORES_FOR_COMPACTION("/get_stores_for_compaction", HttpMethod.GET, Collections.singletonList(CLUSTER)),
   GET_STORE_LARGEST_USED_VERSION("/get_store_largest_used_version", HttpMethod.GET, Arrays.asList(CLUSTER, NAME)),
   LIST_STORE_PUSH_INFO("/list_store_push_info", HttpMethod.GET, Arrays.asList(CLUSTER, NAME, PARTITION_DETAIL_ENABLED)),
   GET_REGION_PUSH_DETAILS(

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/VeniceParentHelixAdminTest.java

@@ -713,7 +713,6 @@ private void testLatestSupersetSchemaIdUpdate(
         });
       }
     }
-
   }
 
   private void testSuperSetSchemaGen(ControllerClient parentControllerClient) {

services/venice-controller/src/main/java/com/linkedin/venice/controller/Admin.java

@@ -60,6 +60,7 @@
 
 
 public interface Admin extends AutoCloseable, Closeable {
+
   // Wrapper to include both overall offline push status and other extra useful info
   class OfflinePushStatusInfo {
     private ExecutionStatus executionStatus;
@@ -935,6 +936,12 @@ default boolean isAdminTopicConsumptionEnabled(String clusterName) {
 
   Map<String, StoreDataAudit> getClusterStaleStores(String clusterName);
 
+  /**
+   * @param clusterName
+   * @return the list of stores ready for compaction
+   */
+  List<StoreInfo> getStoresForCompaction(String clusterName);
+
   /**
    * @return the largest used version number for the given store from store graveyard.
    */

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceHelixAdmin.java

@@ -7467,6 +7467,11 @@ public Map<String, StoreDataAudit> getClusterStaleStores(String clusterName) {
     throw new UnsupportedOperationException("This function has not been implemented.");
   }
 
+  @Override
+  public ArrayList<StoreInfo> getStoresForCompaction(String clusterName) {
+    throw new UnsupportedOperationException("This function has no implementation.");
+  }
+
   @Override
   public Map<String, RegionPushDetails> listStorePushInfo(
       String clusterName,

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java

@@ -287,6 +287,7 @@ public class VeniceParentHelixAdmin implements Admin {
   private static final StackTraceElement[] EMPTY_STACK_TRACE = new StackTraceElement[0];
 
   private static final long TOPIC_DELETION_DELAY_MS = 5 * Time.MS_PER_MINUTE;
+  public static final int COMPACTION_THRESHOLD_HOURS = 24;
 
   final Map<String, Boolean> asyncSetupEnabledMap;
   private final VeniceHelixAdmin veniceHelixAdmin;
@@ -4801,6 +4802,78 @@ public Map<String, StoreDataAudit> getClusterStaleStores(String clusterName) {
     return retMap;
   }
 
+  /**
+   * This function iterates over all of Helix Parent Admin's child controllers,
+   * in order to obtain the stores and respective store information of each child controller,
+   * and then filter out the stores that are ready for compaction.
+   * @param clusterName, the name of the cluster to search for stores that are ready for compaction
+   * @return a list of StoreInfo objects of stores that are ready for compaction
+   */
+  @Override
+  public ArrayList<StoreInfo> getStoresForCompaction(String clusterName) {
+    ArrayList<StoreInfo> compactionReadyStores = new ArrayList<>();
+    try {
+      Map<String, ControllerClient> childControllers = getVeniceHelixAdmin().getControllerClientMap(clusterName);
+      ArrayList<StoreInfo> storeInfoList = new ArrayList<>();
+
+      // iterate through child controllers
+      for (Map.Entry<String, ControllerClient> controller: childControllers.entrySet()) {
+
+        // add all store info to storeInfoList
+        MultiStoreInfoResponse response = controller.getValue().getClusterStores(clusterName);
+        storeInfoList.addAll(response.getStoreInfoList());
+      }
+
+      // filter out
+      filterStoresForCompaction(storeInfoList, compactionReadyStores);
+    } catch (Exception e) {
+      throw new VeniceException("Something went wrong trying to fetch stores for compaction.", e);
+    }
+    return compactionReadyStores;
+  }
+
+  // package exclusive for testing
+  void filterStoresForCompaction(ArrayList<StoreInfo> storeInfoList, ArrayList<StoreInfo> compactionReadyStores) {
+    for (StoreInfo storeInfo: storeInfoList) {
+      if (isCompactionReady(storeInfo)) {
+        compactionReadyStores.add(storeInfo);
+      }
+    }
+  }
+
+  // This function abstracts the criteria for a store to be ready for compaction
+  boolean isCompactionReady(StoreInfo storeInfo) {
+    boolean isHybridStore = storeInfo.getHybridStoreConfig() != null;
+
+    return isHybridStore && isLastCompactionTimeOlderThanThresholdHours(COMPACTION_THRESHOLD_HOURS, storeInfo);
+  }
+
+  // START isCompactionReady() helper methods: each method below encapsulates a log compaction readiness criterion
+  /**
+   * This function checks if the last compaction time is older than the threshold.
+   * @param compactionThresholdHours, the number of hours that the last compaction time should be older than
+   * @param storeInfo, the store to check the last compaction time for
+   * @return true if the last compaction time is older than the threshold, false otherwise
+   */
+  private boolean isLastCompactionTimeOlderThanThresholdHours(int compactionThresholdHours, StoreInfo storeInfo) {
+    // get the last compaction time
+    int currentVersionNumber = storeInfo.getCurrentVersion();
+    Optional<Version> currentVersion = storeInfo.getVersion(currentVersionNumber);
+    if (!currentVersion.isPresent()) {
+      LOGGER.warn("Couldn't find current version: {} from store: {}", currentVersionNumber, storeInfo.getName());
+      return false; // invalid store because no current version, this store is not eligible for compaction
+    }
+
+    // calculate hours since last compaction
+    long lastCompactionTime = currentVersion.get().getCreatedTime();
+    long currentTime = System.currentTimeMillis();
+    long millisecondsSinceLastCompaction = currentTime - lastCompactionTime;
+    long hoursSinceLastCompaction = TimeUnit.MILLISECONDS.toHours(millisecondsSinceLastCompaction);
+
+    return hoursSinceLastCompaction > compactionThresholdHours;
+  }
+  // END isCompactionReady() helper methods
+
   /**
    * @return the largest used version number for the given store from the store graveyard.
    */

services/venice-controller/src/main/java/com/linkedin/venice/controller/server/AdminSparkServer.java

@@ -51,6 +51,7 @@
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STALE_STORES_IN_CLUSTER;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STORAGE_PERSONA;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STORAGE_PERSONA_ASSOCIATED_WITH_STORE;
+import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STORES_FOR_COMPACTION;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STORES_IN_CLUSTER;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STORE_LARGEST_USED_VERSION;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_VALUE_OR_DERIVED_SCHEMA_ID;
@@ -572,6 +573,9 @@ public boolean startInner() throws Exception {
     httpService.get(
         GET_STORES_IN_CLUSTER.getPath(),
         new VeniceParentControllerRegionStateHandler(admin, storesRoutes.getStoresInCluster(admin)));
+    httpService.get(
+        GET_STORES_FOR_COMPACTION.getPath(),
+        new VeniceParentControllerRegionStateHandler(admin, storesRoutes.getStoresForCompaction(admin)));
     httpService.get(
         GET_STORE_LARGEST_USED_VERSION.getPath(),
         new VeniceParentControllerRegionStateHandler(admin, storesRoutes.getStoreLargestUsedVersion(admin)));

services/venice-controller/src/main/java/com/linkedin/venice/controller/server/StoresRoutes.java

@@ -40,6 +40,7 @@
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_REGION_PUSH_DETAILS;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_REPUSH_INFO;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STALE_STORES_IN_CLUSTER;
+import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STORES_FOR_COMPACTION;
 import static com.linkedin.venice.controllerapi.ControllerRoute.GET_STORES_IN_CLUSTER;
 import static com.linkedin.venice.controllerapi.ControllerRoute.LIST_STORES;
 import static com.linkedin.venice.controllerapi.ControllerRoute.LIST_STORE_PUSH_INFO;
@@ -937,6 +938,22 @@ public void internalHandle(Request request, MultiStoreInfoResponse veniceRespons
     };
   }
 
+  /**
+   * @see Admin#getStoresForCompaction(String)
+   */
+  public Route getStoresForCompaction(Admin admin) {
+    return new VeniceRouteHandler<MultiStoreInfoResponse>(MultiStoreInfoResponse.class) {
+      @Override
+      public void internalHandle(Request request, MultiStoreInfoResponse veniceResponse) {
+        AdminSparkServer.validateParams(request, GET_STORES_FOR_COMPACTION.getParams(), admin);
+        String cluster = request.queryParams(CLUSTER);
+        ArrayList<StoreInfo> response = admin.getStoresForCompaction(cluster);
+        veniceResponse.setStoreInfoList(response);
+        veniceResponse.setCluster(cluster);
+      }
+    };
+  }
+
   /**
    * @see Admin#getLargestUsedVersionFromStoreGraveyard(String, String)
    */

services/venice-controller/src/test/java/com/linkedin/venice/controller/TestVeniceParentHelixAdmin.java

@@ -56,6 +56,7 @@
 import com.linkedin.venice.helix.HelixReadWriteStoreRepository;
 import com.linkedin.venice.meta.BufferReplayPolicy;
 import com.linkedin.venice.meta.DataReplicationPolicy;
+import com.linkedin.venice.meta.HybridStoreConfig;
 import com.linkedin.venice.meta.HybridStoreConfigImpl;
 import com.linkedin.venice.meta.OfflinePushStrategy;
 import com.linkedin.venice.meta.PersistenceType;
@@ -2835,4 +2836,65 @@ public void testGetFinalReturnStatus() {
     finalStatus = VeniceParentHelixAdmin.getFinalReturnStatus(statuses, childRegions, 0, new StringBuilder());
     assertEquals(finalStatus, ExecutionStatus.DVC_INGESTION_ERROR_OTHER);
   }
+
+  @Test
+  public void testFilterStoresForCompaction() {
+    VeniceParentHelixAdmin mockAdmin = mock(VeniceParentHelixAdmin.class);
+    ArrayList<StoreInfo> storeInfoList = new ArrayList<>();
+    ArrayList<StoreInfo> compactionReadyStores = new ArrayList<>();
+
+    // Mock StoreInfo instances
+    StoreInfo store1 = mock(StoreInfo.class);
+    StoreInfo store2 = mock(StoreInfo.class);
+    StoreInfo store3 = mock(StoreInfo.class);
+
+    // Mock HybridStoreConfig for the first two StoreInfo instances
+    HybridStoreConfig hybridStoreConfig1 = mock(HybridStoreConfig.class);
+    HybridStoreConfig hybridStoreConfig2 = mock(HybridStoreConfig.class);
+    when(store1.getHybridStoreConfig()).thenReturn(hybridStoreConfig1);
+    when(store2.getHybridStoreConfig()).thenReturn(hybridStoreConfig2);
+    when(store3.getHybridStoreConfig()).thenReturn(null);
+
+    // Mock version numbers with random numbers
+    int store1VersionNumber = 1;
+    int store2VersionNumber = 2;
+    int store3VersionNumber = 3;
+    when(store1.getCurrentVersion()).thenReturn(store1VersionNumber);
+    when(store2.getCurrentVersion()).thenReturn(store2VersionNumber);
+    when(store3.getCurrentVersion()).thenReturn(store3VersionNumber);
+
+    // Mock Version instances
+    Version version1 = mock(Version.class);
+    Version version2 = mock(Version.class);
+    Version version3 = mock(Version.class);
+
+    // Return Version mocks when getVersion() is called
+    when(store1.getVersion(anyInt())).thenReturn(Optional.ofNullable(version1));
+    when(store2.getVersion(anyInt())).thenReturn(Optional.ofNullable(version2));
+    when(store3.getVersion(anyInt())).thenReturn(Optional.ofNullable(version3));
+
+    // Set createTime for Version mocks
+    long currentTime = System.currentTimeMillis();
+    long millisecondsPerHour = 60 * 60 * 1000;
+    when(version1.getCreatedTime()).thenReturn(currentTime - (25 * millisecondsPerHour)); // 25 hours ago
+    when(version2.getCreatedTime()).thenReturn(currentTime - (50 * millisecondsPerHour)); // 50 hours ago
+    when(version3.getCreatedTime()).thenReturn(currentTime - (23 * millisecondsPerHour)); // 23 hours ago
+
+    // Add StoreInfo instances to the list
+    storeInfoList.add(store1);
+    storeInfoList.add(store2);
+    storeInfoList.add(store3);
+
+    // Call the real method to test
+    doCallRealMethod().when(mockAdmin).filterStoresForCompaction(any(), any());
+
+    // Test
+    mockAdmin.filterStoresForCompaction(storeInfoList, compactionReadyStores);
+
+    // Assert the expected outcome
+    assertEquals(compactionReadyStores.size(), 2);
+    assertEquals(compactionReadyStores.contains(store1), true);
+    assertEquals(compactionReadyStores.contains(store2), true);
+    assertEquals(compactionReadyStores.contains(store3), false);
+  }
 }