[orc] Optimize configuration creating in orc file format

paimon-common/src/main/java/org/apache/paimon/fs/ObjectCacheManager.java

@@ -0,0 +1,57 @@
+/*
+ * 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.fs;
+
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
+
+import java.time.Duration;
+import java.util.function.Function;
+
+/**
+ * Sample Object Cache Manager .
+ *
+ * @param <K>
+ * @param <V>
+ */
+public class ObjectCacheManager<K, V> {
+    private final Cache<K, V> cache;
+
+    private ObjectCacheManager(Duration timeout, int maxSize) {
+        this.cache = Caffeine.newBuilder().maximumSize(maxSize).expireAfterWrite(timeout).build();
+    }
+
+    public static <K, V> ObjectCacheManager<K, V> newObjectCacheManager(
+            Duration timeout, int maxSize) {
+        return new ObjectCacheManager<>(timeout, maxSize);
+    }
+
+    public ObjectCacheManager<K, V> put(K k, V v) {
+        this.cache.put(k, v);
+        return this;
+    }
+
+    public V get(K k, Function<? super K, ? extends V> creator) {
+        return this.cache.get(k, creator);
+    }
+
+    public V getIfPresent(K k) {
+        return this.cache.getIfPresent(k);
+    }
+}

paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java

@@ -30,6 +30,7 @@
 import org.apache.paimon.format.orc.filter.OrcSimpleStatsExtractor;
 import org.apache.paimon.format.orc.writer.RowDataVectorizer;
 import org.apache.paimon.format.orc.writer.Vectorizer;
+import org.apache.paimon.fs.ObjectCacheManager;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.predicate.Predicate;
@@ -43,12 +44,14 @@
 import org.apache.paimon.types.MultisetType;
 import org.apache.paimon.types.RowType;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.orc.OrcConf;
 import org.apache.orc.TypeDescription;
 
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
@@ -71,13 +74,29 @@ public class OrcFileFormat extends FileFormat {
     private final int writeBatchSize;
     private final boolean deletionVectorsEnabled;
 
+    private static final org.apache.hadoop.conf.Configuration emptyConf =
+            new org.apache.hadoop.conf.Configuration();
+    private static final ObjectCacheManager<Properties, Configuration> configCache =
+            ObjectCacheManager.newObjectCacheManager(Duration.ofDays(365), 1000);
+
+    static {
+        emptyConf.set("paimon.empty.configuration", "paimon.empty.configuration");
+    }
+
     public OrcFileFormat(FormatContext formatContext) {
         super(IDENTIFIER);
         this.orcProperties = getOrcProperties(formatContext.options(), formatContext);
-        this.readerConf = new org.apache.hadoop.conf.Configuration();
-        this.orcProperties.forEach((k, v) -> readerConf.set(k.toString(), v.toString()));
-        this.writerConf = new org.apache.hadoop.conf.Configuration();
-        this.orcProperties.forEach((k, v) -> writerConf.set(k.toString(), v.toString()));
+        Configuration conf;
+        Configuration cachedConf = configCache.getIfPresent(orcProperties);
+        if (cachedConf != null) {
+            conf = cachedConf;
+        } else {
+            conf = new org.apache.hadoop.conf.Configuration(emptyConf);
+            this.orcProperties.forEach((k, v) -> conf.set(k.toString(), v.toString()));
+            configCache.put(orcProperties, conf);
+        }
+        this.readerConf = conf;
+        this.writerConf = conf;
         this.readBatchSize = formatContext.readBatchSize();
         this.writeBatchSize = formatContext.writeBatchSize();
         this.deletionVectorsEnabled = formatContext.options().get(DELETION_VECTORS_ENABLED);

paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFileFormatTest.java

@@ -83,4 +83,68 @@ public void testSupportedDataTypes() {
         dataFields.add(new DataField(index++, "decimal_type", DataTypes.DECIMAL(10, 3)));
         orc.validateDataFields(new RowType(dataFields));
     }
+
+    @Test
+    public void testCreateCost() {
+        double createConfCost = createConfigCost();
+        for (int i = 0; i < 1000; i++) {
+            create();
+        }
+        int times = 10_000;
+        long start = System.nanoTime();
+        for (int i = 0; i < times; i++) {
+            create();
+        }
+        double cost = ((double) (System.nanoTime() - start)) / 1000_000 / times;
+        assertThat(cost * 500 < createConfCost).isTrue();
+    }
+
+    @Test
+    public void testCreateCostWithRandomConfig() {
+        double createConfCost = createConfigCost();
+        for (int i = 0; i < 1000; i++) {
+            createRandomConfig();
+        }
+        int times = 10_000;
+        long start = System.nanoTime();
+        for (int i = 0; i < times; i++) {
+            createRandomConfig();
+        }
+        double cost = ((double) (System.nanoTime() - start)) / 1000_000 / times;
+        assertThat(cost * 10 < createConfCost).isTrue();
+    }
+
+    private double createConfigCost() {
+        for (int i = 0; i < 1000; i++) {
+            createConfig();
+        }
+        int times = 10_000;
+        long start = System.nanoTime();
+        for (int i = 0; i < times; i++) {
+            createConfig();
+        }
+        return ((double) (System.nanoTime() - start)) / 1000_000 / times;
+    }
+
+    private void createConfig() {
+        org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
+        conf.set("a", "a");
+    }
+
+    private void create() {
+        Options options = new Options();
+        options.setString("haha", "1");
+        options.setString("compress", "zlib");
+        OrcFileFormat orcFileFormat =
+                new OrcFileFormatFactory().create(new FormatContext(options, 1024, 1024));
+    }
+
+    private void createRandomConfig() {
+        Options options = new Options();
+        options.setString("haha", "1");
+        options.setString("compress", "zlib");
+        options.setString("a", Math.random() + "");
+        OrcFileFormat orcFileFormat =
+                new OrcFileFormatFactory().create(new FormatContext(options, 1024, 1024));
+    }
 }