From c2e9e17c7ce4433d152427b9072b3da630a4fdef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Tue, 15 Sep 2026 16:47:17 +0800 Subject: [PATCH 1/4] [core] Add manifest sidecars for partition, row-id and bucket pruning --- .../manifest/ManifestFileMetaSerializer.java | 6 +- .../manifest/ProjectedManifestEntry.java | 9 ++ .../apache/paimon/utils/SegmentsCache.java | 43 ++++++- .../LegacyDataEvolutionRowIdReassigner.java | 3 +- .../ManifestFileMetaSerializerTest.java | 14 +++ .../paimon/utils/SegmentsCacheTest.java | 15 +++ .../src/test/resources/manifest-sidecar.txt | 23 ++++ .../manifest/manifest_file_manager.py | 58 ++++++--- .../pypaimon/manifest/manifest_file_merger.py | 6 +- .../tests/manifest/manifest_sidecar_test.py | 116 ++++++++++++++---- .../pypaimon/write/file_store_commit.py | 6 +- 11 files changed, 244 insertions(+), 55 deletions(-) create mode 100644 paimon-core/src/test/resources/manifest-sidecar.txt diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java index 30f74a305ade..b00135c8edfa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java +++ b/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))); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java index 77486d4cdd53..defd81bcd3e2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java @@ -27,6 +27,7 @@ import javax.annotation.Nullable; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -49,6 +50,7 @@ public final class ProjectedManifestEntry implements ManifestEntry { private static final Projection FULL_PROJECTION = Projection.create(MANIFEST_ROW_TYPE); public static final Projection DELETE_ENTRY_PROJECTION = createDeleteEntryProjection(); public static final Projection ROW_RANGE_PROJECTION = createRowRangeProjection(); + public static final Projection BLOCK_INDEX_PROJECTION = createBlockIndexProjection(); public static final Projection ENTRY_LAYOUT_PROJECTION = createEntryLayoutProjection(); private final Projection projection; @@ -133,6 +135,13 @@ private static Projection createRowRangeProjection() { DataFileMeta.FIRST_ROW_ID))))); } + private static Projection createBlockIndexProjection() { + List fields = new ArrayList<>(ROW_RANGE_PROJECTION.projectedType().getFields()); + fields.add(MANIFEST_ROW_TYPE.getField(ManifestEntry.BUCKET)); + fields.add(MANIFEST_ROW_TYPE.getField(ManifestEntry.TOTAL_BUCKETS)); + return Projection.create(new RowType(false, fields)); + } + private static Projection createEntryLayoutProjection() { RowType manifestType = MANIFEST_ROW_TYPE; return Projection.create( diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java b/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java index cfea1ea219bc..4c0a8cc3b63e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java @@ -27,6 +27,7 @@ import javax.annotation.Nullable; import java.time.Duration; +import java.util.Objects; import static org.apache.paimon.CoreOptions.PAGE_SIZE; @@ -36,7 +37,7 @@ public class SegmentsCache { private static final int OBJECT_MEMORY_SIZE = 1000; private final int pageSize; - private final Cache cache; + private final Cache cache; private final MemorySize maxMemorySize; private final long maxElementSize; @Nullable private final Duration expireAfterAccess; @@ -53,7 +54,7 @@ public SegmentsCache( @Nullable Duration expireAfterAccess, boolean softValues) { this.pageSize = pageSize; - Caffeine builder = + Caffeine builder = Caffeine.newBuilder() .weigher(this::weigh) .maximumWeight(maxMemorySize.getBytes()) @@ -106,10 +107,46 @@ public void put(T key, Segments segments) { cache.put(key, segments); } - private int weigh(T cacheKey, Segments segments) { + @Nullable + public Segments getIfPresents(T key, long offset, long length) { + return cache.getIfPresent(new RangeKey<>(key, offset, length)); + } + + public void put(T key, long offset, long length, Segments segments) { + cache.put(new RangeKey<>(key, offset, length), segments); + } + + private int weigh(Object cacheKey, Segments segments) { return (int) (OBJECT_MEMORY_SIZE + segments.totalMemorySize()); } + /** Separates byte ranges from whole-object keys in the same cache. */ + private static final class RangeKey { + private final T key; + private final long offset; + private final long length; + + private RangeKey(T key, long offset, long length) { + this.key = Objects.requireNonNull(key); + this.offset = offset; + this.length = length; + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof RangeKey)) { + return false; + } + RangeKey that = (RangeKey) other; + return key.equals(that.key) && offset == that.offset && length == that.length; + } + + @Override + public int hashCode() { + return Objects.hash(key, offset, length); + } + } + @Nullable public static SegmentsCache create(MemorySize maxMemorySize, long maxElementSize) { return create((int) PAGE_SIZE.defaultValue().getBytes(), maxMemorySize, maxElementSize); diff --git a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java index 641541499028..e113923f31cd 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/LegacyDataEvolutionRowIdReassigner.java @@ -886,7 +886,8 @@ private List readPlanningManifestEntries( null, Filter.alwaysTrue(), entry -> partitionPredicate == null || partitionPredicate.test(entry.partition()), - ManifestEntry::copyWithoutStats); + ManifestEntry::copyWithoutStats, + null); } private Comparator entryComparator() { diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java index 04f41aa8204f..0b2dc24baf09 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java +++ b/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 serializer() { return new ManifestFileMetaSerializer(); diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java index 5b2e704548f8..73fc8c338370 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java @@ -91,4 +91,19 @@ public void testStrongRefsAreBoundedByWeight() { assertThat(cache.totalCacheBytes()).isLessThanOrEqualTo(budget.getBytes()); assertThat(cache.estimatedSize()).isLessThan(100); } + + @Test + public void testRangeKeysAreDistinctFromWholeObjectsAndOtherRanges() { + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofKibiBytes(64), 100L, null, false); + SingleSegments whole = new SingleSegments(MemorySegment.wrap(new byte[] {1}), 1); + SingleSegments block = new SingleSegments(MemorySegment.wrap(new byte[] {2, 3}), 2); + cache.put("manifest", whole); + cache.put("manifest", 10, 2, block); + assertThat(cache.getIfPresents("manifest")).isSameAs(whole); + assertThat(cache.getIfPresents("manifest", 10, 2)).isSameAs(block); + assertThat(cache.getIfPresents("other", 10, 2)).isNull(); + assertThat(cache.getIfPresents("manifest", 11, 2)).isNull(); + assertThat(cache.getIfPresents("manifest", 10, 3)).isNull(); + } } diff --git a/paimon-core/src/test/resources/manifest-sidecar.txt b/paimon-core/src/test/resources/manifest-sidecar.txt new file mode 100644 index 000000000000..f2199ce2e4f0 --- /dev/null +++ b/paimon-core/src/test/resources/manifest-sidecar.txt @@ -0,0 +1,23 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +avroHeader=T2JqAQQUYXZyby5jb2RlYwhudWxsFmF2cm8uc2NoZW1hDCJsb25nIgAAAAAAAAAAAAAAAAAAAAAA +partitionA=AAAAAgAAAAAAAAAABwAAAAAAAABsZWZ0AAAAhA== +partitionB=AAAAAgACAAAAAAAACQAAAAAAAAAAAAAAAAAAAA== +index=UEFJTVNDQVIAAAABL0eWubkHdL6ioNdIPj3jGtp9TAk83wbtAsmc+xJVwM8AAAAAAAAByQAAAAAAAAAHAAAAOU9iagEEFGF2cm8uY29kZWMIbnVsbBZhdnJvLnNjaGVtYQwibG9uZyIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADAAAAAAAAADkAAAAAAAAAZAAAAAAAAAADAAEAAAAkAAAAAgAAAAAAAAAAAAAAAAAAAAkAAAAAAAAAFAAAAAAAAAAYAAAAAAAAAACdAAAAAAAAAMgAAAAAAAAAAgABAAAAJAAAAAIAAAAA/////gAAAAEAAAACAAAHgcw4bGUAAAeBzDhsZQAAAAAAAAABZQAAAAAAAABkAAAAAAAAAAIAAQAAACQAAAACAAAAAAAAABQAAAAAAAAAGH//////////f/////////8AlOHbHSWYVK7dgM+l6yo0+LjOLGYNf4lnkt2OtValVYA= +indexWithPartitions=UEFJTVNDQVIAAAABL0eWubkHdL6ioNdIPj3jGtp9TAk83wbtAsmc+xJVwM8AAAAAAAAByQAAAAAAAAAHAAAAOU9iagEEFGF2cm8uY29kZWMIbnVsbBZhdnJvLnNjaGVtYQwibG9uZyIAAAAAAAAAAAAAAAAAAAAAAAAAAAIAAAAcAAAAAgAAAAAAAAAABwAAAAAAAABsZWZ0AAAAhAAAABwAAAACAAIAAAAAAAAJAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAAAAAA5AAAAAAAAAGQAAAAAAAAAAwEAAAAMAAAAAgAAAAAAAAABAQAAACQAAAACAAAAAAAAAAAAAAAAAAAACQAAAAAAAAAUAAAAAAAAABgAAAAAAAAAAJ0AAAAAAAAAyAAAAAAAAAACAQAAAAwAAAACAAAAAAAAAAEBAAAAJAAAAAIAAAAA/////gAAAAEAAAACAAAHgcw4bGUAAAeBzDhsZQAAAAAAAAABZQAAAAAAAABkAAAAAAAAAAIBAAAADAAAAAIAAAAAAAAAAQEAAAAkAAAAAgAAAAAAAAAUAAAAAAAAABh//////////3//////////AInpuqbYtRIEgHOW0YNU5V+bvWbXg6ARpSP/hXdwewB8 +indexWithBuckets=UEFJTVNDQVIAAAABL0eWubkHdL6ioNdIPj3jGtp9TAk83wbtAsmc+xJVwM8AAAAAAAAByQAAAAAAAAAHAAAAOU9iagEEFGF2cm8uY29kZWMIbnVsbBZhdnJvLnNjaGVtYQwibG9uZyIAAAAAAAAAAAAAAAAAAAAAAAAAAAIAAAAcAAAAAgAAAAAAAAAABwAAAAAAAABsZWZ0AAAAhAAAABwAAAACAAIAAAAAAAAJAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAAAAAA5AAAAAAAAAGQAAAAAAAAAAwEAAAAMAAAAAgAAAAAAAAABAQAAACQAAAACAAAAAAAAAAAAAAAAAAAACQAAAAAAAAAUAAAAAAAAABgBAAAAFAAAAAIAAAABAAAABAAAAAEAAAAIAAAAAAAAAJ0AAAAAAAAAyAAAAAAAAAACAQAAAAwAAAACAAAAAAAAAAEBAAAAJAAAAAIAAAAA/////gAAAAEAAAACAAAHgcw4bGUAAAeBzDhsZQEAAAAUAAAAAgAAAAIAAAAEAAAAAgAAAAgAAAAAAAABZQAAAAAAAABkAAAAAAAAAAIBAAAADAAAAAIAAAAAAAAAAQEAAAAkAAAAAgAAAAAAAAAUAAAAAAAAABh//////////3//////////AQAAABQAAAACAAAAAAAAAAEAAAADAAAABGCAy0ixcqKywlCau1+E/XLEorySmCmL1zl5q9/dwdBu diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py index 1d833f3d3516..f6d7859615a6 100644 --- a/paimon-python/pypaimon/manifest/manifest_file_manager.py +++ b/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 _sidecar_settings(self): + return Settings.from_options(self.table.options) + def write(self, file_name, entries: List[ManifestEntry]): 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) -> 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, ) diff --git a/paimon-python/pypaimon/manifest/manifest_file_merger.py b/paimon-python/pypaimon/manifest/manifest_file_merger.py index 821b12aef5e9..2f14f44e7736 100644 --- a/paimon-python/pypaimon/manifest/manifest_file_merger.py +++ b/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) diff --git a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py index 405c4c92b14a..7986d73c0124 100644 --- a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py +++ b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py @@ -284,7 +284,7 @@ def test_large_avro_headers(self): self.assertEqual(restored, avro_bytes) self.assertEqual(list(fastavro.reader(BytesIO(restored))), [42]) - def test_disabled_option_is_independent_of_manifest_target_size(self): + def test_disabled_sidecars_do_not_constrain_manifest_target_size(self): for target in ('1 bytes', '1 gb'): with self.subTest(target=target): settings = Settings.from_options(CoreOptions(Options({ @@ -442,30 +442,52 @@ def entry(self, name, first, count=10, kind=0): return ManifestEntry(kind, self._create_file_meta('unused').min_key, 0, 1, replace(self._create_file_meta(name), first_row_id=first, row_count=count)) - def test_enabling_sidecar_reads_does_not_change_python_writes(self): - manager = self.manifest_file_manager - entries = [self.entry('data.parquet', 100)] - self.assertTrue(self.table.options.manifest_sidecar_enabled()) - self.assertIsNone(manager.write('plain-writer', entries)) - self.assertFalse(Path(manager.manifest_path, 'plain-writer' + SUFFIX).exists()) - for meta in manager.rolling_write(entries, 1, 'rolling-writer'): - self.assertIsNone(meta.extra_files) - self.assertFalse(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) - def write_meta(self, name, entries): - # Emulate an externally published sidecar; production Python writes are unchanged. manager = self.manifest_file_manager - manager.write(name, entries) - path = Path(manager.manifest_path, name) - avro_bytes = path.read_bytes() - meta = manager._build_meta(name, entries, len(avro_bytes)) - settings = Settings.from_options(self.table.options) - if settings.enabled: - data = manifest_sidecar.build_from_entries(avro_bytes, entries, settings) - sidecar_path = path.with_name(name + SUFFIX) - sidecar_path.write_bytes(data) - meta = replace(meta, extra_files=[sidecar_path.name]) - return meta + return manager.write(name, entries) + + def test_sidecars_do_not_depend_on_manifest_target_sizes(self): + for i, target in enumerate(('1 bytes', '1 gb', '9223372036854775807 bytes')): + with self.subTest(target=target): + self.table.options.options.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE, target) + entry = self.entry('file.parquet', 100) + meta = self.write_meta(f'budget-{i}', [entry]) + self.assertEqual( + [e.file.file_name for e in self.manifest_file_manager.read(meta.file_name)], + [entry.file.file_name]) + self.assertIsNotNone(sidecar_file_name(meta)) + + def test_payloads_follow_table_metadata(self): + import pyarrow as pa + + for partitioned, data_evolution, bucket in product((False, True), (False, True), (-1, 4)): + with self.subTest(partitioned=partitioned, data_evolution=data_evolution, bucket=bucket): + name = f'default.sidecar_settings_{partitioned}_{data_evolution}_{bucket + 1}' + schema = Schema.from_pyarrow_schema( + pa.schema([('id', pa.int32()), ('value', pa.string())]), + partition_keys=['id'] if partitioned else [], + options={'manifest.sidecar.enabled': 'true', + 'data-evolution.enabled': str(data_evolution).lower(), 'bucket': str(bucket)}) + self.catalog.create_table(name, schema, False) + self.table = self.catalog.get_table(name) + manager = ManifestFileManager(self.table) + entry = replace(self.entry('file.parquet', 100), bucket=1, total_buckets=4, + partition=GenericRow([7] if partitioned else [], self.table.partition_keys_fields)) + metadata = manager.write('settings', [entry]) + with manager.file_io.new_input_stream( + manager.manifest_path + '/' + sidecar_file_name(metadata)) as stream: + data = stream.read() + encoded = manifest_sidecar._Buffer(data) + encoded.take(4) + encoded.uint() + encoded.take(encoded.uint()) + self.assertEqual(encoded.uint(), 1) + selected = select(data, metadata, None, bucket_filter=lambda bucket, total: bucket == 99) + self.assertEqual(len(selected.blocks), 1 if bucket == -1 else 0) + self.assertEqual( + len(select(data, metadata, [Range(99, 99)]).blocks), + 0 if data_evolution else 1) + self.assertEqual([e.file.file_name for e in manager.read(metadata.file_name)], ['file.parquet']) def test_bucket_point_lookup_with_rescale_and_delete_entries(self): self.table.options.options.set(CoreOptions.DATA_EVOLUTION_ENABLED, False) @@ -609,6 +631,10 @@ def test_explicit_reference_and_null_does_not_probe(self): self.assertEqual(len(actual), 1) self.assertEqual([call[0][0] for call in opened.call_args_list], [str(Path(manager.manifest_path, written.file_name))]) + manager.delete(indexed) + self.assertFalse(other_path.exists()) + self.assertFalse(explicit_path.exists()) + self.assertFalse(Path(manager.manifest_path, written.file_name).exists()) def test_manifest_list_index_reference_compatibility(self): indexed = self.write_meta('indexed', [self.entry('data.parquet', 100)]) @@ -799,3 +825,47 @@ def open_stream(path): self.assertEqual([call[0][0] for call in opened.call_args_list], [index_path, body_path]) if stream is not None: self.assertTrue(stream.closed) + + def test_rolling_merge_limits_and_abort_cleanup(self): + entries = [self.entry('file-%d' % i, i * 1000) for i in range(300)] + manager = self.manifest_file_manager + metas = manager.rolling_write(entries, 300, 'rolling') + self.assertGreater(len(metas), 1) + for meta in metas: + actual = manager.read(meta.file_name) + data = Path(manager.manifest_path, meta.file_name + SUFFIX).read_bytes() + for e in actual: + self.assertTrue(intersects(data, meta, [Range(e.file.first_row_id, e.file.first_row_id)], Settings())) + gap = actual[0].file.first_row_id + 10 + self.assertFalse(intersects(data, meta, [Range(gap, gap)], Settings())) + from pypaimon.manifest.manifest_file_merger import ManifestFileMerger + merger = ManifestFileMerger(manager, 1000000, 2) + merged = merger.merge(metas) + # Merger returns both the final manifest list and newly written outputs. + outputs = merged[0] if isinstance(merged, tuple) else merged + for meta in outputs: + self.assertTrue(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) + for meta in metas: + self.assertIsNotNone(sidecar_file_name(meta)) + manager.delete(meta) + self.assertFalse(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) + original = self.table.file_io.new_output_stream + + def fail(path): + if path.endswith(SUFFIX): + raise OSError('sidecar write failed') + return original(path) + with patch.object(self.table.file_io, 'new_output_stream', side_effect=fail): + with self.assertRaises(RuntimeError): + manager.write('failed', entries[:1]) + self.assertFalse(Path(manager.manifest_path, 'failed').exists()) + self.assertFalse(Path(manager.manifest_path, 'failed' + SUFFIX).exists()) + unknown = manager.write('unknown', [self.entry('legacy', None)]) + self.assertIsNotNone(sidecar_file_name(unknown)) + data = Path(manager.manifest_path, sidecar_file_name(unknown)).read_bytes() + self.assertEqual(len(select(data, unknown, [Range(100, 100)]).blocks), 1) + huge = manager.write('huge', [self.entry('one', 0, MAX_ROW_ID), + self.entry('two', MAX_ROW_ID, 1)]) + self.assertIsNotNone(sidecar_file_name(huge)) + data = Path(manager.manifest_path, sidecar_file_name(huge)).read_bytes() + self.assertEqual(len(select(data, huge, [Range(50, 50)]).blocks), 1) diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 27907f8741d1..97e2ac28717a 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -1154,8 +1154,7 @@ def _clean_up_reuse_tmp_manifests( if ml_name: try: for meta in self.manifest_list_manager.read(ml_name): - self.table.file_io.delete_quietly( - f"{self.manifest_file_manager.manifest_path}/{meta.file_name}") + self.manifest_file_manager.delete(meta) except Exception: pass self.table.file_io.delete_quietly(f"{manifest_path}/{ml_name}") @@ -1174,8 +1173,7 @@ def _clean_up_no_reuse_tmp_manifests( if base_manifest_list: self.table.file_io.delete_quietly(f"{manifest_path}/{base_manifest_list}") for meta in merge_new_files: - self.table.file_io.delete_quietly( - f"{self.manifest_file_manager.manifest_path}/{meta.file_name}") + self.manifest_file_manager.delete(meta) def abort(self, commit_messages: List[CommitMessage]): """Abort commit and delete files. Uses external_path if available to ensure proper scheme handling.""" From 3a99b55574863630e834aa73bc76f98f62ee9a92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 16 Sep 2026 17:08:12 +0800 Subject: [PATCH 2/4] [core] Align manifest sidecar integration with merged format --- docs/docs/concepts/spec/manifest.md | 10 ++--- .../manifest/ProjectedManifestEntry.java | 9 ---- .../apache/paimon/utils/SegmentsCache.java | 43 ++----------------- .../paimon/utils/SegmentsCacheTest.java | 15 ------- .../src/test/resources/manifest-sidecar.txt | 23 ---------- 5 files changed, 7 insertions(+), 93 deletions(-) delete mode 100644 paimon-core/src/test/resources/manifest-sidecar.txt diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 6f72db906743..a6ea8a4f2a57 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -90,12 +90,10 @@ 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. +PyPaimon also generates sidecars for newly written manifests and uses their block coverage +during scans. Its `manifest.sidecar.enabled` table option inherits `manifest-sort.enabled` +when unset. Sidecar references are published only after writing succeeds, and cleanup follows +the owning manifest. 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 diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java index defd81bcd3e2..77486d4cdd53 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java @@ -27,7 +27,6 @@ import javax.annotation.Nullable; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -50,7 +49,6 @@ public final class ProjectedManifestEntry implements ManifestEntry { private static final Projection FULL_PROJECTION = Projection.create(MANIFEST_ROW_TYPE); public static final Projection DELETE_ENTRY_PROJECTION = createDeleteEntryProjection(); public static final Projection ROW_RANGE_PROJECTION = createRowRangeProjection(); - public static final Projection BLOCK_INDEX_PROJECTION = createBlockIndexProjection(); public static final Projection ENTRY_LAYOUT_PROJECTION = createEntryLayoutProjection(); private final Projection projection; @@ -135,13 +133,6 @@ private static Projection createRowRangeProjection() { DataFileMeta.FIRST_ROW_ID))))); } - private static Projection createBlockIndexProjection() { - List fields = new ArrayList<>(ROW_RANGE_PROJECTION.projectedType().getFields()); - fields.add(MANIFEST_ROW_TYPE.getField(ManifestEntry.BUCKET)); - fields.add(MANIFEST_ROW_TYPE.getField(ManifestEntry.TOTAL_BUCKETS)); - return Projection.create(new RowType(false, fields)); - } - private static Projection createEntryLayoutProjection() { RowType manifestType = MANIFEST_ROW_TYPE; return Projection.create( diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java b/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java index 4c0a8cc3b63e..cfea1ea219bc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java @@ -27,7 +27,6 @@ import javax.annotation.Nullable; import java.time.Duration; -import java.util.Objects; import static org.apache.paimon.CoreOptions.PAGE_SIZE; @@ -37,7 +36,7 @@ public class SegmentsCache { private static final int OBJECT_MEMORY_SIZE = 1000; private final int pageSize; - private final Cache cache; + private final Cache cache; private final MemorySize maxMemorySize; private final long maxElementSize; @Nullable private final Duration expireAfterAccess; @@ -54,7 +53,7 @@ public SegmentsCache( @Nullable Duration expireAfterAccess, boolean softValues) { this.pageSize = pageSize; - Caffeine builder = + Caffeine builder = Caffeine.newBuilder() .weigher(this::weigh) .maximumWeight(maxMemorySize.getBytes()) @@ -107,46 +106,10 @@ public void put(T key, Segments segments) { cache.put(key, segments); } - @Nullable - public Segments getIfPresents(T key, long offset, long length) { - return cache.getIfPresent(new RangeKey<>(key, offset, length)); - } - - public void put(T key, long offset, long length, Segments segments) { - cache.put(new RangeKey<>(key, offset, length), segments); - } - - private int weigh(Object cacheKey, Segments segments) { + private int weigh(T cacheKey, Segments segments) { return (int) (OBJECT_MEMORY_SIZE + segments.totalMemorySize()); } - /** Separates byte ranges from whole-object keys in the same cache. */ - private static final class RangeKey { - private final T key; - private final long offset; - private final long length; - - private RangeKey(T key, long offset, long length) { - this.key = Objects.requireNonNull(key); - this.offset = offset; - this.length = length; - } - - @Override - public boolean equals(Object other) { - if (!(other instanceof RangeKey)) { - return false; - } - RangeKey that = (RangeKey) other; - return key.equals(that.key) && offset == that.offset && length == that.length; - } - - @Override - public int hashCode() { - return Objects.hash(key, offset, length); - } - } - @Nullable public static SegmentsCache create(MemorySize maxMemorySize, long maxElementSize) { return create((int) PAGE_SIZE.defaultValue().getBytes(), maxMemorySize, maxElementSize); diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java index 73fc8c338370..5b2e704548f8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/SegmentsCacheTest.java @@ -91,19 +91,4 @@ public void testStrongRefsAreBoundedByWeight() { assertThat(cache.totalCacheBytes()).isLessThanOrEqualTo(budget.getBytes()); assertThat(cache.estimatedSize()).isLessThan(100); } - - @Test - public void testRangeKeysAreDistinctFromWholeObjectsAndOtherRanges() { - SegmentsCache cache = - new SegmentsCache<>(1024, MemorySize.ofKibiBytes(64), 100L, null, false); - SingleSegments whole = new SingleSegments(MemorySegment.wrap(new byte[] {1}), 1); - SingleSegments block = new SingleSegments(MemorySegment.wrap(new byte[] {2, 3}), 2); - cache.put("manifest", whole); - cache.put("manifest", 10, 2, block); - assertThat(cache.getIfPresents("manifest")).isSameAs(whole); - assertThat(cache.getIfPresents("manifest", 10, 2)).isSameAs(block); - assertThat(cache.getIfPresents("other", 10, 2)).isNull(); - assertThat(cache.getIfPresents("manifest", 11, 2)).isNull(); - assertThat(cache.getIfPresents("manifest", 10, 3)).isNull(); - } } diff --git a/paimon-core/src/test/resources/manifest-sidecar.txt b/paimon-core/src/test/resources/manifest-sidecar.txt deleted file mode 100644 index f2199ce2e4f0..000000000000 --- a/paimon-core/src/test/resources/manifest-sidecar.txt +++ /dev/null @@ -1,23 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -avroHeader=T2JqAQQUYXZyby5jb2RlYwhudWxsFmF2cm8uc2NoZW1hDCJsb25nIgAAAAAAAAAAAAAAAAAAAAAA -partitionA=AAAAAgAAAAAAAAAABwAAAAAAAABsZWZ0AAAAhA== -partitionB=AAAAAgACAAAAAAAACQAAAAAAAAAAAAAAAAAAAA== -index=UEFJTVNDQVIAAAABL0eWubkHdL6ioNdIPj3jGtp9TAk83wbtAsmc+xJVwM8AAAAAAAAByQAAAAAAAAAHAAAAOU9iagEEFGF2cm8uY29kZWMIbnVsbBZhdnJvLnNjaGVtYQwibG9uZyIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADAAAAAAAAADkAAAAAAAAAZAAAAAAAAAADAAEAAAAkAAAAAgAAAAAAAAAAAAAAAAAAAAkAAAAAAAAAFAAAAAAAAAAYAAAAAAAAAACdAAAAAAAAAMgAAAAAAAAAAgABAAAAJAAAAAIAAAAA/////gAAAAEAAAACAAAHgcw4bGUAAAeBzDhsZQAAAAAAAAABZQAAAAAAAABkAAAAAAAAAAIAAQAAACQAAAACAAAAAAAAABQAAAAAAAAAGH//////////f/////////8AlOHbHSWYVK7dgM+l6yo0+LjOLGYNf4lnkt2OtValVYA= -indexWithPartitions=UEFJTVNDQVIAAAABL0eWubkHdL6ioNdIPj3jGtp9TAk83wbtAsmc+xJVwM8AAAAAAAAByQAAAAAAAAAHAAAAOU9iagEEFGF2cm8uY29kZWMIbnVsbBZhdnJvLnNjaGVtYQwibG9uZyIAAAAAAAAAAAAAAAAAAAAAAAAAAAIAAAAcAAAAAgAAAAAAAAAABwAAAAAAAABsZWZ0AAAAhAAAABwAAAACAAIAAAAAAAAJAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAAAAAA5AAAAAAAAAGQAAAAAAAAAAwEAAAAMAAAAAgAAAAAAAAABAQAAACQAAAACAAAAAAAAAAAAAAAAAAAACQAAAAAAAAAUAAAAAAAAABgAAAAAAAAAAJ0AAAAAAAAAyAAAAAAAAAACAQAAAAwAAAACAAAAAAAAAAEBAAAAJAAAAAIAAAAA/////gAAAAEAAAACAAAHgcw4bGUAAAeBzDhsZQAAAAAAAAABZQAAAAAAAABkAAAAAAAAAAIBAAAADAAAAAIAAAAAAAAAAQEAAAAkAAAAAgAAAAAAAAAUAAAAAAAAABh//////////3//////////AInpuqbYtRIEgHOW0YNU5V+bvWbXg6ARpSP/hXdwewB8 -indexWithBuckets=UEFJTVNDQVIAAAABL0eWubkHdL6ioNdIPj3jGtp9TAk83wbtAsmc+xJVwM8AAAAAAAAByQAAAAAAAAAHAAAAOU9iagEEFGF2cm8uY29kZWMIbnVsbBZhdnJvLnNjaGVtYQwibG9uZyIAAAAAAAAAAAAAAAAAAAAAAAAAAAIAAAAcAAAAAgAAAAAAAAAABwAAAAAAAABsZWZ0AAAAhAAAABwAAAACAAIAAAAAAAAJAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAAAAAA5AAAAAAAAAGQAAAAAAAAAAwEAAAAMAAAAAgAAAAAAAAABAQAAACQAAAACAAAAAAAAAAAAAAAAAAAACQAAAAAAAAAUAAAAAAAAABgBAAAAFAAAAAIAAAABAAAABAAAAAEAAAAIAAAAAAAAAJ0AAAAAAAAAyAAAAAAAAAACAQAAAAwAAAACAAAAAAAAAAEBAAAAJAAAAAIAAAAA/////gAAAAEAAAACAAAHgcw4bGUAAAeBzDhsZQEAAAAUAAAAAgAAAAIAAAAEAAAAAgAAAAgAAAAAAAABZQAAAAAAAABkAAAAAAAAAAIBAAAADAAAAAIAAAAAAAAAAQEAAAAkAAAAAgAAAAAAAAAUAAAAAAAAABh//////////3//////////AQAAABQAAAACAAAAAAAAAAEAAAADAAAABGCAy0ixcqKywlCau1+E/XLEorySmCmL1zl5q9/dwdBu From de2cc0ea87c7c4c419a9ccc31f3b29588ee3c921 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Wed, 16 Sep 2026 19:36:53 +0800 Subject: [PATCH 3/4] [core] Unify manifest sidecar enablement after writer integration --- paimon-python/pypaimon/common/options/core_options.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 7df8ee0ef25f..4256a9dc41f4 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/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] = ( From 2d3115bc2da3596ebf727fa7cd20d4c52029e523 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=9F=E5=BC=8B?= Date: Fri, 18 Sep 2026 16:05:48 +0800 Subject: [PATCH 4/4] [python] Validate sidecar writer failure cleanup --- docs/docs/concepts/spec/manifest.md | 12 ++- .../manifest/manifest_file_manager.py | 4 +- .../tests/manifest/manifest_sidecar_test.py | 101 +++++++++++++++++- 3 files changed, 107 insertions(+), 10 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index a6ea8a4f2a57..497f0e2d1fcb 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -90,10 +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 also generates sidecars for newly written manifests and uses their block coverage -during scans. Its `manifest.sidecar.enabled` table option inherits `manifest-sort.enabled` -when unset. Sidecar references are published only after writing succeeds, and cleanup follows -the owning manifest. +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. 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 diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py index f6d7859615a6..6852099b4a30 100644 --- a/paimon-python/pypaimon/manifest/manifest_file_manager.py +++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py @@ -345,7 +345,7 @@ def _get_value_stats_fields(self, file_dict: dict, file_schema) -> List: def _sidecar_settings(self): return Settings.from_options(self.table.options) - def write(self, file_name, entries: List[ManifestEntry]): + def write(self, file_name, entries: List[ManifestEntry]) -> ManifestFileMeta: buf = BytesIO() fastavro.writer( buf, MANIFEST_ENTRY_SCHEMA, self._to_avro_records(entries), @@ -444,7 +444,7 @@ def delete(self, manifest: ManifestFileMeta): 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) -> ManifestFileMeta: + 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: diff --git a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py index 7986d73c0124..1b64d7b4d527 100644 --- a/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py +++ b/paimon-python/pypaimon/tests/manifest/manifest_sidecar_test.py @@ -18,6 +18,7 @@ import os import unittest from concurrent.futures import CancelledError +from contextlib import contextmanager from copy import deepcopy from io import BytesIO from itertools import product @@ -284,7 +285,7 @@ def test_large_avro_headers(self): self.assertEqual(restored, avro_bytes) self.assertEqual(list(fastavro.reader(BytesIO(restored))), [42]) - def test_disabled_sidecars_do_not_constrain_manifest_target_size(self): + def test_disabled_option_is_independent_of_manifest_target_size(self): for target in ('1 bytes', '1 gb'): with self.subTest(target=target): settings = Settings.from_options(CoreOptions(Options({ @@ -446,6 +447,96 @@ def write_meta(self, name, entries): manager = self.manifest_file_manager return manager.write(name, entries) + def test_explicit_disable_keeps_writes_plain_when_manifest_sort_is_enabled(self): + self.table.options.options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, False) + manager = self.manifest_file_manager + entries = [self.entry('file-%d.parquet' % i, i * 100) for i in range(3)] + metas = [manager.write('plain', entries)] + manager.rolling_write(entries, 1, 'plain-rolling') + for meta in metas: + self.assertIsNone(meta.extra_files) + self.assertFalse(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) + self.assertTrue(manager.read(meta.file_name)) + + @staticmethod + def failing_sidecar_output(original, target, phase, error): + @contextmanager + def output(path): + with original(path) as stream: + if path.endswith(target) and phase == 'write': + wrapped = Mock(wraps=stream) + + def partial_write(data): + stream.write(data[:3]) + raise error + + wrapped.write.side_effect = partial_write + yield wrapped + else: + yield stream + if path.endswith(target) and phase == 'close': + raise error + return output + + def test_sidecar_write_and_close_failures_do_not_publish_references(self): + manager = self.manifest_file_manager + kept = manager.write('kept', [self.entry('kept.parquet', 100)]) + original = manager.file_io.new_output_stream + for phase in ('write', 'close'): + for error in (OSError('failed'), InterruptedError('cancelled'), KeyboardInterrupt()): + with self.subTest(phase=phase, error=type(error).__name__): + stream = self.failing_sidecar_output(original, 'failed' + SUFFIX, phase, error) + expected = type(error) if isinstance(error, (InterruptedError, KeyboardInterrupt)) else RuntimeError + with patch.object(manager.file_io, 'new_output_stream', side_effect=stream), \ + patch.object(manager, '_build_meta', wraps=manager._build_meta) as publish: + with self.assertRaises(expected): + manager.write('failed', [self.entry('new.parquet', 200)]) + publish.assert_not_called() + self.assertFalse(Path(manager.manifest_path, 'failed').exists()) + self.assertFalse(Path(manager.manifest_path, 'failed' + SUFFIX).exists()) + self.assertTrue(Path(manager.manifest_path, kept.file_name).exists()) + self.assertTrue(Path(manager.manifest_path, sidecar_file_name(kept)).exists()) + + def test_later_rolling_failure_cleans_completed_manifest_sidecar_pairs(self): + manager = self.manifest_file_manager + manager.write('kept', [self.entry('kept.parquet', 100)]) + before = {p.name for p in Path(manager.manifest_path).iterdir()} + original = manager.file_io.new_output_stream + stream = self.failing_sidecar_output( + original, 'rolling-failed-1' + SUFFIX, 'close', OSError('second sidecar close failed')) + with patch.object(manager.file_io, 'new_output_stream', side_effect=stream): + with self.assertRaises(RuntimeError): + manager.rolling_write([self.entry('data-%d.parquet' % i, i * 100) for i in range(3)], + 1, 'rolling-failed') + self.assertEqual({p.name for p in Path(manager.manifest_path).iterdir()}, before) + + def test_commit_cleanup_removes_only_owned_manifest_sidecar_pairs(self): + from pypaimon.write.file_store_commit import FileStoreCommit + + manager = self.manifest_file_manager + lists = ManifestListManager(self.table) + kept = manager.write('kept', [self.entry('kept.parquet', 100)]) + delta = manager.write('delta', [self.entry('delta.parquet', 200)]) + changelog = manager.write('changelog', [self.entry('changelog.parquet', 300)]) + merged = manager.write('merged', [self.entry('merged.parquet', 400)]) + lists.write('delta-list', [delta]) + lists.write('changelog-list', [changelog]) + lists.write('base-list', [kept, merged]) + committer = SimpleNamespace(table=self.table, manifest_file_manager=manager, manifest_list_manager=lists) + FileStoreCommit._clean_up_reuse_tmp_manifests(committer, 'delta-list', 'changelog-list') + FileStoreCommit._clean_up_no_reuse_tmp_manifests(committer, 'base-list', [merged]) + self.assertEqual({p.name for p in Path(manager.manifest_path).iterdir()}, + {kept.file_name, sidecar_file_name(kept)}) + + def test_invalid_row_id_ranges_keep_coarse_and_block_coverage_unknown(self): + for first, count in ((-1, 1), (100, 0), (MAX_ROW_ID, 2)): + with self.subTest(first=first, count=count): + manager = self.manifest_file_manager + meta = manager.write('invalid-%s-%s' % (first, count), [self.entry('data.parquet', first, count)]) + self.assertIsNone(meta.min_row_id) + self.assertIsNone(meta.max_row_id) + data = Path(manager.manifest_path, sidecar_file_name(meta)).read_bytes() + self.assertEqual(len(select(data, meta, [Range(50, 50)]).blocks), 1) + def test_sidecars_do_not_depend_on_manifest_target_sizes(self): for i, target in enumerate(('1 bytes', '1 gb', '9223372036854775807 bytes')): with self.subTest(target=target): @@ -840,9 +931,11 @@ def test_rolling_merge_limits_and_abort_cleanup(self): self.assertFalse(intersects(data, meta, [Range(gap, gap)], Settings())) from pypaimon.manifest.manifest_file_merger import ManifestFileMerger merger = ManifestFileMerger(manager, 1000000, 2) - merged = merger.merge(metas) - # Merger returns both the final manifest list and newly written outputs. - outputs = merged[0] if isinstance(merged, tuple) else merged + outputs, new_files = merger.merge(metas) + self.assertTrue(new_files) + self.assertEqual( + sorted(e.file.file_name for meta in outputs for e in manager.read(meta.file_name)), + sorted(e.file.file_name for e in entries)) for meta in outputs: self.assertTrue(Path(manager.manifest_path, meta.file_name + SUFFIX).exists()) for meta in metas: