[core] Custom Partition expire factory should be invoked first

Author: JingsongLi

docs/layouts/shortcodes/generated/core_configuration.html

@@ -895,9 +895,9 @@
         </tr>
         <tr>
             <td><h5>partition.expiration-strategy</h5></td>
-            <td style="word-wrap: break-word;">values-time</td>
-            <td><p>Enum</p></td>
-            <td>The strategy determines how to extract the partition time and compare it with the current time.<br /><br />Possible values:<ul><li>"values-time": This strategy compares the time extracted from the partition value with the current time.</li><li>"update-time": This strategy compares the last update time of the partition with the current time.</li><li>"custom": This strategy use custom class to expire partitions.</li></ul></td>
+            <td style="word-wrap: break-word;">"values-time"</td>
+            <td>String</td>
+            <td>The strategy determines how to extract the partition time and compare it with the current time.<ul><li>"values-time": This strategy compares the time extracted from the partition value with the current time.</li><li>"update-time": This strategy compares the last update time of the partition with the current time.</li></ul></td>
         </tr>
         <tr>
             <td><h5>partition.expiration-time</h5></td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1007,12 +1007,20 @@ public InlineElement getDescription() {
                             "Whether only overwrite dynamic partition when overwriting a partitioned table with "
                                     + "dynamic partition columns. Works only when the table has partition keys.");
 
-    public static final ConfigOption<PartitionExpireStrategy> PARTITION_EXPIRATION_STRATEGY =
+    public static final ConfigOption<String> PARTITION_EXPIRATION_STRATEGY =
             key("partition.expiration-strategy")
-                    .enumType(PartitionExpireStrategy.class)
-                    .defaultValue(PartitionExpireStrategy.VALUES_TIME)
+                    .stringType()
+                    .defaultValue("values-time")
                     .withDescription(
-                            "The strategy determines how to extract the partition time and compare it with the current time.");
+                            Description.builder()
+                                    .text(
+                                            "The strategy determines how to extract the partition time and compare it with the current time.")
+                                    .list(
+                                            text(
+                                                    "\"values-time\": This strategy compares the time extracted from the partition value with the current time."),
+                                            text(
+                                                    "\"update-time\": This strategy compares the last update time of the partition with the current time."))
+                                    .build());
 
     public static final ConfigOption<Duration> PARTITION_EXPIRATION_TIME =
             key("partition.expiration-time")
@@ -2848,7 +2856,7 @@ public int partitionExpireBatchSize() {
                 .orElse(options.get(PARTITION_EXPIRATION_MAX_NUM));
     }
 
-    public PartitionExpireStrategy partitionExpireStrategy() {
+    public String partitionExpireStrategy() {
         return options.get(PARTITION_EXPIRATION_STRATEGY);
     }
 
@@ -3802,38 +3810,6 @@ public InlineElement getDescription() {
         }
     }
 
