This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 9686b2a1c2be4fd10edbd5a4ef94043d8261a2a7 Author: YeJunHao <[email protected]> AuthorDate: Wed Sep 16 19:21:52 2026 +0800 [core] Generate sidecars for newly written manifests (#9889) --- docs/docs/concepts/spec/manifest.md | 13 +- docs/generated/core_configuration.html | 6 + .../main/java/org/apache/paimon/CoreOptions.java | 11 + .../benchmark/ManifestFileSorterBenchmark.java | 3 +- .../java/org/apache/paimon/AbstractFileStore.java | 3 +- .../apache/paimon/manifest/ManifestAvroWriter.java | 52 ++- .../org/apache/paimon/manifest/ManifestFile.java | 20 +- .../java/org/apache/paimon/CoreOptionsTest.java | 18 ++ .../paimon/manifest/ManifestFileMetaTest.java | 6 +- .../paimon/manifest/ManifestFileMetaTestBase.java | 3 +- .../apache/paimon/manifest/ManifestFileTest.java | 3 +- .../paimon/manifest/ManifestIndexTestUtils.java | 1 + .../paimon/manifest/ManifestSidecarTest.java | 4 +- .../paimon/manifest/ManifestSidecarWriteTest.java | 347 +++++++++++++++++++++ .../operation/ManifestRewriteCleanupTest.java | 3 +- 15 files changed, 471 insertions(+), 22 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index ada9baf1ce..29466c1ad6 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -71,11 +71,14 @@ using independent partition, row-ID and bucket coverage. A sidecar uses the `.avro.sidecar` reference in the manifest metadata's `_EXTRA_FILES`, without probing a derived file name. The Avro schemas and `_VERSION` identifiers remain unchanged. -The utility includes construction, validation, block selection and optional caching. Table -writers and scans do not yet invoke it automatically. Callers are responsible for publishing -sidecar references, managing file ownership, applying entry filters and reconciling ADD/DELETE -entries after block selection. `build` reads the completed physical manifest and returns -sidecar bytes; it does not write or publish another file. +The utility includes construction, validation, block selection and optional caching. Java table +writers generate sidecars when `manifest.sidecar.enabled` is true; when unset, it inherits +`manifest-sort.enabled`. Both ordinary writes and raw manifest rewrites build the sidecar from +the completed output manifest and publish its `_EXTRA_FILES` reference only after both files +close successfully. Failed writes and aborted writers clean up their own manifest/sidecar pairs. +Scans do not yet invoke sidecar pruning automatically. Callers remain responsible for applying +entry filters and reconciling ADD/DELETE entries after block selection. The low-level `build` +method returns sidecar bytes without writing or publishing another file. 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/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 16d3339f41..cf48d473c4 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1113,6 +1113,12 @@ Mainly to resolve data skew on primary keys. We recommend starting with 64 mb wh <td>Boolean</td> <td>Whether to skip automatic manifest merging during commit when write-only is true. This also skips automatic manifest sort rewrite. Explicit manifest compaction is not affected.</td> </tr> + <tr> + <td><h5>manifest.sidecar.enabled</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>Boolean</td> + <td>Whether to enable manifest sidecars with independent partition, row-id and bucket coverage. Defaults to manifest-sort.enabled when unset.</td> + </tr> <tr> <td><h5>manifest.target-file-size</h5></td> <td style="word-wrap: break-word;">8 mb</td> diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 68215e18e4..9bb324b780 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -522,6 +522,13 @@ public class CoreOptions implements Serializable { .defaultValue(MemorySize.ofMebiBytes(8)) .withDescription("Suggested file size of a manifest file."); + public static final ConfigOption<Boolean> MANIFEST_SIDECAR_ENABLED = + key("manifest.sidecar.enabled") + .booleanType() + .noDefaultValue() + .withDescription( + "Whether to enable manifest sidecars with independent partition, row-id and bucket coverage. Defaults to manifest-sort.enabled when unset."); + public static final ConfigOption<MemorySize> MANIFEST_FULL_COMPACTION_FILE_SIZE = key("manifest.full-compaction-threshold-size") .memoryType() @@ -3217,6 +3224,10 @@ public class CoreOptions implements Serializable { return options.get(MANIFEST_TARGET_FILE_SIZE); } + public boolean manifestSidecarEnabled() { + return options.getOptional(MANIFEST_SIDECAR_ENABLED).orElseGet(this::manifestSortEnabled); + } + public MemorySize manifestFullCompactionThresholdSize() { return options.get(MANIFEST_FULL_COMPACTION_FILE_SIZE); } diff --git a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java index 5a5f31d875..e805efb694 100644 --- a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java +++ b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/ManifestFileSorterBenchmark.java @@ -300,7 +300,8 @@ public class ManifestFileSorterBenchmark { "zstd", pathFactory, TARGET_MANIFEST_SIZE, - null) + null, + new CoreOptions(new Options())) .create(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index 7399e05778..29c7ab8de0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -211,7 +211,8 @@ abstract class AbstractFileStore<T> implements FileStore<T> { options.manifestCompression(), pathFactory(), options.manifestTargetSize().getBytes(), - readManifestCache); + readManifestCache, + options); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java index 4dabe54f3f..92c81bea7c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java @@ -18,6 +18,7 @@ package org.apache.paimon.manifest; +import org.apache.paimon.CoreOptions; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.SimpleColStats; @@ -43,6 +44,7 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.IdentityHashMap; import java.util.List; import java.util.Map; @@ -68,6 +70,7 @@ public final class ManifestAvroWriter implements AutoCloseable { private final String compression; private final PathFactory pathFactory; private final long targetFileSize; + private final CoreOptions options; private final List<ManifestFileMeta> results = new ArrayList<>(); private final List<Path> completedPaths = new ArrayList<>(); @@ -83,7 +86,8 @@ public final class ManifestAvroWriter implements AutoCloseable { ObjectSerializer<ManifestEntry> serializer, String compression, PathFactory pathFactory, - long targetFileSize) { + long targetFileSize, + CoreOptions options) { this.fileIO = fileIO; this.schemaManager = schemaManager; this.partitionType = partitionType; @@ -92,6 +96,7 @@ public final class ManifestAvroWriter implements AutoCloseable { this.compression = compression; this.pathFactory = pathFactory; this.targetFileSize = targetFileSize; + this.options = options; } public void write(ManifestEntry entry) throws IOException { @@ -218,6 +223,9 @@ public final class ManifestAvroWriter implements AutoCloseable { currentWriter.close(); ManifestFileMeta result = currentWriter.result(); completedPaths.add(currentWriter.path); + if (currentWriter.sidecarCreated) { + completedPaths.add(ManifestSidecar.path(currentWriter.path)); + } results.add(result); currentWriter = null; } @@ -413,6 +421,7 @@ public final class ManifestAvroWriter implements AutoCloseable { private @Nullable RowIdStats rowIdStats = new RowIdStats(); private boolean closed; private boolean aborted; + private boolean sidecarCreated; private FileWriter(Path path) { this.path = path; @@ -488,7 +497,7 @@ public final class ManifestAvroWriter implements AutoCloseable { maxLevel = Math.max(maxLevel, entry.level()); if (rowIdStats != null) { Long firstRowId = entry.file().firstRowId(); - if (firstRowId == null) { + if (!validRowIdRange(firstRowId, entry.file().rowCount())) { rowIdStats = null; } else { rowIdStats.collect(firstRowId, entry.file().rowCount()); @@ -515,7 +524,7 @@ public final class ManifestAvroWriter implements AutoCloseable { minLevel = Math.min(minLevel, entry.level); maxLevel = Math.max(maxLevel, entry.level); if (rowIdStats != null) { - if (!entry.hasRowId) { + if (!entry.hasRowId || !validRowIdRange(entry.firstRowId, entry.rowCount)) { rowIdStats = null; } else { rowIdStats.collect(entry.firstRowId, entry.rowCount); @@ -697,6 +706,14 @@ public final class ManifestAvroWriter implements AutoCloseable { ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure); } } + if (sidecarCreated) { + try { + fileIO.deleteQuietly(ManifestSidecar.path(path)); + } catch (Throwable cleanupFailure) { + primaryFailure = + ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure); + } + } return primaryFailure; } @@ -711,6 +728,7 @@ public final class ManifestAvroWriter implements AutoCloseable { outputBytes = out.getPos(); out.close(); out = null; + writeSidecar(); } catch (IOException | RuntimeException | Error failure) { abortCollecting(failure, true); throw failure; @@ -719,6 +737,26 @@ public final class ManifestAvroWriter implements AutoCloseable { } } + private void writeSidecar() throws IOException { + if (!options.manifestSidecarEnabled()) { + return; + } + byte[] bytes = + ManifestSidecar.build( + fileIO, + path, + outputBytes, + Math.addExact(numAddedFiles, numDeletedFiles), + options.dataEvolutionEnabled(), + options.bucket() != -1); + // Publish result() only after both immutable objects have closed. No rename. + try (PositionOutputStream sidecarOut = + fileIO.newOutputStream(ManifestSidecar.path(path), false)) { + sidecarCreated = true; + sidecarOut.write(bytes); + } + } + private ManifestFileMeta result() { if (!closed || outputBytes == null) { throw new IllegalStateException( @@ -740,10 +778,16 @@ public final class ManifestAvroWriter implements AutoCloseable { rowIdStats == null ? null : rowIdStats.minRowId, rowIdStats == null ? null : rowIdStats.maxRowId, totalBucketsKnown ? totalBuckets : null, - null); + sidecarCreated + ? Collections.singletonList(ManifestSidecar.path(path).getName()) + : null); } } + private static boolean validRowIdRange(@Nullable Long first, long count) { + return first != null && first >= 0 && count > 0 && count - 1 <= Long.MAX_VALUE - first; + } + private static class RowIdStats { private long minRowId = Long.MAX_VALUE; diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java index 9781be4b90..30e0adcf1b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java @@ -18,6 +18,7 @@ package org.apache.paimon.manifest; +import org.apache.paimon.CoreOptions; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.FileFormat; @@ -59,6 +60,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { private final RowType partitionType; private final AvroFileFormat avroFileFormat; private final long suggestedFileSize; + private final CoreOptions options; private ManifestFile( FileIO fileIO, @@ -69,7 +71,8 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { String compression, PathFactory pathFactory, long suggestedFileSize, - @Nullable SegmentsCache<Path> cache) { + @Nullable SegmentsCache<Path> cache, + CoreOptions options) { super( fileIO, serializer, @@ -85,6 +88,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { this.partitionType = partitionType; this.avroFileFormat = avroFileFormat; this.suggestedFileSize = suggestedFileSize; + this.options = options; } @Override @@ -301,7 +305,8 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { serializer, compression, pathFactory, - suggestedFileSize); + suggestedFileSize, + options); } /** Creates an Avro manifest writer for one explicit path. */ @@ -314,7 +319,8 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { serializer, compression, singlePathFactory(manifestPath), - Long.MAX_VALUE); + Long.MAX_VALUE, + options); } private PathFactory singlePathFactory(Path manifestPath) { @@ -357,6 +363,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { private final String compression; private final FileStorePathFactory pathFactory; private final long suggestedFileSize; + private final CoreOptions options; @Nullable private final SegmentsCache<Path> cache; public Factory( @@ -367,7 +374,8 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { String compression, FileStorePathFactory pathFactory, long suggestedFileSize, - @Nullable SegmentsCache<Path> cache) { + @Nullable SegmentsCache<Path> cache, + CoreOptions options) { this.fileIO = fileIO; this.schemaManager = schemaManager; this.partitionType = partitionType; @@ -376,6 +384,7 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { this.pathFactory = pathFactory; this.suggestedFileSize = suggestedFileSize; this.cache = cache; + this.options = options; } public boolean isCacheEnabled() { @@ -392,7 +401,8 @@ public class ManifestFile extends ObjectsFile<ManifestEntry> { compression, pathFactory.manifestFileFactory(), suggestedFileSize, - cache); + cache, + options); } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index 47aa94acd9..e83ca0d4ac 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -29,6 +29,24 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link org.apache.paimon.CoreOptions}. */ public class CoreOptionsTest { + @Test + void testManifestSidecarDefaultsToManifestSort() { + assertThat(CoreOptions.MANIFEST_SIDECAR_ENABLED.defaultValue()).isNull(); + for (Boolean sort : new Boolean[] {null, false, true}) { + for (Boolean configured : new Boolean[] {null, false, true}) { + Options options = new Options(); + if (sort != null) { + options.set(CoreOptions.MANIFEST_SORT_ENABLED, sort); + } + if (configured != null) { + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, configured); + } + assertThat(new CoreOptions(options).manifestSidecarEnabled()) + .isEqualTo(configured == null ? Boolean.TRUE.equals(sort) : configured); + } + } + } + @Test public void testDefaultStartupMode() { Options conf = new Options(); diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java index ad3641b121..16f33f4cee 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java @@ -2744,7 +2744,8 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { false, null), Long.MAX_VALUE, - null) + null, + new CoreOptions(new Options())) .create(); List<ManifestFileMeta> input = new ArrayList<>(); @@ -3204,7 +3205,8 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { false, null), Long.MAX_VALUE, - null) + null, + new CoreOptions(new Options())) .create(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java index 5dd1e4787b..a4516015c2 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java @@ -166,7 +166,8 @@ public abstract class ManifestFileMetaTestBase { false, null), Long.MAX_VALUE, - null) + null, + new CoreOptions(new Options())) .create(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index 50097b041a..0362871ef8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java @@ -1442,7 +1442,8 @@ public class ManifestFileTest { "zstd", pathFactory, suggestedFileSize, - cache) + cache, + new CoreOptions(new Options())) .create(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java index 32a94c9a5b..e9be1afc3f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java @@ -35,6 +35,7 @@ import java.util.List; /** Synthetic index references for manifest serialization and lifecycle tests. */ public final class ManifestIndexTestUtils { + private ManifestIndexTestUtils() {} public static ManifestFileMeta withIndexFileName(ManifestFileMeta meta, String indexFileName) { diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java index b4a7b3e29b..1110c70158 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java @@ -18,6 +18,7 @@ package org.apache.paimon.manifest; +import org.apache.paimon.CoreOptions; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryRowWriter; import org.apache.paimon.data.BinaryString; @@ -276,7 +277,8 @@ class ManifestSidecarTest { new ManifestEntrySerializer(), "zstd", paths, - Long.MAX_VALUE); + Long.MAX_VALUE, + new CoreOptions(new Options())); } @Test diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarWriteTest.java new file mode 100644 index 0000000000..cea952943f --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarWriteTest.java @@ -0,0 +1,347 @@ +/* + * 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. + */ + +package org.apache.paimon.manifest; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.PositionOutputStreamWrapper; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.FileStorePathFactory; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RowRangeIndex; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiPredicate; +import java.util.stream.Stream; + +import static org.apache.paimon.TestKeyValueGenerator.DEFAULT_PART_TYPE; +import static org.apache.paimon.utils.VarLengthIntUtils.decodeInt; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests automatic generation and ownership of manifest sidecars. */ +class ManifestSidecarWriteTest { + + @TempDir java.nio.file.Path temp; + private final ManifestTestDataGenerator gen = ManifestTestDataGenerator.builder().build(); + + @Test + void sidecarOptionControlsWriteIO() { + for (Boolean sort : new Boolean[] {null, false, true}) { + for (Boolean configured : new Boolean[] {null, false, true}) { + Options options = new Options(); + if (sort != null) { + options.set(CoreOptions.MANIFEST_SORT_ENABLED, sort); + } + if (configured != null) { + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, configured); + } + AtomicInteger reads = new AtomicInteger(); + FileIO io = + new LocalFileIO() { + + @Override + public SeekableInputStream newInputStream(Path path) + throws IOException { + reads.incrementAndGet(); + return super.newInputStream(path); + } + }; + Path root = root(sort + "-" + configured); + ManifestFile manifests = + manifests(root, io, DEFAULT_PART_TYPE, Long.MAX_VALUE, options); + ManifestFileMeta meta = + manifests.write(Collections.singletonList(gen.next())).get(0); + boolean enabled = configured == null ? Boolean.TRUE.equals(sort) : configured; + assertThat(reads.get()).isEqualTo(enabled ? 1 : 0); + assertThat(ManifestSidecar.fileName(meta) != null).isEqualTo(enabled); + } + } + } + + @Test + void rollingAndRawRewritesGenerateTheirOwnSidecars() throws Exception { + Options options = enabledOptions(); + options.set(CoreOptions.MANIFEST_SORT_ENABLED, true); + FileIO io = LocalFileIO.create(); + Path root = root("rolling"); + ManifestFile manifests = manifests(root, io, DEFAULT_PART_TYPE, 1, options); + List<ManifestFileMeta> metas = manifests.write(entries(2200)); + assertThat(metas.size()).isGreaterThan(1); + for (ManifestFileMeta meta : metas) { + assertThat(ManifestSidecar.fileName(meta)) + .isEqualTo(meta.fileName() + ManifestSidecar.SUFFIX); + List<ManifestEntry> actual = manifests.read(meta.fileName()); + for (ManifestEntry entry : + Arrays.asList(actual.get(0), actual.get(actual.size() - 1))) { + long row = entry.file().firstRowId(); + assertThat(select(root, io, meta, DEFAULT_PART_TYPE, point(row), null).blocks()) + .isNotEmpty(); + } + long gap = actual.get(0).file().firstRowId() + actual.get(0).file().rowCount(); + assertThat(select(root, io, meta, DEFAULT_PART_TYPE, point(gap), null).blocks()) + .isEmpty(); + } + + ManifestFileMeta source = metas.get(0); + Path rewrittenPath = new Path(new Path(root, "manifest"), "explicit-rewrite"); + ManifestAvroWriter writer = manifests.createAvroWriter(rewrittenPath); + try (ManifestAvroReader reader = + manifests.scanAvroBlocks(source.fileName(), source.fileSize())) { + writer.writeEncodedManifest(reader, source); + } + assertThatThrownBy(writer::result).isInstanceOf(IllegalStateException.class); + writer.close(); + ManifestFileMeta rewritten = writer.result().get(0); + assertThat(ManifestSidecar.fileName(rewritten)) + .isEqualTo("explicit-rewrite" + ManifestSidecar.SUFFIX); + assertThat(manifests.read(rewritten.fileName())) + .isEqualTo(manifests.read(source.fileName())); + long outside = metas.get(metas.size() - 1).maxRowId(); + assertThat(select(root, io, rewritten, DEFAULT_PART_TYPE, point(outside), null).blocks()) + .isEmpty(); + writer.abort(); + assertThat(io.exists(rewrittenPath)).isFalse(); + assertThat(io.exists(ManifestSidecar.path(rewrittenPath))).isFalse(); + assertThat(io.exists(manifestPath(root, source))).isTrue(); + for (ManifestFileMeta meta : metas) { + manifests.delete(meta); + assertThat(io.exists(ManifestSidecar.path(manifestPath(root, meta)))).isFalse(); + } + } + + @Test + void payloadsFollowTableMetadata() throws Exception { + for (boolean partitioned : new boolean[] {false, true}) { + for (boolean evolution : new boolean[] {false, true}) { + for (int bucket : new int[] {-2, -1, 4}) { + Options options = enabledOptions(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, evolution); + options.set(CoreOptions.BUCKET, bucket); + RowType partitionType = partitioned ? DEFAULT_PART_TYPE : RowType.of(); + Path root = root(partitioned + "-" + evolution + "-" + bucket); + FileIO io = LocalFileIO.create(); + ManifestFile manifests = + manifests(root, io, partitionType, Long.MAX_VALUE, options); + ManifestEntry source = gen.next(); + ManifestEntry entry = + ManifestEntry.create( + FileKind.ADD, + partitioned ? source.partition() : BinaryRow.EMPTY_ROW, + 1, + 4, + source.file().newFirstRowId(100L)); + ManifestFileMeta meta = + manifests.write(Collections.singletonList(entry)).get(0); + byte[] bytes = + Files.readAllBytes( + java.nio.file.Paths.get( + ManifestSidecar.path(manifestPath(root, meta)) + .toString())); + ByteBuffer in = ByteBuffer.wrap(bytes); + in.getInt(); + decodeInt(in); + int headerLength = decodeInt(in); + in.position(in.position() + headerLength); + assertThat(decodeInt(in)).isEqualTo(1); + assertThat(select(root, io, meta, partitionType, point(99), null).blocks()) + .hasSize(evolution ? 0 : 1); + assertThat( + select(root, io, meta, partitionType, null, (b, t) -> b == 99) + .blocks()) + .hasSize(bucket == -1 ? 1 : 0); + assertThat(manifests.read(meta.fileName())).containsExactly(entry); + } + } + } + } + + @Test + void invalidRowIdRangeKeepsCoverageUnavailable() throws Exception { + FileIO io = LocalFileIO.create(); + Path root = root("unknown"); + ManifestFile manifests = + manifests(root, io, DEFAULT_PART_TYPE, Long.MAX_VALUE, enabledOptions()); + ManifestEntry source = gen.next(); + ManifestEntry entry = + ManifestEntry.create( + FileKind.ADD, source.partition(), 1, 4, source.file().newFirstRowId(-1L)); + ManifestFileMeta meta = manifests.write(Collections.singletonList(entry)).get(0); + assertThat(meta.minRowId()).isNull(); + assertThat(meta.maxRowId()).isNull(); + assertThat(select(root, io, meta, DEFAULT_PART_TYPE, point(123), null).blocks()).hasSize(1); + } + + @ParameterizedTest + @ValueSource(strings = {"open", "write", "close"}) + void sidecarFailureCleansAllRollingOutputs(String phase) throws Exception { + AtomicInteger sidecars = new AtomicInteger(); + FileIO io = + new LocalFileIO() { + + @Override + public PositionOutputStream newOutputStream(Path path, boolean overwrite) + throws IOException { + if (!path.getName().endsWith(ManifestSidecar.SUFFIX) + || sidecars.incrementAndGet() != 2) { + return super.newOutputStream(path, overwrite); + } + if (phase.equals("open")) { + throw new IOException("sidecar " + phase + " failed"); + } + return new PositionOutputStreamWrapper( + super.newOutputStream(path, overwrite)) { + + @Override + public void write(byte[] bytes) throws IOException { + if (phase.equals("write")) { + throw new IOException("sidecar " + phase + " failed"); + } + super.write(bytes); + } + + @Override + public void close() throws IOException { + super.close(); + if (phase.equals("close")) { + throw new IOException("sidecar " + phase + " failed"); + } + } + }; + } + }; + Path root = root("failure-" + phase); + ManifestFile manifests = manifests(root, io, DEFAULT_PART_TYPE, 1, enabledOptions()); + assertThatThrownBy(() -> manifests.write(entries(2200))) + .hasRootCauseMessage("sidecar " + phase + " failed"); + try (Stream<java.nio.file.Path> files = + Files.list(temp.resolve("failure-" + phase).resolve("manifest"))) { + assertThat(files).isEmpty(); + } + } + + private Options enabledOptions() { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_SIDECAR_ENABLED, true); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + return options; + } + + private List<ManifestEntry> entries(int count) { + List<ManifestEntry> result = new ArrayList<>(); + for (int i = 0; i < count; i++) { + ManifestEntry entry = gen.next(); + result.add( + ManifestEntry.create( + i % 2 == 0 ? FileKind.ADD : FileKind.DELETE, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(i * 100000000L))); + } + return result; + } + + private Path root(String name) { + return new Path(temp.resolve(name).toString()); + } + + private static RowRangeIndex point(long rowId) { + return RowRangeIndex.create(Collections.singletonList(new Range(rowId, rowId))); + } + + private static Path manifestPath(Path root, ManifestFileMeta meta) { + return new Path(new Path(root, "manifest"), meta.fileName()); + } + + private ManifestSidecar.Selection select( + Path root, + FileIO io, + ManifestFileMeta meta, + RowType partitionType, + @Nullable RowRangeIndex rows, + @Nullable BiPredicate<Integer, Integer> buckets) { + ManifestSidecar.Selection result = + ManifestSidecar.read( + io, + manifestPath(root, meta), + meta, + rows, + null, + partitionType, + buckets, + null); + assertThat(result).isNotNull(); + return result; + } + + private ManifestFile manifests( + Path root, FileIO io, RowType partitionType, long targetSize, Options options) { + FileStorePathFactory paths = + new FileStorePathFactory( + root, + partitionType, + "default", + CoreOptions.FILE_FORMAT.defaultValue(), + CoreOptions.DATA_FILE_PREFIX.defaultValue(), + CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(), + CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(), + CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(), + CoreOptions.FILE_COMPRESSION.defaultValue(), + null, + null, + CoreOptions.ExternalPathStrategy.NONE, + null, + false, + null); + return new ManifestFile.Factory( + io, + new FileSystemSchemaManager(io, root), + partitionType, + FileFormat.fromIdentifier("avro", new Options()), + "zstd", + paths, + targetSize, + null, + new CoreOptions(options)) + .create(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java index 8247d23262..20a1b26bee 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java @@ -668,7 +668,8 @@ class ManifestRewriteCleanupTest extends ManifestFileMetaTestBase { false, null), suggestedFileSize, - null) + null, + new CoreOptions(new Options())) .create(); }
