fix

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java

@@ -42,19 +42,13 @@
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.CatalogEnvironment;
 import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.KeyComparatorSupplier;
 import org.apache.paimon.utils.UserDefinedSeqComparator;
 import org.apache.paimon.utils.ValueEqualiserSupplier;
 
 import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
-import java.util.function.Function;
 import java.util.function.Supplier;
 
 import static org.apache.paimon.predicate.PredicateBuilder.and;
@@ -198,7 +192,7 @@ public AbstractFileStoreWrite<KeyValue> newWrite(
                     logDedupEqualSupplier,
                     mfFactory,
                     pathFactory(),
-                    format2PathFactory(options, this::pathFactory),
+                    this::pathFactory,
                     snapshotManager(),
                     newScan(ScanType.FOR_WRITE).withManifestCacheFilter(manifestFilter),
                     indexFactory,
@@ -209,15 +203,6 @@ public AbstractFileStoreWrite<KeyValue> newWrite(
         }
     }
 
-    public static Map<String, FileStorePathFactory> format2PathFactory(
-            CoreOptions options, Function<String, FileStorePathFactory> pathFactory) {
-        Map<String, FileStorePathFactory> pathFactoryMap = new HashMap<>();
-        Set<String> formats = new HashSet<>(options.fileFormatPerLevel().values());
-        formats.add(options.fileFormatString());
-        formats.forEach(format -> pathFactoryMap.put(format, pathFactory.apply(format)));
-        return pathFactoryMap;
-    }
-
     @Override
     protected KeyValueFileStoreScan newScan(ScanType scanType) {
         BucketSelectConverter bucketSelectConverter =

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

@@ -81,7 +81,6 @@
 
 import java.util.Comparator;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -90,6 +89,7 @@
 import static org.apache.paimon.CoreOptions.MergeEngine.DEDUPLICATE;
 import static org.apache.paimon.lookup.LookupStoreFactory.bfGenerator;
 import static org.apache.paimon.mergetree.LookupFile.localFilePrefix;
+import static org.apache.paimon.utils.FileStorePathFactory.createFormatPathFactories;
 
 /** {@link FileStoreWrite} for {@link KeyValueFileStore}. */
 public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
@@ -124,7 +124,7 @@ public KeyValueFileStoreWrite(
             Supplier<RecordEqualiser> logDedupEqualSupplier,
             MergeFunctionFactory<KeyValue> mfFactory,
             FileStorePathFactory pathFactory,
-            Map<String, FileStorePathFactory> format2PathFactory,
+            Function<String, FileStorePathFactory> formatPathFactory,
             SnapshotManager snapshotManager,
             FileStoreScan scan,
             @Nullable IndexMaintainer.Factory<KeyValue> indexFactory,
@@ -166,7 +166,7 @@ public KeyValueFileStoreWrite(
                         keyType,
                         valueType,
                         options.fileFormat(),
-                        format2PathFactory,
+                        createFormatPathFactories(options, formatPathFactory),
                         options.targetFileSize(true));
         this.keyComparatorSupplier = keyComparatorSupplier;
         this.logDedupEqualSupplier = logDedupEqualSupplier;

paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java

@@ -42,7 +42,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
 
-import static org.apache.paimon.KeyValueFileStore.format2PathFactory;
+import static org.apache.paimon.utils.FileStorePathFactory.createFormatPathFactories;
 
 /** {@link FileStoreWrite} for {@code bucket = -2} tables. */
 public class PostponeBucketFileStoreWrite extends AbstractFileStoreWrite<KeyValue> {
@@ -56,7 +56,7 @@ public PostponeBucketFileStoreWrite(
             RowType partitionType,
             RowType keyType,
             RowType valueType,
-            Function<String, FileStorePathFactory> pathFactory,
+            Function<String, FileStorePathFactory> formatPathFactory,
             SnapshotManager snapshotManager,
             FileStoreScan scan,
             CoreOptions options,
@@ -87,7 +87,7 @@ public PostponeBucketFileStoreWrite(
                         keyType,
                         valueType,
                         this.options.fileFormat(),
-                        format2PathFactory(this.options, pathFactory),
+                        createFormatPathFactories(this.options, formatPathFactory),
                         options.targetFileSize(true));
 
         // Ignoring previous files saves scanning time.

paimon-core/src/main/java/org/apache/paimon/utils/FileStorePathFactory.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.utils;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.fs.ExternalPathProvider;
@@ -29,9 +30,14 @@
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /** Factory which produces {@link Path}s for manifest files. */
@@ -290,4 +296,13 @@ public Path toPath(String fileName) {
             }
         };
     }
+
+    public static Map<String, FileStorePathFactory> createFormatPathFactories(
+            CoreOptions options, Function<String, FileStorePathFactory> formatPathFactory) {
+        Map<String, FileStorePathFactory> pathFactoryMap = new HashMap<>();
+        Set<String> formats = new HashSet<>(options.fileFormatPerLevel().values());
+        formats.add(options.fileFormatString());
+        formats.forEach(format -> pathFactoryMap.put(format, formatPathFactory.apply(format)));
+        return pathFactoryMap;
+    }
 }