[core][flink] Refactor partition time resolution using PartitionTimeResolver

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java

@@ -37,7 +37,7 @@
 import org.apache.paimon.memory.MemoryPoolFactory;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.operation.metrics.CompactionMetrics;
-import org.apache.paimon.partition.PartitionTimeExtractor;
+import org.apache.paimon.partition.PartitionTimeResolvable;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.CommitMessageImpl;
 import org.apache.paimon.types.RowType;
@@ -714,15 +714,15 @@ public CompactionMetrics compactionMetrics() {
 
     private static class PartitionTimestampValidator {
 
-        private final PartitionTimeExtractor timeExtractor;
+        private final PartitionTimeResolvable timeResolver;
         private final RowDataToObjectArrayConverter partitionConverter;
         private final List<String> partitionKeys;
 
         private PartitionTimestampValidator(
-                PartitionTimeExtractor timeExtractor,
+                PartitionTimeResolvable timeResolver,
                 RowDataToObjectArrayConverter partitionConverter,
                 List<String> partitionKeys) {
-            this.timeExtractor = timeExtractor;
+            this.timeResolver = timeResolver;
             this.partitionConverter = partitionConverter;
             this.partitionKeys = partitionKeys;
         }
@@ -738,7 +738,8 @@ private static PartitionTimestampValidator create(
             if ((timeFormatter != null || timePattern != null)
                     && partitionType.getFieldCount() > 0) {
                 return new PartitionTimestampValidator(
-                        new PartitionTimeExtractor(timePattern, timeFormatter),
+                        PartitionTimeResolvable.create(
+                                partitionType.getFieldNames(), timePattern, timeFormatter),
                         new RowDataToObjectArrayConverter(partitionType),
                         partitionType.getFieldNames());
             }
@@ -748,7 +749,7 @@ private static PartitionTimestampValidator create(
         private void validate(BinaryRow partition) {
             Object[] array = partitionConverter.convert(partition);
             try {
-                timeExtractor.extract(partitionKeys, Arrays.asList(array));
+                timeResolver.parsePartitionValues(Arrays.asList(array));
             } catch (DateTimeParseException e) {
                 String partitionInfo =
                         IntStream.range(0, partitionKeys.size())

paimon-core/src/main/java/org/apache/paimon/operation/ChainTablePartitionExpire.java

@@ -25,7 +25,7 @@
 import org.apache.paimon.codegen.RecordComparator;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.manifest.PartitionEntry;
-import org.apache.paimon.partition.PartitionTimeExtractor;
+import org.apache.paimon.partition.PartitionTimeResolver;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.PartitionModification;
 import org.apache.paimon.table.sink.BatchTableCommit;
@@ -85,7 +85,7 @@ public class ChainTablePartitionExpire implements PartitionExpire {
     private final Duration checkInterval;
     private final FileStoreTable snapshotTable;
     private final FileStoreTable deltaTable;
-    private final PartitionTimeExtractor timeExtractor;
+    private final PartitionTimeResolver timeResolver;
     private final ChainPartitionProjector projector;
     private final RecordComparator chainPartitionComparator;
     private final InternalRowPartitionComputer partitionComputer;
@@ -126,9 +126,11 @@ public ChainTablePartitionExpire(
         this.projector = new ChainPartitionProjector(partitionType, chainFieldCount);
         this.chainPartitionComparator =
                 CodeGenUtils.newRecordComparator(projector.chainPartitionType().getFieldTypes());
-        this.timeExtractor =
-                new PartitionTimeExtractor(
-                        options.partitionTimestampPattern(), options.partitionTimestampFormatter());
+        this.timeResolver =
+                new PartitionTimeResolver(
+                        this.chainPartitionKeys,
+                        options.partitionTimestampPattern(),
+                        options.partitionTimestampFormatter());
         this.partitionComputer =
                 new InternalRowPartitionComputer(
                         options.partitionDefaultName(),
@@ -408,7 +410,7 @@ private LocalDateTime extractPartitionTime(BinaryRow partition) {
             for (String key : chainPartitionKeys) {
                 chainValues.add(partValues.get(key));
             }
-            return timeExtractor.extract(chainPartitionKeys, chainValues);
+            return timeResolver.parsePartitionValues(chainValues);
         } catch (Exception e) {
             LOG.warn("Failed to extract partition time from {}", partition, e);
             return null;

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

@@ -0,0 +1,67 @@
+/*
+ * 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.paimon.partition;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+import java.time.temporal.TemporalAmount;
+import java.util.LinkedHashMap;
+import java.util.List;
+
+/**
+ * Resolves partition values to/from timestamp and extracts the minimum time step.
+ *
+ * <p>Use {@link #create(List, String, String)} to obtain an instance. If both {@code pattern} and
+ * {@code formatter} are provided, a full pattern-based resolver is returned; otherwise a fallback
+ * resolver that handles the unconfigured case is returned.
+ */
+public interface PartitionTimeResolvable extends Serializable {
+
+    /** Returns the partition keys used by this resolver. */
+    List<String> partitionKeys();
+
+    /** Parses partition column values into a {@link LocalDateTime}. */
+    LocalDateTime parsePartitionValues(List<?> partitionValues);
+
+    /** Formats a {@link LocalDateTime} into partition column values. */
+    default LinkedHashMap<String, String> resolvePartitionValues(LocalDateTime dateTime) {
+        throw new UnsupportedOperationException(
+                "resolvePartitionValues is not supported by this resolver");
+    }
+
+    /** Extracts the minimum time step covered by the partition pattern and formatter. */
+    default TemporalAmount extractMinStep() {
+        throw new UnsupportedOperationException("extractMinStep is not supported by this resolver");
+    }
+
+    /**
+     * Creates a {@link PartitionTimeResolvable}.
+     *
+     * <p>If both {@code pattern} and {@code formatter} are non-null, returns a normal {@link
+     * PartitionTimeResolver}. If either is null, returns a fallback resolver that handles the
+     * unconfigured case.
+     */
+    static PartitionTimeResolvable create(
+            List<String> partitionKeys, String pattern, String formatter) {
+        if (pattern == null || formatter == null) {
+            return PartitionTimeResolver.createFallback(partitionKeys, pattern, formatter);
+        }
+        return new PartitionTimeResolver(partitionKeys, pattern, formatter);
+    }
+}