[format] Resolve the ORC compression kind independent of the default locale

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

@@ -45,6 +45,7 @@
 
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
@@ -58,6 +59,8 @@
  */
 public class OrcWriterFactory implements FormatWriterFactory, SupportsShreddingWritePlan {
 
+    private static final String COMPRESS_ATTRIBUTE = OrcConf.COMPRESS.getAttribute();
+
     private final Vectorizer<InternalRow> vectorizer;
     private final Properties writerProperties;
     private final Map<String, String> confMap;
@@ -84,23 +87,45 @@ public OrcWriterFactory(
             MemorySize writeBatchMemory,
             boolean legacyTimestampLtzType) {
         this.vectorizer = checkNotNull(vectorizer);
-        this.writerProperties = checkNotNull(writerProperties);
+        this.writerProperties = upperCaseCompression(checkNotNull(writerProperties));
         this.confMap = new HashMap<>();
         this.legacyTimestampLtzType = legacyTimestampLtzType;
 
         // Todo: Replace the Map based approach with a better approach
         for (Map.Entry<String, String> entry : configuration) {
             confMap.put(entry.getKey(), entry.getValue());
         }
+        String compress = confMap.get(COMPRESS_ATTRIBUTE);
+        if (compress != null) {
+            confMap.put(COMPRESS_ATTRIBUTE, compress.toUpperCase(Locale.ROOT));
+        }
         this.writeBatchSize = writeBatchSize;
         this.writeBatchMemory = writeBatchMemory;
     }
 
+    /**
+     * ORC resolves the compression kind through a locale sensitive {@code toUpperCase}, which turns
+     * the 'i' of zlib into 'İ' under a Turkish or Azeri default locale and then matches no {@link
+     * CompressionKind}. Upper case the option once here instead.
+     */
+    private static Properties upperCaseCompression(Properties properties) {
+        String compress = properties.getProperty(COMPRESS_ATTRIBUTE);
+        if (compress == null) {
+            return properties;
+        }
+        Properties normalized = new Properties();
+        for (String name : properties.stringPropertyNames()) {
+            normalized.setProperty(name, properties.getProperty(name));
+        }
+        normalized.setProperty(COMPRESS_ATTRIBUTE, compress.toUpperCase(Locale.ROOT));
+        return normalized;
+    }
+
     @Override
     public FormatWriter create(PositionOutputStream out, String compression) throws IOException {
         OrcFile.WriterOptions opts = getWriterOptions();
-        if (!writerProperties.containsKey(OrcConf.COMPRESS.getAttribute())) {
-            opts.compress(CompressionKind.valueOf(compression.toUpperCase()));
+        if (!writerProperties.containsKey(COMPRESS_ATTRIBUTE)) {
+            opts.compress(CompressionKind.valueOf(compression.toUpperCase(Locale.ROOT)));
         }
 
         opts.physicalWriter(

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

@@ -22,10 +22,13 @@
 import org.apache.paimon.format.orc.writer.RowDataVectorizer;
 import org.apache.paimon.format.orc.writer.Vectorizer;
 import org.apache.paimon.fs.local.LocalFileIO.LocalPositionOutputStream;
+import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataTypes;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.orc.CompressionKind;
 import org.apache.orc.MemoryManager;
 import org.apache.orc.OrcFile;
 import org.apache.orc.TypeDescription;
@@ -37,6 +40,8 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
+import java.util.Properties;
 
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -113,4 +118,78 @@ void testWriterOptionsNotSharedBetweenCalls() {
                                 false));
         assertThat(factory.getWriterOptions()).isNotSameAs(factory.getWriterOptions());
     }
+
+    @Test
+    void testLowerCaseFileCompressionUnderTurkishLocale(@TempDir java.nio.file.Path tmpDir)
+            throws IOException {
+        // 'i' uppercases to 'İ' in Turkish, so a locale sensitive conversion turns zlib into
+        // ZLİB and CompressionKind.valueOf rejects it
+        CapturingOrcWriterFactory factory = capturingFactory(new Properties());
+        withTurkishLocale(
+                () -> factory.create(outputStream(tmpDir, "file-compression.orc"), "zlib").close());
+        assertThat(factory.captured.getCompress()).isEqualTo(CompressionKind.ZLIB);
+    }
+
+    @Test
+    void testLowerCaseOrcCompressPropertyUnderTurkishLocale(@TempDir java.nio.file.Path tmpDir)
+            throws IOException {
+        // the orc.compress table option is resolved by OrcFile.WriterOptions instead, one call
+        // earlier than the branch above
+        Properties properties = new Properties();
+        properties.setProperty("orc.compress", "zlib");
+        CapturingOrcWriterFactory factory = capturingFactory(properties);
+        withTurkishLocale(
+                () -> factory.create(outputStream(tmpDir, "orc-compress.orc"), "zstd").close());
+        assertThat(factory.captured.getCompress()).isEqualTo(CompressionKind.ZLIB);
+    }
+
+    private static LocalPositionOutputStream outputStream(java.nio.file.Path tmpDir, String name)
+            throws IOException {
+        return new LocalPositionOutputStream(tmpDir.resolve(name).toFile());
+    }
+
+    private static CapturingOrcWriterFactory capturingFactory(Properties writerProperties) {
+        return new CapturingOrcWriterFactory(
+                new RowDataVectorizer(
+                        TypeDescription.createString(),
+                        Collections.singletonList(new DataField(0, "f0", DataTypes.STRING())),
+                        false),
+                writerProperties);
+    }
+
+    private static void withTurkishLocale(ThrowingRunnable body) throws IOException {
+        Locale original = Locale.getDefault();
+        try {
+            Locale.setDefault(new Locale("tr", "TR"));
+            body.run();
+        } finally {
+            Locale.setDefault(original);
+        }
+    }
+
+    private interface ThrowingRunnable {
+        void run() throws IOException;
+    }
+
+    private static class CapturingOrcWriterFactory extends OrcWriterFactory {
+
+        private OrcFile.WriterOptions captured;
+
+        private CapturingOrcWriterFactory(
+                Vectorizer<InternalRow> vectorizer, Properties writerProperties) {
+            super(
+                    vectorizer,
+                    writerProperties,
+                    new Configuration(false),
+                    1024,
+                    MemorySize.ZERO,
+                    false);
+        }
+
+        @Override
+        protected OrcFile.WriterOptions getWriterOptions() {
+            captured = super.getWriterOptions();
+            return captured;
+        }
+    }
 }