[python] Refactor to File Scanner and add _filter_manifest_files_by_row_ranges

Author: JingsongLi

paimon-python/pypaimon/globalindex/__init__.py

@@ -20,8 +20,8 @@
 from pypaimon.globalindex.global_index_reader import GlobalIndexReader, FieldRef
 from pypaimon.globalindex.vector_search import VectorSearch
 from pypaimon.globalindex.vector_search_result import (
-    VectorSearchGlobalIndexResult,
-    DictBasedVectorSearchResult,
+    ScoredGlobalIndexResult,
+    DictBasedScoredIndexResult,
     ScoreGetter,
 )
 from pypaimon.globalindex.global_index_meta import GlobalIndexMeta, GlobalIndexIOMeta
@@ -38,8 +38,8 @@
     'GlobalIndexReader',
     'FieldRef',
     'VectorSearch',
-    'VectorSearchGlobalIndexResult',
-    'DictBasedVectorSearchResult',
+    'ScoredGlobalIndexResult',
+    'DictBasedScoredIndexResult',
     'ScoreGetter',
     'GlobalIndexMeta',
     'GlobalIndexIOMeta',

paimon-python/pypaimon/globalindex/faiss/faiss_vector_reader.py

@@ -30,7 +30,7 @@
 from pypaimon.globalindex.global_index_reader import GlobalIndexReader
 from pypaimon.globalindex.global_index_result import GlobalIndexResult
 from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
