[python] Align atomic write temporary file naming

paimon-core/src/main/java/org/apache/paimon/utils/FileUtils.java

@@ -43,7 +43,17 @@ public class FileUtils {
      */
     public static Stream<Long> listVersionedFiles(FileIO fileIO, Path dir, String prefix)
             throws IOException {
-        return listOriginalVersionedFiles(fileIO, dir, prefix).map(Long::parseLong);
+        // Python temporary files may share the versioned-file prefix, for example
+        // snapshot-1<UUID>.tmp, so ignore entries which are not valid version IDs.
+        return listOriginalVersionedFiles(fileIO, dir, prefix)
+                .flatMap(
+                        version -> {
+                            try {
+                                return Stream.of(Long.parseLong(version));
+                            } catch (NumberFormatException ignored) {
+                                return Stream.empty();
+                            }
+                        });
     }
 
     /**

paimon-core/src/test/java/org/apache/paimon/utils/FileUtilsTest.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.utils;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FileUtils}. */
+public class FileUtilsTest {
+
+    @TempDir java.nio.file.Path tempDir;
+
+    @Test
+    public void testListVersionedFilesIgnoresInvalidVersions() throws IOException {
+        FileIO fileIO = LocalFileIO.create();
+        Path directory = new Path(tempDir.toString(), "snapshot");
+        fileIO.mkdirs(directory);
+        fileIO.writeFile(new Path(directory, "snapshot-1"), "", false);
+        String uuid = "d686aba1-b44a-40a4-a4f1-d854830aa5cb";
+        fileIO.writeFile(new Path(directory, "snapshot-2" + uuid + ".tmp"), "", false);
+        fileIO.writeFile(new Path(directory, "snapshot-3." + uuid + ".tmp"), "", false);
+        fileIO.writeFile(new Path(directory, "snapshot-999999999999999999999999999"), "", false);
+        fileIO.writeFile(new Path(directory, "unrelated"), "", false);
+
+        List<Long> versions =
+                FileUtils.listVersionedFiles(fileIO, directory, "snapshot-")
+                        .collect(Collectors.toList());
+
+        assertThat(versions).containsExactly(1L);
+    }
+}

paimon-python/pypaimon/common/file_io.py

@@ -55,6 +55,12 @@ def pread(stream, length: int, offset: int) -> bytes:
 _COALESCE_VIEW_MAX_RETAINED_AMPLIFICATION = 2.0
 
 
+def create_temp_path(path: str) -> str:
+    """Create the hidden temporary path used for an atomic write."""
+    separator = max(path.rfind('/'), path.rfind('\\'))
+    return f"{path[:separator + 1]}.{path[separator + 1:]}.{uuid.uuid4()}.tmp"
+
+
 def _coalesce_ranges(items, max_gap, max_span):
     """Group ``(idx, path, offset, length)`` (length >= 0) into merged spans:
     ``[(path, span_offset, span_length, [(idx, offset, length), ...])]``."""
@@ -292,7 +298,7 @@ def try_to_write_atomic(self, path: str, content: str) -> bool:
             if self.is_dir(path):
                 return False
 
-        temp_path = path + str(uuid.uuid4()) + ".tmp"
+        temp_path = create_temp_path(path)
         success = False
         try:
             self.write_file(temp_path, content, False)

paimon-python/pypaimon/filesystem/local_file_io.py

@@ -19,7 +19,6 @@
 import os
 import shutil
 import threading
-import uuid
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, Optional
@@ -28,7 +27,7 @@
 import pyarrow
 import pyarrow.fs as pafs
 
-from pypaimon.common.file_io import FileIO
+from pypaimon.common.file_io import FileIO, create_temp_path
 from pypaimon.common.options import Options
 from pypaimon.common.uri_reader import UriReaderFactory
 from pypaimon.filesystem.local import PaimonLocalFileSystem
@@ -243,7 +242,7 @@ def try_to_write_atomic(self, path: str, content: str) -> bool:
         if parent and not parent.exists():
             parent.mkdir(parents=True, exist_ok=True)
 
-        temp_path = file_path.parent / f"{file_path.name}.{uuid.uuid4()}.tmp"
+        temp_path = Path(create_temp_path(str(file_path)))
         success = False
         try:
             with open(temp_path, 'w', encoding='utf-8') as f:

paimon-python/pypaimon/filesystem/pyarrow_file_io.py

@@ -20,7 +20,6 @@
 import re
 import subprocess
 import threading
-import uuid
 from datetime import datetime, timezone
 from pathlib import PurePosixPath
 from typing import Any, Dict, List, Optional
@@ -31,7 +30,7 @@
 from packaging.version import parse
 from pyarrow._fs import FileSystem
 
-from pypaimon.common.file_io import FileIO
+from pypaimon.common.file_io import FileIO, create_temp_path
 from pypaimon.common.options import Options
 from pypaimon.common.options.config import OssOptions, S3Options, SecurityOptions
 from pypaimon.common.options.options_utils import OptionsUtils
@@ -581,7 +580,7 @@ def try_to_write_atomic(self, path: str, content: str) -> bool:
             if file_info.type == pafs.FileType.Directory:
                 return False
 
-        temp_path = path + str(uuid.uuid4()) + ".tmp"
+        temp_path = create_temp_path(path)
         success = False
         try:
             self.write_file(temp_path, content, False)

paimon-python/pypaimon/tests/file_io_test.py

@@ -24,6 +24,7 @@
 
 import pyarrow.fs as pafs
 
+from pypaimon.common.file_io import create_temp_path
 from pypaimon.common.options import Options
 from pypaimon.common.options.config import OssOptions
 from pypaimon.filesystem.local_file_io import LocalFileIO
@@ -33,6 +34,18 @@
 class FileIOTest(unittest.TestCase):
     """Test cases for FileIO.to_filesystem_path method."""
 
+    @patch('pypaimon.common.file_io.uuid.uuid4', return_value='test-uuid')
+    def test_create_temp_path(self, _):
+        self.assertEqual(
+            create_temp_path("oss://bucket/table/snapshot/snapshot-1"),
+            "oss://bucket/table/snapshot/.snapshot-1.test-uuid.tmp")
+        self.assertEqual(
+            create_temp_path("snapshot-1"),
+            ".snapshot-1.test-uuid.tmp")
+        self.assertEqual(
+            create_temp_path(r"C:\table\snapshot\snapshot-1"),
+            r"C:\table\snapshot\.snapshot-1.test-uuid.tmp")
+
     def test_filesystem_path_conversion(self):
         """Test S3FileSystem path conversion with various formats."""
         file_io = PyArrowFileIO("s3://bucket/warehouse", Options({}))