[python] Address PR apache#7347 review feedback: use @cachedmethod, trim docs, remove ChangelogProducer

- Upgrade cachetools to >=7,<8 for cachedmethod(info=True) support
- Remove ChangelogProducer enum (belongs in apache#7348 scanners branch)
- Replace manual cache hit/miss counters with @cachedmethod(info=True)
  decorator on ManifestFileManager, ManifestListManager, SnapshotManager
- Trim verbose docstrings across identifier, file_io, pyarrow_file_io,
  manifest_list_manager, and snapshot_manager
- Update cache tests to use cache_info() instead of manual counters

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: tub

paimon-python/dev/requirements.txt

@@ -17,7 +17,7 @@
 ################################################################################
 # Core dependencies for pypaimon
 cachetools>=4.2,<6; python_version=="3.6"
-cachetools>=5,<6; python_version>"3.6"
+cachetools>=7,<8; python_version>"3.6"
 dataclasses>=0.8; python_version < "3.7"
 fastavro>=1.4,<2
 fsspec>=2021.10,<2026; python_version<"3.8"

paimon-python/pypaimon/common/file_io.py

@@ -53,18 +53,7 @@ def exists(self, path: str) -> bool:
         pass
 
     def exists_batch(self, paths: List[str]) -> Dict[str, bool]:
-        """
-        Check existence of multiple paths.
-
-        Default implementation calls exists() for each path sequentially.
-        Subclasses may override to provide more efficient batch operations.
-
-        Args:
-            paths: List of file paths to check
-
-        Returns:
-            Dictionary mapping each path to its existence status
-        """
+        """Check existence of multiple paths, returning {path: bool}."""
         return {path: self.exists(path) for path in paths}
 
     @abstractmethod

paimon-python/pypaimon/common/identifier.py

@@ -37,29 +37,7 @@ def create(cls, database: str, object: str) -> "Identifier":
 
     @classmethod
     def from_string(cls, full_name: str) -> "Identifier":
-        """
-        Parse an identifier from a string.
-
-        Supports two modes:
-        1. Backtick-quoted (for names with periods):
-           - `database.name`.table → database="database.name", object="table"
-           - `db`.`table` → database="db", object="table"
-
-        2. Simple split on first period (Java-compatible fallback):
-           - database.table → database="database", object="table"
-
-        For database names containing periods, use backticks or
-        Identifier.create("database.name", "table") directly.
-
-        Args:
-            full_name: The full identifier string
-
-        Returns:
-            Identifier instance
-
-        Raises:
-            ValueError: If the format is invalid
-        """
+        """Parse a 'database.object' identifier, with optional backtick quoting."""
         if not full_name or not full_name.strip():
             raise ValueError("fullName cannot be null or empty")
 
@@ -79,13 +57,6 @@ def from_string(cls, full_name: str) -> "Identifier":
 
     @classmethod
     def _parse_with_backticks(cls, full_name: str) -> "Identifier":
-        """
-        Parse identifier with backtick-quoted segments.
-
-        Examples:
-        - `db.name`.table → database="db.name", object="table"
-        - `db`.`table` → database="db", object="table"
-        """
         parts = []
         current = ""
         in_backticks = False

paimon-python/pypaimon/common/options/core_options.py

@@ -45,21 +45,6 @@ class MergeEngine(str, Enum):
     FIRST_ROW = "first-row"
 
 
-class ChangelogProducer(str, Enum):
-    """
-    Specifies the changelog producer for tables with primary key.
-
-    - NONE: No changelog file (default for append-only tables)
-    - INPUT: Double write changelog from input
-    - FULL_COMPACTION: Generate changelog with each full compaction
-    - LOOKUP: Generate changelog through lookup compaction
-    """
-    NONE = "none"
-    INPUT = "input"
-    FULL_COMPACTION = "full-compaction"
-    LOOKUP = "lookup"
-
-
 class CoreOptions:
     """Core options for Paimon tables."""
     # File format constants
@@ -276,15 +261,6 @@ class CoreOptions:
                           "Options: deduplicate, partial-update, aggregation, first-row.")
     )
 
