diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/FileUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/FileUtils.java index 8c419537611f..1eef93b5815a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/FileUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/FileUtils.java @@ -43,7 +43,17 @@ public class FileUtils { */ public static Stream 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.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(); + } + }); } /** diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/FileUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/FileUtilsTest.java new file mode 100644 index 000000000000..c3170b3f86c1 --- /dev/null +++ b/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 versions = + FileUtils.listVersionedFiles(fileIO, directory, "snapshot-") + .collect(Collectors.toList()); + + assertThat(versions).containsExactly(1L); + } +} diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index f5b772d2256d..bfde96dc9058 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/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) diff --git a/paimon-python/pypaimon/filesystem/local_file_io.py b/paimon-python/pypaimon/filesystem/local_file_io.py index c35e34d19d3e..b315226181a6 100644 --- a/paimon-python/pypaimon/filesystem/local_file_io.py +++ b/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: diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index b789a5f8195f..4423c9559e30 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/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) diff --git a/paimon-python/pypaimon/tests/file_io_test.py b/paimon-python/pypaimon/tests/file_io_test.py index c7d2ebc6c7eb..f261b0a73bc9 100644 --- a/paimon-python/pypaimon/tests/file_io_test.py +++ b/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({}))