[core] Add manifest sidecars for partition, row-id and bucket pruning

docs/docs/concepts/spec/manifest.md

@@ -90,12 +90,14 @@ 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.
+PyPaimon generates sidecars for newly written manifests and can prune manifest blocks using
+partition, row-ID and bucket filters. Its `manifest.sidecar.enabled` option controls reads and
+writes and inherits `manifest-sort.enabled` when unset. Ordinary and rolling writes publish
+`_EXTRA_FILES` references only after both the manifest and sidecar close successfully. Failed
+writes, merges and commit cleanup remove the sidecars with their owning new manifests.
 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.
+explain/statistics scans do not perform sidecar I/O.
 
 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

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java

@@ -96,7 +96,9 @@ private ManifestFileMeta fromDataRow(InternalRow row) {
                 row.isNullAt(9) ? null : row.getInt(9),
                 row.isNullAt(10) ? null : row.getLong(10),
                 row.isNullAt(11) ? null : row.getLong(11),
-                row.isNullAt(12) ? null : row.getInt(12),
-                row.isNullAt(13) ? null : fromStringArrayData(row.getArray(13)));
+                row.getFieldCount() <= 12 || row.isNullAt(12) ? null : row.getInt(12),
+                row.getFieldCount() <= 13 || row.isNullAt(13)
+                        ? null
+                        : fromStringArrayData(row.getArray(13)));
     }
 }

paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java

@@ -886,7 +886,8 @@ private List<ManifestEntry> readPlanningManifestEntries(
                 null,
                 Filter.alwaysTrue(),
                 entry -> partitionPredicate == null || partitionPredicate.test(entry.partition()),
-                ManifestEntry::copyWithoutStats);
+                ManifestEntry::copyWithoutStats,
+                null);
     }
 
     private Comparator<ManifestEntry> entryComparator() {

paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.manifest;
 
+import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.utils.ObjectSerializer;
 import org.apache.paimon.utils.ObjectSerializerTestBase;
 
@@ -86,6 +87,19 @@ void testExtraFiles() throws IOException {
         }
     }
 