-    CHANGELOG_PRODUCER: ConfigOption[ChangelogProducer] = (
-        ConfigOptions.key("changelog-producer")
-        .enum_type(ChangelogProducer)
-        .default_value(ChangelogProducer.NONE)
-        .with_description(
-            "Whether to double write to a changelog file. "
-            "Options: none, input, full-compaction, lookup."
-        )
-    )
     # Commit options
     COMMIT_USER_PREFIX: ConfigOption[str] = (
         ConfigOptions.key("commit.user-prefix")
@@ -502,9 +478,6 @@ def deletion_vectors_enabled(self, default=None):
     def merge_engine(self, default=None):
         return self.options.get(CoreOptions.MERGE_ENGINE, default)
 
-    def changelog_producer(self, default=None):
-        return self.options.get(CoreOptions.CHANGELOG_PRODUCER, default)
-
     def data_file_external_paths(self, default=None):
         external_paths_str = self.options.get(CoreOptions.DATA_FILE_EXTERNAL_PATHS, default)
         if not external_paths_str:

paimon-python/pypaimon/filesystem/pyarrow_file_io.py

@@ -235,18 +235,7 @@ def exists(self, path: str) -> bool:
         return self._get_file_info(path_str).type != pafs.FileType.NotFound
 
     def exists_batch(self, paths: List[str]) -> Dict[str, bool]:
-        """
-        Check existence of multiple paths in one S3 API call.
-
-        This is more efficient than calling exists() for each path individually
-        as it batches the file info requests.
-
-        Args:
-            paths: List of file paths to check
-
-        Returns:
-            Dictionary mapping each path to its existence status
-        """
+        """Check existence of multiple paths in a single batched API call."""
         if not paths:
             return {}
 

paimon-python/pypaimon/manifest/manifest_file_manager.py

@@ -18,10 +18,10 @@
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from io import BytesIO
-from typing import List, Optional
+from typing import List
 
 import fastavro
-from cachetools import LRUCache
+from cachetools import LRUCache, cachedmethod
 
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
 from pypaimon.manifest.schema.manifest_entry import (MANIFEST_ENTRY_SCHEMA,
@@ -47,12 +47,7 @@ def __init__(self, table, cache_max_size: int = 100):
         self.primary_keys_fields = self.table.primary_keys_fields
         self.trimmed_primary_keys_fields = self.table.trimmed_primary_keys_fields
 
-        # LRU cache for manifest file contents
-        self._cache: Optional[LRUCache] = (
-            LRUCache(maxsize=cache_max_size) if cache_max_size > 0 else None
-        )
-        self._cache_hits = 0
-        self._cache_misses = 0
+        self._cache: LRUCache = LRUCache(maxsize=max(cache_max_size, 0))
 
     def read_entries_parallel(self, manifest_files: List[ManifestFileMeta], manifest_entry_filter=None,
                               drop_stats=True, max_workers=8) -> List[ManifestEntry]:
@@ -89,21 +84,7 @@ def _entry_identifier(e: ManifestEntry) -> tuple:
         return final_entries
 
     def read(self, manifest_file_name: str, manifest_entry_filter=None, drop_stats=True) -> List[ManifestEntry]:
-        # Check cache first (cache stores unfiltered entries with stats)
-        if self._cache is not None and manifest_file_name in self._cache:
-            self._cache_hits += 1
-            entries = self._cache[manifest_file_name]
-            return self._apply_post_read(entries, manifest_entry_filter, drop_stats)
-
-        self._cache_misses += 1
-
-        # Read from storage
         entries = self._read_from_storage(manifest_file_name)
-
-        # Cache the unfiltered entries
-        if self._cache is not None:
-            self._cache[manifest_file_name] = entries
-
         return self._apply_post_read(entries, manifest_entry_filter, drop_stats)
 
     def _apply_post_read(self, entries: List[ManifestEntry], manifest_entry_filter,
@@ -119,6 +100,7 @@ def _apply_post_read(self, entries: List[ManifestEntry], manifest_entry_filter,
                 result.append(entry)
         return result
 
+    @cachedmethod(lambda self: self._cache, key=lambda self, name: name, info=True)
     def _read_from_storage(self, manifest_file_name: str) -> List[ManifestEntry]:
         """Read manifest entries from storage (no filtering, with stats)."""
         manifest_file_path = f"{self.manifest_path}/{manifest_file_name}"

paimon-python/pypaimon/manifest/manifest_list_manager.py

@@ -20,7 +20,7 @@
 from typing import List, Optional
 
 import fastavro
-from cachetools import LRUCache
+from cachetools import LRUCache, cachedmethod
 
 from pypaimon.manifest.schema.manifest_file_meta import (
     MANIFEST_FILE_META_SCHEMA, ManifestFileMeta)
@@ -41,12 +41,7 @@ def __init__(self, table, cache_max_size: int = 50):
         self.manifest_path = f"{manifest_path}/manifest"
         self.file_io = self.table.file_io
 
-        # LRU cache for manifest list contents
-        self._cache: Optional[LRUCache] = (
-            LRUCache(maxsize=cache_max_size) if cache_max_size > 0 else None
-        )
-        self._cache_hits = 0
-        self._cache_misses = 0
+        self._cache: LRUCache = LRUCache(maxsize=max(cache_max_size, 0))
 
     def read_all(self, snapshot: Optional[Snapshot]) -> List[ManifestFileMeta]:
         if snapshot is None:
@@ -59,57 +54,22 @@ def read_all(self, snapshot: Optional[Snapshot]) -> List[ManifestFileMeta]:
         return manifest_files
 
     def read_base(self, snapshot: Snapshot) -> List[ManifestFileMeta]:
-        """
-        Read only base manifest list (all files at this snapshot point).
-
-        This is useful for computing diffs between two snapshots, where
-        we only need the base state and not the delta changes.
-
-        Args:
-            snapshot: The snapshot to read base manifest from
-
-        Returns:
-            List of ManifestFileMeta representing all files at this snapshot
-        """
+        """Read only the base manifest list for the given snapshot."""
         return self.read(snapshot.base_manifest_list)
 
     def read_delta(self, snapshot: Snapshot) -> List[ManifestFileMeta]:
         return self.read(snapshot.delta_manifest_list)
 
     def read_changelog(self, snapshot: Snapshot) -> List[ManifestFileMeta]:
-        """
-        Read changelog manifest files from snapshot.
-
-        For primary key tables with changelog-producer=input/full-compaction/lookup,
-        changelog files are stored in a separate manifest list.
-
-        Args:
-            snapshot: The snapshot to read changelog from
-
-        Returns:
-            List of ManifestFileMeta for changelog files, or empty list if no changelog
-        """
+        """Read changelog manifest files from snapshot, or empty list if none."""
         if snapshot.changelog_manifest_list is None:
             return []
         return self.read(snapshot.changelog_manifest_list)
 
     def read(self, manifest_list_name: str) -> List[ManifestFileMeta]:
-        # Check cache first
-        if self._cache is not None and manifest_list_name in self._cache:
-            self._cache_hits += 1
-            return self._cache[manifest_list_name]
-
-        self._cache_misses += 1
-
-        # Read from storage
-        result = self._read_from_storage(manifest_list_name)
-
-        # Cache the result
-        if self._cache is not None:
-            self._cache[manifest_list_name] = result
-
-        return result
+        return self._read_from_storage(manifest_list_name)
 
+    @cachedmethod(lambda self: self._cache, key=lambda self, name: name, info=True)
     def _read_from_storage(self, manifest_list_name: str) -> List[ManifestFileMeta]:
         """Read manifest list from storage."""
         manifest_files = []