[python] Address review feedback for partition expiration

- Fix dead code in _parse_with_formatter: fallback now strips time
  directives and retries as date-only, matching Java's LocalDate.parse
- Replace fragile str.replace() with regex tokenizer for Java→Python
  format conversion to avoid overlapping token issues
- Skip partitions with last_file_creation_time=0 in update-time strategy
  to prevent false expiration of unknown-state partitions
- Remove unrelated statistics()/analyze() methods from file_store_table
- Fix _parse_duration type hint to Optional[str]
- Add tests for date-only fallback and zero-creation-time guard

Author: JunRuiLee

paimon-python/pypaimon/partition/partition_expire.py

@@ -292,7 +292,7 @@ def _read_partition_entries(table) -> List[PartitionEntry]:
     ]
 
 
-def _parse_duration(duration_str: str) -> Optional[timedelta]:
+def _parse_duration(duration_str: Optional[str]) -> Optional[timedelta]:
     """
     Parse a duration string like '7d', '1h', '30m', '10s', '1d 2h'.
     Also supports ISO-like formats: '7 days', '1 hour'.

paimon-python/pypaimon/partition/partition_expire_strategy.py

@@ -150,7 +150,14 @@ def select_expired_partitions(
         self, partition_entries: List[PartitionEntry], expiration_time: datetime
     ) -> List[PartitionEntry]:
         expiration_millis = int(expiration_time.timestamp() * 1000)
-        return [
-            entry for entry in partition_entries
-            if expiration_millis > entry.last_file_creation_time
-        ]
+        expired = []
+        for entry in partition_entries:
+            if entry.last_file_creation_time == 0:
+                logger.warning(
+                    "Partition %s has last_file_creation_time=0 (unknown), skipping expiration.",
+                    entry.spec,
+                )
+                continue
+            if expiration_millis > entry.last_file_creation_time:
+                expired.append(entry)
+        return expired

paimon-python/pypaimon/partition/partition_time_extractor.py

@@ -37,32 +37,41 @@
     "%Y-%m-%d",
 ]
 
-# Java DateTimeFormatter → Python strptime conversion (order matters: longest first)
-_JAVA_TO_PYTHON_PATTERNS = [
-    ("yyyy", "%Y"),
-    ("yy", "%y"),
-    ("MM", "%m"),
-    ("dd", "%d"),
-    ("HH", "%H"),
-    ("mm", "%M"),
-    ("SSS", "%f"),
-    ("SS", "%f"),
-    ("ss", "%S"),
-]
+# Java DateTimeFormatter → Python strptime token mapping.
+# Note: Python's %f is microseconds (6 digits); Java's SSS is milliseconds (3 digits).
+# This works for parsing because strptime(%f) accepts 1-6 digits, but the semantic
+# precision differs. Callers should be aware of this when formatting output.
+_JAVA_TO_PYTHON_TOKENS = {
+    "yyyy": "%Y",
+    "yy": "%y",
+    "MM": "%m",
+    "dd": "%d",
+    "HH": "%H",
+    "mm": "%M",
+    "SSS": "%f",
+    "SS": "%f",
+    "ss": "%S",
+}
+
+# Regex that matches Java tokens (longest first) or quoted literals
+_JAVA_TOKEN_RE = re.compile(
+    r"'([^']*)'|(" + "|".join(re.escape(k) for k in sorted(_JAVA_TO_PYTHON_TOKENS, key=len, reverse=True)) + r")"
+)
 
 
 def _java_to_python_format(java_fmt: str) -> str:
     """Convert Java DateTimeFormatter pattern to Python strptime format.
 