-    /** Specifies the expiration strategy for partition expiration. */
-    public enum PartitionExpireStrategy implements DescribedEnum {
-        VALUES_TIME(
-                "values-time",
-                "This strategy compares the time extracted from the partition value with the current time."),
-
-        UPDATE_TIME(
-                "update-time",
-                "This strategy compares the last update time of the partition with the current time."),
-
-        CUSTOM("custom", "This strategy use custom class to expire partitions.");
-
-        private final String value;
-
-        private final String description;
-
-        PartitionExpireStrategy(String value, String description) {
-            this.value = value;
-            this.description = description;
-        }
-
-        @Override
-        public String toString() {
-            return value;
-        }
-
-        @Override
-        public InlineElement getDescription() {
-            return text(description);
-        }
-    }
-
     /** Specifies the strategy for selecting external storage paths. */
     public enum ExternalPathStrategy implements DescribedEnum {
         NONE(

paimon-api/src/main/java/org/apache/paimon/factories/FactoryUtil.java

@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 import java.util.ServiceLoader;
 import java.util.stream.Collectors;
 
@@ -149,13 +150,11 @@ public static <T> List<T> discoverFactories(ClassLoader classLoader, Class<T> kl
      * @param <T> the type of the factory
      * @return the factory
      */
-    public static <T> T discoverSingletonFactory(ClassLoader classLoader, Class<T> klass) {
+    public static <T> Optional<T> discoverSingletonFactory(
+            ClassLoader classLoader, Class<T> klass) {
         List<T> factories = FactoryUtil.discoverFactories(classLoader, klass);
         if (factories.isEmpty()) {
-            throw new FactoryException(
-                    String.format(
-                            "Could not find any factories that implement '%s' in the classpath.",
-                            klass.getName()));
+            return Optional.empty();
         }
 
         if (factories.size() > 1) {
@@ -171,6 +170,6 @@ public static <T> T discoverSingletonFactory(ClassLoader classLoader, Class<T> k
                                     .collect(Collectors.joining("\n"))));
         }
 
-        return factories.get(0);
+        return Optional.of(factories.get(0));
     }
 }

paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java

@@ -34,6 +34,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 /** Strategy for partition expiration. */
 public abstract class PartitionExpireStrategy {
@@ -80,18 +81,23 @@ public static PartitionExpireStrategy createPartitionExpireStrategy(
             RowType partitionType,
             @Nullable CatalogLoader catalogLoader,
             @Nullable Identifier identifier) {
-        switch (options.partitionExpireStrategy()) {
-            case UPDATE_TIME:
+        Optional<PartitionExpireStrategyFactory> custom =
+                PartitionExpireStrategyFactory.INSTANCE.get();
+        if (custom.isPresent()) {
+            try {
+                return custom.get().create(catalogLoader, identifier, options, partitionType);
+            } catch (UnsupportedOperationException ignored) {
+            }
+        }
+
+        String strategy = options.partitionExpireStrategy();
+        switch (strategy) {
+            case "update-time":
                 return new PartitionUpdateTimeExpireStrategy(options, partitionType);
-            case VALUES_TIME:
+            case "values-time":
                 return new PartitionValuesTimeExpireStrategy(options, partitionType);
-            case CUSTOM:
-                return PartitionExpireStrategyFactory.INSTANCE
-                        .get()
-                        .create(catalogLoader, identifier, options, partitionType);
             default:
-                throw new IllegalArgumentException(
-                        "Unknown partitionExpireStrategy: " + options.partitionExpireStrategy());
+                throw new IllegalArgumentException("Unknown partitionExpireStrategy: " + strategy);
         }
     }
 }

paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategyFactory.java

@@ -27,6 +27,8 @@
 import org.apache.paimon.shade.guava30.com.google.common.base.Supplier;
 import org.apache.paimon.shade.guava30.com.google.common.base.Suppliers;
 
+import java.util.Optional;
+
 /** Factory to create a {@link PartitionExpireStrategy}. */
 public interface PartitionExpireStrategyFactory {
 
@@ -36,7 +38,7 @@ PartitionExpireStrategy create(
             CoreOptions options,
             RowType partitionType);
 
-    Supplier<PartitionExpireStrategyFactory> INSTANCE =
+    Supplier<Optional<PartitionExpireStrategyFactory>> INSTANCE =
             Suppliers.memoize(
                     () ->
                             FactoryUtil.discoverSingletonFactory(

paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java

@@ -36,6 +36,8 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import static org.apache.paimon.CoreOptions.PARTITION_EXPIRATION_STRATEGY;
+
 /**
  * A partition expiration policy that compare the time extracted from the partition with the current
  * time.
@@ -86,19 +88,17 @@ public boolean test(BinaryRow partition) {
                                 + "  1. Check the expiration configuration.\n"
                                 + "  2. Manually delete the partition using the drop-partition command if the partition"
                                 + " value is non-date formatted.\n"
-                                + "  3. Use '{}' expiration strategy by set '{}', which supports non-date formatted partition.",
+                                + "  3. Use 'update-time' expiration strategy by set '{}', which supports non-date formatted partition.",
                         formatPartitionInfo(array),
-                        CoreOptions.PartitionExpireStrategy.UPDATE_TIME,
-                        CoreOptions.PARTITION_EXPIRATION_STRATEGY.key());
+                        PARTITION_EXPIRATION_STRATEGY.key());
                 return false;
             } catch (NullPointerException e) {
                 // there might exist NULL partition value
                 LOG.warn(
                         "This partition {} cannot be expired because it contains null value. "
-                                + "You can try to drop it manually or use '{}' expiration strategy by set '{}'.",
+                                + "You can try to drop it manually or use 'update-time' expiration strategy by set '{}'.",
                         formatPartitionInfo(array),
-                        CoreOptions.PartitionExpireStrategy.UPDATE_TIME,
-                        CoreOptions.PARTITION_EXPIRATION_STRATEGY.key());
+                        PARTITION_EXPIRATION_STRATEGY.key());
                 return false;
             }
         }

paimon-core/src/test/java/org/apache/paimon/partition/CustomPartitionExpirationFactory.java

@@ -43,6 +43,10 @@ public PartitionExpireStrategy create(
             Identifier identifier,
             CoreOptions options,
             RowType partitionType) {
+        String strategy = options.partitionExpireStrategy();
+        if (!"custom".equals(strategy)) {
+            throw new UnsupportedOperationException();
+        }
         return new PartitionExpireStrategy(partitionType, options.partitionDefaultName()) {
             @Override
             public List<PartitionEntry> selectExpiredPartitions(