[python] Add manifest sidecar codec and block-pruned reads

docs/docs/concepts/spec/manifest.md

@@ -88,6 +88,13 @@ Selected block bytes still share the manifest content cache without populating t
 whole-manifest entry cache with partial results. The low-level `build` method returns
 sidecar bytes without writing or publishing another file.
 
+PyPaimon can read these sidecars and prune manifest blocks using partition, row-ID and bucket
+filters. Its `manifest.sidecar.enabled` option inherits `manifest-sort.enabled` when unset.
+Entry filters and ADD/DELETE reconciliation still apply after block selection. Missing or
+unusable sidecars fall back to full manifest reads; scans without pruning filters and
+explain/statistics scans do not perform sidecar I/O. The standalone codec can build sidecar
+bytes, but automatic Python writer publication and cleanup are not integrated yet.
+
 Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches.
 `build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent
 payload generation. Partition generation is always enabled,

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

@@ -301,6 +301,20 @@ class CoreOptions:
         .with_description("The parallelism for scanning manifest files.")
     )
 
+    MANIFEST_SIDECAR_ENABLED: ConfigOption[bool] = (
+        ConfigOptions.key("manifest.sidecar.enabled")
+        .boolean_type()
+        .no_default_value()
+        .with_description("Enable sidecar pruning on reads. Defaults to manifest-sort.enabled when unset.")
+    )
+
+    MANIFEST_SORT_ENABLED: ConfigOption[bool] = (
+        ConfigOptions.key("manifest-sort.enabled")
+        .boolean_type()
+        .default_value(False)
+        .with_description("Manifest sort setting. Also supplies the default for manifest sidecar reads.")
+    )
+
     MANIFEST_COMPRESSION: ConfigOption[str] = (
         ConfigOptions.key("manifest.compression")
         .string_type()
@@ -1266,6 +1280,13 @@ def manifest_target_size(self, default=None):
             default = MemorySize.of_bytes(default) if isinstance(default, int) else MemorySize.parse(default)
         return self.options.get(CoreOptions.MANIFEST_TARGET_FILE_SIZE, default).get_bytes()
 
+    def manifest_sidecar_enabled(self):
+        enabled = self.options.get(CoreOptions.MANIFEST_SIDECAR_ENABLED)
+        return self.manifest_sort_enabled() if enabled is None else enabled
+
+    def manifest_sort_enabled(self):
+        return self.options.get(CoreOptions.MANIFEST_SORT_ENABLED)
+
     def manifest_merge_skip_on_write_only(self, default=None):
         return self.options.get(CoreOptions.MANIFEST_MERGE_SKIP_ON_WRITE_ONLY, default)
 

paimon-python/pypaimon/manifest/manifest_file_manager.py

@@ -31,6 +31,9 @@
 
 from datetime import datetime
 
+from pypaimon.manifest.manifest_sidecar import (
+    Query, read_sidecar, read_selected_bytes,
+)
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
 from pypaimon.manifest.schema.manifest_entry import (MANIFEST_ENTRY_SCHEMA,
                                                      ManifestEntry)
@@ -128,13 +131,27 @@ def read_entries_parallel(self, manifest_files: List[ManifestFileMeta], manifest
                               early_entry_filter: Optional[Callable[[int, int], bool]] = None,
                               early_record_filter: Optional[Callable[[dict], bool]] = None,
                               partition_filter=None,
+                              row_ranges=None,
                               ) -> List[ManifestEntry]:
 
-        def _process_single_manifest(manifest_file: ManifestFileMeta) -> List[ManifestEntry]:
-            return self.read(manifest_file.file_name, manifest_entry_filter, drop_stats,
-                             early_entry_filter=early_entry_filter,
-                             early_record_filter=early_record_filter,
-                             partition_filter=partition_filter)
+        enabled = self.table.options.manifest_sidecar_enabled()
+        query = Query(row_ranges) if enabled and row_ranges is not None else None
+
+        def _process_single_manifest(manifest_file: ManifestFileMeta):
+            path = f"{self.manifest_path}/{manifest_file.file_name}"
+            selected = None
+            if enabled and (
+                    query is not None or partition_filter is not None or early_entry_filter is not None):
+                selected = read_sidecar(self.file_io, path, manifest_file, query,
+                                        partition_filter, self.partition_keys_fields, early_entry_filter)
+                if selected is not None and not selected.blocks:
+                    return []
+            return self.read(
+                manifest_file.file_name, manifest_entry_filter, drop_stats,
+                early_entry_filter=early_entry_filter,
+                early_record_filter=early_record_filter,
+                partition_filter=partition_filter,
+                selected_blocks=selected)
 
         def _entry_identifier(e: ManifestEntry) -> tuple:
             return (
@@ -177,6 +194,7 @@ def read(self, manifest_file_name: str, manifest_entry_filter=None, drop_stats=T
              early_entry_filter: Optional[Callable[[int, int], bool]] = None,
              early_record_filter: Optional[Callable[[dict], bool]] = None,
              partition_filter=None,
+             selected_blocks=None,
              ) -> List[ManifestEntry]:
         """
         early_entry_filter: ``(bucket, total_buckets) -> bool``, skip before deserializing _FILE.
@@ -190,8 +208,11 @@ def read(self, manifest_file_name: str, manifest_entry_filter=None, drop_stats=T
         manifest_file_path = f"{self.manifest_path}/{manifest_file_name}"
 
         entries = []
-        with self.file_io.new_input_stream(manifest_file_path) as input_stream:
-            avro_bytes = input_stream.read()
+        if selected_blocks is not None:
+            avro_bytes = read_selected_bytes(self.file_io, manifest_file_path, selected_blocks)
+        else:
+            with self.file_io.new_input_stream(manifest_file_path) as input_stream:
+                avro_bytes = input_stream.read()
         buffer = BytesIO(avro_bytes)
         records = _read_manifest_records(
             buffer, early_entry_filter, partition_filter,