-    Handles:
-    - Standard tokens: yyyy, MM, dd, HH, mm, ss, SSS
-    - Quoted literals: 'T' → T (strips single quotes)
+    Uses regex tokenization to avoid overlapping-replacement issues with
+    sequential str.replace().
     """
-    import re
-    # Strip Java quoted literals: 'T' → T, '' → '
-    result = re.sub(r"'([^']*)'", r"\1", java_fmt)
-    for java_pat, py_pat in _JAVA_TO_PYTHON_PATTERNS:
-        result = result.replace(java_pat, py_pat)
+    def _replace_token(m):
+        if m.group(1) is not None:
+            return m.group(1)
+        return _JAVA_TO_PYTHON_TOKENS[m.group(2)]
+
+    # Replace known tokens; leave unmatched characters (literals like '-', ' ', 'T') as-is
+    result = _JAVA_TOKEN_RE.sub(_replace_token, java_fmt)
     return result
 
 
@@ -129,12 +138,23 @@ def _to_local_datetime(self, timestamp_string: str) -> datetime:
 
     @staticmethod
     def _parse_with_formatter(timestamp_string: str, formatter: str) -> datetime:
-        """Parse using the converted Python strptime pattern."""
+        """Parse using the converted Python strptime pattern.
+
+        Mirrors Java behavior: tries LocalDateTime.parse first, then falls back
+        to LocalDate.parse (date-only) with the same formatter pattern.
+        """
         try:
             return datetime.strptime(timestamp_string, formatter)
         except ValueError:
-            parsed_date = datetime.strptime(timestamp_string, formatter).date()
-            return datetime.combine(parsed_date, dt_time.min)
+            # Fallback: strip time directives (%H, %M, %S, %f) and surrounding
+            # literals to get a date-only format, mirroring Java's LocalDate.parse fallback.
+            date_only_format = re.split(r'(?=%H|%M|%S|%f)', formatter)[0].rstrip()
+            # Also strip any trailing non-directive separators (e.g. trailing space or 'T')
+            date_only_format = date_only_format.rstrip(' T-:')
+            if date_only_format and date_only_format != formatter:
+                parsed_date = datetime.strptime(timestamp_string, date_only_format).date()
+                return datetime.combine(parsed_date, dt_time.min)
+            raise
 
     @staticmethod
     def _parse_default(timestamp_string: str) -> datetime:

paimon-python/pypaimon/table/file_store_table.py

@@ -395,52 +395,6 @@ def new_batch_write_builder(self) -> BatchWriteBuilder:
     def new_stream_write_builder(self) -> StreamWriteBuilder:
         return StreamWriteBuilder(self)
 
-    def statistics(self) -> 'Optional[Statistics]':
-        """Read existing statistics for this table from the latest snapshot.
-
-        Returns:
-            A Statistics instance, or None if no statistics exist.
-        """
-        from pypaimon.stats.stats_file_handler import StatsFileHandler
-        snapshot_mgr = self.snapshot_manager()
-        snapshot = snapshot_mgr.get_latest_snapshot()
-        if snapshot is None:
-            return None
-        handler = StatsFileHandler(self.file_io, self.table_path)
-        return handler.read_stats(snapshot)
-
-    def analyze(self, columns: 'Optional[List[str]]' = None) -> 'Statistics':
-        """Compute column-level statistics from manifest metadata.
-
-        This method scans manifest file entries to aggregate per-file statistics
-        (min, max, null_count) into table-level column statistics. It does NOT
-        scan actual data files, making it very fast.
-
-        Args:
-            columns: Optional list of column names to analyze. If None, all
-                     columns are analyzed.
-
-        Returns:
-            A Statistics instance with the computed stats.
-
-        Raises:
-            ValueError: If the table has no snapshots or specified columns
-                       do not exist.
-        """
-        from pypaimon.stats.statistics_collector import StatisticsCollector
-        from pypaimon.stats.stats_file_handler import StatsFileHandler
-
-        collector = StatisticsCollector(self)
-        stats = collector.collect(columns=columns)
-        if stats is None:
-            raise ValueError("Cannot analyze table: no snapshots exist.")
-
-        # Write statistics file
-        handler = StatsFileHandler(self.file_io, self.table_path)
-        handler.write_stats(stats)
-
-        return stats
-
     def new_full_text_search_builder(self) -> 'FullTextSearchBuilder':
         from pypaimon.table.source.full_text_search_builder import FullTextSearchBuilderImpl
         return FullTextSearchBuilderImpl(self)

paimon-python/pypaimon/tests/partition_expire_test.py

@@ -81,6 +81,12 @@ def test_extract_single_digit_month_day(self):
         result = extractor.extract(["dt"], ["2024-1-5"])
         self.assertEqual(result, datetime(2024, 1, 5, 0, 0, 0))
 
+    def test_formatter_date_only_fallback(self):
+        """Formatter with time components should fall back to date-only parsing."""
+        extractor = PartitionTimeExtractor(formatter="yyyy-MM-dd HH:mm:ss")
+        result = extractor.extract(["dt"], ["2024-03-15"])
+        self.assertEqual(result, datetime(2024, 3, 15, 0, 0, 0))
+
 
 class TestPartitionValuesTimeExpireStrategy(unittest.TestCase):
     """Tests for PartitionValuesTimeExpireStrategy."""
@@ -183,6 +189,24 @@ def test_no_expired_when_all_recent(self):
         expired = strategy.select_expired_partitions(entries, expiration_time)
         self.assertEqual(len(expired), 0)
 
+    def test_zero_creation_time_skipped(self):
+        """Partitions with last_file_creation_time=0 should be skipped (unknown state)."""
+        strategy = PartitionUpdateTimeExpireStrategy(
+            partition_keys=["dt"],
+        )
+        now = datetime(2024, 6, 1)
+        old_time_millis = int((now - timedelta(days=30)).timestamp() * 1000)
+
+        entries = [
+            PartitionEntry(spec={"dt": "2024-05-01"}, last_file_creation_time=old_time_millis),
+            PartitionEntry(spec={"dt": "2024-04-01"}, last_file_creation_time=0),
+        ]
+
+        expiration_time = now - timedelta(days=7)
+        expired = strategy.select_expired_partitions(entries, expiration_time)
+        self.assertEqual(len(expired), 1)
+        self.assertEqual(expired[0].spec["dt"], "2024-05-01")
+
 
 class TestPartitionExpire(unittest.TestCase):
     """Tests for PartitionExpire orchestration class."""