[common] Support unescaped characters in blob file paths

paimon-common/src/main/java/org/apache/paimon/utils/UriReaderFactory.java

@@ -60,6 +60,15 @@ private static URI parseUri(String input) {
         try {
             return URI.create(input);
         } catch (IllegalArgumentException e) {
+            // File paths may contain unescaped characters accepted by Path.
+            try {
+                URI pathUri = new Path(input).toUri();
+                if (!isHttp(pathUri)) {
+                    return pathUri;
+                }
+            } catch (IllegalArgumentException ignored) {
+                // Throw the sanitized exception below.
+            }
             throw SensitiveConfigUtils.invalidUri(input);
         }
     }
@@ -94,7 +103,8 @@ protected UriReader newReader(URI uri) {
     }
 
     private static boolean isHttp(URI uri) {
-        return "http".equals(uri.getScheme()) || "https".equals(uri.getScheme());
+        return "http".equalsIgnoreCase(uri.getScheme())
+                || "https".equalsIgnoreCase(uri.getScheme());
     }
 
     private static final class ProvidedFileIOUriReaderFactory extends UriReaderFactory {

paimon-common/src/test/java/org/apache/paimon/utils/UriReaderFactoryTest.java

@@ -85,6 +85,12 @@ public void testInvalidUriDoesNotLeakCredentials() {
                         });
     }
 
+    @Test
+    public void testInvalidUpperCaseHttpUriDoesNotFallBackToFileReader() {
+        assertThatThrownBy(() -> factory.create("HTTPS://example.com/bad path"))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+
     @Test
     public void testCreateHttpsUriReader() {
         UriReader reader = factory.create("https://example.com/file.txt");
@@ -97,6 +103,21 @@ public void testCreateFileUriReader() {
         assertThat(reader).isInstanceOf(FileUriReader.class);
     }
 
+    @Test
+    public void testReadFileUriWithUnescapedCharacters() throws Exception {
+        java.nio.file.Path file = tempPath.resolve("\u4ed5\u5e9c\u516c\u9986 (2).jpg");
+        Files.write(file, new byte[] {1, 2});
+        String fileUri = "file://" + file.toAbsolutePath();
+
+        UriReader reader = factory.create(fileUri);
+
+        assertThat(reader).isInstanceOf(FileUriReader.class);
+        try (SeekableInputStream inputStream = reader.newInputStream(fileUri)) {
+            assertThat(inputStream.read()).isEqualTo(1);
+            assertThat(inputStream.read()).isEqualTo(2);
+        }
+    }
+
     @Test
     public void testProvidedFileIOSurvivesSerialization() throws Exception {
         java.nio.file.Path file = tempPath.resolve("file.txt");

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java

@@ -23,11 +23,15 @@
 import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.data.BlobRef;
 import org.apache.paimon.data.BlobViewStruct;
+import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.rest.TestHttpWebServer;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.types.DataTypeRoot;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.UriReader;
@@ -690,6 +694,55 @@ public void testDescriptorToPresignedUrlBuiltInFunctions() {
                 .hasStackTraceContaining("does not support creating blob presigned URLs");
     }
 
+    @Test
+    public void testWriteBlobWithUnicodeAndSpaceInPath() throws Exception {
+        byte[] blobData = "image-content".getBytes();
+        FileIO fileIO = new LocalFileIO();
+        String uri = "file://" + warehouse + "/\u4ed5\u5e9c\u516c\u9986 (2).jpg";
+        try (OutputStream outputStream =
+                fileIO.newOutputStream(new org.apache.paimon.fs.Path(uri), true)) {
+            outputStream.write(blobData);
+        }
+
+        batchSql(
+                "INSERT INTO blob_table_descriptor VALUES"
+                        + " (1, 'paimon', sys.path_to_descriptor('"
+                        + uri
+                        + "'))");
+        batchSql("ALTER TABLE blob_table_descriptor SET ('blob-as-descriptor'='false')");
+
+        assertThat(batchSql("SELECT picture FROM blob_table_descriptor"))
+                .containsExactly(Row.of(blobData));
+    }
+
+    @Test
+    public void testReadDescriptorBlobWithUnicodeAndSpaceInPath() throws Exception {
+        byte[] blobData = "image-content".getBytes();
+        FileIO fileIO = new LocalFileIO();
+        String uri = "file://" + warehouse + "/\u4ed5\u5e9c\u516c\u9986 (2).jpg";
+        try (OutputStream outputStream =
+                fileIO.newOutputStream(new org.apache.paimon.fs.Path(uri), true)) {
+            outputStream.write(blobData);
+        }
+
+        tEnv.executeSql(
+                "CREATE TABLE external_blob_source (id INT, picture BYTES)"
+                        + " WITH ('row-tracking.enabled'='true',"
+                        + " 'data-evolution.enabled'='true',"
+                        + " 'blob-descriptor-field'='picture')");
+        FileStoreTable table = paimonTable("external_blob_source");
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite();
+                BatchTableCommit commit = writeBuilder.newCommit()) {
+            BlobDescriptor descriptor = new BlobDescriptor(uri, 0, blobData.length);
+            write.write(GenericRow.of(1, new BlobRef(UriReader.fromFile(fileIO), descriptor)));
+            commit.commit(write.prepareCommit());
+        }
+
+        assertThat(batchSql("SELECT picture FROM external_blob_source"))
+                .containsExactly(Row.of(blobData));
+    }
+
     @Test
     public void testWriteBlobViewWithBuiltInFunction() throws Exception {
         tEnv.executeSql(