-from pypaimon.globalindex.vector_search_result import DictBasedVectorSearchResult
+from pypaimon.globalindex.vector_search_result import DictBasedScoredIndexResult
 from pypaimon.globalindex.roaring_bitmap import RoaringBitmap64
 from pypaimon.globalindex.faiss.faiss_options import (
     FaissVectorIndexOptions,
@@ -170,7 +170,7 @@ def _search(self, vector_search: 'VectorSearch') -> Optional[GlobalIndexResult]:
         if not results:
             return None
 
-        return DictBasedVectorSearchResult(results)
+        return DictBasedScoredIndexResult(results)
 
     def _configure_search_params(self, index: FaissIndex) -> None:
         """Configure search parameters based on index type."""

paimon-python/pypaimon/globalindex/vector_search_result.py

@@ -29,7 +29,7 @@
 ScoreGetter = Callable[[int], Optional[float]]
 
 
-class VectorSearchGlobalIndexResult(GlobalIndexResult):
+class ScoredGlobalIndexResult(GlobalIndexResult):
     """
     Vector search global index result for vector index.
     
@@ -41,7 +41,7 @@ def score_getter(self) -> ScoreGetter:
         """Returns a function to get the score for a given row ID."""
         pass
 
-    def offset(self, offset: int) -> 'VectorSearchGlobalIndexResult':
+    def offset(self, offset: int) -> 'ScoredGlobalIndexResult':
         """Returns a new result with row IDs offset by the given amount."""
         if offset == 0:
             return self
@@ -53,14 +53,14 @@ def offset(self, offset: int) -> 'VectorSearchGlobalIndexResult':
         for row_id in bitmap:
             offset_bitmap.add(row_id + offset)
 
-        return SimpleVectorSearchGlobalIndexResult(
+        return SimpleScoredGlobalIndexResult(
             offset_bitmap,
             lambda row_id: this_score_getter(row_id - offset)
         )
 
     def or_(self, other: GlobalIndexResult) -> GlobalIndexResult:
         """Returns the union of this result and the other result."""
-        if not isinstance(other, VectorSearchGlobalIndexResult):
+        if not isinstance(other, ScoredGlobalIndexResult):
             return super().or_(other)
 
         this_row_ids = self.results()
@@ -76,18 +76,18 @@ def combined_score_getter(row_id: int) -> Optional[float]:
                 return this_score_getter(row_id)
             return other_score_getter(row_id)
 
-        return SimpleVectorSearchGlobalIndexResult(result_or, combined_score_getter)
+        return SimpleScoredGlobalIndexResult(result_or, combined_score_getter)
 
     @staticmethod
     def create(
         supplier: Callable[[], RoaringBitmap64],
         score_getter: ScoreGetter
-    ) -> 'VectorSearchGlobalIndexResult':
+    ) -> 'ScoredGlobalIndexResult':
         """Creates a new VectorSearchGlobalIndexResult from supplier."""
-        return LazyVectorSearchGlobalIndexResult(supplier, score_getter)
+        return LazyScoredGlobalIndexResult(supplier, score_getter)
 
 
-class SimpleVectorSearchGlobalIndexResult(VectorSearchGlobalIndexResult):
+class SimpleScoredGlobalIndexResult(ScoredGlobalIndexResult):
     """Simple implementation of VectorSearchGlobalIndexResult."""
 
     def __init__(self, bitmap: RoaringBitmap64, score_getter_fn: ScoreGetter):
@@ -101,7 +101,7 @@ def score_getter(self) -> ScoreGetter:
         return self._score_getter_fn
 
 
-class LazyVectorSearchGlobalIndexResult(VectorSearchGlobalIndexResult):
+class LazyScoredGlobalIndexResult(ScoredGlobalIndexResult):
     """Lazy implementation of VectorSearchGlobalIndexResult."""
 
     def __init__(self, supplier: Callable[[], RoaringBitmap64], score_getter_fn: ScoreGetter):
@@ -118,7 +118,7 @@ def score_getter(self) -> ScoreGetter:
         return self._score_getter_fn
 
 
-class DictBasedVectorSearchResult(VectorSearchGlobalIndexResult):
+class DictBasedScoredIndexResult(ScoredGlobalIndexResult):
     """Vector search result backed by a dictionary of row_id -> score."""
 
     def __init__(self, id_to_scores: Dict[int, float]):

paimon-python/pypaimon/manifest/manifest_list_manager.py

@@ -17,7 +17,7 @@
 ################################################################################
 
 from io import BytesIO
-from typing import List
+from typing import List, Optional
 
 import fastavro
 
@@ -40,7 +40,9 @@ def __init__(self, table):
         self.manifest_path = f"{manifest_path}/manifest"
         self.file_io = self.table.file_io
 
-    def read_all(self, snapshot: Snapshot) -> List[ManifestFileMeta]:
+    def read_all(self, snapshot: Optional[Snapshot]) -> List[ManifestFileMeta]:
+        if snapshot is None:
+            return []
         manifest_files = []
         base_manifests = self.read(snapshot.base_manifest_list)
         manifest_files.extend(base_manifests)
@@ -79,6 +81,8 @@ def read(self, manifest_list_name: str) -> List[ManifestFileMeta]:
                 num_deleted_files=record['_NUM_DELETED_FILES'],
                 partition_stats=partition_stats,
                 schema_id=record['_SCHEMA_ID'],
+                min_row_id=record['_MIN_ROW_ID'],
+                max_row_id=record['_MAX_ROW_ID'],
             )
             manifest_files.append(manifest_file_meta)
 

paimon-python/pypaimon/manifest/schema/manifest_file_meta.py

@@ -18,6 +18,7 @@
 
 from dataclasses import dataclass
 
+from typing import Optional
 from pypaimon.manifest.schema.simple_stats import (PARTITION_STATS_SCHEMA,
                                                    SimpleStats)
 
@@ -31,6 +32,8 @@ class ManifestFileMeta:
     partition_stats: SimpleStats
     schema_id: int
 
+    min_row_id: Optional[int] = None
+    max_row_id: Optional[int] = None
 
 MANIFEST_FILE_META_SCHEMA = {
     "type": "record",
@@ -43,5 +46,7 @@ class ManifestFileMeta:
         {"name": "_NUM_DELETED_FILES", "type": "long"},
         {"name": "_PARTITION_STATS", "type": PARTITION_STATS_SCHEMA},
         {"name": "_SCHEMA_ID", "type": "long"},
+        {"name": "_MIN_ROW_ID", "type": ["null", "long"], "default": None},
+        {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": None},
     ]
 }