[python] file_scanner should use with_shard instead of new interfaces

Author: JingsongLi

paimon-python/pypaimon/read/scanner/file_scanner.py

@@ -168,10 +168,7 @@ def __init__(
         manifest_scanner: Callable[[], List[ManifestFileMeta]],
         predicate: Optional[Predicate] = None,
         limit: Optional[int] = None,
-        vector_search: Optional['VectorSearch'] = None,
-        shard_index: Optional[int] = None,
-        shard_count: Optional[int] = None,
-        bucket_filter: Optional[Callable[[int], bool]] = None
+        vector_search: Optional['VectorSearch'] = None
     ):
         from pypaimon.table.file_store_table import FileStoreTable
 
@@ -182,11 +179,6 @@ def __init__(
         self.limit = limit
         self.vector_search = vector_search
 
-        # Bucket-level sharding for parallel consumption
-        self._shard_index = shard_index
-        self._shard_count = shard_count
-        self._bucket_filter = bucket_filter
-
         self.snapshot_manager = SnapshotManager(table)
         self.manifest_list_manager = ManifestListManager(table)
         self.manifest_file_manager = ManifestFileManager(table)
@@ -422,13 +414,6 @@ def _filter_manifest_entry(self, entry: ManifestEntry) -> bool:
             return False
         if self.partition_key_predicate and not self.partition_key_predicate.test(entry.partition):
             return False
-        # Apply bucket-level sharding for parallel consumption
-        if self._shard_index is not None and self._shard_count is not None:
-            if entry.bucket % self._shard_count != self._shard_index:
-                return False
-        elif self._bucket_filter is not None:
-            if not self._bucket_filter(entry.bucket):
-                return False
         # Get SimpleStatsEvolution for this schema
         evolution = self.simple_stats_evolutions.get_or_create(entry.file.schema_id)