+    @Test
+    void testOldRowWithoutExtraFilesField() {
+        ManifestFileMeta meta = object();
+        ManifestFileMetaSerializer serializer = new ManifestFileMetaSerializer();
+        GenericRow current = (GenericRow) serializer.toRow(meta);
+        GenericRow legacy = new GenericRow(current.getFieldCount() - 1);
+        for (int field = 0; field < legacy.getFieldCount(); field++) {
+            legacy.setField(field, current.getField(field));
+        }
+        assertThat(serializer.fromRow(legacy)).isEqualTo(meta);
+        assertThat(serializer.fromRow(legacy).extraFiles()).isNull();
+    }
+
     @Override
     protected ObjectSerializer<ManifestFileMeta> serializer() {
         return new ManifestFileMetaSerializer();

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

@@ -305,7 +305,7 @@ class CoreOptions:
         ConfigOptions.key("manifest.sidecar.enabled")
         .boolean_type()
         .no_default_value()
-        .with_description("Enable sidecar pruning on reads. Defaults to manifest-sort.enabled when unset.")
+        .with_description("Enable manifest sidecar reads and writes. Defaults to manifest-sort.enabled when unset.")
     )
 
     MANIFEST_SORT_ENABLED: ConfigOption[bool] = (

paimon-python/pypaimon/manifest/manifest_file_manager.py

@@ -32,7 +32,7 @@
 from datetime import datetime
 
 from pypaimon.manifest.manifest_sidecar import (
-    Query, read_sidecar, read_selected_bytes,
+    Settings, SUFFIX, Query, build_from_entries, read_sidecar, read_selected_bytes,
 )
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
 from pypaimon.manifest.schema.manifest_entry import (MANIFEST_ENTRY_SCHEMA,
@@ -342,12 +342,15 @@ def _get_value_stats_fields(self, file_dict: dict, file_schema) -> List:
             fields = [data_field_dict[col] for col in file_dict['_VALUE_STATS_COLS']]
         return fields
 
-    def write(self, file_name, entries: List[ManifestEntry]):
+    def _sidecar_settings(self):
+        return Settings.from_options(self.table.options)
+
+    def write(self, file_name, entries: List[ManifestEntry]) -> ManifestFileMeta:
         buf = BytesIO()
         fastavro.writer(
             buf, MANIFEST_ENTRY_SCHEMA, self._to_avro_records(entries),
             codec=self._codec)
-        self._flush(file_name, buf.getvalue())
+        return self._flush(file_name, buf.getvalue(), entries)
 
     def rolling_write(self, entries: List[ManifestEntry],
                       suggested_file_size: int,
@@ -371,10 +374,9 @@ def rolling_write(self, entries: List[ManifestEntry],
                     writer.flush()
                     avro_bytes = buf.getvalue()
                     file_name = f"{name_prefix}-{len(result)}"
-                    self._flush(file_name, avro_bytes)
-                    written_files.append(file_name)
-                    result.append(self._build_meta(
-                        file_name, entries[chunk_start:i + 1], len(avro_bytes)))
+                    meta = self._flush(file_name, avro_bytes, entries[chunk_start:i + 1])
+                    written_files.append(meta)
+                    result.append(meta)
                     chunk_start = i + 1
                     buf = BytesIO()
                     writer = Writer(
@@ -385,13 +387,12 @@ def rolling_write(self, entries: List[ManifestEntry],
                 writer.flush()
                 avro_bytes = buf.getvalue()
                 file_name = f"{name_prefix}-{len(result)}"
-                self._flush(file_name, avro_bytes)
-                written_files.append(file_name)
-                result.append(self._build_meta(
-                    file_name, entries[chunk_start:], len(avro_bytes)))
-        except Exception:
-            for fname in written_files:
-                self.file_io.delete_quietly(f"{self.manifest_path}/{fname}")
+                meta = self._flush(file_name, avro_bytes, entries[chunk_start:])
+                written_files.append(meta)
+                result.append(meta)
+        except BaseException:
+            for meta in written_files:
+                self.delete(meta)
             raise
         return result
 
@@ -438,17 +439,37 @@ def _to_avro_record(entry: ManifestEntry) -> dict:
     def _to_avro_records(self, entries: List[ManifestEntry]) -> List[dict]:
         return [self._to_avro_record(e) for e in entries]
 
-    def _flush(self, file_name: str, avro_bytes: bytes):
+    def delete(self, manifest: ManifestFileMeta):
+        self.file_io.delete_quietly(f"{self.manifest_path}/{manifest.file_name}")
+        for extra_file in manifest.extra_files or []:
+            self.file_io.delete_quietly(f"{self.manifest_path}/{extra_file}")
+
+    def _flush(self, file_name: str, avro_bytes: bytes, entries: List[ManifestEntry]) -> ManifestFileMeta:
         manifest_path = f"{self.manifest_path}/{file_name}"
+        sidecar_file_name = None
         try:
             with self.file_io.new_output_stream(manifest_path) as output_stream:
                 output_stream.write(avro_bytes)
-        except Exception as e:
+            settings = self._sidecar_settings()
+            if settings.enabled:
+                data = build_from_entries(avro_bytes, entries, settings)
+                if data is not None:
+                    sidecar_file_name = file_name + SUFFIX
+                    with self.file_io.new_output_stream(f"{self.manifest_path}/{sidecar_file_name}") as output_stream:
+                        output_stream.write(data)
+            # Publish the reference only after both objects close successfully.
+            return self._build_meta(file_name, entries, len(avro_bytes),
+                                    [sidecar_file_name] if sidecar_file_name is not None else None)
+        except BaseException as e:
             self.file_io.delete_quietly(manifest_path)
+            if sidecar_file_name is not None:
+                self.file_io.delete_quietly(f"{self.manifest_path}/{sidecar_file_name}")
+            if not isinstance(e, Exception) or isinstance(e, InterruptedError):
+                raise
             raise RuntimeError(f"Failed to write manifest file: {e}") from e
 
     def _build_meta(self, file_name: str, entries: List[ManifestEntry],
-                    file_size: int = None) -> ManifestFileMeta:
+                    file_size: int = None, extra_files: Optional[List[str]] = None) -> ManifestFileMeta:
         added_file_count = 0
         deleted_file_count = 0
         schema_id = None
@@ -480,7 +501,9 @@ def _build_meta(self, file_name: str, entries: List[ManifestEntry],
         min_row_id = None
         max_row_id = None
         for entry in entries:
-            if entry.file.first_row_id is None:
+            if (entry.file.first_row_id is None or entry.file.first_row_id < 0
+                    or entry.file.row_count <= 0
+                    or entry.file.row_count - 1 > (1 << 63) - 1 - entry.file.first_row_id):
                 min_row_id = None
                 max_row_id = None
                 break
@@ -516,5 +539,6 @@ def _build_meta(self, file_name: str, entries: List[ManifestEntry],
             max_level=max((e.file.level for e in entries), default=None),
             min_row_id=min_row_id,
             max_row_id=max_row_id,
+            extra_files=extra_files,
             total_buckets=total_buckets,
         )

paimon-python/pypaimon/manifest/manifest_file_merger.py

@@ -93,8 +93,4 @@ def _merge_candidates(self, candidates: List[ManifestFileMeta],
 
     def _delete_manifests(self, manifests: List[ManifestFileMeta]):
         for manifest in manifests:
-            manifest_path = "{}/{}".format(
-                self.manifest_file_manager.manifest_path,
-                manifest.file_name,
-            )
-            self.manifest_file_manager.file_io.delete_quietly(manifest_path)
+            self.manifest_file_manager.delete(manifest)