[core] Use Locale.ROOT for every case conversion in main sources

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -3273,7 +3273,7 @@ public Map<Integer, String> statsModePerLevel() {
     }
 
     public static String normalizeFileFormat(String fileFormat) {
-        return StringUtils.isEmpty(fileFormat) ? fileFormat : fileFormat.toLowerCase();
+        return StringUtils.isEmpty(fileFormat) ? fileFormat : fileFormat.toLowerCase(Locale.ROOT);
     }
 
     public String dataFilePrefix() {
@@ -4238,7 +4238,10 @@ public String partitionMarkDoneCustomClass() {
 
     public Set<PartitionMarkDoneAction> partitionMarkDoneActions() {
         return Arrays.stream(options.get(PARTITION_MARK_DONE_ACTION).split(","))
-                .map(x -> PartitionMarkDoneAction.valueOf(x.replace('-', '_').toUpperCase()))
+                .map(
+                        x ->
+                                PartitionMarkDoneAction.valueOf(
+                                        x.replace('-', '_').toUpperCase(Locale.ROOT)))
                 .collect(Collectors.toCollection(HashSet::new));
     }
 

paimon-api/src/main/java/org/apache/paimon/options/OptionsUtils.java

@@ -199,7 +199,7 @@ static Boolean convertToBoolean(Object o) {
             return (Boolean) o;
         }
 
-        switch (o.toString().toUpperCase()) {
+        switch (o.toString().toUpperCase(Locale.ROOT)) {
             case "TRUE":
                 return true;
             case "FALSE":

paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java

@@ -46,6 +46,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Function;
@@ -221,7 +222,9 @@ private static String extractRequestId(ClassicHttpResponse response) {
                 .filter(
                         h ->
                                 h.getName() != null
-                                        && h.getName().toLowerCase().contains("request-id"))
+                                        && h.getName()
+                                                .toLowerCase(Locale.ROOT)
+                                                .contains("request-id"))
                 .map(Header::getValue)
                 .filter(Objects::nonNull)
                 .findFirst()

paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFAuthProviderFactory.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.rest.RESTCatalogOptions;
 import org.apache.paimon.utils.StringUtils;
 
+import java.util.Locale;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -96,7 +97,7 @@ protected static String parseSigningAlgoFromUri(String uri) {
         }
 
         // Check for aliyun openapi endpoints
-        if (uri.toLowerCase().contains("dlfnext")) {
+        if (uri.toLowerCase(Locale.ROOT).contains("dlfnext")) {
             return DLFOpenApiSigner.IDENTIFIER;
         }
 

paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFDefaultSigner.java

@@ -34,6 +34,7 @@
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -62,12 +63,12 @@ public class DLFDefaultSigner implements DLFRequestSigner {
     private static final String NEW_LINE = "\n";
     private static final List<String> SIGNED_HEADERS =
             Arrays.asList(
-                    DLF_CONTENT_MD5_HEADER_KEY.toLowerCase(),
-                    DLF_CONTENT_TYPE_KEY.toLowerCase(),
-                    DLF_CONTENT_SHA56_HEADER_KEY.toLowerCase(),
-                    DLF_DATE_HEADER_KEY.toLowerCase(),
-                    DLF_AUTH_VERSION_HEADER_KEY.toLowerCase(),
-                    DLF_SECURITY_TOKEN_HEADER_KEY.toLowerCase());
+                    DLF_CONTENT_MD5_HEADER_KEY.toLowerCase(Locale.ROOT),
+                    DLF_CONTENT_TYPE_KEY.toLowerCase(Locale.ROOT),
+                    DLF_CONTENT_SHA56_HEADER_KEY.toLowerCase(Locale.ROOT),
+                    DLF_DATE_HEADER_KEY.toLowerCase(Locale.ROOT),
+                    DLF_AUTH_VERSION_HEADER_KEY.toLowerCase(Locale.ROOT),
+                    DLF_SECURITY_TOKEN_HEADER_KEY.toLowerCase(Locale.ROOT));
 
     private final String region;
 
@@ -215,7 +216,7 @@ private static TreeMap<String, String> buildSortedSignedHeadersMap(
         TreeMap<String, String> orderMap = new TreeMap<>();
         if (headers != null) {
             for (Map.Entry<String, String> header : headers.entrySet()) {
-                String key = header.getKey().toLowerCase();
+                String key = header.getKey().toLowerCase(Locale.ROOT);
                 if (SIGNED_HEADERS.contains(key)) {
                     orderMap.put(key, StringUtils.trim(header.getValue()));
                 }

paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFOpenApiSigner.java

@@ -184,7 +184,7 @@ public String identifier() {
     private static String buildCanonicalizedHeaders(Map<String, String> headers) {
         TreeMap<String, String> sortedHeaders = new TreeMap<>();
         for (Map.Entry<String, String> entry : headers.entrySet()) {
-            String key = entry.getKey().toLowerCase();
+            String key = entry.getKey().toLowerCase(Locale.ROOT);
             if (key.startsWith("x-acs-")) {
                 sortedHeaders.put(key, StringUtils.trim(entry.getValue()));
             }

paimon-api/src/main/java/org/apache/paimon/types/DataTypeJsonParser.java

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
@@ -213,7 +214,7 @@ private static List<Token> tokenize(String chars) {
                     builder.setLength(0);
                     cursor = consumeIdentifier(builder, chars, cursor);
                     final String token = builder.toString();
-                    final String normalizedToken = token.toUpperCase();
+                    final String normalizedToken = token.toUpperCase(Locale.ROOT);
                     if (KEYWORDS.contains(normalizedToken)) {
                         tokens.add(new Token(TokenType.KEYWORD, cursor, normalizedToken));
                     } else {
@@ -344,7 +345,7 @@ private enum Keyword {
 
     private static final Set<String> KEYWORDS =
             Stream.of(Keyword.values())
-                    .map(k -> k.toString().toUpperCase())
+                    .map(k -> k.toString().toUpperCase(Locale.ROOT))
                     .collect(Collectors.toSet());
 
     private static class Token {

paimon-api/src/main/java/org/apache/paimon/types/RowKind.java

@@ -20,6 +20,8 @@
 
 import org.apache.paimon.annotation.Public;
 
+import java.util.Locale;
+
 /**
  * Lists all kinds of changes that a row can describe in a changelog.
  *
@@ -135,7 +137,7 @@ public static RowKind fromByteValue(byte value) {
      * @see #shortString() for mapping of string and {@link RowKind}.
      */
     public static RowKind fromShortString(String value) {
-        switch (value.toUpperCase()) {
+        switch (value.toUpperCase(Locale.ROOT)) {
             case "+I":
                 return INSERT;
             case "-U":

paimon-api/src/main/java/org/apache/paimon/utils/StringUtils.java

@@ -22,6 +22,7 @@
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Objects;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
@@ -668,7 +669,9 @@ public static String quote(String str) {
     }
 
     public static String toLowerCaseIfNeed(String str, boolean caseSensitive) {
-        return caseSensitive ? str : str.toLowerCase();
+        // Locale.ROOT: identifier matching must not depend on the JVM default locale
+        // (e.g. Turkish lowercases 'I' to a dotless glyph and breaks column mapping)
+        return caseSensitive ? str : str.toLowerCase(Locale.ROOT);
     }
 
     public static boolean isNumeric(final CharSequence cs) {
@@ -733,14 +736,14 @@ public static String toUpperCase(String value) {
         if (value == null) {
             return null;
         }
-        return value.toUpperCase();
+        return value.toUpperCase(Locale.ROOT);
     }
 
     public static String toLowerCase(String value) {
         if (value == null) {
             return null;
         }
-        return value.toLowerCase();
+        return value.toLowerCase(Locale.ROOT);
     }
 
     public static boolean isOpenBracket(char c) {

paimon-api/src/test/java/org/apache/paimon/TurkishLocaleParsingTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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;
+
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.RowKind;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Locale;
+
+import static org.apache.paimon.CoreOptions.PARTITION_MARK_DONE_ACTION;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Parsing an option value, an enum name or a protocol token uppercases or lowercases it first.
+ * Under a Turkish default locale 'i' maps to a dotted capital and 'I' to a dotless small letter, so
+ * those conversions must pin {@link Locale#ROOT} or the token no longer matches what it is compared
+ * against.
+ */
+class TurkishLocaleParsingTest {
+
+    private Locale original;
+
+    @BeforeEach
+    void setUp() {
+        original = Locale.getDefault();
+        Locale.setDefault(new Locale("tr", "TR"));
+    }
+
+    @AfterEach
+    void tearDown() {
+        Locale.setDefault(original);
+    }
+
+    @Test
+    void partitionMarkDoneActionsParse() {
+        // SUCCESS_FILE and DONE_PARTITION both contain an 'i': a locale-sensitive uppercase
+        // turns them into names no enum constant has, and valueOf throws
+        Options options = new Options();
+        options.set(PARTITION_MARK_DONE_ACTION, "success-file,done-partition");
+
+        assertThat(new CoreOptions(options).partitionMarkDoneActions())
+                .containsExactlyInAnyOrder(
+                        CoreOptions.PartitionMarkDoneAction.SUCCESS_FILE,
+                        CoreOptions.PartitionMarkDoneAction.DONE_PARTITION);
+    }
+
+    @Test
+    void rowKindFromLowerCaseShortString() {
+        // "+i" is the only short string this can catch: Turkish differs from ROOT on 'i' and
+        // 'I' alone, so "-d" or "-u" would pass whichever conversion the code uses
+        assertThat(RowKind.fromShortString("+i")).isEqualTo(RowKind.INSERT);
+    }
+}

paimon-benchmark/paimon-cluster-benchmark/src/main/java/org/apache/paimon/benchmark/metric/cpu/OperatingSystem.java

@@ -18,6 +18,8 @@
 
 package org.apache.paimon.benchmark.metric.cpu;
 
+import java.util.Locale;
+
 /** An enumeration indicating the operating system that the JVM runs on. */
 public enum OperatingSystem {
     LINUX,
@@ -115,7 +117,7 @@ private static OperatingSystem readOSFromSystemProperties() {
         if (osName.startsWith(FREEBSD_OS_PREFIX)) {
             return FREE_BSD;
         }
-        String osNameLowerCase = osName.toLowerCase();
+        String osNameLowerCase = osName.toLowerCase(Locale.ROOT);
         if (osNameLowerCase.contains(SOLARIS_OS_INFIX_1)
                 || osNameLowerCase.contains(SOLARIS_OS_INFIX_2)) {
             return SOLARIS;

paimon-common/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java

@@ -23,6 +23,8 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Locale;
+
 /**
  * Each compression codec has an implementation of {@link BlockCompressionFactory} to create
  * compressors and decompressors.
@@ -38,7 +40,7 @@ public interface BlockCompressionFactory {
     /** Creates {@link BlockCompressionFactory} according to the configuration. */
     @Nullable
     static BlockCompressionFactory create(CompressOptions compression) {
-        switch (compression.compress().toUpperCase()) {
+        switch (compression.compress().toUpperCase(Locale.ROOT)) {
             case "NONE":
                 return null;
             case "ZSTD":

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

@@ -30,6 +30,7 @@
 
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 
@@ -88,16 +89,19 @@ 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())
+                        FileFormat.class.getClassLoader(), identifier.toLowerCase(Locale.ROOT))
                 .create(context);
     }
 
     protected Options getIdentifierPrefixOptions(Options options) {
         Map<String, String> result = new HashMap<>();
-        String prefix = formatIdentifier.toLowerCase() + ".";
+        // match against the identifier as written so the suffix is sliced at an offset the key
+        // actually has: lower-casing can lengthen a string, and U+0130 lower-cases to two chars
+        String prefix = formatIdentifier + ".";
+        String lowerCasePrefix = formatIdentifier.toLowerCase(Locale.ROOT) + ".";
         for (String key : options.keySet()) {
-            if (key.toLowerCase().startsWith(prefix)) {
-                result.put(prefix + key.substring(prefix.length()), options.get(key));
+            if (key.regionMatches(true, 0, prefix, 0, prefix.length())) {
+                result.put(lowerCasePrefix + key.substring(prefix.length()), options.get(key));
             }
         }
         return new Options(result);

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

@@ -48,6 +48,7 @@
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;
@@ -562,7 +563,7 @@ static FileIO get(Path path, CatalogContext config) throws IOException {
             for (String[] keys : loader.requiredOptions()) {
                 boolean found = false;
                 for (String key : keys) {
-                    if (options.contains(key.toLowerCase())) {
+                    if (options.contains(key.toLowerCase(Locale.ROOT))) {
                         found = true;
                         break;
                     }

paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java

@@ -25,6 +25,7 @@
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.PriorityQueue;
 
@@ -63,7 +64,7 @@ public static String normalizeRanker(String ranker) {
         if (ranker == null || ranker.trim().isEmpty()) {
             return RRF_RANKER;
         }
-        String normalized = ranker.trim().toLowerCase();
+        String normalized = ranker.trim().toLowerCase(Locale.ROOT);
         if (!RRF_RANKER.equals(normalized)
                 && !WEIGHTED_SCORE_RANKER.equals(normalized)
                 && !MRR_RANKER.equals(normalized)) {

paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java

@@ -46,6 +46,7 @@
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.PriorityQueue;
 import java.util.TreeMap;
@@ -946,7 +947,8 @@ public void accept(MemorySlice key, byte[] value) throws IOException {
 
     private File newSstFile() {
         long sequence = fileSequence.getAndIncrement();
-        return new File(dataDirectory, String.format("sst-%s-%06d.db", uuid, sequence));
+        return new File(
+                dataDirectory, String.format(Locale.ROOT, "sst-%s-%06d.db", uuid, sequence));
     }
 
     private void ensureOpen() {