[core][format] op format factory mv to cache .

paimon-common/src/main/java/org/apache/paimon/factories/FactoryNotFoundException.java

@@ -0,0 +1,26 @@
+/*
+ * 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.factories;
+
+/** Exception for factory not found. */
+public class FactoryNotFoundException extends FactoryException {
+    public FactoryNotFoundException(String message) {
+        super(message);
+    }
+}

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

@@ -28,6 +28,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.ServiceLoader;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 /** Utility for working with {@link Factory}s. */
@@ -38,15 +39,12 @@ public class FactoryUtil {
     private static final Cache<ClassLoader, List<Factory>> FACTORIES =
             Caffeine.newBuilder().softValues().maximumSize(100).executor(Runnable::run).build();
 
-    /** Discovers a factory using the given factory base class and identifier. */
-    @SuppressWarnings("unchecked")
     public static <T extends Factory> T discoverFactory(
             ClassLoader classLoader, Class<T> factoryClass, String identifier) {
-        final List<Factory> factories = getFactories(classLoader);
-
-        final List<Factory> foundFactories =
-                factories.stream()
+        final List<T> foundFactories =
+                getFactories(classLoader).stream()
                         .filter(f -> factoryClass.isAssignableFrom(f.getClass()))
+                        .map(factoryClass::cast)
                         .collect(Collectors.toList());
 
         if (foundFactories.isEmpty()) {
@@ -56,31 +54,37 @@ public static <T extends Factory> T discoverFactory(
                             factoryClass.getName()));
         }
 
-        final List<Factory> matchingFactories =
-                foundFactories.stream()
-                        .filter(f -> f.identifier().equals(identifier))
+        return matchFactory(foundFactories, Factory::identifier, identifier);
+    }
+
+    /** Discovers a factory using the given factory base class and identifier. */
+    @SuppressWarnings("unchecked")
+    public static <T extends Factory> T matchFactory(
+            List<T> factories, Function<T, String> identifierFunction, String identifier) {
+
+        final List<T> matchingFactories =
+                factories.stream()
+                        .filter(f -> identifierFunction.apply(f).equals(identifier))
                         .collect(Collectors.toList());
 
         if (matchingFactories.isEmpty()) {
-            throw new FactoryException(
+            throw new FactoryNotFoundException(
                     String.format(
-                            "Could not find any factory for identifier '%s' that implements '%s' in the classpath.\n\n"
+                            "Could not find any factory for identifier '%s' in the classpath.\n\n"
                                     + "Available factory identifiers are:\n\n"
                                     + "%s",
                             identifier,
-                            factoryClass.getName(),
-                            foundFactories.stream()
-                                    .map(Factory::identifier)
+                            factories.stream()
+                                    .map(identifierFunction)
                                     .collect(Collectors.joining("\n"))));
         }
         if (matchingFactories.size() > 1) {
             throw new FactoryException(
                     String.format(
-                            "Multiple factories for identifier '%s' that implement '%s' found in the classpath.\n\n"
+                            "Multiple factories for identifier '%s' in the classpath.\n\n"
                                     + "Ambiguous factory classes are:\n\n"
                                     + "%s",
                             identifier,
-                            factoryClass.getName(),
                             matchingFactories.stream()
                                     .map(f -> f.getClass().getName())
                                     .sorted()

paimon-common/src/main/java/org/apache/paimon/factories/FormatFactoryUtil.java

@@ -24,7 +24,6 @@
 import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
 
 import java.util.List;
-import java.util.stream.Collectors;
 
 import static org.apache.paimon.factories.FactoryUtil.discoverFactories;
 
@@ -40,24 +39,8 @@ public static <T extends FileFormatFactory> T discoverFactory(
             ClassLoader classLoader, String identifier) {
         final List<FileFormatFactory> foundFactories = getFactories(classLoader);
 
-        final List<FileFormatFactory> matchingFactories =
-                foundFactories.stream()
-                        .filter(f -> f.identifier().equals(identifier))
-                        .collect(Collectors.toList());
-
-        if (matchingFactories.isEmpty()) {
-            throw new FactoryException(
-                    String.format(
-                            "Could not find any factory for identifier '%s' that implements FileFormatFactory in the classpath.\n\n"
-                                    + "Available factory identifiers are:\n\n"
-                                    + "%s",
-                            identifier,
-                            foundFactories.stream()
-                                    .map(FileFormatFactory::identifier)
-                                    .collect(Collectors.joining("\n"))));
-        }
-
-        return (T) matchingFactories.get(0);
+        return (T)
+                FactoryUtil.matchFactory(foundFactories, FileFormatFactory::identifier, identifier);
     }
 
     private static List<FileFormatFactory> getFactories(ClassLoader classLoader) {

paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java

@@ -19,6 +19,8 @@
 package org.apache.paimon.format;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.factories.FactoryNotFoundException;
+import org.apache.paimon.factories.FactoryUtil;
 import org.apache.paimon.factories.FormatFactoryUtil;
 import org.apache.paimon.format.FileFormatFactory.FormatContext;
 import org.apache.paimon.options.Options;
@@ -88,9 +90,21 @@ public static FileFormat fromIdentifier(String identifier, Options options) {
 
     /** Create a {@link FileFormat} from format identifier and format options. */
     public static FileFormat fromIdentifier(String identifier, FormatContext context) {
-        return FormatFactoryUtil.discoverFactory(
-                        FileFormat.class.getClassLoader(), identifier.toLowerCase())
-                .create(context);
+        if (identifier != null) {
+            identifier = identifier.toLowerCase();
+        }
+        try {
+            FileFormatFactory fileFormatFactory =
+                    FactoryUtil.discoverFactory(
+                            FileFormatFactory.class.getClassLoader(),
+                            FileFormatFactory.class,
+                            identifier);
+            return fileFormatFactory.create(context);
+        } catch (FactoryNotFoundException e) {
+            // Compatible with existing third-party factory implementations .
+            return FormatFactoryUtil.discoverFactory(FileFormat.class.getClassLoader(), identifier)
+                    .create(context);
+        }
     }
 
     protected Options getIdentifierPrefixOptions(Options options) {

paimon-common/src/main/java/org/apache/paimon/format/FileFormatFactory.java

@@ -19,13 +19,14 @@
 package org.apache.paimon.format;
 
 import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.factories.Factory;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
 
 import javax.annotation.Nullable;
 
 /** Factory to create {@link FileFormat}. */
-public interface FileFormatFactory {
+public interface FileFormatFactory extends Factory {
 
     String identifier();
 

paimon-common/src/test/java/org/apache/paimon/factories/FactoryUtilTest.java

@@ -40,9 +40,6 @@ public void testDiscoverFactory() {
                                         Thread.currentThread().getContextClassLoader(),
                                         DummyFactory.class,
                                         "non-exist-factory"))
-                .isInstanceOf(FactoryException.class)
-                .hasMessageContaining(
-                        "Could not find any factory for identifier '%s' that implements '%s' in the classpath.",
-                        "non-exist-factory", DummyFactory.class.getName());
+                .isInstanceOf(FactoryNotFoundException.class);
     }
 }

paimon-core/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory

@@ -13,4 +13,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+org.apache.paimon.format.FileStatsExtractingAvroFormatFactory
 org.apache.paimon.mergetree.compact.aggregate.TestCustomAggFactory

paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory

@@ -82,3 +82,8 @@ org.apache.paimon.flink.procedure.MarkPartitionDoneProcedure
 org.apache.paimon.flink.procedure.CloneProcedure
 org.apache.paimon.flink.procedure.CompactManifestProcedure
 org.apache.paimon.flink.procedure.RefreshObjectTableProcedure
+
+### fileFormat factories
+org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat$OrcFactory
+org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat$ParquetFactory
+org.apache.paimon.flink.compact.changelog.format.CompactedChangelogReadOnlyFormat$AvroFactory

paimon-format/src/main/resources/META-INF/services/org.apache.paimon.format.FileFormatFactory → paimon-format/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory

@@ -15,4 +15,4 @@
 
 org.apache.paimon.format.avro.AvroFileFormatFactory
 org.apache.paimon.format.orc.OrcFileFormatFactory
-org.apache.paimon.format.parquet.ParquetFileFormatFactory
+org.apache.paimon.format.parquet.ParquetFileFormatFactory