diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 6ad271bb5a89..8895415b1f8b 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -63,6 +63,26 @@ skip manifests before opening them. Each extra file belongs exclusively to one manifest. It is retained and cleaned up together with that manifest during snapshot, tag, or changelog deletion. +### Row-ID Block Index + +With `manifest.row-id-index.write` enabled, a manifest writer can create a binary +`.row-id-index` sidecar. Its name is stored in the manifest-list +record's `_EXTRA_FILES`; the existing Avro schemas and `_VERSION` identifiers are unchanged. +Readers identify the row-ID index by the `.row-id-index` suffix among these explicit +references, not by probing for a derived file name. Other extra-file references are preserved. + +With `manifest.row-id-index.read` enabled and a row-ID filter available, readers can use +the sidecar to select complete Avro blocks before reading manifest entries. Both options +default to `false`. Old manifests, null or empty extra-file lists, and lists containing only +other extra-file types use the normal manifest read path. Missing, unsupported, corrupt, +or over-budget indexes also fall back to that path. Writers omit the sidecar if complete +row-ID coverage cannot be established within the configured range and byte budgets. +Cancellation and interruption errors propagate instead of triggering a full-manifest fallback. + +Selected blocks still pass through entry filtering and ADD/DELETE reconciliation. Snapshot, +tag, changelog, orphan-file and failed-commit cleanup retain or remove the sidecar through +its extra-file reference together with the owning manifest. + ## Manifest Data manifests record **ADD** (`0`) and **DELETE** (`1`) entries. Readers reconcile these entries 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 7b0665c50296..5561d31abf5d 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -510,6 +510,34 @@ public InlineElement getDescription() { + "in the previous file. This must not exceed " + "'variant.shredding.minFieldCardinalityRatio'."); + public static final ConfigOption MANIFEST_ROW_ID_INDEX_WRITE = + key("manifest.row-id-index.write") + .booleanType() + .defaultValue(false) + .withDescription( + "Write complete row-id block indexes for newly created manifests."); + + public static final ConfigOption MANIFEST_ROW_ID_INDEX_READ = + key("manifest.row-id-index.read") + .booleanType() + .defaultValue(false) + .withDescription( + "Read optional row-id sidecars after coarse manifest pruning. Missing or invalid indexes fall back to manifest reads."); + + public static final ConfigOption MANIFEST_ROW_ID_INDEX_MAX_RANGES = + key("manifest.row-id-index.max-ranges") + .intType() + .defaultValue(131072) + .withDescription( + "Maximum disjoint row-id intervals across all Avro blocks in a manifest. Exceeding the limit disables the entire index. Range: 1 to 1048576."); + + public static final ConfigOption MANIFEST_ROW_ID_INDEX_MAX_BYTES = + key("manifest.row-id-index.max-bytes") + .intType() + .defaultValue(8388608) + .withDescription( + "Maximum serialized row-id sidecar bytes, including header and checksum. Exceeding the limit disables the entire index. Range: 128 to 67108864."); + public static final ConfigOption MANIFEST_COMPRESSION = key("manifest.compression") .stringType() 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 7399e057783c..df0cccb7d9bd 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -204,14 +204,15 @@ public ChangelogManager changelogManager() { @Override public ManifestFile.Factory manifestFileFactory() { return new ManifestFile.Factory( - fileIO, - schemaManager, - partitionType, - FileFormat.manifestFormat(options), - options.manifestCompression(), - pathFactory(), - options.manifestTargetSize().getBytes(), - readManifestCache); + fileIO, + schemaManager, + partitionType, + FileFormat.manifestFormat(options), + options.manifestCompression(), + pathFactory(), + options.manifestTargetSize().getBytes(), + readManifestCache) + .withRowIdIndexOptions(options.toConfiguration()); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java index b1863358779e..79500c97a60a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java @@ -70,6 +70,19 @@ public final class ManifestAvroReader implements AutoCloseable { } } + @Nullable + public byte[] headerBytes() { + return blockReader.headerBytes(); + } + + public long blockOffset() { + return blockReader.blockOffset(); + } + + public long blockLength() { + return blockReader.blockLength(); + } + /** Returns whether another raw Avro block is available. */ public boolean hasNext() throws IOException { return blockReader.hasNextBlock(); 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 e78f273e29f1..04b14d55a579 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 @@ -43,6 +43,7 @@ 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 +69,7 @@ public final class ManifestAvroWriter implements AutoCloseable { private final String compression; private final PathFactory pathFactory; private final long targetFileSize; + private final ManifestRowIdIndex.Settings rowIdIndexSettings; private final List results = new ArrayList<>(); private final List completedPaths = new ArrayList<>(); @@ -83,7 +85,8 @@ public final class ManifestAvroWriter implements AutoCloseable { ObjectSerializer serializer, String compression, PathFactory pathFactory, - long targetFileSize) { + long targetFileSize, + ManifestRowIdIndex.Settings rowIdIndexSettings) { this.fileIO = fileIO; this.schemaManager = schemaManager; this.partitionType = partitionType; @@ -92,6 +95,7 @@ public final class ManifestAvroWriter implements AutoCloseable { this.compression = compression; this.pathFactory = pathFactory; this.targetFileSize = targetFileSize; + this.rowIdIndexSettings = rowIdIndexSettings; } public void write(ManifestEntry entry) throws IOException { @@ -218,6 +222,9 @@ private void closeCurrentWriter() throws IOException { currentWriter.close(); ManifestFileMeta result = currentWriter.result(); completedPaths.add(currentWriter.path); + if (currentWriter.sidecarCreated) { + completedPaths.add(ManifestRowIdIndex.path(currentWriter.path)); + } results.add(result); currentWriter = null; } @@ -403,6 +410,7 @@ private final class FileWriter { private @Nullable RowIdStats rowIdStats = new RowIdStats(); private boolean closed; private boolean aborted; + private boolean sidecarCreated; private FileWriter(Path path) { this.path = path; @@ -477,7 +485,7 @@ private void collectStats(ManifestEntry entry) { 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()); @@ -503,7 +511,7 @@ private void collectStats(EncodedEntry entry) { 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); @@ -668,6 +676,14 @@ private Throwable abortCollecting(@Nullable Throwable primaryFailure, boolean de ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure); } } + if (sidecarCreated) { + try { + fileIO.deleteQuietly(ManifestRowIdIndex.path(path)); + } catch (Throwable cleanupFailure) { + primaryFailure = + ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure); + } + } return primaryFailure; } @@ -682,6 +698,7 @@ private void close() throws IOException { outputBytes = out.getPos(); out.close(); out = null; + writeRowIdIndex(); } catch (IOException | RuntimeException | Error failure) { abortCollecting(failure, true); throw failure; @@ -690,6 +707,27 @@ private void close() throws IOException { } } + private void writeRowIdIndex() throws IOException { + if (!rowIdIndexSettings.write) { + return; + } + byte[] bytes = + ManifestRowIdIndex.build( + fileIO, + path, + outputBytes, + Math.addExact(numAddedFiles, numDeletedFiles), + rowIdIndexSettings); + if (bytes != null) { + // Publish result() only after both immutable objects have closed. No rename. + try (PositionOutputStream indexOut = + fileIO.newOutputStream(ManifestRowIdIndex.path(path), false)) { + sidecarCreated = true; + indexOut.write(bytes); + } + } + } + private ManifestFileMeta result() { if (!closed || outputBytes == null) { throw new IllegalStateException( @@ -709,10 +747,17 @@ private ManifestFileMeta result() { levelStatsKnown ? minLevel : null, levelStatsKnown ? maxLevel : null, rowIdStats == null ? null : rowIdStats.minRowId, - rowIdStats == null ? null : rowIdStats.maxRowId); + rowIdStats == null ? null : rowIdStats.maxRowId, + sidecarCreated + ? Collections.singletonList(ManifestRowIdIndex.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 0dc99a047076..f14860f48bfb 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 @@ -27,6 +27,7 @@ import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.manifest.ProjectedManifestEntry.Projection; import org.apache.paimon.operation.metrics.CacheMetrics; +import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.types.RowType; @@ -36,6 +37,7 @@ import org.apache.paimon.utils.Filter; import org.apache.paimon.utils.ObjectsFile; import org.apache.paimon.utils.PathFactory; +import org.apache.paimon.utils.RowRangeIndex; import org.apache.paimon.utils.SegmentsCache; import javax.annotation.Nullable; @@ -59,6 +61,7 @@ public class ManifestFile extends ObjectsFile { private final RowType partitionType; private final AvroFileFormat avroFileFormat; private final long suggestedFileSize; + private final ManifestRowIdIndex.Settings rowIdIndexSettings; private ManifestFile( FileIO fileIO, @@ -69,7 +72,8 @@ private ManifestFile( String compression, PathFactory pathFactory, long suggestedFileSize, - @Nullable SegmentsCache cache) { + @Nullable SegmentsCache cache, + ManifestRowIdIndex.Settings rowIdIndexSettings) { super( fileIO, serializer, @@ -85,6 +89,7 @@ private ManifestFile( this.partitionType = partitionType; this.avroFileFormat = avroFileFormat; this.suggestedFileSize = suggestedFileSize; + this.rowIdIndexSettings = rowIdIndexSettings; } @Override @@ -136,9 +141,33 @@ public List read( Filter readFilter, Filter readTFilter, Function convertor) { + return read( + fileName, + fileSize, + partitionFilter, + bucketFilter, + readFilter, + readTFilter, + convertor, + null); + } + + public List read( + String fileName, + @Nullable Long fileSize, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + Filter readFilter, + Filter readTFilter, + Function convertor, + @Nullable ManifestRowIdIndex.Selection selected) { + if (selected != null && selected.blocks().isEmpty()) { + return java.util.Collections.emptyList(); + } try { Path path = pathFactory.toPath(fileName); - if (cache != null) { + // A partial manifest must never enter the cache under the full manifest's key. + if (cache != null && selected == null) { ManifestEntryFilters filters = new ManifestEntryFilters( partitionFilter, bucketFilter, readFilter, readTFilter); @@ -151,7 +180,8 @@ public List read( path, ManifestEntry.MANIFEST_ROW_TYPE, partitionFilter, - bucketFilter); + bucketFilter, + selected); return readFromIterator(iterator, serializer, readFilter, readTFilter, convertor); } catch (IOException e) { throw new UncheckedIOException(e); @@ -205,8 +235,21 @@ private static CloseableIterator createManifestIterator( @Nullable PartitionPredicate partitionFilter, @Nullable BucketFilter bucketFilter) throws IOException { + return createManifestIterator( + fileIO, path, projectedType, partitionFilter, bucketFilter, null); + } + + private static CloseableIterator createManifestIterator( + FileIO fileIO, + Path path, + RowType projectedType, + @Nullable PartitionPredicate partitionFilter, + @Nullable BucketFilter bucketFilter, + @Nullable ManifestRowIdIndex.Selection selected) + throws IOException { try { - ManifestAvroReader reader = new ManifestAvroReader(fileIO.newInputStream(path)); + ManifestAvroReader reader = + new ManifestAvroReader(ManifestRowIdIndex.openManifest(fileIO, path, selected)); return reader.read(projectedType, partitionFilter, bucketFilter); } catch (IOException e) { FileUtils.checkExists(fileIO, path); @@ -301,7 +344,8 @@ public ManifestAvroWriter createAvroWriter() { serializer, compression, pathFactory, - suggestedFileSize); + suggestedFileSize, + rowIdIndexSettings); } /** Creates an Avro manifest writer for one explicit path. */ @@ -314,7 +358,8 @@ public ManifestAvroWriter createAvroWriter(Path manifestPath) { serializer, compression, singlePathFactory(manifestPath), - Long.MAX_VALUE); + Long.MAX_VALUE, + rowIdIndexSettings); } private PathFactory singlePathFactory(Path manifestPath) { @@ -339,6 +384,32 @@ public Path toPath(String fileName) { }; } + @Nullable + public ManifestRowIdIndex.Selection selectBlocks( + ManifestFileMeta manifest, @Nullable RowRangeIndex query) { + return !rowIdIndexSettings.read || query == null + ? null + : ManifestRowIdIndex.read( + fileIO, + pathFactory.toPath(manifest.fileName()), + manifest, + query, + rowIdIndexSettings); + } + + public boolean mayContainRowIds(ManifestFileMeta manifest, @Nullable RowRangeIndex query) { + ManifestRowIdIndex.Selection selected = selectBlocks(manifest, query); + return selected == null || !selected.blocks().isEmpty(); + } + + /** Deletes an unreferenced manifest and its explicitly referenced extra files. */ + public void delete(ManifestFileMeta manifest) { + delete(manifest.fileName()); + if (manifest.extraFiles() != null) { + manifest.extraFiles().forEach(this::delete); + } + } + /** Creator of {@link ManifestFile}. */ public static class Factory { @@ -349,6 +420,8 @@ public static class Factory { private final String compression; private final FileStorePathFactory pathFactory; private final long suggestedFileSize; + private ManifestRowIdIndex.Settings rowIdIndexSettings = + new ManifestRowIdIndex.Settings(new Options()); @Nullable private final SegmentsCache cache; public Factory( @@ -370,6 +443,11 @@ public Factory( this.cache = cache; } + public Factory withRowIdIndexOptions(Options options) { + rowIdIndexSettings = new ManifestRowIdIndex.Settings(options); + return this; + } + public boolean isCacheEnabled() { return cache != null; } @@ -384,7 +462,8 @@ public ManifestFile create() { compression, pathFactory.manifestFileFactory(), suggestedFileSize, - cache); + cache, + rowIdIndexSettings); } } } 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 4c2ab90c96ff..9e35e93873dc 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 @@ -95,6 +95,8 @@ 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 : fromStringArrayData(row.getArray(12))); + row.getFieldCount() <= 12 || row.isNullAt(12) + ? null + : fromStringArrayData(row.getArray(12))); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestRowIdIndex.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestRowIdIndex.java new file mode 100644 index 000000000000..ab3c91cdd918 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestRowIdIndex.java @@ -0,0 +1,559 @@ +/* + * 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.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.options.Options; +import org.apache.paimon.utils.RowRangeIndex; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.UncheckedIOException; +import java.net.SocketTimeoutException; +import java.nio.ByteBuffer; +import java.nio.channels.ClosedByInterruptException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.CancellationException; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Complete row-id interval unions and physical locations of a manifest's Avro blocks. */ +public final class ManifestRowIdIndex { + public static final String SUFFIX = ".row-id-index"; + private static final Logger LOG = LoggerFactory.getLogger(ManifestRowIdIndex.class); + private static final long MAGIC = 0x5041494d52494458L; + private static final int HEADER_BYTES = 68; + private static final int DIGEST_BYTES = 32; + private static final int MAX_AVRO_HEADER = 1024 * 1024; + private static final int READ_BUFFER_BYTES = 1024 * 1024; + + private ManifestRowIdIndex() {} + + public static Path path(Path manifest) { + return new Path(manifest.toString() + SUFFIX); + } + + @Nullable + public static String fileName(ManifestFileMeta manifest) { + if (manifest.extraFiles() != null) { + for (String extraFile : manifest.extraFiles()) { + if (extraFile.endsWith(SUFFIX)) { + return extraFile; + } + } + } + return null; + } + + /** Independent read/write switches and construction/serialization bounds. */ + public static final class Settings { + public final boolean write; + public final boolean read; + public final int maxRanges; + public final int maxBytes; + + public Settings(Options options) { + write = options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE); + read = options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_READ); + maxRanges = options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_RANGES); + maxBytes = options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_BYTES); + checkArgument( + maxRanges > 0 && maxRanges <= 1048576, + "manifest.row-id-index.max-ranges must be in [1, 1048576]"); + checkArgument( + maxBytes >= 128 && maxBytes <= 64 * 1024 * 1024, + "manifest.row-id-index.max-bytes must be in [128, 67108864]"); + } + } + + /** Original file offset/length and zero-based manifest entry ordinal, not table row id. */ + public static final class Block { + public final long offset; + public final long length; + public final long firstRecord; + public final long recordCount; + + public Block(long offset, long length, long firstRecord, long recordCount) { + this.offset = offset; + this.length = length; + this.firstRecord = firstRecord; + this.recordCount = recordCount; + } + } + + /** Selected blocks in original file order. Empty means the manifest can be excluded. */ + public static final class Selection { + private final byte[] header; + private final List blocks; + + private Selection(byte[] header, List blocks) { + this.header = header; + this.blocks = Collections.unmodifiableList(blocks); + } + + public List blocks() { + return blocks; + } + } + + /** Bounded range union. No row-id enumeration, even for a range ending at Long.MAX_VALUE. */ + public static final class Builder { + private final Settings settings; + private final TreeMap ranges = new TreeMap<>(); + private final ByteArrayOutputStream payload = new ByteArrayOutputStream(); + private final DataOutputStream out = new DataOutputStream(payload); + private final int countPosition; + private boolean complete; + private long nextOffset; + private long nextRecord; + private Block current; + private long entriesInBlock; + private int blocks; + private int rangeCount; + + public Builder(Settings settings, @Nullable byte[] header) throws IOException { + this.settings = settings; + this.complete = + header != null + && header.length <= MAX_AVRO_HEADER + && header.length + HEADER_BYTES + DIGEST_BYTES + 8 <= settings.maxBytes; + countPosition = complete ? 4 + header.length : 0; + if (complete) { + out.writeInt(header.length); + out.write(header); + out.writeInt(0); + nextOffset = header.length; + } + } + + public boolean complete() { + return complete; + } + + private void disable(String reason) { + complete = false; + ranges.clear(); + payload.reset(); + LOG.debug("Omitting manifest row-id block index: {}", reason); + } + + public void beginBlock(long offset, long length, long records) throws IOException { + if (!complete) { + return; + } + require(current == null && offset == nextOffset && length > 0 && records > 0); + current = new Block(offset, length, nextRecord, records); + entriesInBlock = 0; + } + + public void add(@Nullable Long first, long count) { + if (!complete) { + return; + } + if (current == null) { + throw new IllegalStateException("No current Avro block"); + } + entriesInBlock++; + if (first == null || first < 0 || count <= 0 || count - 1 > Long.MAX_VALUE - first) { + disable("unknown or invalid row-id coverage"); + return; + } + long start = first; + long end = first + (count - 1); + Map.Entry before = ranges.floorEntry(start); + if (before != null && before.getValue() >= start - 1) { + start = before.getKey(); + end = Math.max(end, before.getValue()); + ranges.remove(before.getKey()); + } + Map.Entry next; + while ((next = ranges.ceilingEntry(start)) != null + && (next.getKey() <= end || next.getKey() - end == 1)) { + end = Math.max(end, next.getValue()); + ranges.remove(next.getKey()); + } + if (rangeCount + ranges.size() >= settings.maxRanges) { + disable("range budget exceeded"); + return; + } + ranges.put(start, end); + } + + public void endBlock() throws IOException { + if (!complete) { + return; + } + require(current != null && entriesInBlock == current.recordCount && !ranges.isEmpty()); + if (HEADER_BYTES + DIGEST_BYTES + (long) payload.size() + 36 + 16L * ranges.size() + > settings.maxBytes) { + disable("serialized byte budget exceeded"); + return; + } + out.writeLong(current.offset); + out.writeLong(current.length); + out.writeLong(current.firstRecord); + out.writeLong(current.recordCount); + out.writeInt(ranges.size()); + for (Map.Entry range : ranges.entrySet()) { + out.writeLong(range.getKey()); + out.writeLong(range.getValue()); + } + nextOffset = Math.addExact(current.offset, current.length); + nextRecord = Math.addExact(current.firstRecord, current.recordCount); + rangeCount += ranges.size(); + blocks++; + ranges.clear(); + current = null; + } + + @Nullable + public byte[] serialize(String name, long fileSize, long entryCount) throws IOException { + if (!complete) { + return null; + } + require(current == null && nextOffset == fileSize && nextRecord == entryCount); + byte[] body = payload.toByteArray(); + ByteBuffer.wrap(body).putInt(countPosition, blocks); + ByteArrayOutputStream buffer = + new ByteArrayOutputStream(HEADER_BYTES + body.length + DIGEST_BYTES); + DataOutputStream envelope = new DataOutputStream(buffer); + envelope.writeLong(MAGIC); + envelope.writeShort(2); + envelope.writeShort(2); // sorted inclusive interval unions per Avro block + envelope.writeInt(1); // COMPLETE; all other bits reserved + envelope.write(digest(name.getBytes(StandardCharsets.UTF_8))); + envelope.writeLong(fileSize); + envelope.writeLong(entryCount); + envelope.writeInt(body.length); + envelope.write(body); + envelope.write(digest(buffer.toByteArray())); + return buffer.toByteArray(); + } + } + + /** Rebuild from the final physical blocks, including raw-copy and encoded rewrite paths. */ + @Nullable + public static byte[] build(FileIO io, Path path, long size, long records, Settings settings) + throws IOException { + try (ManifestAvroReader reader = new ManifestAvroReader(io.newInputStream(path))) { + Builder builder = new Builder(settings, reader.headerBytes()); + ProjectedManifestEntry.Projection projection = + ProjectedManifestEntry.ROW_RANGE_PROJECTION; + ProjectedManifestEntry entry = projection.createEntry(); + while (builder.complete() && reader.hasNext()) { + ManifestAvroReader.RawBlock block = reader.next(); + builder.beginBlock(reader.blockOffset(), reader.blockLength(), block.recordCount()); + ManifestAvroReader.RowIterator rows = block.toRows(projection.projectedType()); + while (builder.complete() && rows.hasNext()) { + entry.replace(rows.next()); + builder.add(entry.file().firstRowId(), entry.file().rowCount()); + } + builder.endBlock(); + } + return builder.serialize(path.getName(), size, records); + } + } + + /** Validate the complete index before allowing any negative decision. */ + public static Selection select( + byte[] data, ManifestFileMeta manifest, RowRangeIndex query, Settings settings) + throws IOException { + require(data.length >= 128 && data.length <= settings.maxBytes); + int checksumOffset = data.length - DIGEST_BYTES; + require( + MessageDigest.isEqual( + digest(Arrays.copyOf(data, checksumOffset)), + Arrays.copyOfRange(data, checksumOffset, data.length))); + DataInputStream in = new DataInputStream(new ByteArrayInputStream(data, 0, checksumOffset)); + require( + in.readLong() == MAGIC + && in.readUnsignedShort() == 2 + && in.readUnsignedShort() == 2 + && in.readInt() == 1); + byte[] nameHash = new byte[DIGEST_BYTES]; + in.readFully(nameHash); + require( + MessageDigest.isEqual( + nameHash, digest(manifest.fileName().getBytes(StandardCharsets.UTF_8)))); + require(in.readLong() == manifest.fileSize()); + long entries = Math.addExact(manifest.numAddedFiles(), manifest.numDeletedFiles()); + require(in.readLong() == entries && in.readInt() == checksumOffset - HEADER_BYTES); + int headerLength = in.readInt(); + require( + headerLength >= 21 + && headerLength <= MAX_AVRO_HEADER + && headerLength <= in.available() - 4); + byte[] header = new byte[headerLength]; + in.readFully(header); + require(header[0] == 'O' && header[1] == 'b' && header[2] == 'j' && header[3] == 1); + int blocks = in.readInt(); + require(blocks >= 0 && blocks <= in.available() / 52); + long nextOffset = headerLength; + long nextRecord = 0; + int totalRanges = 0; + List selected = new ArrayList<>(); + ByteBuffer view = ByteBuffer.wrap(data); + for (int i = 0; i < blocks; i++) { + long offset = in.readLong(); + long length = in.readLong(); + long first = in.readLong(); + long count = in.readLong(); + int ranges = in.readInt(); + require(offset == nextOffset && length > 0 && length <= manifest.fileSize() - offset); + require(first == nextRecord && count > 0 && count <= entries - first); + require( + ranges > 0 + && ranges <= settings.maxRanges - totalRanges + && ranges <= in.available() / 16); + totalRanges += ranges; + int rangesEnd = checksumOffset - in.available() + 16 * ranges; + long minRowId = in.readLong(); + long firstEnd = in.readLong(); + // Sorted intervals already encode the envelope. Peek at the final endpoint without + // adding redundant fields to the format or materializing the interval list. + long maxRowId = ranges == 1 ? firstEnd : view.getLong(rangesEnd - Long.BYTES); + require(minRowId >= 0 && firstEnd >= minRowId && maxRowId >= firstEnd); + boolean candidate = query.intersects(minRowId, maxRowId); + boolean hit = candidate && (ranges == 1 || query.intersects(minRowId, firstEnd)); + long previousEnd = firstEnd; + for (int j = 1; j < ranges; j++) { + long start = in.readLong(); + long end = in.readLong(); + // Validate even rejected blocks: a checksummed but malformed interval list must + // still cause a conservative fallback, not a false negative from its envelope. + require(start >= 0 && end >= start && start > previousEnd); + previousEnd = end; + if (candidate && !hit) { + hit = query.intersects(start, end); + } + } + if (hit) { + selected.add(new Block(offset, length, first, count)); + } + nextOffset = offset + length; + nextRecord = first + count; + } + require(in.available() == 0 && nextOffset == manifest.fileSize() && nextRecord == entries); + return new Selection(header, selected); + } + + /** Bounded, bulk index reads. Null means read the original manifest. */ + @Nullable + public static Selection read( + FileIO io, + Path path, + ManifestFileMeta manifest, + RowRangeIndex query, + Settings settings) { + String indexFileName = fileName(manifest); + if (indexFileName == null) { + return null; + } + try { + byte[] data; + try (InputStream in = io.newInputStream(new Path(path.getParent(), indexFileName))) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buffer = new byte[Math.min(READ_BUFFER_BYTES, settings.maxBytes + 1)]; + int n; + while ((n = + in.read( + buffer, + 0, + Math.min( + buffer.length, settings.maxBytes + 1 - out.size()))) + != -1) { + out.write(buffer, 0, n); + require(out.size() <= settings.maxBytes); + } + data = out.toByteArray(); + } + return select(data, manifest, query, settings); + } catch (CancellationException failure) { + throw failure; + } catch (IOException | RuntimeException failure) { + List pending = new ArrayList<>(); + Set visited = Collections.newSetFromMap(new IdentityHashMap<>()); + pending.add(failure); + for (int position = 0; position < pending.size(); position++) { + Throwable cause = pending.get(position); + if (!visited.add(cause)) { + continue; + } + if (cause instanceof Error) { + throw (Error) cause; + } + if (cause instanceof CancellationException) { + throw (CancellationException) cause; + } + if (cause instanceof InterruptedException + || cause instanceof ClosedByInterruptException + || (cause instanceof InterruptedIOException + && !(cause instanceof SocketTimeoutException))) { + Thread.currentThread().interrupt(); + throw interrupted(failure); + } + if (cause.getCause() != null) { + pending.add(cause.getCause()); + } + Collections.addAll(pending, cause.getSuppressed()); + } + if (Thread.currentThread().isInterrupted()) { + throw interrupted(failure); + } + LOG.debug("Cannot use row-id block index for {}; reading manifest", path, failure); + return null; + } + } + + private static UncheckedIOException interrupted(Throwable failure) { + InterruptedIOException interrupted = + new InterruptedIOException("Interrupted reading row-id index"); + interrupted.initCause(failure); + return new UncheckedIOException(interrupted); + } + + static InputStream openManifest(FileIO io, Path path, @Nullable Selection selected) + throws IOException { + SeekableInputStream input = io.newInputStream(path); + return selected == null ? input : new SelectedBlockInput(input, selected); + } + + /** An OCF stream comprising the original header and selected complete compressed blocks. */ + private static final class SelectedBlockInput extends InputStream { + private final SeekableInputStream input; + private final Selection selected; + private int headerPosition; + private int blockPosition; + private long remaining; + private byte[] buffer; + private int bufferPosition; + private int bufferLimit; + + private SelectedBlockInput(SeekableInputStream input, Selection selected) { + this.input = input; + this.selected = selected; + } + + @Override + public int read() throws IOException { + if (headerPosition < selected.header.length) { + return selected.header[headerPosition++] & 255; + } + return fillBuffer() ? buffer[bufferPosition++] & 255 : -1; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + if (length == 0) { + return 0; + } + if (headerPosition < selected.header.length) { + int n = Math.min(length, selected.header.length - headerPosition); + System.arraycopy(selected.header, headerPosition, bytes, offset, n); + headerPosition += n; + return n; + } + if (!fillBuffer()) { + return -1; + } + int copied = Math.min(length, bufferLimit - bufferPosition); + System.arraycopy(buffer, bufferPosition, bytes, offset, copied); + bufferPosition += copied; + return copied; + } + + private boolean fillBuffer() throws IOException { + if (bufferPosition < bufferLimit) { + return true; + } + if (remaining == 0) { + if (blockPosition == selected.blocks.size()) { + return false; + } + Block block = selected.blocks.get(blockPosition++); + long end = block.offset + block.length; + while (blockPosition < selected.blocks.size() + && selected.blocks.get(blockPosition).offset == end) { + end += selected.blocks.get(blockPosition++).length; + } + input.seek(block.offset); + remaining = end - block.offset; + } + int requested = (int) Math.min(READ_BUFFER_BYTES, remaining); + if (buffer == null || buffer.length < requested) { + buffer = new byte[requested]; + } + bufferPosition = 0; + bufferLimit = 0; + while (bufferLimit < requested) { + int count = input.read(buffer, bufferLimit, requested - bufferLimit); + if (count < 0) { + throw new EOFException("Truncated manifest block"); + } + bufferLimit += count; + } + remaining -= requested; + return true; + } + + @Override + public void close() throws IOException { + input.close(); + } + } + + private static void require(boolean valid) throws IOException { + if (!valid) { + throw new IOException( + "Invalid, unsupported, mismatched or over-budget manifest row-id block index"); + } + } + + private static byte[] digest(byte[] bytes) { + try { + return MessageDigest.getInstance("SHA-256").digest(bytes); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException(e); + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java index a9ef5902ec9e..934c6151e5cb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java @@ -29,6 +29,7 @@ import org.apache.paimon.manifest.ManifestEntrySerializer; import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestRowIdIndex; import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.manifest.SimpleFileEntry; import org.apache.paimon.operation.metrics.ScanMetrics; @@ -496,13 +497,17 @@ private List readManifest( @Nullable Filter additionalFilter, @Nullable Filter additionalTFilter) { + ManifestFile manifestFile = manifestFileFactory.create(); + ManifestRowIdIndex.Selection selected = manifestFile.selectBlocks(manifest, rowRangeIndex); + if (selected != null && selected.blocks().isEmpty()) { + return Collections.emptyList(); + } Filter entryRowFilter = createEntryRowFilter(); Function finalConverter = dropStats ? e -> converter.apply(dropStats(e)) : converter; List entries = - manifestFileFactory - .create() + manifestFile .withCacheMetrics( scanMetrics != null ? scanMetrics.getCacheMetrics() : null) .read( @@ -516,7 +521,8 @@ private List readManifest( && (manifestEntryFilter == null || manifestEntryFilter.test(entry)) && filterByStats(entry), - finalConverter); + finalConverter, + selected); LOG.info("Read {} manifest entries from {}", entries.size(), manifest.fileName()); return entries; } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java index 9689f272e2eb..eec6090a3635 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java @@ -26,7 +26,6 @@ import org.apache.paimon.manifest.ExpireFileEntry; import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestFile; -import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.manifest.ManifestList; import org.apache.paimon.stats.StatsFileHandler; import org.apache.paimon.utils.FileStorePathFactory; @@ -100,17 +99,17 @@ public Set manifestSkippingSet(List skippingSnapshots) { // base manifests if (manifestList.exists(skippingSnapshot.baseManifestList())) { skippingSet.add(skippingSnapshot.baseManifestList()); - manifestList.read(skippingSnapshot.baseManifestList()).stream() - .map(ManifestFileMeta::fileName) - .forEach(skippingSet::add); + manifestList + .read(skippingSnapshot.baseManifestList()) + .forEach(manifest -> addManifestToSkippingSet(skippingSet, manifest)); } // delta manifests if (manifestList.exists(skippingSnapshot.deltaManifestList())) { skippingSet.add(skippingSnapshot.deltaManifestList()); - manifestList.read(skippingSnapshot.deltaManifestList()).stream() - .map(ManifestFileMeta::fileName) - .forEach(skippingSet::add); + manifestList + .read(skippingSnapshot.deltaManifestList()) + .forEach(manifest -> addManifestToSkippingSet(skippingSet, manifest)); } // index manifests diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java index 6494411a8c4b..c9b6f0c47012 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java @@ -489,9 +489,9 @@ private Set manifestSkippingSet(Snapshot skippingSnapshot) { // data manifests skippingSet.add(skippingSnapshot.baseManifestList()); skippingSet.add(skippingSnapshot.deltaManifestList()); - manifestList.readDataManifests(skippingSnapshot).stream() - .map(ManifestFileMeta::fileName) - .forEach(skippingSet::add); + manifestList + .readDataManifests(skippingSnapshot) + .forEach(manifest -> addManifestToSkippingSet(skippingSet, manifest)); // index manifests String indexManifest = skippingSnapshot.indexManifest(); @@ -511,6 +511,14 @@ private Set manifestSkippingSet(Snapshot skippingSnapshot) { return skippingSet; } + protected static void addManifestToSkippingSet( + Set skippingSet, ManifestFileMeta manifest) { + skippingSet.add(manifest.fileName()); + if (manifest.extraFiles() != null) { + skippingSet.addAll(manifest.extraFiles()); + } + } + private boolean tryDeleteEmptyDirectory(Path path) { try { fileIO.delete(path, false); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java index e3f8c7af7671..fb6e7aec89ec 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java @@ -79,7 +79,7 @@ public static List merge( // exception occurs, clean up and rethrow for (ManifestFileMeta manifest : newFilesForAbort) { try { - manifestFile.delete(manifest.fileName()); + manifestFile.delete(manifest); } catch (Throwable cleanupFailure) { primaryFailure = ExceptionUtils.firstOrSuppressed(cleanupFailure, primaryFailure); diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java index a24b5c4c6e9b..735a706937cb 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java @@ -52,14 +52,14 @@ public void cleanUpReuseTmpManifests( String newIndexManifest) { if (deltaManifestList != null) { for (ManifestFileMeta manifest : manifestList.read(deltaManifestList.getKey())) { - manifestFile.delete(manifest.fileName()); + manifestFile.delete(manifest); } manifestList.delete(deltaManifestList.getKey()); } if (changelogManifestList != null) { for (ManifestFileMeta manifest : manifestList.read(changelogManifestList.getKey())) { - manifestFile.delete(manifest.fileName()); + manifestFile.delete(manifest); } manifestList.delete(changelogManifestList.getKey()); } @@ -80,7 +80,7 @@ public void cleanUpNoReuseTmpManifests( .collect(Collectors.toSet()); for (ManifestFileMeta suspect : mergeAfterManifests) { if (!oldMetaSet.contains(suspect.fileName())) { - manifestFile.delete(suspect.fileName()); + manifestFile.delete(suspect); } } } 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 2f4e32cd2759..cc68219a4aad 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; @@ -53,7 +54,8 @@ void testExtraFiles() throws IOException { Arrays.asList( null, Collections.emptyList(), - Arrays.asList("extra-1", "extra-2"))) { + Arrays.asList("extra-1", "extra-2"), + Arrays.asList("partition-index", "manifest" + ManifestRowIdIndex.SUFFIX))) { ManifestFileMeta meta = new ManifestFileMeta( original.fileName(), @@ -84,6 +86,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/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index c3a50f4ef1de..3783c667f643 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 @@ -28,13 +28,19 @@ import org.apache.paimon.fs.FileIOFinder; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.SeekableInputStreamWrapper; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFileMetaWriteColsLegacySerializer; +import org.apache.paimon.operation.AppendOnlyFileStoreScan; +import org.apache.paimon.operation.ManifestsReader; +import org.apache.paimon.operation.commit.CommitCleaner; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.TableSchema; import org.apache.paimon.stats.StatsTestUtils; import org.apache.paimon.types.DataField; import org.apache.paimon.types.RowType; @@ -42,6 +48,9 @@ import org.apache.paimon.utils.FailingFileIO; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.Filter; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RowRangeIndex; import org.apache.paimon.utils.SegmentsCache; import org.junit.jupiter.api.RepeatedTest; @@ -63,13 +72,18 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; import java.util.stream.Stream; import static org.apache.paimon.TestKeyValueGenerator.DEFAULT_PART_TYPE; +import static org.apache.paimon.manifest.ManifestIndexTestUtils.withExtraFiles; import static org.apache.paimon.stats.StatsTestUtils.convertWithoutSchemaEvolution; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** Tests for {@link ManifestFile}. */ public class ManifestFileTest { @@ -1224,16 +1238,449 @@ private static int indexOf(byte[] bytes, byte[] target, int from, int limit) { return -1; } + @Test + void testRowIdSidecarRollingRawRewriteAndDelete() throws Exception { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE, true); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, true); + ManifestFile manifests = createManifestFile(tempDir.toString(), 1, options); + List entries = new ArrayList<>(); + for (int i = 0; i < 2200; i++) { + ManifestEntry source = gen.next(); + entries.add( + ManifestEntry.create( + i % 2 == 0 ? FileKind.ADD : FileKind.DELETE, + source.partition(), + source.bucket(), + source.totalBuckets(), + source.file().newFirstRowId(i * 100000000L))); + } + List metas = manifests.write(entries); + assertThat(metas.size()).isGreaterThan(1); + + for (ManifestFileMeta meta : metas) { + assertThat(ManifestRowIdIndex.fileName(meta)) + .isEqualTo(meta.fileName() + ManifestRowIdIndex.SUFFIX); + List actual = manifests.read(meta.fileName()); + for (ManifestEntry entry : + Arrays.asList(actual.get(0), actual.get(actual.size() - 1))) { + assertThat( + manifests.mayContainRowIds( + meta, + RowRangeIndex.create( + Collections.singletonList( + new Range( + entry.file().firstRowId(), + entry.file().firstRowId()))))) + .isTrue(); + } + long gap = actual.get(0).file().firstRowId() + actual.get(0).file().rowCount(); + assertThat( + manifests.mayContainRowIds( + meta, + RowRangeIndex.create( + Collections.singletonList(new Range(gap, gap))))) + .isFalse(); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest") + .resolve(meta.fileName() + ManifestRowIdIndex.SUFFIX))) + .isTrue(); + } + ManifestFileMeta source = metas.get(0); + ManifestAvroWriter rewrite = manifests.createAvroWriter(); + try (ManifestAvroReader reader = + manifests.scanAvroBlocks(source.fileName(), source.fileSize())) { + rewrite.writeEncodedManifest(reader, source); + } + rewrite.close(); + ManifestFileMeta rewritten = rewrite.result().get(0); + assertThat(ManifestRowIdIndex.fileName(rewritten)) + .isEqualTo(rewritten.fileName() + ManifestRowIdIndex.SUFFIX); + assertThat(manifests.read(rewritten.fileName())) + .isEqualTo(manifests.read(source.fileName())); + long outside = metas.get(metas.size() - 1).maxRowId(); + assertThat( + manifests.mayContainRowIds( + rewritten, + RowRangeIndex.create( + Collections.singletonList(new Range(outside, outside))))) + .isFalse(); + rewrite.abort(); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest") + .resolve(rewritten.fileName() + ManifestRowIdIndex.SUFFIX))) + .isFalse(); + for (ManifestFileMeta meta : metas) { + manifests.delete(meta); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest") + .resolve(meta.fileName() + ManifestRowIdIndex.SUFFIX))) + .isFalse(); + } + } + + @Test + void testReadsOnlySelectedBlocksAndPreservesPhysicalOrdinals() throws Exception { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE, true); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, true); + RecordingFileIO fileIO = new RecordingFileIO(); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, fileIO); + ManifestFile manifests = factory.create(); + List entries = new ArrayList<>(); + for (int i = 0; i < 4000; i++) { + ManifestEntry entry = gen.next(); + entries.add( + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(i * 1000000L))); + } + ManifestFileMeta meta = manifests.write(entries).get(0); + RowRangeIndex query = + RowRangeIndex.create( + Arrays.asList( + new Range(1000000000L, 1000000000L), + new Range(3000000000L, 3000000000L))); + + ManifestRowIdIndex.Selection selected = manifests.selectBlocks(meta, query); + assertThat(selected.blocks()).hasSize(2); + + fileIO.reset(); + List actual = + factory.create() + .read( + meta.fileName(), + meta.fileSize(), + null, + null, + row -> true, + entry -> true, + java.util.function.Function.identity(), + selected); + List expected = new ArrayList<>(); + for (ManifestRowIdIndex.Block block : selected.blocks()) { + expected.addAll( + entries.subList( + (int) block.firstRecord, + (int) (block.firstRecord + block.recordCount))); + } + assertThat(actual).containsExactlyElementsOf(expected); + assertThat(actual).contains(entries.get(1000), entries.get(3000)); + assertThat(fileIO.bytes.get()).isLessThan(meta.fileSize() / 4); + assertThat(fileIO.seeks) + .containsExactlyElementsOf( + selected.blocks().stream() + .map(block -> block.offset) + .collect(Collectors.toList())); + assertThat(fileIO.opened) + .containsExactly(new Path(tempDir.toString(), "manifest/" + meta.fileName())); + // Block selections cannot populate the ordinary full-manifest read cache. + assertThat(manifests.read(meta.fileName())).containsExactlyElementsOf(entries); + } + + @Test + void testScannerPreservesDeletesAndColumnGroups() { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE, true); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, true); + RecordingFileIO fileIO = new RecordingFileIO(); + ManifestFile.Factory factory = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, fileIO); + ManifestFile manifests = factory.create(); + ManifestEntry entry = gen.next(); + ManifestEntry add = + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(100L)); + ManifestEntry delete = + ManifestEntry.create( + FileKind.DELETE, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + add.file()); + ManifestEntry other = gen.next(); + ManifestEntry live = + ManifestEntry.create( + FileKind.ADD, + other.partition(), + other.bucket(), + other.totalBuckets(), + other.file().newFirstRowId(100L)); + List metas = new ArrayList<>(); + metas.addAll(manifests.write(Arrays.asList(add, live))); + metas.addAll(manifests.write(Collections.singletonList(delete))); + metas.addAll( + manifests.write( + Collections.singletonList( + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(0L))))); + AppendOnlyFileStoreScan scan = + new AppendOnlyFileStoreScan( + mock(ManifestsReader.class), + null, + null, + null, + mock(TableSchema.class), + factory, + 2, + false, + false, + false); + scan.withRowRanges(Collections.singletonList(new Range(100, 100))); + fileIO.reset(); + List result = new ArrayList<>(); + scan.readManifestEntries(metas, false).forEachRemaining(result::add); + assertThat(result).containsExactly(live); + assertThat( + fileIO.opened.stream() + .filter(path -> !path.getName().endsWith(ManifestRowIdIndex.SUFFIX)) + .map(Path::getName)) + .containsExactlyInAnyOrder(metas.get(0).fileName(), metas.get(1).fileName()); + } + + @Test + void testSidecarWriteFailureAbortsAllRollingOutputs() { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE, true); + AtomicInteger indexes = new AtomicInteger(); + FileIO failing = + new LocalFileIO() { + @Override + public PositionOutputStream newOutputStream(Path path, boolean overwrite) + throws IOException { + if (path.getName().endsWith(ManifestRowIdIndex.SUFFIX) + && indexes.incrementAndGet() == 2) { + throw new IOException("sidecar write failed"); + } + return super.newOutputStream(path, overwrite); + } + }; + ManifestFile manifests = + createManifestFileFactory(tempDir.toString(), 1, options, failing).create(); + List entries = new ArrayList<>(); + for (int i = 0; i < 2200; i++) { + ManifestEntry entry = gen.next(); + entries.add( + ManifestEntry.create( + FileKind.ADD, + entry.partition(), + entry.bucket(), + entry.totalBuckets(), + entry.file().newFirstRowId(0L))); + } + assertThatThrownBy(() -> manifests.write(entries)) + .hasRootCauseMessage("sidecar write failed"); + try (java.util.stream.Stream files = + java.nio.file.Files.list(tempDir.resolve("manifest"))) { + assertThat(files).isEmpty(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + @Test + void testUnknownRowIdDisablesIndexAndNoQueryDoesNotReadSidecar() { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE, true); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, true); + RecordingFileIO fileIO = new RecordingFileIO(); + ManifestFile manifests = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, fileIO) + .create(); + ManifestFileMeta meta = manifests.write(Collections.singletonList(gen.next())).get(0); + assertThat(ManifestRowIdIndex.fileName(meta)).isNull(); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest") + .resolve(meta.fileName() + ManifestRowIdIndex.SUFFIX))) + .isFalse(); + + fileIO.reset(); + assertThat(manifests.mayContainRowIds(meta, null)).isTrue(); + assertThat(fileIO.opened).isEmpty(); + } + + @Test + void testExplicitIndexReferenceAndNullDoesNotProbe() throws Exception { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE, true); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, true); + RecordingFileIO io = new RecordingFileIO(); + ManifestFile manifests = + createManifestFileFactory(tempDir.toString(), Long.MAX_VALUE, options, io).create(); + ManifestEntry original = gen.next(); + ManifestEntry entry = + ManifestEntry.create( + FileKind.ADD, + original.partition(), + original.bucket(), + original.totalBuckets(), + original.file().newFirstRowId(100L)); + ManifestFileMeta written = manifests.write(Collections.singletonList(entry)).get(0); + assertThat(ManifestRowIdIndex.fileName(written)).isNotNull(); + RowRangeIndex query = RowRangeIndex.create(Collections.singletonList(new Range(0, 0))); + io.reset(); + for (List extraFiles : + Arrays.asList( + null, + Collections.emptyList(), + Collections.singletonList("other-partition-index"))) { + ManifestFileMeta unindexed = withExtraFiles(written, extraFiles); + assertThat(manifests.selectBlocks(unindexed, query)).isNull(); + assertThat(io.opened).isEmpty(); + } + // An existing suffix-named object must not be inferred as a reference. + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest") + .resolve(ManifestRowIdIndex.fileName(written)))) + .isTrue(); + String explicitName = "custom-index-name" + ManifestRowIdIndex.SUFFIX; + java.nio.file.Files.move( + tempDir.resolve("manifest").resolve(ManifestRowIdIndex.fileName(written)), + tempDir.resolve("manifest").resolve(explicitName)); + String otherName = "other-partition-index"; + java.nio.file.Path otherPath = tempDir.resolve("manifest").resolve(otherName); + java.nio.file.Files.write(otherPath, new byte[] {1, 2, 3}); + ManifestFileMeta indexed = withExtraFiles(written, Arrays.asList(otherName, explicitName)); + assertThat(manifests.selectBlocks(indexed, query).blocks()).isEmpty(); + assertThat(io.opened) + .containsExactly(new Path(tempDir.toString(), "manifest/" + explicitName)); + manifests.delete(indexed); + assertThat(java.nio.file.Files.exists(otherPath)).isFalse(); + assertThat(java.nio.file.Files.exists(tempDir.resolve("manifest").resolve(explicitName))) + .isFalse(); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest").resolve(written.fileName()))) + .isFalse(); + } + + @Test + void testCommitCleanerDeletesExplicitIndexReferences() throws Exception { + ManifestFile manifests = createManifestFile(tempDir.toString(), Long.MAX_VALUE); + ManifestList lists = mock(ManifestList.class); + CommitCleaner cleaner = new CommitCleaner(lists, manifests, mock(IndexManifestFile.class)); + for (int mode = 0; mode < 2; mode++) { + ManifestFileMeta meta = manifests.write(Collections.singletonList(gen.next())).get(0); + String indexName = "commit-index-" + mode + ManifestRowIdIndex.SUFFIX; + Path indexPath = new Path(tempDir.toString(), "manifest/" + indexName); + LocalFileIO.create().newOutputStream(indexPath, false).close(); + String otherName = "commit-other-" + mode; + Path otherPath = new Path(tempDir.toString(), "manifest/" + otherName); + LocalFileIO.create().newOutputStream(otherPath, false).close(); + ManifestFileMeta indexed = withExtraFiles(meta, Arrays.asList(otherName, indexName)); + if (mode == 0) { + when(lists.read("delta-list")).thenReturn(Collections.singletonList(indexed)); + cleaner.cleanUpReuseTmpManifests(Pair.of("delta-list", 1L), null, null, null); + } else { + cleaner.cleanUpNoReuseTmpManifests( + null, Collections.emptyList(), Collections.singletonList(indexed)); + } + assertThat(LocalFileIO.create().exists(indexPath)).isFalse(); + assertThat(LocalFileIO.create().exists(otherPath)).isFalse(); + assertThat( + java.nio.file.Files.exists( + tempDir.resolve("manifest").resolve(meta.fileName()))) + .isFalse(); + } + } + + /** Observes actual file access without adding counters to production readers. */ + private static final class RecordingFileIO extends LocalFileIO { + private final List opened = Collections.synchronizedList(new ArrayList<>()); + private final List seeks = Collections.synchronizedList(new ArrayList<>()); + private final AtomicLong bytes = new AtomicLong(); + + private void reset() { + opened.clear(); + seeks.clear(); + bytes.set(0); + } + + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + opened.add(path); + return new SeekableInputStreamWrapper(super.newInputStream(path)) { + @Override + public void seek(long desired) throws IOException { + seeks.add(desired); + super.seek(desired); + } + + @Override + public int read() throws IOException { + int value = super.read(); + if (value >= 0) { + bytes.incrementAndGet(); + } + return value; + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + int n = super.read(buffer, offset, length); + if (n > 0) { + bytes.addAndGet(n); + } + return n; + } + }; + } + } + private ManifestFile createManifestFile(String pathStr) { return createManifestFile(pathStr, ThreadLocalRandom.current().nextInt(8192) + 1024); } private ManifestFile createManifestFile(String pathStr, long suggestedFileSize) { - return createManifestFile(pathStr, suggestedFileSize, null); + return createManifestFile(pathStr, suggestedFileSize, new Options()); } private ManifestFile createManifestFile( String pathStr, long suggestedFileSize, @Nullable SegmentsCache cache) { + return createManifestFileFactory( + pathStr, + suggestedFileSize, + new Options(), + FileIOFinder.find(new Path(pathStr)), + cache) + .create(); + } + + private ManifestFile createManifestFile( + String pathStr, long suggestedFileSize, Options options) { + return createManifestFileFactory( + pathStr, suggestedFileSize, options, FileIOFinder.find(new Path(pathStr))) + .create(); + } + + private ManifestFile.Factory createManifestFileFactory( + String pathStr, long suggestedFileSize, Options options, FileIO fileIO) { + return createManifestFileFactory(pathStr, suggestedFileSize, options, fileIO, null); + } + + private ManifestFile.Factory createManifestFileFactory( + String pathStr, + long suggestedFileSize, + Options options, + FileIO fileIO, + @Nullable SegmentsCache cache) { Path path = new Path(pathStr); FileStorePathFactory pathFactory = new FileStorePathFactory( @@ -1252,7 +1699,6 @@ private ManifestFile createManifestFile( null, false, null); - FileIO fileIO = FileIOFinder.find(path); return new ManifestFile.Factory( fileIO, new FileSystemSchemaManager(fileIO, path), @@ -1262,7 +1708,7 @@ private ManifestFile createManifestFile( pathFactory, suggestedFileSize, cache) - .create(); + .withRowIdIndexOptions(options); } @Test 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 new file mode 100644 index 000000000000..34461785818e --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestIndexTestUtils.java @@ -0,0 +1,98 @@ +/* + * 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.FileStore; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.utils.JsonSerdeUtil; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.SnapshotManager; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +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) { + return withExtraFiles( + meta, indexFileName == null ? null : Collections.singletonList(indexFileName)); + } + + public static ManifestFileMeta withExtraFiles(ManifestFileMeta meta, List extraFiles) { + return new ManifestFileMeta( + meta.fileName(), + meta.fileSize(), + meta.numAddedFiles(), + meta.numDeletedFiles(), + meta.partitionStats(), + meta.schemaId(), + meta.minBucket(), + meta.maxBucket(), + meta.minLevel(), + meta.maxLevel(), + meta.minRowId(), + meta.maxRowId(), + extraFiles); + } + + /** Replaces only synthetic snapshot fixtures, using newly written manifest lists. */ + public static void registerIndexReferences(FileStore store, long snapshotId) + throws IOException { + SnapshotManager manager = store.snapshotManager(); + FileIO io = manager.fileIO(); + Path snapshotPath = manager.snapshotPath(snapshotId); + ObjectNode snapshot = + (ObjectNode) + JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree( + io.readFileUtf8(snapshotPath)); + ManifestList lists = store.manifestListFactory().create(); + for (String field : + new String[] {"baseManifestList", "deltaManifestList", "changelogManifestList"}) { + JsonNode value = snapshot.get(field); + if (value == null || value.isNull()) { + continue; + } + List indexed = new ArrayList<>(); + for (ManifestFileMeta meta : lists.read(value.asText())) { + // Deliberately use a name which cannot be derived by appending the sidecar suffix. + String name = "index-for-" + meta.fileName() + ManifestRowIdIndex.SUFFIX; + Path index = store.pathFactory().toManifestFilePath(name); + if (!io.exists(index)) { + // GC treats index bytes as opaque; unsupported/partial files are still owned. + io.newOutputStream(index, false).close(); + } + indexed.add(withIndexFileName(meta, name)); + } + Pair replacement = lists.write(indexed); + snapshot.put(field, replacement.getLeft()); + snapshot.put(field + "Size", replacement.getRight()); + } + io.overwriteFileUtf8( + snapshotPath, JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.writeValueAsString(snapshot)); + manager.invalidateCache(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestRowIdIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestRowIdIndexTest.java new file mode 100644 index 000000000000..6a5f5f0a9764 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestRowIdIndexTest.java @@ -0,0 +1,670 @@ +/* + * 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.fs.ByteArraySeekableStream; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.utils.IOUtils; +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 java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Base64; +import java.util.Collections; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.CancellationException; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Cross-language format, physical block positions, completeness and allocation bounds. */ +class ManifestRowIdIndexTest { + @TempDir java.nio.file.Path temp; + private final ManifestRowIdIndex.Settings settings = + new ManifestRowIdIndex.Settings(new Options()); + + static ManifestFileMeta meta(String name, long size, long entries) { + ManifestFileMeta meta = mock(ManifestFileMeta.class); + when(meta.fileName()).thenReturn(name); + when(meta.fileSize()).thenReturn(size); + when(meta.extraFiles()) + .thenReturn(Collections.singletonList(name + ManifestRowIdIndex.SUFFIX)); + when(meta.numAddedFiles()).thenReturn(entries); + return meta; + } + + private Properties fixture() throws IOException { + Properties properties = new Properties(); + try (java.io.InputStream input = + getClass().getResourceAsStream("/manifest-row-id-index-v2.txt")) { + properties.load(input); + } + return properties; + } + + private byte[] header() throws IOException { + return Base64.getDecoder().decode(fixture().getProperty("avroHeader")); + } + + private byte[] golden() throws IOException { + return Base64.getDecoder().decode(fixture().getProperty("index")); + } + + private ManifestFileMeta goldenMeta() throws IOException { + return meta("manifest-golden", header().length + 400, 7); + } + + @Test + void crossLanguageFormatAndBlockOrdinals() throws Exception { + byte[] header = header(); + ManifestRowIdIndex.Builder builder = new ManifestRowIdIndex.Builder(settings, header); + builder.beginBlock(header.length, 100, 3); + builder.add(0L, 10); + builder.add(5L, 5); + builder.add(20L, 5); + builder.endBlock(); + builder.beginBlock(header.length + 100, 200, 2); + builder.add((1L << 32) - 2, 5); + builder.add(8254058425445L, 1); + builder.endBlock(); + builder.beginBlock(header.length + 300, 100, 2); + builder.add(20L, 5); + builder.add(Long.MAX_VALUE, 1); + builder.endBlock(); + byte[] data = builder.serialize("manifest-golden", header.length + 400, 7); + assertThat(data).isEqualTo(golden()); + ManifestFileMeta meta = goldenMeta(); + for (long point : + new long[] { + 0, + 9, + 20, + 24, + (1L << 32) - 2, + 1L << 32, + (1L << 32) + 2, + 8254058425445L, + Long.MAX_VALUE + }) { + assertThat(select(data, meta, point).blocks()).as("row %s", point).isNotEmpty(); + } + for (long point : + new long[] { + 10, 19, 25, (1L << 32) - 3, (1L << 32) + 3, 8254058425444L, Long.MAX_VALUE - 1 + }) { + assertThat(select(data, meta, point).blocks()).as("row %s", point).isEmpty(); + } + ManifestRowIdIndex.Selection selected = select(data, meta, 20); + assertThat(selected.blocks()).extracting(b -> b.firstRecord).containsExactly(0L, 5L); + assertThat(selected.blocks()) + .extracting(b -> b.offset) + .containsExactly((long) header.length, header.length + 300L); + assertThat(selected.blocks()).extracting(b -> b.length).containsExactly(100L, 100L); + + ManifestRowIdIndex.Selection gap = select(data, meta, 16); + + assertThat(gap.blocks()).isEmpty(); + RowRangeIndex query = + RowRangeIndex.create(Arrays.asList(new Range(10, 19), new Range(25, 40))); + assertThat(ManifestRowIdIndex.select(data, meta, query, settings).blocks()).isEmpty(); + assertThat(query.ranges()).containsExactly(new Range(10, 19), new Range(25, 40)); + } + + @Test + void minMaxSkipsExactIntersectionChecksAndHandlesOneInterval() throws Exception { + byte[] header = header(); + ManifestRowIdIndex.Builder builder = new ManifestRowIdIndex.Builder(settings, header); + builder.beginBlock(header.length, 100, 2); + builder.add(0L, 10); + builder.add(20L, 10); + builder.endBlock(); + builder.beginBlock(header.length + 100, 100, 2); + builder.add(100L, 10); + builder.add(200L, 10); + builder.endBlock(); + builder.beginBlock(header.length + 200, 100, 1); + builder.add(1L << 32, 10); + builder.endBlock(); + byte[] data = builder.serialize("m", header.length + 300, 5); + ManifestFileMeta meta = meta("m", header.length + 300, 5); + RowRangeIndex outside = + spy(RowRangeIndex.create(Collections.singletonList(new Range(50, 59)))); + ManifestRowIdIndex.Selection none = + ManifestRowIdIndex.select(data, meta, outside, settings); + assertThat(none.blocks()).isEmpty(); + + // Only the three envelopes are tested; no individual interval intersection is evaluated. + verify(outside, times(3)).intersects(anyLong(), anyLong()); + verify(outside).intersects(0, 29); + verify(outside).intersects(100, 209); + verify(outside).intersects(1L << 32, (1L << 32) + 9); + + RowRangeIndex one = + spy( + RowRangeIndex.create( + Collections.singletonList( + new Range((1L << 32) + 9, (1L << 32) + 9)))); + ManifestRowIdIndex.Selection hit = ManifestRowIdIndex.select(data, meta, one, settings); + assertThat(hit.blocks()).extracting(b -> b.firstRecord).containsExactly(4L); + + // A one-interval block needs no second intersection check after its envelope matches. + verify(one, times(3)).intersects(anyLong(), anyLong()); + } + + @Test + void malformedIntervalsStillFallbackAfterMinMaxRejectionOrAnEarlyHit() throws Exception { + byte[] data = golden(); + int firstBlockIntervals = 68 + 4 + header().length + 4 + 36; + // Make the second interval overlap the first, keeping the envelope unchanged. + ByteBuffer.wrap(data).putLong(firstBlockIntervals + 16, 9L); + byte[] hash = + MessageDigest.getInstance("SHA-256").digest(Arrays.copyOf(data, data.length - 32)); + System.arraycopy(hash, 0, data, data.length - 32, 32); + Files.write(temp.resolve("manifest-golden" + ManifestRowIdIndex.SUFFIX), data); + ManifestFileMeta meta = goldenMeta(); + + for (long point : new long[] {30, 0}) { + RowRangeIndex query = + RowRangeIndex.create(Collections.singletonList(new Range(point, point))); + assertThat( + ManifestRowIdIndex.read( + LocalFileIO.create(), + new Path(temp.toString(), "manifest-golden"), + meta, + query, + settings)) + .isNull(); + } + } + + @Test + void hugeRangesAreNotExpandedAndInvalidCoverageDisablesIndex() throws Exception { + byte[] header = header(); + ManifestRowIdIndex.Builder builder = new ManifestRowIdIndex.Builder(settings, header); + builder.beginBlock(header.length, 100, 2); + builder.add(0L, Long.MAX_VALUE); + builder.add(Long.MAX_VALUE, 1); + builder.endBlock(); + byte[] data = builder.serialize("m", header.length + 100, 2); + assertThat(data.length).isLessThan(512); + assertThat(select(data, meta("m", header.length + 100, 2), Long.MAX_VALUE).blocks()) + .hasSize(1); + for (Long first : Arrays.asList(null, -1L, Long.MAX_VALUE)) { + builder = new ManifestRowIdIndex.Builder(settings, header); + builder.beginBlock(header.length, 100, 1); + builder.add(first, 2); + builder.endBlock(); + assertThat(builder.serialize("m", header.length + 100, 1)).isNull(); + } + for (long count : new long[] {0, -1}) { + builder = new ManifestRowIdIndex.Builder(settings, header); + builder.beginBlock(header.length, 100, 1); + builder.add(0L, count); + assertThat(builder.serialize("m", 1, 1)).isNull(); + } + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_RANGES, 1); + builder = new ManifestRowIdIndex.Builder(new ManifestRowIdIndex.Settings(options), header); + builder.beginBlock(header.length, 100, 2); + builder.add(0L, 1); + builder.add(10L, 1); + assertThat(builder.serialize("m", 1, 2)).isNull(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_BYTES, 128); + builder = new ManifestRowIdIndex.Builder(new ManifestRowIdIndex.Settings(options), header); + assertThat(builder.serialize("m", 1, 2)).isNull(); + } + + @Test + void corruptMissingIncompleteAndMismatchedIndexesFallback() throws Exception { + Path manifest = new Path(temp.toString(), "manifest-golden"); + java.nio.file.Path index = temp.resolve("manifest-golden" + ManifestRowIdIndex.SUFFIX); + ManifestFileMeta meta = goldenMeta(); + RowRangeIndex query = RowRangeIndex.create(Collections.singletonList(new Range(11, 11))); + + assertThat(ManifestRowIdIndex.read(LocalFileIO.create(), manifest, meta, query, settings)) + .isNull(); + byte[] good = golden(); + for (int position : new int[] {0, 9, 11, 15, 16, 55, 63, 67, 75, good.length - 1}) { + byte[] bad = good.clone(); + bad[position] ^= 2; + Files.write(index, bad); + assertThat( + ManifestRowIdIndex.read( + LocalFileIO.create(), manifest, meta, query, settings)) + .isNull(); + } + // Self-consistent checksum cannot turn an unsupported or incomplete envelope into an index. + for (int position : new int[] {9, 11, 15}) { + byte[] bad = good.clone(); + bad[position] = 0; + byte[] hash = + MessageDigest.getInstance("SHA-256") + .digest(Arrays.copyOf(bad, bad.length - 32)); + System.arraycopy(hash, 0, bad, bad.length - 32, 32); + assertThatThrownBy(() -> ManifestRowIdIndex.select(bad, meta, query, settings)) + .isInstanceOf(IOException.class); + } + Files.write(index, Arrays.copyOf(good, good.length - 1)); + assertThat(ManifestRowIdIndex.read(LocalFileIO.create(), manifest, meta, query, settings)) + .isNull(); + Files.write(index, good); + assertThat( + ManifestRowIdIndex.read( + LocalFileIO.create(), manifest, meta, query, settings) + .blocks()) + .isEmpty(); + assertThatThrownBy( + () -> + ManifestRowIdIndex.select( + good, meta("other", meta.fileSize(), 7), query, settings)) + .isInstanceOf(IOException.class); + } + + @Test + void ioTimeoutFallsBackButInterruptionAndFatalErrorsPropagate() { + ManifestFileMeta manifest = meta("m", 1, 1); + RowRangeIndex query = RowRangeIndex.create(Collections.singletonList(new Range(1, 1))); + Path path = new Path(temp.toString(), "m"); + + LocalFileIO timedOut = + new LocalFileIO() { + @Override + public org.apache.paimon.fs.SeekableInputStream newInputStream(Path path) + throws IOException { + throw new java.net.SocketTimeoutException("timeout"); + } + }; + assertThat(ManifestRowIdIndex.read(timedOut, path, manifest, query, settings)).isNull(); + assertThat(Thread.currentThread().isInterrupted()).isFalse(); + LocalFileIO interrupted = + new LocalFileIO() { + @Override + public org.apache.paimon.fs.SeekableInputStream newInputStream(Path path) + throws IOException { + throw new java.io.InterruptedIOException("stop"); + } + }; + try { + assertThatThrownBy( + () -> + ManifestRowIdIndex.read( + interrupted, path, manifest, query, settings)) + .isInstanceOf(java.io.UncheckedIOException.class); + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + } finally { + Thread.interrupted(); + } + LocalFileIO failed = + new LocalFileIO() { + @Override + public org.apache.paimon.fs.SeekableInputStream newInputStream(Path path) { + throw new AssertionError("fatal"); + } + }; + assertThatThrownBy(() -> ManifestRowIdIndex.read(failed, path, manifest, query, settings)) + .isInstanceOf(AssertionError.class); + } + + @Test + void suppressedCancellationInterruptionAndFatalErrorsPropagate() { + ManifestFileMeta manifest = meta("m", 1, 1); + RowRangeIndex query = RowRangeIndex.create(Collections.singletonList(new Range(1, 1))); + for (Throwable closeFailure : + Arrays.asList( + new CancellationException("cancelled"), + new java.io.InterruptedIOException("interrupted"), + new AssertionError("fatal"))) { + AtomicBoolean closed = new AtomicBoolean(); + LocalFileIO fileIO = + new LocalFileIO() { + @Override + public org.apache.paimon.fs.SeekableInputStream newInputStream(Path path) { + return new ByteArraySeekableStream(new byte[0]) { + @Override + public int read(byte[] bytes, int offset, int length) + throws IOException { + throw new IOException("read failed"); + } + + @Override + public void close() throws IOException { + super.close(); + closed.set(true); + if (closeFailure instanceof IOException) { + throw (IOException) closeFailure; + } + if (closeFailure instanceof Error) { + throw (Error) closeFailure; + } + throw (RuntimeException) closeFailure; + } + }; + } + }; + try { + assertThatThrownBy( + () -> + ManifestRowIdIndex.read( + fileIO, + new Path(temp.toString(), "m"), + manifest, + query, + settings)) + .isInstanceOf( + closeFailure instanceof java.io.InterruptedIOException + ? java.io.UncheckedIOException.class + : closeFailure.getClass()); + assertThat(Thread.currentThread().isInterrupted()) + .isEqualTo(closeFailure instanceof java.io.InterruptedIOException); + assertThat(closed).isTrue(); + } finally { + Thread.interrupted(); + } + } + } + + @Test + void ordinaryExceptionCyclesFallBack() { + IOException first = new IOException("first"); + IOException second = new IOException("second"); + first.initCause(second); + second.addSuppressed(first); + LocalFileIO fileIO = + new LocalFileIO() { + @Override + public org.apache.paimon.fs.SeekableInputStream newInputStream(Path path) + throws IOException { + throw first; + } + }; + assertThat( + ManifestRowIdIndex.read( + fileIO, + new Path(temp.toString(), "m"), + meta("m", 1, 1), + RowRangeIndex.create(Collections.singletonList(new Range(1, 1))), + settings)) + .isNull(); + assertThat(Thread.currentThread().isInterrupted()).isFalse(); + } + + @Test + void indexReadsUseBoundedBulkRequests() throws Exception { + byte[] header = header(); + for (int blockCount : new int[] {5000, 25000}) { + ManifestRowIdIndex.Builder builder = new ManifestRowIdIndex.Builder(settings, header); + for (int blockNumber = 0; blockNumber < blockCount; blockNumber++) { + builder.beginBlock(header.length + blockNumber * 100L, 100, 1); + builder.add((long) blockNumber, 1); + builder.endBlock(); + } + long size = header.length + blockCount * 100L; + byte[] data = builder.serialize("manifest-large", size, blockCount); + ManifestFileMeta meta = meta("manifest-large", size, blockCount); + CountingInput stream = new CountingInput(data, Integer.MAX_VALUE); + Path path = new Path(temp.toString(), meta.fileName()); + FileIO io = mock(FileIO.class); + when(io.newInputStream(ManifestRowIdIndex.path(path))).thenReturn(stream); + ManifestRowIdIndex.Selection actual = + ManifestRowIdIndex.read( + io, + path, + meta, + RowRangeIndex.create(Collections.singletonList(new Range(0, 0))), + settings); + assertThat(actual.blocks()).hasSize(1); + assertThat(actual.blocks().get(0).offset).isEqualTo(header.length); + assertThat(stream.readLengths).hasSize((data.length + (1 << 20) - 1) / (1 << 20)); + assertThat(stream.requests).allMatch(request -> request <= 1 << 20); + assertThat(stream.closed).isTrue(); + } + } + + @Test + void indexShortReadsAndExactBudget() throws Exception { + byte[] data = golden(); + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_BYTES, data.length); + Path path = new Path(temp.toString(), "manifest-golden"); + for (int maxRead : new int[] {Integer.MAX_VALUE, 7}) { + CountingInput stream = new CountingInput(data, maxRead); + FileIO io = mock(FileIO.class); + when(io.newInputStream(ManifestRowIdIndex.path(path))).thenReturn(stream); + ManifestRowIdIndex.Selection actual = + ManifestRowIdIndex.read( + io, + path, + goldenMeta(), + RowRangeIndex.create(Collections.singletonList(new Range(20, 20))), + new ManifestRowIdIndex.Settings(options)); + assertThat(actual.blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 5L); + assertThat(stream.closed).isTrue(); + } + } + + @Test + void indexOverBudgetStopsAfterOneExtraByte() throws Exception { + Options options = new Options(); + options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_BYTES, 128); + Path path = new Path(temp.toString(), "manifest-golden"); + CountingInput stream = new CountingInput(golden(), Integer.MAX_VALUE); + FileIO io = mock(FileIO.class); + when(io.newInputStream(ManifestRowIdIndex.path(path))).thenReturn(stream); + assertThat( + ManifestRowIdIndex.read( + io, + path, + goldenMeta(), + RowRangeIndex.create(Collections.singletonList(new Range(20, 20))), + new ManifestRowIdIndex.Settings(options))) + .isNull(); + assertThat(stream.readLengths).containsExactly(129); + assertThat(stream.closed).isTrue(); + } + + @Test + void adjacentBlocksShareReadsForSingleByteConsumers() throws Exception { + byte[] header = header(); + byte[] body = new byte[400]; + for (int position = 0; position < body.length; position++) { + body[position] = (byte) position; + } + ManifestRowIdIndex.Selection selected = + ManifestRowIdIndex.select( + golden(), + goldenMeta(), + RowRangeIndex.create( + Arrays.asList( + new Range(0, 0), + new Range(8254058425445L, 8254058425445L))), + settings); + byte[] manifest = Arrays.copyOf(header, header.length + body.length); + System.arraycopy(body, 0, manifest, header.length, body.length); + CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); + Path path = new Path(temp.toString(), "manifest-golden"); + FileIO io = mock(FileIO.class); + when(io.newInputStream(path)).thenReturn(stream); + ByteArrayOutputStream actual = new ByteArrayOutputStream(); + try (InputStream input = ManifestRowIdIndex.openManifest(io, path, selected)) { + int value; + while ((value = input.read()) != -1) { + actual.write(value); + } + assertThat(input.read(new byte[1], 0, 0)).isZero(); + } + assertThat(actual.toByteArray()).isEqualTo(Arrays.copyOf(manifest, header.length + 300)); + assertThat(stream.readLengths).containsExactly(300); + assertThat(stream.seeks).containsExactly((long) header.length); + assertThat(stream.closed).isTrue(); + } + + @Test + void blockReadsSkipGapsAndEmptySelections() throws Exception { + byte[] header = header(); + byte[] manifest = Arrays.copyOf(header, header.length + 400); + Arrays.fill(manifest, header.length + 100, header.length + 300, (byte) 7); + Path path = new Path(temp.toString(), "manifest-golden"); + for (long point : new long[] {20, 16}) { + CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); + FileIO io = mock(FileIO.class); + when(io.newInputStream(path)).thenReturn(stream); + byte[] actual; + try (InputStream input = + ManifestRowIdIndex.openManifest( + io, path, select(golden(), goldenMeta(), point))) { + actual = IOUtils.readFully(input, false); + } + if (point == 20) { + assertThat(actual).isEqualTo(Arrays.copyOf(header, header.length + 200)); + assertThat(stream.readLengths).containsExactly(100, 100); + assertThat(stream.seeks) + .containsExactly((long) header.length, header.length + 300L); + } else { + assertThat(actual).isEqualTo(header); + assertThat(stream.readLengths).isEmpty(); + assertThat(stream.seeks).isEmpty(); + } + assertThat(stream.closed).isTrue(); + } + } + + @Test + void largeBlockSpansUseBoundedReads() throws Exception { + byte[] header = header(); + ManifestRowIdIndex.Builder builder = new ManifestRowIdIndex.Builder(settings, header); + long offset = header.length; + for (int length : new int[] {512 * 1024, 512 * 1024, 257}) { + builder.beginBlock(offset, length, 1); + builder.add(20L, 1); + builder.endBlock(); + offset += length; + } + byte[] data = builder.serialize("manifest-large", offset, 3); + byte[] manifest = Arrays.copyOf(header, (int) offset); + CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); + FileIO io = mock(FileIO.class); + Path path = new Path(temp.toString(), "manifest-large"); + when(io.newInputStream(path)).thenReturn(stream); + try (InputStream input = + ManifestRowIdIndex.openManifest( + io, path, select(data, meta("manifest-large", offset, 3), 20))) { + assertThat(IOUtils.readFully(input, false)).isEqualTo(manifest); + } + assertThat(stream.readLengths).containsExactly(1 << 20, 257); + assertThat(stream.seeks).containsExactly((long) header.length); + assertThat(stream.closed).isTrue(); + } + + @Test + void blockShortReadsAndTruncation() throws Exception { + byte[] header = header(); + Path path = new Path(temp.toString(), "manifest-golden"); + ManifestRowIdIndex.Selection selected = + ManifestRowIdIndex.select( + golden(), + goldenMeta(), + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE))), + settings); + for (int bodyLength : new int[] {400, 399}) { + byte[] manifest = Arrays.copyOf(header, header.length + bodyLength); + CountingInput stream = new CountingInput(manifest, 7); + FileIO io = mock(FileIO.class); + when(io.newInputStream(path)).thenReturn(stream); + try (InputStream input = ManifestRowIdIndex.openManifest(io, path, selected)) { + if (bodyLength == 400) { + assertThat(IOUtils.readFully(input, false)).isEqualTo(manifest); + } else { + assertThatThrownBy(() -> IOUtils.readFully(input, false)) + .isInstanceOf(EOFException.class); + } + } + assertThat(stream.closed).isTrue(); + } + } + + private static class CountingInput extends ByteArraySeekableStream { + private final int maxRead; + private final List requests = new ArrayList<>(); + private final List readLengths = new ArrayList<>(); + private final List seeks = new ArrayList<>(); + private boolean closed; + + private CountingInput(byte[] data, int maxRead) { + super(data); + this.maxRead = maxRead; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + requests.add(length); + int count = super.read(bytes, offset, Math.min(length, maxRead)); + if (count > 0) { + readLengths.add(count); + } + return count; + } + + @Override + public void seek(long position) throws IOException { + seeks.add(position); + super.seek(position); + } + + @Override + public void close() throws IOException { + closed = true; + super.close(); + } + } + + private ManifestRowIdIndex.Selection select(byte[] data, ManifestFileMeta meta, long point) + throws IOException { + return ManifestRowIdIndex.select( + data, + meta, + RowRangeIndex.create(Collections.singletonList(new Range(point, point))), + settings); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java index 8dfcb1c21ab9..e86f602bcd39 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java @@ -38,8 +38,11 @@ import org.apache.paimon.manifest.FileSource; import org.apache.paimon.manifest.ManifestEntry; import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestIndexTestUtils; +import org.apache.paimon.manifest.ManifestRowIdIndex; import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction; import org.apache.paimon.options.ExpireConfig; +import org.apache.paimon.options.MemorySize; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaManager; @@ -853,6 +856,75 @@ public void testExpirePlansManifestsConcurrentlyWithSkippingSet() throws Excepti store.assertCleaned(); } + @Test + void testSidecarsFollowSnapshotAndTagRetention() throws Exception { + store.options().toConfiguration().set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 2); + store.options() + .toConfiguration() + .set(CoreOptions.MANIFEST_TARGET_FILE_SIZE, MemorySize.parse("8 mb")); + List allData = new ArrayList<>(); + List snapshotPositions = new ArrayList<>(); + commit(8, allData, snapshotPositions); + int latest = requireNonNull(snapshotManager.latestSnapshotId()).intValue(); + Set manifests = new HashSet<>(); + Set retainedManifests = new HashSet<>(); + for (int snapshotId = 1; snapshotId <= latest; snapshotId++) { + rewriteSnapshotTime(snapshotId, 0); + ManifestIndexTestUtils.registerIndexReferences(store, snapshotId); + snapshotManager.invalidateCache(); + for (ManifestFileMeta meta : + store.manifestListFactory() + .create() + .readDataManifests(snapshotManager.snapshot(snapshotId))) { + Path manifest = store.pathFactory().toManifestFilePath(meta.fileName()); + manifests.add(manifest); + if (snapshotId == 3 || snapshotId == latest) { + retainedManifests.add(manifest); + } + } + } + store.newTagManager() + .createTag( + snapshotManager.snapshot(3), + "keep-sidecars", + store.options().tagDefaultTimeRetained(), + Collections.emptyList(), + false); + Set expiredManifests = new HashSet<>(manifests); + expiredManifests.removeAll(retainedManifests); + assertThat(expiredManifests).isNotEmpty(); + ExpireSnapshotsImpl expire = + (ExpireSnapshotsImpl) store.newExpire(expireAllButLatestConfig()); + expire.setCurrentTimeMillis(() -> 1000L); + expire.expire(); + for (Path manifest : manifests) { + boolean retained = retainedManifests.contains(manifest); + assertThat(fileIO.exists(manifest)).as("manifest %s", manifest).isEqualTo(retained); + assertThat( + fileIO.exists( + new Path( + manifest.getParent(), + "index-for-" + + manifest.getName() + + ManifestRowIdIndex.SUFFIX))) + .as("sidecar for %s", manifest) + .isEqualTo(retained); + } + for (ManifestFileMeta meta : + store.manifestListFactory() + .create() + .readDataManifests( + store.newTagManager() + .getOrThrow("keep-sidecars") + .trimToSnapshot())) { + assertThat( + fileIO.exists( + store.pathFactory() + .toManifestFilePath(ManifestRowIdIndex.fileName(meta)))) + .isTrue(); + } + } + @Test public void testExpireWithTagsAndConcurrentPlanningKeepsTaggedSnapshotsReadable() throws Exception { diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java index 36499bb4a632..005d8ec222d8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java @@ -32,7 +32,9 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestIndexTestUtils; import org.apache.paimon.manifest.ManifestList; +import org.apache.paimon.manifest.ManifestRowIdIndex; import org.apache.paimon.mergetree.compact.ConcatRecordReader; import org.apache.paimon.options.Options; import org.apache.paimon.reader.ReaderSupplier; @@ -158,6 +160,39 @@ public void testNormallyRemoving() throws Throwable { normallyRemoving(tablePath); } + @Test + void testOrphanCleanupProtectsReferencedSidecars() throws Exception { + commit(Collections.singletonList(TestPojo.next())); + ManifestIndexTestUtils.registerIndexReferences( + table.store(), table.snapshotManager().latestSnapshotId()); + table.snapshotManager().invalidateCache(); + table.createTag("sidecar-tag", table.snapshotManager().latestSnapshotId()); + List sidecars = new ArrayList<>(); + List unreferenced = new ArrayList<>(); + for (ManifestFileMeta meta : + table.store() + .manifestListFactory() + .create() + .readDataManifests(table.snapshotManager().latestSnapshot())) { + Path sidecar = new Path(manifestDir, ManifestRowIdIndex.fileName(meta)); + sidecars.add(sidecar); + Path guessed = new Path(manifestDir, meta.fileName() + ManifestRowIdIndex.SUFFIX); + fileIO.newOutputStream(guessed, false).close(); + unreferenced.add(guessed); + } + Path orphan = new Path(manifestDir, "manifest-orphan" + ManifestRowIdIndex.SUFFIX); + fileIO.newOutputStream(orphan, false).close(); + new LocalOrphanFilesClean(table, System.currentTimeMillis() + 2000).clean(); + assertThat(fileIO.exists(orphan)).isFalse(); + assertThat(sidecars).isNotEmpty(); + for (Path sidecar : sidecars) { + assertThat(fileIO.exists(sidecar)).isTrue(); + } + for (Path guessed : unreferenced) { + assertThat(fileIO.exists(guessed)).isFalse(); + } + } + @Test public void testKeepManagedBlobPack() throws Exception { commit(Collections.singletonList(TestPojo.next())); diff --git a/paimon-core/src/test/resources/manifest-row-id-index-v2.txt b/paimon-core/src/test/resources/manifest-row-id-index-v2.txt new file mode 100644 index 000000000000..c58cf599b276 --- /dev/null +++ b/paimon-core/src/test/resources/manifest-row-id-index-v2.txt @@ -0,0 +1,19 @@ +# 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 +index=UEFJTVJJRFgAAgACAAAAAS9Hlrm5B3S+oqDXSD494xrafUwJPN8G7QLJnPsSVcDPAAAAAAAAAckAAAAAAAAABwAAAQ0AAAA5T2JqAQQUYXZyby5jb2RlYwhudWxsFmF2cm8uc2NoZW1hDCJsb25nIgAAAAAAAAAAAAAAAAAAAAAAAAAAAwAAAAAAAAA5AAAAAAAAAGQAAAAAAAAAAAAAAAAAAAADAAAAAgAAAAAAAAAAAAAAAAAAAAkAAAAAAAAAFAAAAAAAAAAYAAAAAAAAAJ0AAAAAAAAAyAAAAAAAAAADAAAAAAAAAAIAAAACAAAAAP////4AAAABAAAAAgAAB4HMOGxlAAAHgcw4bGUAAAAAAAABZQAAAAAAAABkAAAAAAAAAAUAAAAAAAAAAgAAAAIAAAAAAAAAFAAAAAAAAAAYf/////////9//////////xekCXjgYJlWPiPlQ80IyyKSmIPH5z5iMhyRa8BhBph5 diff --git a/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java b/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java index 43a68c54a44d..382f55f93d60 100644 --- a/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java +++ b/paimon-format/src/main/java/org/apache/avro/file/RawBlockReader.java @@ -22,27 +22,107 @@ import org.apache.avro.io.DatumReader; import org.apache.avro.io.Decoder; +import java.io.ByteArrayOutputStream; +import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; +import java.util.Arrays; /** Package bridge exposing Avro's compressed blocks without reflection. */ public final class RawBlockReader extends DataFileStream { + private final CountingInput input; + private final byte[] headerBytes; + private long blockOffset; + private long blockLength; + private boolean pending; + public RawBlockReader(InputStream input) throws IOException { + this(new CountingInput(input)); + } + + private RawBlockReader(CountingInput input) throws IOException { super(input, new NoOpDatumReader()); + this.input = input; + long length = position(); + this.headerBytes = + length <= CountingInput.MAX_HEADER + ? Arrays.copyOf(input.prefix.toByteArray(), (int) length) + : null; + input.prefix = null; + } + + public byte[] headerBytes() { + return headerBytes == null ? null : headerBytes.clone(); + } + + public long blockOffset() { + return blockOffset; + } + + public long blockLength() { + return blockLength; } - public boolean hasNextRawBlock() { - return super.hasNextBlock(); + private long position() throws IOException { + // This is the same read-ahead adjustment used by DataFileReader.blockFinished(). + return input.position - vin.inputStream().available(); + } + + public boolean hasNextRawBlock() throws IOException { + if (!pending) { + blockOffset = position(); + pending = super.hasNextBlock(); + } + return pending; } public RawBlock nextRawBlock(RawBlock reuse) throws IOException { + if (!hasNextRawBlock()) { + throw new java.util.NoSuchElementException(); + } DataBlock raw = super.nextRawBlock(reuse == null ? null : reuse.dataBlock()); + blockLength = position() - blockOffset; + pending = false; return reuse == null ? new RawBlock(raw, resolveCodec(), getSchema()) : reuse.replace(raw, resolveCodec(), getSchema()); } + private static final class CountingInput extends FilterInputStream { + private static final int MAX_HEADER = 1024 * 1024; + private long position; + private ByteArrayOutputStream prefix = new ByteArrayOutputStream(); + + private CountingInput(InputStream input) { + super(input); + } + + @Override + public int read() throws IOException { + int value = in.read(); + if (value >= 0) { + position++; + if (prefix != null && prefix.size() < MAX_HEADER) { + prefix.write(value); + } + } + return value; + } + + @Override + public int read(byte[] bytes, int offset, int length) throws IOException { + int n = in.read(bytes, offset, length); + if (n > 0) { + position += n; + if (prefix != null && prefix.size() < MAX_HEADER) { + prefix.write(bytes, offset, Math.min(n, MAX_HEADER - prefix.size())); + } + } + return n; + } + } + private static final class NoOpDatumReader implements DatumReader { @Override diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java index c4359afcda43..eeca62bb18c8 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroBlockReader.java @@ -54,6 +54,19 @@ public AvroBlockReader(InputStream input) throws IOException { } } + @Nullable + public byte[] headerBytes() { + return reader.headerBytes(); + } + + public long blockOffset() { + return reader.blockOffset(); + } + + public long blockLength() { + return reader.blockLength(); + } + /** Creates a record decoder from the writer schema stored in the Avro file header. */ public AvroRecordDecoder createRecordDecoder() { return new AvroRecordDecoder(reader.getSchema()); diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 903697a94ff2..3d500fd7a5a7 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -297,6 +297,30 @@ class CoreOptions: .with_description("The parallelism for scanning manifest files.") ) + MANIFEST_ROW_ID_INDEX_WRITE: ConfigOption[bool] = ( + ConfigOptions.key("manifest.row-id-index.write") + .boolean_type() + .default_value(False) + ) + + MANIFEST_ROW_ID_INDEX_READ: ConfigOption[bool] = ( + ConfigOptions.key("manifest.row-id-index.read") + .boolean_type() + .default_value(False) + ) + + MANIFEST_ROW_ID_INDEX_MAX_RANGES: ConfigOption[int] = ( + ConfigOptions.key("manifest.row-id-index.max-ranges") + .int_type() + .default_value(131072) + ) + + MANIFEST_ROW_ID_INDEX_MAX_BYTES: ConfigOption[int] = ( + ConfigOptions.key("manifest.row-id-index.max-bytes") + .int_type() + .default_value(8388608) + ) + MANIFEST_COMPRESSION: ConfigOption[str] = ( ConfigOptions.key("manifest.compression") .string_type() diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py index a3ac5cfe7434..9ec5a1ef8ba7 100644 --- a/paimon-python/pypaimon/manifest/manifest_file_manager.py +++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py @@ -19,6 +19,10 @@ from io import BytesIO from typing import Callable, List, Optional +from pypaimon.manifest.row_id_index import ( + Settings, SUFFIX, Query, build_from_entries, read_index, read_selected_bytes, +) + import fastavro from datetime import datetime @@ -57,13 +61,25 @@ 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) + settings = Settings.from_options(self.table.options) + query = Query(row_ranges) if settings.read 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 query is not None: + selected = read_index(self.file_io, path, manifest_file, query, settings) + 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 ( @@ -97,6 +113,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. @@ -110,8 +127,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) reader = fastavro.reader(buffer) @@ -244,7 +264,7 @@ def write(self, file_name, entries: List[ManifestEntry]): 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, @@ -268,10 +288,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( @@ -282,13 +301,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 @@ -335,17 +353,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}" + index_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 = Settings.from_options(self.table.options) + if settings.write: + data = build_from_entries(avro_bytes, entries, file_name, settings) + if data is not None: + index_file_name = file_name + SUFFIX + with self.file_io.new_output_stream(f"{self.manifest_path}/{index_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), + [index_file_name] if index_file_name is not None else None) + except BaseException as e: self.file_io.delete_quietly(manifest_path) + if index_file_name is not None: + self.file_io.delete_quietly(f"{self.manifest_path}/{index_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 @@ -370,7 +408,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 @@ -402,4 +442,5 @@ def _build_meta(self, file_name: str, entries: List[ManifestEntry], schema_id=schema_id, min_row_id=min_row_id, max_row_id=max_row_id, + extra_files=extra_files, ) 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/manifest/row_id_index.py b/paimon-python/pypaimon/manifest/row_id_index.py new file mode 100644 index 000000000000..3a3bba51ee41 --- /dev/null +++ b/paimon-python/pypaimon/manifest/row_id_index.py @@ -0,0 +1,330 @@ +# 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. + +"""Complete row-id interval unions with Avro block offsets and entry ordinals. + +Version 2 uses fixed-width big-endian integers; no library-specific bitmap encoding. +""" + +import hashlib +import logging +import struct +from bisect import bisect_left +from concurrent.futures import CancelledError +from dataclasses import dataclass +from io import BytesIO +from typing import Tuple + +from pyarrow import ArrowCancelled + +from pypaimon.common.options.core_options import CoreOptions +from pypaimon.utils.range import Range + +LOG = logging.getLogger(__name__) +SUFFIX = '.row-id-index' +MAGIC = b'PAIMRIDX' +MAX_ROW_ID = (1 << 63) - 1 +MAX_AVRO_HEADER = 1024 * 1024 +READ_BUFFER_BYTES = 1024 * 1024 +HEADER = struct.Struct('>8sHHI32sqqI') +BLOCK = struct.Struct('>qqqqI') +PAIR = struct.Struct('>qq') +LONG = struct.Struct('>q') +_PROPAGATED_ERRORS = (InterruptedError, CancelledError, ArrowCancelled, MemoryError, RecursionError) + + +@dataclass +class Settings: + write: bool = False + read: bool = False + max_ranges: int = 131072 + max_bytes: int = 8 * 1024 * 1024 + + def __post_init__(self): + if not 1 <= self.max_ranges <= 1048576: + raise ValueError('manifest.row-id-index.max-ranges must be in [1, 1048576]') + if not 128 <= self.max_bytes <= 64 * 1024 * 1024: + raise ValueError('manifest.row-id-index.max-bytes must be in [128, 67108864]') + + @classmethod + def from_options(cls, options): + return cls( + options.options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE), + options.options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_READ), + options.options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_RANGES), + options.options.get(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_BYTES)) + + +@dataclass(frozen=True) +class Block: + offset: int + length: int + first_record: int + record_count: int + + +@dataclass(frozen=True) +class Selection: + header: bytes + blocks: Tuple[Block, ...] + + +class Query: + def __init__(self, ranges): + normalized = Range.sort_and_merge_overlap(list(ranges), True) + self.starts = [r.from_ for r in normalized] + self.ends = [r.to for r in normalized] + + def intersects(self, first, last): + candidate = bisect_left(self.ends, first) + return candidate < len(self.starts) and self.starts[candidate] <= last + + +class Builder: + def __init__(self, settings, header): + self.settings = settings + self.complete = (header is not None and len(header) <= MAX_AVRO_HEADER + and len(header) + HEADER.size + 40 <= settings.max_bytes) + self.payload = bytearray() + self.ranges = [] + self.count_position = 4 + len(header) if self.complete else 0 + self.next_offset = len(header) if self.complete else 0 + self.next_record = 0 + self.range_count = 0 + self.blocks = 0 + self.current = None + self.entries_in_block = 0 + if self.complete: + self.payload.extend(struct.pack('>I', len(header))) + self.payload.extend(header) + self.payload.extend(struct.pack('>I', 0)) + + def _disable(self, reason): + self.complete = False + self.payload.clear() + self.ranges.clear() + LOG.debug('Omitting manifest row-id block index: %s', reason) + + def begin_block(self, offset, length, records): + if not self.complete: + return + _require(self.current is None and offset == self.next_offset and length > 0 and records > 0) + self.current = Block(offset, length, self.next_record, records) + self.entries_in_block = 0 + + def add(self, first, count): + if not self.complete: + return + _require(self.current is not None) + self.entries_in_block += 1 + if (first is None or first < 0 or count <= 0 + or first > MAX_ROW_ID or count - 1 > MAX_ROW_ID - first): + self._disable('unknown or invalid row-id coverage') + return + end = first + count - 1 + left = bisect_left(self.ranges, (first, -1)) + if left and self.ranges[left - 1][1] >= first - 1: + left -= 1 + right = left + while right < len(self.ranges) and self.ranges[right][0] <= end + 1: + first = min(first, self.ranges[right][0]) + end = max(end, self.ranges[right][1]) + right += 1 + if self.range_count + len(self.ranges) - (right - left) >= self.settings.max_ranges: + self._disable('range budget exceeded') + return + self.ranges[left:right] = [(first, end)] + + def end_block(self): + if not self.complete: + return + block = self.current + _require(block is not None and self.entries_in_block == block.record_count and self.ranges) + if HEADER.size + 32 + len(self.payload) + BLOCK.size + 16 * len(self.ranges) > self.settings.max_bytes: + self._disable('serialized byte budget exceeded') + return + self.payload.extend(BLOCK.pack(block.offset, block.length, block.first_record, + block.record_count, len(self.ranges))) + for first, end in self.ranges: + self.payload.extend(PAIR.pack(first, end)) + self.next_offset = block.offset + block.length + self.next_record = block.first_record + block.record_count + self.range_count += len(self.ranges) + self.blocks += 1 + self.ranges.clear() + self.current = None + + def serialize(self, name, file_size, entry_count): + if not self.complete: + return None + _require(self.current is None and self.next_offset == file_size and self.next_record == entry_count) + struct.pack_into('>I', self.payload, self.count_position, self.blocks) + header = HEADER.pack(MAGIC, 2, 2, 1, hashlib.sha256(name.encode('utf-8')).digest(), + file_size, entry_count, len(self.payload)) + data = header + self.payload + return data + hashlib.sha256(data).digest() + + +def build_from_entries(avro_bytes, entries, name, settings): + import fastavro + blocks = iter(fastavro.block_reader(BytesIO(avro_bytes))) + first_block = next(blocks, None) + header = avro_bytes[:first_block.offset] if first_block else avro_bytes + builder = Builder(settings, header) + position = 0 + block = first_block + while block is not None and builder.complete: + builder.begin_block(block.offset, block.size, block.num_records) + end = position + block.num_records + _require(end <= len(entries)) + for i in range(position, end): + entry = entries[i] + builder.add(entry.file.first_row_id, entry.file.row_count) + if not builder.complete: + break + builder.end_block() + position = end + block = next(blocks, None) if builder.complete else None + return builder.serialize(name, len(avro_bytes), len(entries)) + + +def _require(condition): + if not condition: + raise ValueError('Invalid, unsupported, mismatched or over-budget manifest row-id block index') + + +def select(data, manifest, query, settings): + if not isinstance(query, Query): + query = Query(query) + _require(128 <= len(data) <= settings.max_bytes) + _require(hashlib.sha256(data[:-32]).digest() == data[-32:]) + magic, version, codec, flags, name_hash, size, entries, length = HEADER.unpack_from(data) + _require((magic, version, codec, flags) == (MAGIC, 2, 2, 1)) + _require(name_hash == hashlib.sha256(manifest.file_name.encode('utf-8')).digest()) + _require(size == manifest.file_size and entries == manifest.num_added_files + manifest.num_deleted_files) + _require(length == len(data) - HEADER.size - 32) + offset = HEADER.size + header_length, = struct.unpack_from('>I', data, offset) + offset += 4 + _require(21 <= header_length <= MAX_AVRO_HEADER and header_length <= len(data) - offset - 36) + header = bytes(data[offset:offset + header_length]) + _require(header[:4] == b'Obj\x01') + offset += header_length + blocks, = struct.unpack_from('>I', data, offset) + offset += 4 + _require(blocks <= (len(data) - 32 - offset) // 52) + next_offset = header_length + next_record = 0 + total_ranges = 0 + selected = [] + for _ in range(blocks): + file_offset, block_length, first, count, ranges = BLOCK.unpack_from(data, offset) + offset += BLOCK.size + _require(file_offset == next_offset and 0 < block_length <= size - file_offset) + _require(first == next_record and 0 < count <= entries - first) + _require(0 < ranges <= settings.max_ranges - total_ranges and ranges <= (len(data) - 32 - offset) // 16) + total_ranges += ranges + ranges_end = offset + PAIR.size * ranges + min_row_id, first_end = PAIR.unpack_from(data, offset) + offset += PAIR.size + # The sorted interval list already contains min/max; no format change or extra fields. + max_row_id = first_end if ranges == 1 else LONG.unpack_from(data, ranges_end - LONG.size)[0] + _require(min_row_id >= 0 and first_end >= min_row_id and max_row_id >= first_end) + candidate = query.intersects(min_row_id, max_row_id) + hit = candidate and (ranges == 1 or query.intersects(min_row_id, first_end)) + previous_end = first_end + for _ in range(1, ranges): + start, end = PAIR.unpack_from(data, offset) + offset += PAIR.size + # Retain validation even when min/max rejects the block or an earlier interval hit. + _require(start >= 0 and end >= start and start > previous_end) + previous_end = end + if candidate and not hit: + hit = query.intersects(start, end) + if hit: + selected.append(Block(file_offset, block_length, first, count)) + next_offset = file_offset + block_length + next_record = first + count + _require(offset == len(data) - 32 and next_offset == size and next_record == entries) + return Selection(header, tuple(selected)) + + +def index_file_name(manifest): + return next((name for name in manifest.extra_files or [] if name.endswith(SUFFIX)), None) + + +def read_index(file_io, manifest_path, manifest, query, settings): + name = index_file_name(manifest) + if name is None: + return None + index_path = manifest_path.rsplit('/', 1)[0] + '/' + name + try: + with file_io.new_input_stream(index_path) as stream: + data = bytearray() + while True: + chunk = stream.read(min(READ_BUFFER_BYTES, settings.max_bytes + 1 - len(data))) + if not chunk: + break + data.extend(chunk) + _require(len(data) <= settings.max_bytes) + return select(data, manifest, query, settings) + except _PROPAGATED_ERRORS: + raise + except Exception as error: + pending = [error] + visited = set() + while pending: + cause = pending.pop() + if id(cause) in visited: + continue + visited.add(id(cause)) + if not isinstance(cause, Exception) or isinstance(cause, _PROPAGATED_ERRORS): + raise cause + if cause.__cause__ is not None: + pending.append(cause.__cause__) + if cause.__context__ is not None: + pending.append(cause.__context__) + LOG.debug('Cannot use row-id block index for %s; reading manifest: %s', manifest_path, error) + return None + + +def read_selected_bytes(file_io, manifest_path, selected): + """Read complete selected blocks with seek; adjacent blocks share one contiguous span. + + The concatenated original header and blocks form a valid Avro OCF. Partial entries + must not be stored in a cache keyed by the complete manifest. + """ + data = bytearray(selected.header) + with file_io.new_input_stream(manifest_path) as stream: + block_position = 0 + while block_position < len(selected.blocks): + block = selected.blocks[block_position] + block_position += 1 + end = block.offset + block.length + while (block_position < len(selected.blocks) + and selected.blocks[block_position].offset == end): + end += selected.blocks[block_position].length + block_position += 1 + stream.seek(block.offset) + remaining = end - block.offset + while remaining: + chunk = stream.read(min(remaining, READ_BUFFER_BYTES)) + if not chunk: + raise EOFError('Truncated manifest block') + data.extend(chunk) + remaining -= len(chunk) + return bytes(data) diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py b/paimon-python/pypaimon/read/scanner/file_scanner.py index c114fd656e1c..477d0f2c0772 100755 --- a/paimon-python/pypaimon/read/scanner/file_scanner.py +++ b/paimon-python/pypaimon/read/scanner/file_scanner.py @@ -597,6 +597,7 @@ def read_manifest_entries(self, manifest_files: List[ManifestFileMeta], early_entry_filter=self._build_early_bucket_filter(), early_record_filter=early_row_filter, partition_filter=partition_filter, + row_ranges=row_ranges, ) def _build_early_bucket_filter(self): diff --git a/paimon-python/pypaimon/tests/manifest/row_id_index_test.py b/paimon-python/pypaimon/tests/manifest/row_id_index_test.py new file mode 100644 index 000000000000..3bf398a471c5 --- /dev/null +++ b/paimon-python/pypaimon/tests/manifest/row_id_index_test.py @@ -0,0 +1,582 @@ +# 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. + +import base64 +import hashlib +import os +import struct +import unittest +from concurrent.futures import CancelledError +from copy import deepcopy +from io import BytesIO + +import fastavro +from pyarrow import ArrowCancelled +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from pypaimon.common.options.core_options import CoreOptions +from pypaimon.globalindex.global_index_result import GlobalIndexResult +from pypaimon.manifest.row_id_index import ( + Block, Builder, Selection, Settings, SUFFIX, MAX_ROW_ID, Query, select, read_index, + read_selected_bytes, index_file_name, +) +from pypaimon.manifest.schema.manifest_entry import ManifestEntry +from pypaimon.manifest.manifest_list_manager import ManifestListManager +from pypaimon.manifest.schema.manifest_file_meta import MANIFEST_FILE_META_SCHEMA +from pypaimon.read.scanner.file_scanner import FileScanner +from pypaimon.tests.manifest import manifest_entry_identifier_test as existing +from pypaimon.utils.range import Range + + +def fixture(): + path = (Path(__file__).resolve().parents[4] / 'paimon-core/src/test/resources' / + 'manifest-row-id-index-v2.txt') + return dict(line.split('=', 1) for line in path.read_text().splitlines() + if line.startswith(('index=', 'avroHeader='))) + + +def golden(): + return base64.b64decode(fixture()['index']) + + +def avro_header(): + return base64.b64decode(fixture()['avroHeader']) + + +def golden_meta(): + return SimpleNamespace(file_name='manifest-golden', file_size=len(avro_header()) + 400, + num_added_files=7, num_deleted_files=0) + + +def intersects(data, meta, ranges, settings): + return bool(select(data, meta, ranges, settings).blocks) + + +class CountingInput(BytesIO): + def __init__(self, data, max_read=None): + super().__init__(data) + self.max_read = max_read + self.reads = [] + self.requests = [] + self.seeks = [] + + def read(self, size=-1): + if size < 0: + raise AssertionError('Unbounded read') + self.requests.append(size) + position = self.tell() + data = super().read(size if self.max_read is None else min(size, self.max_read)) + if data: + self.reads.append((position, len(data))) + return data + + def seek(self, offset, whence=0): + self.seeks.append(offset) + return super().seek(offset, whence) + + +class FailingIndexInput(BytesIO): + def __init__(self, data, failure, phase, close_failure=None): + super().__init__(data) + self.failure = failure + self.phase = phase + self.close_failure = close_failure + + def read(self, size=-1): + if self.phase == 'read': + raise self.failure + return super().read(size) + + def close(self): + was_closed = self.closed + super().close() + if not was_closed and self.close_failure is not None: + raise self.close_failure + if self.phase == 'close' and not was_closed: + raise self.failure + + +class RowIdIndexReadTest(unittest.TestCase): + def test_index_reads_use_bounded_bulk_requests(self): + header = avro_header() + for block_count in (5000, 25000): + with self.subTest(block_count=block_count): + builder = Builder(Settings(), header) + for block_number in range(block_count): + builder.begin_block(len(header) + block_number * 100, 100, 1) + builder.add(block_number, 1) + builder.end_block() + size = len(header) + block_count * 100 + data = builder.serialize('manifest-large', size, block_count) + meta = SimpleNamespace(file_name='manifest-large', file_size=size, + num_added_files=block_count, num_deleted_files=0, + extra_files=['manifest-large' + SUFFIX]) + stream = CountingInput(data) + file_io = SimpleNamespace(new_input_stream=lambda path: stream) + actual = read_index(file_io, '/manifest/manifest-large', meta, + [Range(0, 0)], Settings()) + self.assertEqual(actual, select(data, meta, [Range(0, 0)], Settings())) + self.assertEqual(len(stream.reads), (len(data) + (1 << 20) - 1) // (1 << 20)) + self.assertLessEqual(max(stream.requests), 1 << 20) + self.assertTrue(stream.closed) + + def test_index_short_reads_and_exact_budget(self): + data, meta = golden(), golden_meta() + meta.extra_files = [meta.file_name + SUFFIX] + for max_read in (None, 7): + with self.subTest(max_read=max_read): + stream = CountingInput(data, max_read) + file_io = SimpleNamespace(new_input_stream=lambda path: stream) + settings = Settings(max_bytes=len(data)) + actual = read_index(file_io, '/manifest/manifest-golden', meta, + [Range(20, 20)], settings) + self.assertEqual(actual, select(data, meta, [Range(20, 20)], settings)) + self.assertTrue(stream.closed) + + def test_index_over_budget_stops_after_one_extra_byte(self): + data, meta = golden(), golden_meta() + meta.extra_files = [meta.file_name + SUFFIX] + stream = CountingInput(data) + file_io = SimpleNamespace(new_input_stream=lambda path: stream) + self.assertIsNone(read_index(file_io, '/manifest/manifest-golden', meta, + [Range(20, 20)], Settings(max_bytes=128))) + self.assertEqual(stream.reads, [(0, 129)]) + self.assertTrue(stream.closed) + + def test_adjacent_blocks_share_reads_without_reading_gaps(self): + header = avro_header() + body = bytes(range(200)) * 2 + for points, spans in [([0, 8254058425445], [(0, 300)]), + ([20], [(0, 100), (300, 100)]), ([16], [])]: + with self.subTest(points=points): + selected = select(golden(), golden_meta(), + [Range(point, point) for point in points], Settings()) + stream = CountingInput(header + body) + file_io = SimpleNamespace(new_input_stream=lambda path: stream) + actual = read_selected_bytes(file_io, '/manifest/manifest-golden', selected) + expected = header + b''.join(body[start:start + size] for start, size in spans) + self.assertEqual(actual, expected) + self.assertEqual(stream.reads, [(len(header) + start, size) for start, size in spans]) + self.assertEqual(stream.seeks, [len(header) + start for start, _ in spans]) + self.assertTrue(stream.closed) + + def test_large_block_spans_use_bounded_reads(self): + header = avro_header() + block_size = 512 * 1024 + body = bytes(2 * block_size + 257) + selected = Selection(header, (Block(len(header), block_size, 0, 1), + Block(len(header) + block_size, block_size, 1, 1), + Block(len(header) + 2 * block_size, 257, 2, 1))) + stream = CountingInput(header + body) + file_io = SimpleNamespace(new_input_stream=lambda path: stream) + self.assertEqual(read_selected_bytes(file_io, '/manifest/manifest-large', selected), header + body) + self.assertEqual(stream.reads, [(len(header), 1 << 20), (len(header) + (1 << 20), 257)]) + self.assertEqual(stream.seeks, [len(header)]) + self.assertTrue(stream.closed) + + def test_block_short_reads_and_truncation(self): + header = avro_header() + body = bytes(range(200)) * 2 + selected = select(golden(), golden_meta(), [Range(0, MAX_ROW_ID)], Settings()) + stream = CountingInput(header + body, 7) + file_io = SimpleNamespace(new_input_stream=lambda path: stream) + self.assertEqual(read_selected_bytes(file_io, '/manifest/manifest-golden', selected), header + body) + self.assertTrue(stream.closed) + stream = CountingInput(header + body[:-1], 7) + with self.assertRaises(EOFError): + read_selected_bytes(file_io, '/manifest/manifest-golden', selected) + self.assertTrue(stream.closed) + + +class RowIdIndexFormatTest(unittest.TestCase): + def test_cross_language_and_block_ordinals(self): + data, meta, header = golden(), golden_meta(), avro_header() + for point in (0, 9, 20, 24, (1 << 32) - 2, 1 << 32, (1 << 32) + 2, + 8254058425445, MAX_ROW_ID): + self.assertTrue(intersects(data, meta, [Range(point, point)], Settings())) + for point in (10, 19, 25, (1 << 32) - 3, (1 << 32) + 3, 8254058425444, MAX_ROW_ID - 1): + self.assertFalse(intersects(data, meta, [Range(point, point)], Settings())) + selected = select(data, meta, [Range(20, 20)], Settings()) + self.assertEqual([b.first_record for b in selected.blocks], [0, 5]) + self.assertEqual([b.offset for b in selected.blocks], [len(header), len(header) + 300]) + self.assertEqual([b.length for b in selected.blocks], [100, 100]) + + gap = select(data, meta, [Range(16, 16)], Settings()) + + self.assertFalse(gap.blocks) + ranges = [Range(10, 19), Range(25, 40)] + self.assertFalse(intersects(data, meta, ranges, Settings())) + self.assertEqual(ranges, [Range(10, 19), Range(25, 40)]) + b = Builder(Settings(), header) + for offset, length, values in [ + (len(header), 100, [(0, 10), (5, 5), (20, 5)]), + (len(header) + 100, 200, [((1 << 32) - 2, 5), (8254058425445, 1)]), + (len(header) + 300, 100, [(20, 5), (MAX_ROW_ID, 1)])]: + b.begin_block(offset, length, len(values)) + for first, count in values: + b.add(first, count) + b.end_block() + self.assertEqual(b.serialize(meta.file_name, meta.file_size, 7), golden()) + + def test_minmax_skips_exact_checks_and_one_interval_is_already_exact(self): + header = avro_header() + builder = Builder(Settings(), header) + for offset, values in [(0, [(0, 10), (20, 10)]), + (100, [(100, 10), (200, 10)]), + (200, [(1 << 32, 10)])]: + builder.begin_block(len(header) + offset, 100, len(values)) + for first, count in values: + builder.add(first, count) + builder.end_block() + data = builder.serialize('m', len(header) + 300, 5) + meta = SimpleNamespace(file_name='m', file_size=len(header) + 300, + num_added_files=5, num_deleted_files=0) + for point, expected in [(50, 0), ((1 << 32) + 9, 1)]: + query = Query([Range(point, point)]) + with patch.object(query, 'intersects', wraps=query.intersects) as check: + selected = select(data, meta, query, Settings()) + + self.assertEqual(len(selected.blocks), expected) + self.assertEqual(check.call_count, 3) + check.assert_any_call(0, 29) + check.assert_any_call(100, 209) + check.assert_any_call(1 << 32, (1 << 32) + 9) + + def test_rejected_and_early_hit_blocks_still_validate_every_interval(self): + data = bytearray(golden()) + first_block_intervals = 68 + 4 + len(avro_header()) + 4 + 36 + struct.pack_into('>q', data, first_block_intervals + 16, 9) + data[-32:] = hashlib.sha256(data[:-32]).digest() + for point in (30, 0): + with self.assertRaises(ValueError): + select(data, golden_meta(), [Range(point, point)], Settings()) + + def test_coverage_and_budgets(self): + header = avro_header() + for first, count in [(None, 1), (-1, 1), (10, 0), (10, -1), (MAX_ROW_ID, 2)]: + b = Builder(Settings(), header) + b.begin_block(len(header), 100, 1) + b.add(first, count) + self.assertIsNone(b.serialize('m', 1, 1)) + b = Builder(Settings(), header) + b.begin_block(len(header), 100, 2) + b.add(0, MAX_ROW_ID) + b.add(MAX_ROW_ID, 1) + b.end_block() + self.assertLess(len(b.serialize('m', len(header) + 100, 2)), 512) + b = Builder(Settings(max_ranges=1), header) + b.begin_block(len(header), 100, 2) + b.add(1, 1) + b.add(1 << 32, 1) + self.assertIsNone(b.serialize('m', 1, 2)) + b = Builder(Settings(max_bytes=128), header) + self.assertIsNone(b.serialize('m', 1, 1)) + + def test_invalid_envelopes(self): + meta, data = golden_meta(), golden() + for index in (0, 9, 11, 15, 16, 55, 63, 67, 75, len(data) - 1): + bad = bytearray(data) + bad[index] ^= 2 + with self.assertRaises(ValueError): + select(bad, meta, [Range(10, 10)], Settings()) + for index in (9, 11, 15): + bad = bytearray(data[:-32]) + bad[index] = 0 + bad.extend(hashlib.sha256(bad).digest()) + with self.assertRaises(ValueError): + select(bad, meta, [Range(10, 10)], Settings()) + with self.assertRaises(ValueError): + select(data[:-1], meta, [Range(10, 10)], Settings()) + meta.file_name = 'mismatch' + with self.assertRaises(ValueError): + select(data, meta, [Range(10, 10)], Settings()) + + +class RowIdIndexScanTest(existing.ManifestEntryIdentifierTest): + def setUp(self): + super().setUp() + self.table.options.options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_WRITE, True) + self.table.options.options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, True) + + 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 write_meta(self, name, entries): + manager = self.manifest_file_manager + return manager.write(name, entries) + + def test_explicit_reference_and_null_does_not_probe(self): + manager = self.manifest_file_manager + written = self.write_meta('explicit', [self.entry('data.parquet', 100)]) + self.assertEqual(index_file_name(written), written.file_name + SUFFIX) + index_path = Path(manager.manifest_path, index_file_name(written)) + explicit_path = index_path.with_name('independent-index' + SUFFIX) + index_path.rename(explicit_path) + other_path = index_path.with_name('other-partition-index') + other_path.write_bytes(b'not a row-id index') + indexed = replace(written, extra_files=[other_path.name, explicit_path.name]) + with patch.object(self.table.file_io, 'new_input_stream', + wraps=self.table.file_io.new_input_stream) as opened: + self.assertEqual(manager.read_entries_parallel([indexed], row_ranges=[Range(0, 0)]), []) + self.assertEqual([call[0][0] for call in opened.call_args_list], [str(explicit_path)]) + + for extra_files in (None, [], [other_path.name]): + with self.subTest(extra_files=extra_files): + unindexed = replace(indexed, extra_files=extra_files) + with patch.object(self.table.file_io, 'new_input_stream', + wraps=self.table.file_io.new_input_stream) as opened: + actual = manager.read_entries_parallel([unindexed], row_ranges=[Range(0, 0)]) + 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)]) + indexed = replace(indexed, extra_files=['other-index'] + indexed.extra_files) + unindexed = self.write_meta('legacy-entry', [self.entry('old.parquet', None)]) + self.assertIsNone(index_file_name(unindexed)) + lists = ManifestListManager(self.table) + lists.write('references', [indexed, unindexed]) + actual = lists.read('references') + self.assertEqual([meta.extra_files for meta in actual], [indexed.extra_files, None]) + self.assertEqual([index_file_name(meta) for meta in actual], [index_file_name(indexed), None]) + self.assertEqual([meta.file_name for meta in actual], [indexed.file_name, unindexed.file_name]) + + data = Path(lists.manifest_path, 'references').read_bytes() + legacy_schema = deepcopy(MANIFEST_FILE_META_SCHEMA) + legacy_schema['fields'] = [field for field in legacy_schema['fields'] + if field['name'] != '_EXTRA_FILES'] + legacy_records = list(fastavro.reader(BytesIO(data), reader_schema=legacy_schema)) + self.assertTrue(all('_EXTRA_FILES' not in record for record in legacy_records)) + self.assertEqual([record['_FILE_NAME'] for record in legacy_records], + [indexed.file_name, unindexed.file_name]) + with self.table.file_io.new_output_stream(str(Path(lists.manifest_path, 'old-list'))) as stream: + fastavro.writer(stream, legacy_schema, legacy_records) + self.assertTrue(all(index_file_name(meta) is None for meta in lists.read('old-list'))) + + def test_skips_blocks_inside_a_matching_manifest(self): + entries = [self.entry('file-%d.parquet' % i, i * 1000) for i in range(4000)] + meta = self.write_meta('many-blocks', entries) + outputs = [] + reader = fastavro.reader + for enabled in (False, True): + self.table.options.options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, enabled) + scanner = FileScanner(self.table, lambda: ([meta], None)) + scanner.with_global_index_result(GlobalIndexResult.from_ranges([Range(2000005, 2000005)])) + decoded = [] + + def observed_reader(stream): + for record in reader(stream): + decoded.append(record) + yield record + + with patch('pypaimon.manifest.manifest_file_manager.fastavro.reader', side_effect=observed_reader), \ + patch('pypaimon.manifest.manifest_file_manager.read_selected_bytes', + wraps=read_selected_bytes) as selected_read: + actual, _ = scanner._create_data_evolution_split_generator() + outputs.append([e.file.file_name for e in actual]) + if enabled: + self.assertEqual(selected_read.call_count, 1) + selected = selected_read.call_args[0][2] + self.assertEqual(len(selected.blocks), 1) + self.assertLess(sum(block.length for block in selected.blocks), meta.file_size // 10) + self.assertLess(len(decoded), 200) + self.assertEqual(len(decoded), selected.blocks[0].record_count) + else: + self.assertEqual(selected_read.call_count, 0) + self.assertEqual(len(decoded), 4000) + self.assertEqual(outputs, [['file-2000.parquet']] * 2) + + def test_actual_global_index_scanner_72_to_2(self): + metas = [] + for i in range(72): + entries = [self.entry('a%d.parquet' % i, 0), self.entry('b%d.parquet' % i, 100)] + if i < 2: + entries.append(self.entry('hit.' + ('parquet' if i == 0 else 'blob'), 45)) + metas.append(self.write_meta('manifest-%d' % i, entries)) + results = [] + for enabled in (False, True): + self.table.options.options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, enabled) + scanner = FileScanner(self.table, lambda: (metas, None)) + scanner.with_global_index_result(GlobalIndexResult.from_ranges([Range(50, 50)])) + manager = scanner.manifest_file_manager + with patch.object(manager, 'read', wraps=manager.read) as read_body, \ + patch('pypaimon.manifest.manifest_file_manager.read_index', wraps=read_index) as read_sidecar: + entries, _ = scanner._create_data_evolution_split_generator() + results.append(sorted(e.file.file_name for e in entries)) + self.assertEqual(len(read_body.call_args_list), 2 if enabled else 72) + self.assertEqual(len(read_sidecar.call_args_list), 72 if enabled else 0) + self.assertEqual(results, [['hit.blob', 'hit.parquet']] * 2) + + def test_delete_union_no_resurrection_and_no_query_no_index_io(self): + add = self.entry('data.parquet', 45) + blob = self.entry('data.blob', 45) + metas = [self.write_meta('add', [add, blob]), + self.write_meta('delete', [replace(add, kind=1), replace(blob, kind=1)]), + self.write_meta('gap', [self.entry('lo', 0), self.entry('hi', 100)])] + manager = self.manifest_file_manager + for enabled in (False, True): + self.table.options.options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_READ, enabled) + with patch.object(manager, 'read', wraps=manager.read) as read_body: + entries = manager.read_entries_parallel(metas[:2], row_ranges=[Range(50, 50)]) + self.assertEqual(entries, []) + self.assertEqual(len(read_body.call_args_list), 2) + with patch.object(self.table.file_io, 'new_input_stream', + wraps=self.table.file_io.new_input_stream) as opened: + manager.read_entries_parallel(metas) + self.assertTrue(all(not call[0][0].endswith(SUFFIX) for call in opened.call_args_list)) + # Missing and corrupt objects retain their manifests and read the full body. + path = manager.manifest_path + '/gap' + SUFFIX + for bad in (None, b'partial'): + if bad is None: + os.unlink(path) + else: + Path(path).write_bytes(bad) + with patch.object(manager, 'read', wraps=manager.read) as read_body: + entries = manager.read_entries_parallel(metas[2:], row_ranges=[Range(50, 50)]) + self.assertEqual(len(entries), 2) + self.assertEqual(read_body.call_count, 1) + self.assertIsNone(read_body.call_args[1]['selected_blocks']) + with patch.object(self.table.file_io, 'new_input_stream', side_effect=InterruptedError('stop')): + with self.assertRaises(InterruptedError): + read_index(self.table.file_io, path, metas[0], [Range(0, 0)], Settings()) + + def test_sidecar_cancellation_during_open(self): + self._check_sidecar_cancellation('open') + + def test_sidecar_cancellation_during_read(self): + self._check_sidecar_cancellation('read') + + def test_sidecar_cancellation_during_close(self): + self._check_sidecar_cancellation('close') + + def _check_sidecar_cancellation(self, phase): + for failure_type in (ArrowCancelled, CancelledError, InterruptedError): + with self.subTest(phase=phase, failure_type=failure_type): + self._check_sidecar_io_failure(phase, failure_type('cancelled'), cancelled=True) + + def test_sidecar_io_failures_fall_back_to_manifest(self): + for phase in ('open', 'read', 'close'): + for failure_type in (FileNotFoundError, TimeoutError, OSError): + with self.subTest(phase=phase, failure_type=failure_type): + self._check_sidecar_io_failure(phase, failure_type('unavailable'), cancelled=False) + + def test_sidecar_cancellation_survives_close_failure(self): + for failure_type in (ArrowCancelled, CancelledError, InterruptedError): + with self.subTest(failure_type=failure_type): + self._check_sidecar_io_failure('read', failure_type('cancelled'), cancelled=True, + close_failure=OSError('close failed')) + + def test_sidecar_wrapped_cancellation_propagates(self): + for failure_type in (ArrowCancelled, CancelledError, InterruptedError): + with self.subTest(failure_type=failure_type): + cancellation = failure_type('cancelled') + wrapped = OSError('wrapped failure') + wrapped.__cause__ = cancellation + self._check_sidecar_io_failure('open', wrapped, cancelled=True, + expected_failure=cancellation) + + def test_sidecar_exception_cycle_falls_back(self): + first = OSError('first') + second = OSError('second') + first.__cause__ = second + second.__cause__ = first + self._check_sidecar_io_failure('open', first, cancelled=False) + + def _check_sidecar_io_failure(self, phase, failure, cancelled, close_failure=None, + expected_failure=None): + manager = self.manifest_file_manager + meta = self.write_meta('failure-' + phase + '-' + type(failure).__name__, + [self.entry('data.parquet', 100)]) + index_path = str(Path(manager.manifest_path, index_file_name(meta))) + body_path = str(Path(manager.manifest_path, meta.file_name)) + stream = (FailingIndexInput(Path(index_path).read_bytes(), failure, phase, close_failure) + if phase != 'open' else None) + original_open = self.table.file_io.new_input_stream + + def open_stream(path): + if path == index_path: + if phase == 'open': + raise failure + return stream + return original_open(path) + + with patch.object(self.table.file_io, 'new_input_stream', side_effect=open_stream) as opened, \ + patch.object(manager, 'read', wraps=manager.read) as read_body: + if cancelled: + expected = failure if expected_failure is None else expected_failure + with self.assertRaises(type(expected)) as raised: + manager.read_entries_parallel([meta], row_ranges=[Range(100, 100)]) + self.assertIs(raised.exception, expected) + read_body.assert_not_called() + self.assertEqual([call.args[0] for call in opened.call_args_list], [index_path]) + else: + entries = manager.read_entries_parallel([meta], row_ranges=[Range(100, 100)]) + self.assertEqual([entry.file.file_name for entry in entries], ['data.parquet']) + read_body.assert_called_once() + self.assertIsNone(read_body.call_args.kwargs['selected_blocks']) + self.assertEqual([call.args[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(index_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()) + manager.write('unknown', [self.entry('legacy', None)]) + self.assertFalse(Path(manager.manifest_path, 'unknown' + SUFFIX).exists()) + self.table.options.options.set(CoreOptions.MANIFEST_ROW_ID_INDEX_MAX_RANGES, 1) + manager.write('huge', [self.entry('one', 0, 10), self.entry('two', 100, 10)]) + self.assertFalse(Path(manager.manifest_path, 'huge' + SUFFIX).exists()) diff --git a/paimon-python/pypaimon/tests/reader_append_only_test.py b/paimon-python/pypaimon/tests/reader_append_only_test.py index 480cd4ebbe4b..1c783bfe0611 100644 --- a/paimon-python/pypaimon/tests/reader_append_only_test.py +++ b/paimon-python/pypaimon/tests/reader_append_only_test.py @@ -1090,7 +1090,8 @@ def test_is_in_with_partitions(self): def counting_read(self_mgr, manifest_file_name, manifest_entry_filter=None, drop_stats=True, early_entry_filter=None, - early_record_filter=None, partition_filter=None): + early_record_filter=None, partition_filter=None, + selected_blocks=None): # avro_total = every entry in the manifest (no manifest-file pruning # here: single file, is_in spans its partition stats). path = f"{self_mgr.manifest_path}/{manifest_file_name}" @@ -1100,7 +1101,8 @@ def counting_read(self_mgr, manifest_file_name, return original_read( self_mgr, manifest_file_name, manifest_entry_filter, drop_stats, - early_entry_filter, early_record_filter, partition_filter) + early_entry_filter, early_record_filter, partition_filter, + selected_blocks=selected_blocks) def counting_dfm_init(self_dfm, *args, **kwargs): entry_counts['constructed'] += 1 diff --git a/paimon-python/pypaimon/write/file_store_commit.py b/paimon-python/pypaimon/write/file_store_commit.py index 66e88f2f5a68..2fd81fda2cc7 100644 --- a/paimon-python/pypaimon/write/file_store_commit.py +++ b/paimon-python/pypaimon/write/file_store_commit.py @@ -1140,8 +1140,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}") @@ -1160,8 +1159,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."""