diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index 6ad271bb5a89..ada9baf1cee1 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -63,6 +63,242 @@ 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. +### Manifest Sidecar + +`ManifestSidecar` provides a binary sidecar for selecting complete Avro manifest blocks +using independent partition, row-ID and bucket coverage. A sidecar uses the +`.avro.sidecar` naming convention. Readers find it through an explicit +`.avro.sidecar` reference in the manifest metadata's `_EXTRA_FILES`, without probing a +derived file name. The Avro schemas and `_VERSION` identifiers remain unchanged. + +The utility includes construction, validation, block selection and optional caching. Table +writers and scans do not yet invoke it automatically. Callers are responsible for publishing +sidecar references, managing file ownership, applying entry filters and reconciling ADD/DELETE +entries after block selection. `build` reads the completed physical manifest and returns +sidecar bytes; it does not write or publish another file. + +Callers decide whether to invoke `build` and `read`; these utilities have no read/write switches. +`build` and `Builder` accept `rowIdEnabled` and `bucketEnabled` arguments for independent +payload generation. Partition generation is always enabled, +including the empty partition tuple for unpartitioned tables. Missing or invalid +metadata makes only the affected block's dimension unavailable. There is no sidecar byte budget: +construction keeps complete coverage and `read` consumes the entire file once it is opened. + +`read` returns null for an absent sidecar reference or an `IOException`, allowing the caller +to fall back to the manifest. If the thread is interrupted, the I/O failure is propagated as +`UncheckedIOException`. Other exceptions and errors propagate unchanged. `select` validates +supplied bytes directly and reports invalid containers with `IOException`. + +Version 1 uses the following layout. Counts, lengths, offsets and the version use canonical +nonnegative unsigned LEB128 varints. Counts and payload lengths are bounded by `Integer.MAX_VALUE`; +block offsets, lengths and record counts are bounded by `Long.MAX_VALUE`. Row-ID envelope +endpoints remain fixed-width, eight-byte big-endian longs. Encoding IDs are unsigned bytes +with separate namespaces. The existing serialized partition tuple bytes are unchanged. + +```text +magic : 4 bytes // ASCII PMSC +formatVersion : varint // 1 +avroHeaderLength : varint +avroHeader : bytes // original schema, codec and sync marker +partitionCount : varint +partitionDictionary[] + partitionByteLength : varint + partitionBytes : bytes // existing manifest BinaryRow serialization +blockCount : varint +blocks[] // original physical order + offset : varint + length : varint // complete encoded block, including sync marker + recordCount : varint + partitionEncoding : byte + if partitionEncoding != 0: + partitionPayloadLength : varint + partitionPayload : bytes + rowIdEncoding : byte + if rowIdEncoding != 0: + rowIdPayloadLength : varint + rowIdPayload : bytes + bucketEncoding : byte + if bucketEncoding != 0: + bucketPayloadLength : varint + bucketPayload : bytes +checksum : 4 bytes // big-endian CRC32 of all preceding bytes +``` + +The block ID is its position. Its first entry ordinal is the sum of preceding record counts +and is not stored. Each complete partition tuple appears once in the dictionary, including +all its fields and nulls. The scan's partition type interprets the existing serialized tuple. +Partition predicates are evaluated once per dictionary entry. + +| Dimension | Encoding | Payload | +| --- | --- | --- | +| Any | `0` | Unavailable; only the encoding byte is present. | +| Partition | `1` | `intsDeltaPayload` of sorted unique dictionary IDs. | +| Row ID | `1` | Minimum, maximum, and `intsDeltaPayload` of sorted interior interval endpoints. | +| Bucket | `1` | Two paired `intsDeltaPayload` sequences: sorted bucket IDs and their recorded total bucket counts. | +| Any | Other nonzero ID | Skip the declared payload length; treat only this dimension as unavailable. | + +Only nonzero encodings are followed by a length and payload. Payload lengths exclude the +encoding and length fields, but include the count and other fields within the payload. +Partition IDs and bucket pairs have positive counts no greater than the block's record count. +Row-ID coverage contains one or more intervals; its interior endpoint count can be zero for +a single interval. Encoding 0 represents unavailable coverage, not an empty known set. + +#### Integer Delta Payload + +Partition, row-ID and both bucket sequences share this structure: + +```text +intsDeltaPayload + count : varint + deltas[count] : varint +``` + +The count is in `[0, Integer.MAX_VALUE]`. Nonnegative deltas use unsigned LEB128 varints, +occupying one to nine bytes for values from 0 through +`Long.MAX_VALUE`. Seven value bits are stored per byte, least significant group first; the +high bit indicates another byte follows. +Encodings use the shortest representation without padding. + +A nondecreasing sequence is delta-encoded from a specified base. Each value contributes one +unsigned varint containing its difference from the preceding value. The first difference +is relative to the base: + +```text +deltas[] : varint +value[0] = base + deltas[0] +value[i] = value[i - 1] + deltas[i] +``` + +The shared `DeltaVarintCodec` utility writes the count and then each delta immediately, +and reads values on demand using `VarLengthIntUtils`. A reader consumes exactly the declared +number of values, leaving any following sequence available in the buffer. Callers check +their enclosing payload boundaries. Counts, overflow and value bounds are checked without +materializing arrays. Reads may stop early. + +Only the `totalBuckets` sequence uses signed differences, because totals need not increase +when pairs are sorted by bucket. Its differences use ZigZag before unsigned varint encoding: +`encoded = (delta << 1) ^ (delta >> 63)` and +`delta = (encoded >>> 1) ^ -(encoded & 1)`. Values are nonnegative ints, so encoded deltas +are at most `2 * Integer.MAX_VALUE` and require at most five bytes. The field defines this +signed mode; no additional mode byte is stored. Other sequences use nonnegative differences. + +#### Partition Payload + +When `partitionEncoding == 1`, the block stores IDs of all distinct partition tuples +represented by its entries: + +```text +partitionPayload + intsDeltaPayload // dictionary IDs, base = 0 +``` + +An ID is the zero-based position of a complete tuple in the sidecar's shared dictionary. +IDs satisfy `0 <= id < partitionCount` and are strictly increasing. Tuple bytes appear only +in the dictionary and are not repeated in each block. For IDs `[0, 1, 2, 3, 4]`, the deltas +are `[0, 1, 1, 1, 1]`. The payload contains a one-byte count of 5 followed by these five +varint bytes: 6 bytes, or 8 bytes including the encoding and length fields. + +With a partition filter, a block matches if any referenced tuple matches. A tuple containing +a null partition value still has a dictionary ID. Unpartitioned tables record the empty +tuple. If any entry's entire partition tuple is unavailable, the block uses encoding 0, +so a dictionary miss cannot exclude that block. Later blocks can still use existing IDs. + +#### Row-ID Payload + +The writer merges overlapping and adjacent inclusive intervals contributed by entries. +An entry contributes `[firstRowId, firstRowId + rowCount - 1]`. The resulting intervals are +sorted and disjoint; they are never expanded into individual row IDs or coarsened to include gaps. + +```text +rowIdPayload + minRowId : long // first interval's start + maxRowId : long // last interval's inclusive end + intsDeltaPayload // stores longs: 2 * (N - 1) sorted interior endpoints, base = minRowId +``` + +The envelope satisfies `0 <= minRowId <= maxRowId <= Long.MAX_VALUE`. Flatten the intervals +as `[start0, end0, start1, end1, ...]`. The first start is supplied by `minRowId`, and the last +end by `maxRowId`; only the remaining `2 * (N - 1)` interior endpoints are delta/varint encoded. +Their count must be even, and the derived interval count `N = count / 2 + 1` must not exceed +the block's record count. There is no separate stored interval count. +Pairing the reconstructed endpoints recovers the intervals. Each pair satisfies +`0 <= start <= end <= Long.MAX_VALUE`; each following start must exceed the preceding end. + +For `[(10, 19), (30, 39)]`, the minimum is 10 and maximum is 39. The interior +endpoints `[19, 30]` have deltas `[9, 11]` from base 10, each encoded as one varint byte. +The payload starts with an eight-byte minimum of 10 and an eight-byte maximum of 39, +followed by `intsDeltaPayload` bytes `[2, 9, 11]`: 19 bytes, or 21 bytes with framing. +For a single interval, the two eight-byte endpoints and a one-byte zero count define the +interval: 17 payload bytes and no deltas. + +The reader first tests the envelope without decoding any deltas. A query for row ID 25 +passes the example's envelope check but matches neither interval. Unknown or invalid row-ID +metadata makes that block's row-ID payload unavailable; partition and bucket coverage remain usable. + +#### Bucket Payload + +When `bucketEncoding == 1`, the block stores distinct bucket/count pairs: + +```text +bucketPayload + buckets : intsDeltaPayload // N > 0, base = 0, nonnegative deltas + totalBuckets : intsDeltaPayload // N values, base = 0, ZigZag signed deltas +``` + +Pairs are sorted first by bucket, then by total bucket count, with duplicates removed. +Each pair satisfies `0 <= bucket < totalBuckets <= Integer.MAX_VALUE`. Both sequences have +the same count, and values at the same position form one pair. The totals must not be sorted +independently. The same bucket may occur with different totals after rescaling. + +For `[(1, 4), (1, 8), (3, 4)]`, bucket values `[1, 1, 3]` have deltas `[1, 0, 2]`. +Paired totals `[4, 8, 4]` have signed deltas `[4, 4, -4]`, ZigZag-encoded as `[8, 8, 7]`. +The two payloads are `[3, 1, 0, 2]` and `[3, 8, 8, 7]`: 8 bytes total, or 10 bytes with framing. + +Missing, invalid or negative/synthetic bucket metadata makes the block's bucket coverage +unavailable. A caller can supply a predicate on `(bucket, totalBuckets)` which conservatively +retains every potentially matching pair. Filters requiring an entry's partition belong at +the entry-filtering stage; omit the bucket predicate if no safe check is available. + +#### Validation and Reading + +Readers validate the checksum, container fields, payload lengths and leading count headers, +and the complete physical block directory regardless of the query. Byte spans must cover +the whole original manifest after its header; record counts must sum to the manifest entry +count. Manifest length and entry count come from the supplied `ManifestFileMeta` rather +than being duplicated in the sidecar. Unknown nonzero encodings skip their declared bytes +without interpreting a count. + +Compressed contents are decoded only for dimensions needed by the filters. A row-ID envelope +rejection skips all its deltas; a matching interval or partition ID skips remaining values. +Bucket filtering first walks the bucket sequence to locate the paired totals without +allocating arrays, then decodes pairs until a match. Unused totals may be skipped. +Invalid varints, value counts, overflows, +out-of-range values or ordering encountered while decoding invalidate the container. Delta +contents skipped by short-circuiting are not individually validated. + +For conjunctive filters a block is retained only if every dimension is unavailable or matches. +Matching tests row ID, partition, then bucket coverage. Absent filters are skipped, and a +rejection skips the remaining dimensions. Matches in different dimensions can come from +different entries, so entry filtering and ADD/DELETE reconciliation remain necessary. + +The entire sidecar is read in chunks of at most 1 MiB, including payloads unused by a query. +There is no size-based fallback or payload dropping. Payload lengths save decoding work, +not sidecar storage I/O. Selected compressed Avro blocks are read by byte range with adjacent +spans coalesced and individual read requests bounded to 4 MiB. Building a sidecar does not +modify the original manifest. + +`read` and `openManifest` accept an optional caller-supplied `SegmentsCache`. Complete +sidecar bytes are keyed by their explicit `Path`. Only successful reads and selections +populate the cache; query-specific selections are not cached. Cache entry-size limits affect +admission only: larger sidecars are still fully read, validated and used. + +Selected Avro blocks also share this cache. Each entry contains one complete compressed block, +keyed by the manifest's full path, original offset and encoded length, separately from whole-file +keys. Different selections reuse the same blocks. Only complete reads populate the cache; +oversized blocks stream through the read buffer. Adjacent uncached blocks fitting the buffer +are read together and cached individually. Fully cached selections do not open the manifest. +The cache retains its configured memory budget, entry-size limit, expiration and eviction policy. + ## Manifest Data manifests record **ADD** (`0`) and **DELETE** (`1`) entries. Readers reconcile these entries diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCodec.java b/paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCodec.java new file mode 100644 index 000000000000..9c6575bb4f91 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/utils/DeltaVarintCodec.java @@ -0,0 +1,126 @@ +/* + * 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.utils; + +import java.io.DataOutput; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Objects; + +/** Streaming count-prefixed delta/varint encoding of nonnegative integer sequences. */ +public final class DeltaVarintCodec { + + private DeltaVarintCodec() {} + + /** Writes a varint count followed by deltas, without buffering the sequence. */ + public static final class Writer { + private final DataOutput out; + private final boolean signedDeltas; + private long remaining; + private long previous; + + public Writer(DataOutput out, int count, long base) throws IOException { + this(out, count, base, false); + } + + /** Signed deltas support non-monotonic nonnegative int values using ZigZag. */ + public Writer(DataOutput out, int count, long base, boolean signedDeltas) + throws IOException { + if (base < 0 || count < 0 || (signedDeltas && base > Integer.MAX_VALUE)) { + throw new IllegalArgumentException("Invalid delta/varint count or base"); + } + this.out = Objects.requireNonNull(out); + this.signedDeltas = signedDeltas; + this.remaining = count; + previous = base; + VarLengthIntUtils.encodeInt(out, count); + } + + public void write(long value) throws IOException { + require(remaining > 0 && value >= 0); + long delta = value - previous; + if (signedDeltas) { + require(value <= Integer.MAX_VALUE); + delta = (delta << 1) ^ (delta >> 63); + } else { + require(delta >= 0); + } + VarLengthIntUtils.encodeLong(out, delta); + previous = value; + remaining--; + } + } + + /** + * Reads the count and then values on demand. A complete read leaves the buffer at the next + * payload, allowing count-prefixed sequences to be concatenated. Reading may stop early. + */ + public static final class Reader { + private final ByteBuffer data; + private final long max; + private final int count; + private final boolean signedDeltas; + private long remaining; + private long value; + + public Reader(ByteBuffer data, long base, long max) throws IOException { + this(data, base, max, false); + } + + public Reader(ByteBuffer data, long base, long max, boolean signedDeltas) + throws IOException { + this.data = Objects.requireNonNull(data); + long encodedCount = VarLengthIntUtils.decodeLong(data); + require(encodedCount <= Integer.MAX_VALUE && encodedCount <= data.remaining()); + require(base >= 0 && max >= base && (!signedDeltas || max <= Integer.MAX_VALUE)); + count = (int) encodedCount; + remaining = count; + value = base; + this.max = max; + this.signedDeltas = signedDeltas; + } + + public int count() { + return count; + } + + public boolean hasNext() { + return remaining > 0; + } + + public long next() throws IOException { + require(remaining > 0); + long delta = VarLengthIntUtils.decodeLong(data); + if (signedDeltas) { + require(delta <= 2L * Integer.MAX_VALUE); + delta = (delta >>> 1) ^ -(delta & 1); + } + require(delta >= -value && delta <= max - value); + value += delta; + remaining--; + return value; + } + } + + private static void require(boolean valid) throws IOException { + if (!valid) { + throw new IOException("Invalid delta/varint sequence"); + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/VarLengthIntUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/VarLengthIntUtils.java index a05eb8a99484..cc7a56e5c526 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/VarLengthIntUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/VarLengthIntUtils.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.nio.ByteBuffer; /* This file is based on source code of LongPacker from the PalDB Project (https://github.com/linkedin/PalDB), licensed by the Apache * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for @@ -82,6 +83,25 @@ public static long decodeLong(DataInput is) throws IOException { throw new Error("Malformed long."); } + /** Decodes a canonical nonnegative long from the buffer and advances its position. */ + public static long decodeLong(ByteBuffer in) throws IOException { + long value = 0; + for (int shift = 0; shift < 63; shift += 7) { + if (!in.hasRemaining()) { + throw new EOFException("Truncated variable-length long"); + } + int b = Byte.toUnsignedInt(in.get()); + value |= (long) (b & 0x7f) << shift; + if ((b & 0x80) == 0) { + if (shift != 0 && (b & 0x7f) == 0) { + throw new IOException("Noncanonical variable-length long"); + } + return value; + } + } + throw new IOException("Invalid variable-length long"); + } + public static long decodeLong(byte[] ba, int index) { long result = 0; for (int offset = 0; offset < 64; offset += 7) { @@ -158,6 +178,15 @@ public static int decodeInt(DataInput is) throws IOException { throw new Error("Malformed integer."); } + /** Decodes a canonical nonnegative int from the buffer and advances its position. */ + public static int decodeInt(ByteBuffer in) throws IOException { + long value = decodeLong(in); + if (value > Integer.MAX_VALUE) { + throw new IOException("Variable-length integer exceeds Integer.MAX_VALUE"); + } + return (int) value; + } + public static int decodeInt(InputStream is) throws IOException { for (int offset = 0, result = 0; offset < 32; offset += 7) { int b = is.read(); diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/DeltaVarintCodecTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/DeltaVarintCodecTest.java new file mode 100644 index 000000000000..dd667f475d35 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/utils/DeltaVarintCodecTest.java @@ -0,0 +1,225 @@ +/* + * 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.utils; + +import org.apache.paimon.io.DataOutputSerializer; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Random; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for count-prefixed streaming delta/varint payloads. */ +class DeltaVarintCodecTest { + @Test + void fixedBytesAndSlicedBuffer() throws Exception { + long[] values = {10, 12, 14, 16, 16, 1024}; + byte[] encoded = encode(10, values); + assertThat(encoded).containsExactly(new byte[] {6, 0, 2, 2, 2, 0, (byte) 0xf0, 7}); + ByteBuffer data = ByteBuffer.allocate(encoded.length + 4); + data.position(2); + data.put(encoded); + data.limit(data.position()).position(2); + DeltaVarintCodec.Reader reader = new DeltaVarintCodec.Reader(data, 10, 1024); + assertThat(reader.count()).isEqualTo(values.length); + for (long value : values) { + assertThat(reader.hasNext()).isTrue(); + assertThat(reader.next()).isEqualTo(value); + } + assertThat(reader.hasNext()).isFalse(); + assertThat(data.hasRemaining()).isFalse(); + assertThatThrownBy(reader::next).isInstanceOf(IOException.class); + } + + @Test + void longBoundariesAndEmptySequence() throws Exception { + byte[] encoded = encode(0, Long.MAX_VALUE); + assertThat(encoded).hasSize(10); + assertThat(new DeltaVarintCodec.Reader(ByteBuffer.wrap(encoded), 0, Long.MAX_VALUE).next()) + .isEqualTo(Long.MAX_VALUE); + assertThat(encode(Long.MAX_VALUE - 1, Long.MAX_VALUE, Long.MAX_VALUE)) + .containsExactly(new byte[] {2, 1, 0}); + DeltaVarintCodec.Reader empty = + new DeltaVarintCodec.Reader(ByteBuffer.wrap(new byte[] {0}), 0, 0); + assertThat(empty.hasNext()).isFalse(); + assertThatThrownBy(empty::next).isInstanceOf(IOException.class); + assertThat(encode(0)).containsExactly((byte) 0); + } + + @Test + void stopsBeforeUnusedMalformedData() throws Exception { + DeltaVarintCodec.Reader reader = + new DeltaVarintCodec.Reader( + ByteBuffer.wrap(new byte[] {2, 0, (byte) 0x80}), 10, 100); + assertThat(reader.next()).isEqualTo(10); + assertThat(reader.hasNext()).isTrue(); + assertThatThrownBy(reader::next).isInstanceOf(IOException.class); + } + + @Test + void rejectsInvalidCountsBoundsAndOrdering() throws Exception { + assertThatThrownBy(() -> new DeltaVarintCodec.Reader(ByteBuffer.allocate(0), 0, 1)) + .isInstanceOf(IOException.class); + assertThatThrownBy( + () -> new DeltaVarintCodec.Reader(ByteBuffer.wrap(new byte[] {2, 0}), 0, 1)) + .isInstanceOf(IOException.class); + assertThatThrownBy( + () -> + new DeltaVarintCodec.Reader( + ByteBuffer.wrap(new byte[] {(byte) 0x81, 0, 0}), 0, 1)) + .isInstanceOf(IOException.class); + assertThatThrownBy(() -> new DeltaVarintCodec.Reader(ByteBuffer.wrap(new byte[] {0}), 2, 1)) + .isInstanceOf(IOException.class); + assertThatThrownBy(() -> new DeltaVarintCodec.Writer(new DataOutputSerializer(8), 0, -1)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> encode(10, 9)).isInstanceOf(IOException.class); + assertThatThrownBy(() -> encode(0, 2, 1)).isInstanceOf(IOException.class); + DeltaVarintCodec.Reader overflow = + new DeltaVarintCodec.Reader( + ByteBuffer.wrap(encode(0, Long.MAX_VALUE)), 1, Long.MAX_VALUE); + assertThatThrownBy(overflow::next).isInstanceOf(IOException.class); + } + + @Test + void rejectsMalformedVarints() throws Exception { + byte[] overlong = new byte[10]; + Arrays.fill(overlong, (byte) 0x80); + for (byte[] bytes : + Arrays.asList(new byte[] {(byte) 0x80}, new byte[] {(byte) 0x81, 0}, overlong)) { + byte[] payload = new byte[bytes.length + 1]; + payload[0] = 1; + System.arraycopy(bytes, 0, payload, 1, bytes.length); + DeltaVarintCodec.Reader reader = + new DeltaVarintCodec.Reader(ByteBuffer.wrap(payload), 0, Long.MAX_VALUE); + assertThatThrownBy(reader::next).isInstanceOf(IOException.class); + } + } + + @Test + void randomizedRoundTrips() throws Exception { + Random random = new Random(9845); + for (int trial = 0; trial < 1000; trial++) { + long base = random.nextInt(10000); + long[] values = new long[random.nextInt(128)]; + long value = base; + for (int i = 0; i < values.length; i++) { + value += random.nextInt(10000); + values[i] = value; + } + DeltaVarintCodec.Reader reader = + new DeltaVarintCodec.Reader(ByteBuffer.wrap(encode(base, values)), base, value); + for (long expected : values) { + assertThat(reader.next()).isEqualTo(expected); + } + assertThat(reader.hasNext()).isFalse(); + } + } + + private static byte[] encode(long base, long... values) throws IOException { + DataOutputSerializer out = new DataOutputSerializer(32); + DeltaVarintCodec.Writer writer = new DeltaVarintCodec.Writer(out, values.length, base); + for (long value : values) { + writer.write(value); + } + return out.getCopyOfBuffer(); + } + + @Test + void concatenatedPayloadsRetainTheirOwnBoundaries() throws Exception { + byte[] first = encode(0, 1, 1, 3); + byte[] second = encode(0, 4, 8, 16); + ByteBuffer buffer = ByteBuffer.allocate(first.length + second.length); + buffer.put(first).put(second).flip(); + DeltaVarintCodec.Reader reader = new DeltaVarintCodec.Reader(buffer, 0, 3); + assertThat(reader.count()).isEqualTo(3); + assertThat(reader.next()).isEqualTo(1); + assertThat(reader.next()).isEqualTo(1); + assertThat(reader.next()).isEqualTo(3); + assertThat(reader.hasNext()).isFalse(); + assertThat(buffer.position()).isEqualTo(first.length); + reader = new DeltaVarintCodec.Reader(buffer, 0, 16); + assertThat(reader.next()).isEqualTo(4); + assertThat(reader.next()).isEqualTo(8); + assertThat(reader.next()).isEqualTo(16); + assertThat(buffer.hasRemaining()).isFalse(); + } + + @Test + void signedIntDeltasPreserveDecreasingTotals() throws Exception { + DataOutputSerializer out = new DataOutputSerializer(32); + DeltaVarintCodec.Writer writer = new DeltaVarintCodec.Writer(out, 3, 0, true); + for (int value : new int[] {4, 8, 4}) { + writer.write(value); + } + assertThat(out.getCopyOfBuffer()).containsExactly(new byte[] {3, 8, 8, 7}); + DeltaVarintCodec.Reader reader = + new DeltaVarintCodec.Reader(ByteBuffer.wrap(out.getCopyOfBuffer()), 0, 8, true); + assertThat(reader.next()).isEqualTo(4); + assertThat(reader.next()).isEqualTo(8); + assertThat(reader.next()).isEqualTo(4); + assertThat(reader.hasNext()).isFalse(); + assertThatThrownBy(() -> writer.write(4)).isInstanceOf(IOException.class); + } + + @Test + void signedIntBoundariesAndRandomSequences() throws Exception { + Random random = new Random(20260916); + for (int trial = 0; trial < 100; trial++) { + int[] values = new int[128]; + for (int i = 0; i < values.length; i++) { + values[i] = random.nextInt(Integer.MAX_VALUE); + } + values[0] = Integer.MAX_VALUE; + values[1] = 0; + values[2] = Integer.MAX_VALUE; + DataOutputSerializer out = new DataOutputSerializer(32); + DeltaVarintCodec.Writer writer = + new DeltaVarintCodec.Writer(out, values.length, 0, true); + for (int value : values) { + writer.write(value); + } + ByteBuffer buffer = ByteBuffer.wrap(out.getCopyOfBuffer()); + DeltaVarintCodec.Reader reader = + new DeltaVarintCodec.Reader(buffer, 0, Integer.MAX_VALUE, true); + for (int value : values) { + assertThat(reader.next()).isEqualTo(value); + } + assertThat(buffer.hasRemaining()).isFalse(); + } + DeltaVarintCodec.Reader negative = + new DeltaVarintCodec.Reader(ByteBuffer.wrap(new byte[] {1, 1}), 0, 10, true); + assertThatThrownBy(negative::next).isInstanceOf(IOException.class); + DeltaVarintCodec.Reader tooLarge = + new DeltaVarintCodec.Reader(ByteBuffer.wrap(new byte[] {1, 22}), 0, 10, true); + assertThatThrownBy(tooLarge::next).isInstanceOf(IOException.class); + assertThatThrownBy( + () -> + new DeltaVarintCodec.Writer( + new DataOutputSerializer(8), + 0, + (long) Integer.MAX_VALUE + 1, + true)) + .isInstanceOf(IllegalArgumentException.class); + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/VarLengthIntUtilsTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/VarLengthIntUtilsTest.java new file mode 100644 index 000000000000..76a8f4f8d5ec --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/utils/VarLengthIntUtilsTest.java @@ -0,0 +1,101 @@ +/* + * 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.utils; + +import org.apache.paimon.io.DataOutputSerializer; + +import org.junit.jupiter.api.Test; + +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; + +import static org.apache.paimon.utils.VarLengthIntUtils.decodeInt; +import static org.apache.paimon.utils.VarLengthIntUtils.encodeInt; +import static org.apache.paimon.utils.VarLengthIntUtils.encodeLong; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for variable-length integer buffer decoding. */ +class VarLengthIntUtilsTest { + + @Test + void decodeIntBoundariesAndBufferPosition() throws Exception { + int[] values = { + 0, + 1, + 127, + 128, + 16383, + 16384, + (1 << 21) - 1, + 1 << 21, + (1 << 28) - 1, + 1 << 28, + Integer.MAX_VALUE + }; + DataOutputSerializer out = new DataOutputSerializer(64); + for (int value : values) { + encodeInt(out, value); + } + byte[] bytes = out.getCopyOfBuffer(); + for (ByteBuffer buffer : + Arrays.asList( + ByteBuffer.allocate(bytes.length + 4), + ByteBuffer.allocateDirect(bytes.length + 4))) { + buffer.position(2); + buffer.put(bytes).put((byte) 42).flip(); + buffer.position(2); + ByteBuffer in = buffer.slice().asReadOnlyBuffer(); + for (int value : values) { + assertThat(decodeInt(in)).isEqualTo(value); + } + assertThat(in.position()).isEqualTo(bytes.length); + assertThat(in.get()).isEqualTo((byte) 42); + } + } + + @Test + void decodeIntRejectsOverflow() throws Exception { + for (long value : new long[] {(long) Integer.MAX_VALUE + 1, 1L << 32, Long.MAX_VALUE}) { + DataOutputSerializer out = new DataOutputSerializer(16); + encodeLong(out, value); + assertThatThrownBy(() -> decodeInt(ByteBuffer.wrap(out.getCopyOfBuffer()))) + .isInstanceOf(IOException.class) + .hasMessageContaining("Integer.MAX_VALUE"); + } + } + + @Test + void decodeIntRejectsTruncatedAndMalformedValues() { + for (byte[] bytes : + Arrays.asList(new byte[0], new byte[] {(byte) 0x80}, new byte[] {(byte) 0xff})) { + assertThatThrownBy(() -> decodeInt(ByteBuffer.wrap(bytes))) + .isInstanceOf(EOFException.class); + } + byte[] overlong = new byte[10]; + Arrays.fill(overlong, (byte) 0x80); + for (byte[] bytes : + Arrays.asList(new byte[] {(byte) 0x80, 0}, new byte[] {(byte) 0x81, 0}, overlong)) { + assertThatThrownBy(() -> decodeInt(ByteBuffer.wrap(bytes))) + .isInstanceOf(IOException.class); + } + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java new file mode 100644 index 000000000000..c6ee9e9f5615 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestSidecar.java @@ -0,0 +1,942 @@ +/* + * 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.annotation.VisibleForTesting; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.Segments; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.DeltaVarintCodec; +import org.apache.paimon.utils.RowRangeIndex; +import org.apache.paimon.utils.SegmentsCache; +import org.apache.paimon.utils.SerializationUtils; +import org.apache.paimon.utils.VarLengthIntUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; +import java.util.function.BiPredicate; +import java.util.zip.CRC32; + +import static org.apache.paimon.utils.VarLengthIntUtils.decodeInt; +import static org.apache.paimon.utils.VarLengthIntUtils.encodeLong; + +/** Independently usable partition, row-id and bucket coverage for each manifest block. */ +public final class ManifestSidecar { + + public static final String SUFFIX = ".avro.sidecar"; + private static final Logger LOG = LoggerFactory.getLogger(ManifestSidecar.class); + private static final int MAGIC = 0x504d5343; + private static final int FORMAT_VERSION = 1; + private static final int MIN_HEADER_BYTES = 29; + private static final int MIN_BLOCK_BYTES = 6; + private static final byte[] EMPTY = new byte[0]; + private static final int CHECKSUM_BYTES = Integer.BYTES; + private static final int SIDECAR_READ_BUFFER_BYTES = 1024 * 1024; + private static final int BLOCK_READ_BUFFER_BYTES = 4 * 1024 * 1024; + private static final ProjectedManifestEntry.Projection BLOCK_INDEX_PROJECTION = + createBlockIndexProjection(); + + private ManifestSidecar() {} + + 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; + } + + /** 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; + } + } + + /** Builds a complete block directory with independently available coverage. */ + public static final class Builder { + private final boolean rowIdEnabled; + private final boolean bucketEnabled; + private final byte[] header; + private final TreeMap ranges = new TreeMap<>(); + private final Map dictionary = new LinkedHashMap<>(); + private final TreeSet partitionIds = new TreeSet<>(); + private final TreeSet bucketPairs = new TreeSet<>(); + private final List blocks = new ArrayList<>(); + private long nextOffset; + private long nextRecord; + private Block current; + private long entriesInBlock; + private boolean rowAvailable; + private boolean partitionAvailable; + private boolean bucketAvailable; + + public Builder(byte[] header, boolean rowIdEnabled, boolean bucketEnabled) { + this.rowIdEnabled = rowIdEnabled; + this.bucketEnabled = bucketEnabled; + this.header = Objects.requireNonNull(header); + nextOffset = header.length; + } + + public void beginBlock(long offset, long length, long records) throws IOException { + require(current == null && offset == nextOffset && length > 0 && records > 0); + current = new Block(offset, length, nextRecord, records); + entriesInBlock = 0; + rowAvailable = rowIdEnabled; + partitionAvailable = true; + bucketAvailable = bucketEnabled; + ranges.clear(); + partitionIds.clear(); + bucketPairs.clear(); + } + + @VisibleForTesting + public void add(@Nullable Long first, long count) { + add(first, count, null); + } + + @VisibleForTesting + public void add(@Nullable Long first, long count, @Nullable byte[] partition) { + add(first, count, partition, null, null); + } + + public void add( + @Nullable Long first, + long count, + @Nullable byte[] partition, + @Nullable Integer bucket, + @Nullable Integer totalBuckets) { + if (current == null) { + throw new IllegalStateException("No current Avro block"); + } + entriesInBlock++; + addPartition(partition); + addBucket(bucket, totalBuckets); + if (!rowAvailable) { + return; + } + if (first == null || first < 0 || count <= 0 || count - 1 > Long.MAX_VALUE - first) { + rowAvailable = false; + ranges.clear(); + 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()); + } + ranges.put(start, end); + } + + private void addBucket(@Nullable Integer bucket, @Nullable Integer totalBuckets) { + if (!bucketAvailable) { + return; + } + if (bucket == null || totalBuckets == null || bucket < 0 || totalBuckets <= bucket) { + bucketAvailable = false; + bucketPairs.clear(); + return; + } + bucketPairs.add(((long) bucket << 32) | totalBuckets); + } + + private void addPartition(@Nullable byte[] bytes) { + if (!partitionAvailable) { + return; + } + if (bytes == null) { + partitionAvailable = false; + partitionIds.clear(); + return; + } + Integer id = dictionary.get(ByteBuffer.wrap(bytes)); + if (id == null) { + id = dictionary.size(); + dictionary.put(ByteBuffer.wrap(bytes.clone()), id); + } + partitionIds.add(id); + } + + public void endBlock() throws IOException { + require(current != null && entriesInBlock == current.recordCount); + blocks.add( + new IndexedBlock( + current, + partitionAvailable + ? encodeValues(partitionIds, partitionIds.size()) + : EMPTY, + rowAvailable ? encodeRanges() : EMPTY, + bucketAvailable ? encodeBuckets() : EMPTY)); + nextOffset = Math.addExact(current.offset, current.length); + nextRecord = Math.addExact(current.firstRecord, current.recordCount); + ranges.clear(); + partitionIds.clear(); + bucketPairs.clear(); + current = null; + } + + private byte[] encodeRanges() throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(buffer); + long min = ranges.firstKey(); + long max = ranges.lastEntry().getValue(); + out.writeLong(min); + out.writeLong(max); + // The envelope supplies the first start and last end. Encode only interior endpoints. + DeltaVarintCodec.Writer encoder = + new DeltaVarintCodec.Writer(out, Math.multiplyExact(ranges.size() - 1, 2), min); + int index = 0; + for (Map.Entry range : ranges.entrySet()) { + if (index > 0) { + encoder.write(range.getKey()); + } + if (++index < ranges.size()) { + encoder.write(range.getValue()); + } + } + return buffer.toByteArray(); + } + + private byte[] encodeBuckets() throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(buffer); + DeltaVarintCodec.Writer buckets = + new DeltaVarintCodec.Writer(out, bucketPairs.size(), 0); + for (long pair : bucketPairs) { + buckets.write(pair >>> 32); + } + DeltaVarintCodec.Writer totals = + new DeltaVarintCodec.Writer(out, bucketPairs.size(), 0, true); + for (long pair : bucketPairs) { + totals.write((int) pair); + } + return buffer.toByteArray(); + } + + public byte[] serialize(long fileSize, long entryCount) throws IOException { + require(current == null && nextOffset == fileSize && nextRecord == entryCount); + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(buffer); + out.writeInt(MAGIC); + encodeLong(out, FORMAT_VERSION); + encodeLong(out, header.length); + out.write(header); + encodeLong(out, dictionary.size()); + for (ByteBuffer bytes : dictionary.keySet()) { + encodeLong(out, bytes.remaining()); + out.write(bytes.array()); + } + encodeLong(out, blocks.size()); + for (IndexedBlock block : blocks) { + encodeLong(out, block.block.offset); + encodeLong(out, block.block.length); + encodeLong(out, block.block.recordCount); + writePayload(out, block.partitions); + writePayload(out, block.rowIds); + writePayload(out, block.buckets); + } + CRC32 crc = new CRC32(); + crc.update(buffer.toByteArray()); + out.writeInt((int) crc.getValue()); + return buffer.toByteArray(); + } + + private static byte[] encodeValues(Iterable values, int count) + throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(buffer); + DeltaVarintCodec.Writer encoder = new DeltaVarintCodec.Writer(out, count, 0); + for (Number value : values) { + encoder.write(value.longValue()); + } + return buffer.toByteArray(); + } + + private static void writePayload(DataOutputStream out, byte[] payload) throws IOException { + out.writeByte(payload.length == 0 ? 0 : 1); + if (payload.length > 0) { + encodeLong(out, payload.length); + out.write(payload); + } + } + } + + private static final class IndexedBlock { + private final Block block; + private final byte[] partitions; + private final byte[] rowIds; + private final byte[] buckets; + + private IndexedBlock(Block block, byte[] partitions, byte[] rowIds, byte[] buckets) { + this.block = block; + this.partitions = partitions; + this.rowIds = rowIds; + this.buckets = buckets; + } + } + + private static ProjectedManifestEntry.Projection createBlockIndexProjection() { + List fields = + new ArrayList<>( + ProjectedManifestEntry.ROW_RANGE_PROJECTION.projectedType().getFields()); + fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.BUCKET)); + fields.add(ManifestEntry.MANIFEST_ROW_TYPE.getField(ManifestEntry.TOTAL_BUCKETS)); + return ProjectedManifestEntry.Projection.create(new RowType(false, fields)); + } + + /** + * Rebuild from the final physical blocks, including raw-copy and encoded rewrite paths. Callers + * decide whether to build a sidecar. Partition coverage is always generated. + */ + public static byte[] build( + FileIO io, + Path path, + long size, + long records, + boolean rowIdEnabled, + boolean bucketEnabled) + throws IOException { + try (ManifestAvroReader reader = new ManifestAvroReader(io.newInputStream(path))) { + Builder builder = new Builder(reader.headerBytes(), rowIdEnabled, bucketEnabled); + ProjectedManifestEntry.Projection projection = BLOCK_INDEX_PROJECTION; + ProjectedManifestEntry entry = projection.createEntry(); + while (reader.hasNext()) { + ManifestAvroReader.RawBlock block = reader.next(); + builder.beginBlock(reader.blockOffset(), reader.blockLength(), block.recordCount()); + ManifestAvroReader.RowIterator rows = block.toRows(projection.projectedType()); + while (rows.hasNext()) { + entry.replace(rows.next()); + builder.add( + rowIdEnabled ? entry.file().firstRowId() : null, + rowIdEnabled ? entry.file().rowCount() : 0, + entry.partitionBytes(), + bucketEnabled ? entry.bucket() : null, + bucketEnabled ? entry.totalBuckets() : null); + } + builder.endBlock(); + } + return builder.serialize(size, records); + } + } + + /** Selects blocks using row-ID coverage. A null query retains every block after validation. */ + public static Selection select( + byte[] data, ManifestFileMeta manifest, @Nullable RowRangeIndex query) + throws IOException { + return select(data, manifest, query, null, null); + } + + /** Validates framing and tests row ID, partition, then bucket coverage. */ + public static Selection select( + byte[] data, + ManifestFileMeta manifest, + @Nullable RowRangeIndex query, + @Nullable PartitionPredicate partitionFilter, + @Nullable RowType partitionType) + throws IOException { + return select(data, manifest, query, partitionFilter, partitionType, null); + } + + /** + * Selects blocks using independent filters. The bucket predicate must conservatively test only + * the bucket and recorded total bucket count; omit it if filtering requires an entry partition. + */ + public static Selection select( + byte[] data, + ManifestFileMeta manifest, + @Nullable RowRangeIndex query, + @Nullable PartitionPredicate partitionFilter, + @Nullable RowType partitionType, + @Nullable BiPredicate bucketFilter) + throws IOException { + require(data.length >= MIN_HEADER_BYTES + CHECKSUM_BYTES); + int limit = data.length - CHECKSUM_BYTES; + CRC32 crc = new CRC32(); + crc.update(data, 0, limit); + require((int) crc.getValue() == ByteBuffer.wrap(data, limit, CHECKSUM_BYTES).getInt()); + ByteBuffer in = ByteBuffer.wrap(data, 0, limit).slice(); + require(in.getInt() == MAGIC); + require(decodeInt(in) == FORMAT_VERSION); + long entries = Math.addExact(manifest.numAddedFiles(), manifest.numDeletedFiles()); + require(entries >= 0); + int headerLength = decodeInt(in); + require(headerLength >= 21 && headerLength <= in.remaining() - 2); + require(headerLength <= manifest.fileSize()); + byte[] header = new byte[headerLength]; + in.get(header); + require(header[0] == 'O' && header[1] == 'b' && header[2] == 'j' && header[3] == 1); + int partitions = decodeInt(in); + require(partitions <= in.remaining() / 13); + boolean[] matches = partitionFilter == null ? null : new boolean[partitions]; + Set unique = new java.util.HashSet<>(); + for (int id = 0; id < partitions; id++) { + int length = decodeInt(in); + require(length >= 12 && length <= in.remaining()); + ByteBuffer encoded = in.slice(); + encoded.limit(length); + int arity = encoded.getInt(0); + require(arity >= 0 && 4L + ((arity + 71L) / 64) * 8 + arity * 8L <= length); + require(partitionType == null || arity == partitionType.getFieldCount()); + require(unique.add(encoded.asReadOnlyBuffer())); + if (partitionFilter != null) { + byte[] bytes = new byte[length]; + encoded.get(bytes); + BinaryRow partition = SerializationUtils.deserializeBinaryRow(bytes); + matches[id] = partitionFilter.test(partition); + } + in.position(in.position() + length); + } + int count = decodeInt(in); + require(count <= in.remaining() / MIN_BLOCK_BYTES); + long nextOffset = headerLength; + long firstRecord = 0; + List selected = new ArrayList<>(); + for (int i = 0; i < count; i++) { + require(in.remaining() >= MIN_BLOCK_BYTES); + long offset = VarLengthIntUtils.decodeLong(in); + long length = VarLengthIntUtils.decodeLong(in); + long records = VarLengthIntUtils.decodeLong(in); + require(offset == nextOffset && length > 0 && length <= manifest.fileSize() - offset); + require(records > 0 && records <= entries - firstRecord); + ByteBuffer partitionPayload = payload(in); + ByteBuffer rowPayload = payload(in); + ByteBuffer bucketPayload = payload(in); + DeltaVarintCodec.Reader ids = null; + if (partitionPayload != null) { + ids = new DeltaVarintCodec.Reader(partitionPayload, 0, partitions - 1L); + require(ids.count() > 0 && ids.count() <= records && ids.count() <= partitions); + } + long min = 0; + long max = 0; + DeltaVarintCodec.Reader endpoints = null; + if (rowPayload != null) { + require(rowPayload.remaining() >= 2 * Long.BYTES + 1); + min = rowPayload.getLong(); + max = rowPayload.getLong(); + endpoints = new DeltaVarintCodec.Reader(rowPayload, min, max); + require(endpoints.count() % 2 == 0 && endpoints.count() / 2L < records); + require(endpoints.count() != 0 || !rowPayload.hasRemaining()); + } + if (bucketPayload != null) { + ByteBuffer prefix = bucketPayload.duplicate(); + int pairs = decodeInt(prefix); + require(pairs > 0 && pairs <= records && 2L * pairs + 1 <= prefix.remaining()); + } + long blockFirstRecord = firstRecord; + nextOffset = offset + length; + firstRecord += records; + + if (query != null && rowPayload != null) { + if (!query.intersects(min, max)) { + continue; + } + int rangeCount = endpoints.count() / 2 + 1; + boolean rowHit = rangeCount == 1; + long start = min; + for (int range = 0; !rowHit && range < rangeCount; range++) { + long end = range + 1 == rangeCount ? max : endpoints.next(); + require(end >= start); + rowHit = query.intersects(start, end); + if (!rowHit && range + 1 < rangeCount) { + start = endpoints.next(); + require(start > end); + } + require(endpoints.hasNext() || !rowPayload.hasRemaining()); + } + if (!rowHit) { + continue; + } + } + + if (partitionFilter != null && partitionPayload != null) { + boolean partitionHit = false; + long previous = -1; + while (!partitionHit && ids.hasNext()) { + long id = ids.next(); + require(id > previous); + require(ids.hasNext() || !partitionPayload.hasRemaining()); + previous = id; + partitionHit = matches[(int) id]; + } + if (!partitionHit) { + continue; + } + } + + if (bucketFilter != null && bucketPayload != null) { + // Locate the second count-prefixed sequence without allocating value arrays. + ByteBuffer totalsData = bucketPayload.duplicate(); + DeltaVarintCodec.Reader directory = + new DeltaVarintCodec.Reader(totalsData, 0, Integer.MAX_VALUE); + while (directory.hasNext()) { + directory.next(); + } + bucketPayload.limit(totalsData.position()); + DeltaVarintCodec.Reader buckets = + new DeltaVarintCodec.Reader(bucketPayload, 0, Integer.MAX_VALUE); + DeltaVarintCodec.Reader totals = + new DeltaVarintCodec.Reader(totalsData, 0, Integer.MAX_VALUE, true); + require(totals.count() == buckets.count()); + boolean bucketHit = false; + int previousBucket = -1; + int previousTotal = -1; + while (!bucketHit && buckets.hasNext()) { + int bucket = (int) buckets.next(); + int totalBuckets = (int) totals.next(); + require(totalBuckets > bucket); + require( + bucket > previousBucket + || (bucket == previousBucket && totalBuckets > previousTotal)); + require( + buckets.hasNext() + || (!bucketPayload.hasRemaining() + && !totalsData.hasRemaining())); + previousBucket = bucket; + previousTotal = totalBuckets; + bucketHit = bucketFilter.test(bucket, totalBuckets); + } + if (!bucketHit) { + continue; + } + } + selected.add(new Block(offset, length, blockFirstRecord, records)); + } + require(!in.hasRemaining() && nextOffset == manifest.fileSize() && firstRecord == entries); + return new Selection(header, selected); + } + + /** Reads framing without expanding the compressed contents. */ + @Nullable + private static ByteBuffer payload(ByteBuffer in) throws IOException { + require(in.hasRemaining()); + int encoding = Byte.toUnsignedInt(in.get()); + if (encoding == 0) { + return null; + } + int length = decodeInt(in); + require(length <= in.remaining()); + ByteBuffer result = in.slice(); + result.limit(length); + in.position(in.position() + length); + if (encoding != 1) { + return null; + } + return result; + } + + /** Reads the complete sidecar. Null means read the original manifest. */ + @Nullable + public static Selection read( + FileIO io, Path path, ManifestFileMeta manifest, @Nullable RowRangeIndex query) { + return read(io, path, manifest, query, null, null); + } + + @Nullable + public static Selection read( + FileIO io, + Path path, + ManifestFileMeta manifest, + @Nullable RowRangeIndex query, + @Nullable PartitionPredicate partitionFilter, + @Nullable RowType partitionType) { + return read(io, path, manifest, query, partitionFilter, partitionType, null, null); + } + + @Nullable + public static Selection read( + FileIO io, + Path path, + ManifestFileMeta manifest, + @Nullable RowRangeIndex query, + @Nullable PartitionPredicate partitionFilter, + @Nullable RowType partitionType, + @Nullable BiPredicate bucketFilter, + @Nullable SegmentsCache cache) { + String sidecarFileName = fileName(manifest); + if (sidecarFileName == null) { + return null; + } + try { + Path sidecarPath = new Path(path.getParent(), sidecarFileName); + Segments cached = cache == null ? null : cache.getIfPresents(sidecarPath); + boolean cacheHit = cached instanceof ManifestSidecarSegment; + byte[] data = + cacheHit + ? ((ManifestSidecarSegment) cached).bytes() + : readBytes(io, sidecarPath); + Selection selection = + select(data, manifest, query, partitionFilter, partitionType, bucketFilter); + if (cache != null && !cacheHit && data.length <= cache.maxElementSize()) { + cache.put(sidecarPath, new ManifestSidecarSegment(data)); + } + return selection; + } catch (IOException failure) { + if (Thread.currentThread().isInterrupted()) { + throw new UncheckedIOException(failure); + } + LOG.debug("Cannot use manifest sidecar for {}; reading manifest", path, failure); + return null; + } + } + + /** Complete sidecar bytes stored in the shared manifest cache. */ + static final class ManifestSidecarSegment implements Segments { + private final byte[] bytes; + + public ManifestSidecarSegment(byte[] bytes) { + this.bytes = bytes; + } + + public byte[] bytes() { + return bytes; + } + + @Override + public long totalMemorySize() { + return bytes.length; + } + } + + private static byte[] readBytes(FileIO io, Path path) throws IOException { + try (InputStream in = io.newInputStream(path)) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buffer = new byte[SIDECAR_READ_BUFFER_BYTES]; + int n; + while ((n = in.read(buffer, 0, buffer.length)) != -1) { + out.write(buffer, 0, n); + } + return out.toByteArray(); + } + } + + static InputStream openManifest(FileIO io, Path path, @Nullable Selection selected) + throws IOException { + return openManifest(io, path, selected, null); + } + + static InputStream openManifest( + FileIO io, + Path path, + @Nullable Selection selected, + @Nullable SegmentsCache cache) + throws IOException { + return selected == null + ? io.newInputStream(path) + : new SelectedBlockInput(io, path, selected, cache); + } + + /** Separates physical byte ranges from whole-file cache keys. */ + static final class BlockCacheKey { + private final Path path; + private final long offset; + private final long length; + + BlockCacheKey(Path path, long offset, long length) { + this.path = Objects.requireNonNull(path); + this.offset = offset; + this.length = length; + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof BlockCacheKey)) { + return false; + } + BlockCacheKey that = (BlockCacheKey) other; + return path.equals(that.path) && offset == that.offset && length == that.length; + } + + @Override + public int hashCode() { + return Objects.hash(path, offset, length); + } + } + + /** Complete encoded Avro blocks, distinct from cached manifest entries and sidecar bytes. */ + private static final class ManifestBlockSegment implements Segments { + private final byte[] bytes; + + private ManifestBlockSegment(byte[] bytes) { + this.bytes = bytes; + } + + @Override + public long totalMemorySize() { + return bytes.length; + } + } + + /** An OCF stream comprising the original header and selected complete compressed blocks. */ + private static final class SelectedBlockInput extends InputStream { + private final FileIO io; + private final Path path; + private final Selection selected; + @Nullable private final SegmentsCache cache; + @Nullable private SeekableInputStream input; + private boolean closed; + private int headerPosition; + private int blockPosition; + private long remaining; + private byte[] buffer; + private int bufferPosition; + private int bufferLimit; + + private SelectedBlockInput( + FileIO io, Path path, Selection selected, @Nullable SegmentsCache cache) { + this.io = io; + this.path = path; + this.selected = selected; + this.cache = cache; + } + + @Override + public int read() throws IOException { + ensureOpen(); + 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 { + ensureOpen(); + 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 next = selected.blocks.get(blockPosition); + if (cache != null && next.length <= cache.maxElementSize()) { + readCachedBlocks(next); + return true; + } + Block block = selected.blocks.get(blockPosition++); + long end = block.offset + block.length; + while (blockPosition < selected.blocks.size() + && selected.blocks.get(blockPosition).offset == end + && (cache == null + || selected.blocks.get(blockPosition).length + > cache.maxElementSize())) { + end += selected.blocks.get(blockPosition++).length; + } + seekInput(block.offset); + remaining = end - block.offset; + } + int requested = (int) Math.min(BLOCK_READ_BUFFER_BYTES, remaining); + // A previous buffer may be shared with other readers through the block cache. + if (cache != null || buffer == null || buffer.length < requested) { + buffer = new byte[requested]; + } + bufferPosition = 0; + bufferLimit = 0; + readFully(buffer, requested); + bufferLimit = requested; + remaining -= requested; + return true; + } + + private void readCachedBlocks(Block first) throws IOException { + byte[] cached = cachedBlock(first); + if (cached != null) { + blockPosition++; + buffer = cached; + } else { + int firstPosition = blockPosition++; + long end = first.offset + first.length; + while (blockPosition < selected.blocks.size()) { + Block next = selected.blocks.get(blockPosition); + if (next.offset != end + || next.length > cache.maxElementSize() + || end - first.offset + next.length > BLOCK_READ_BUFFER_BYTES + || cachedBlock(next) != null) { + break; + } + end += next.length; + blockPosition++; + } + + byte[] bytes = new byte[(int) (end - first.offset)]; + seekInput(first.offset); + readFully(bytes, bytes.length); + // Publish only complete reads, and use individual block keys so overlapping + // selections can share data even when their coalesced read spans differ. + int offset = 0; + for (int i = firstPosition; i < blockPosition; i++) { + Block block = selected.blocks.get(i); + int length = (int) block.length; + byte[] blockBytes = + length == bytes.length + ? bytes + : Arrays.copyOfRange(bytes, offset, offset + length); + cache.put( + new BlockCacheKey(path, block.offset, block.length), + new ManifestBlockSegment(blockBytes)); + offset += length; + } + buffer = bytes; + } + bufferPosition = 0; + bufferLimit = buffer.length; + } + + @Nullable + private byte[] cachedBlock(Block block) { + Segments cached = + cache.getIfPresents(new BlockCacheKey(path, block.offset, block.length)); + if (cached instanceof ManifestBlockSegment) { + byte[] bytes = ((ManifestBlockSegment) cached).bytes; + if (bytes.length == block.length) { + return bytes; + } + } + return null; + } + + private void seekInput(long offset) throws IOException { + if (input == null) { + input = io.newInputStream(path); + } + input.seek(offset); + } + + private void readFully(byte[] bytes, int length) throws IOException { + int position = 0; + while (position < length) { + int count = + input.read( + bytes, + position, + Math.min(BLOCK_READ_BUFFER_BYTES, length - position)); + if (count < 0) { + throw new EOFException("Truncated manifest block"); + } + position += count; + } + } + + private void ensureOpen() throws IOException { + if (closed) { + throw new IOException("Manifest stream is closed"); + } + } + + @Override + public void close() throws IOException { + if (!closed) { + closed = true; + if (input != null) { + input.close(); + } + } + } + } + + private static void require(boolean valid) throws IOException { + if (!valid) { + throw new IOException("Invalid, unsupported or mismatched manifest sidecar"); + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestBlockIndexTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestBlockIndexTest.java new file mode 100644 index 000000000000..77c3cc9c9e69 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestBlockIndexTest.java @@ -0,0 +1,891 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RowRangeIndex; +import org.apache.paimon.utils.SerializationUtils; +import org.apache.paimon.utils.VarLengthIntUtils; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.function.BiPredicate; +import java.util.zip.CRC32; + +import static org.apache.paimon.manifest.ManifestSidecarTest.header; +import static org.apache.paimon.manifest.ManifestSidecarTest.meta; +import static org.apache.paimon.manifest.ManifestSidecarTest.partition; +import static org.apache.paimon.manifest.ManifestSidecarTest.testSidecar; +import static org.apache.paimon.manifest.ManifestSidecarTest.verifyGoldenFile; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +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.verifyNoInteractions; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +/** Complete sidecar payload format, compression and independent filtering. */ +class ManifestBlockIndexTest { + + private final RowType type = RowType.of(DataTypes.INT(), DataTypes.STRING()); + + private RowRangeIndex query(long point) { + return RowRangeIndex.create(Collections.singletonList(new Range(point, point))); + } + + private static BiPredicate bucketFilter(int expected) { + return new BiPredicate() { + @Override + public boolean test(Integer bucket, Integer totalBuckets) { + return bucket == expected; + } + }; + } + + private PartitionPredicate part(int value) { + return PartitionPredicate.fromPredicate(type, new PredicateBuilder(type).equal(0, value)); + } + + @Test + void testV1GoldenFile() throws Exception { + int blockCount = 10; + int entriesPerBlock = 16; + int blockLength = 1024 * 1024 + 17; + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + for (int block = 0; block < blockCount; block++) { + builder.beginBlock( + header.length + (long) block * blockLength, blockLength, entriesPerBlock); + long rowBase = (long) block << 40; + int step = block % 4 == 0 ? 16400 : 16; + for (int entry = 0; entry < entriesPerBlock; entry++) { + int p = (block * 7 + entry * 3) % 16; + builder.add( + rowBase + (long) entry * step, + 3, + partition(p, p % 7 == 0 ? null : "partition-" + p), + block + entry * 13, + entry % 2 == 0 ? 512 : 1024); + } + builder.endBlock(); + } + long fileSize = header.length + (long) blockCount * blockLength; + long entryCount = (long) blockCount * entriesPerBlock; + byte[] current = builder.serialize(fileSize, entryCount); + assertThat(current.length).isBetween(2 * 1024, 2 * 1024 + 256); + byte[] data = verifyGoldenFile(current); + ManifestFileMeta meta = meta("manifest-golden-v1", fileSize, entryCount); + assertThat(partitionCount(data)).isEqualTo(16); + List blocks = ManifestSidecar.select(data, meta, null).blocks(); + assertThat(blocks).hasSize(blockCount); + for (int block = 0; block < blockCount; block++) { + ManifestSidecar.Block expected = blocks.get(block); + assertThat(expected.offset).isEqualTo(header.length + (long) block * blockLength); + assertThat(expected.length).isEqualTo(blockLength); + assertThat(expected.firstRecord).isEqualTo((long) block * entriesPerBlock); + assertThat(expected.recordCount).isEqualTo(entriesPerBlock); + int step = block % 4 == 0 ? 16400 : 16; + for (int entry = 0; entry < entriesPerBlock; entry++) { + long rowId = ((long) block << 40) + (long) entry * step; + int p = (block * 7 + entry * 3) % 16; + int bucket = block + entry * 13; + int total = entry % 2 == 0 ? 512 : 1024; + assertThat( + ManifestSidecar.select( + data, + meta, + query(rowId), + part(p), + type, + (b, t) -> b == bucket && t == total) + .blocks()) + .extracting(b -> b.firstRecord) + .containsExactly((long) block * entriesPerBlock); + assertThat(ManifestSidecar.select(data, meta, query(rowId + 3)).blocks()).isEmpty(); + } + } + assertThat(ManifestSidecar.select(data, meta, null, part(99), type).blocks()).isEmpty(); + assertThat( + ManifestSidecar.select(data, meta, null, null, type, (b, t) -> t == 4096) + .blocks()) + .isEmpty(); + } + + @Test + void partitionCoverageIsAlwaysGenerated() throws Exception { + byte[] header = header(); + for (int mask = 0; mask < 4; mask++) { + boolean rowIdEnabled = (mask & 1) != 0; + boolean bucketEnabled = (mask & 2) != 0; + ManifestSidecar.Builder builder = + new ManifestSidecar.Builder(header, rowIdEnabled, bucketEnabled); + for (int block = 0; block < 2; block++) { + builder.beginBlock(header.length + block * 100L, 100, 1); + builder.add(100L + block * 100L, 10, partition(7 + block, "p"), 1, 4); + builder.endBlock(); + } + byte[] data = builder.serialize(header.length + 200, 2); + assertThat(partitionCount(data)).isEqualTo(2); + for (int[] position : positions(data)) { + assertThat(data[position[1]]).isEqualTo((byte) 1); + assertThat(data[position[2]]).isEqualTo((byte) (rowIdEnabled ? 1 : 0)); + assertThat(data[position[3]]).isEqualTo((byte) (bucketEnabled ? 1 : 0)); + } + ManifestFileMeta meta = meta("m", header.length + 200, 2); + assertThat(ManifestSidecar.select(data, meta, query(999)).blocks()) + .hasSize(rowIdEnabled ? 0 : 2); + assertThat(ManifestSidecar.select(data, meta, null, part(99), type).blocks()).isEmpty(); + BiPredicate buckets = bucketFilter(99); + assertThat(ManifestSidecar.select(data, meta, null, null, type, buckets).blocks()) + .hasSize(bucketEnabled ? 0 : 2); + + // Generation settings do not disable payloads already stored in a sidecar. + byte[] existing = testSidecar(); + ManifestFileMeta existingMeta = meta("manifest-golden", header.length + 400, 7); + assertThat(ManifestSidecar.select(existing, existingMeta, query(999)).blocks()) + .isEmpty(); + assertThat( + ManifestSidecar.select(existing, existingMeta, null, part(99), type) + .blocks()) + .isEmpty(); + assertThat( + ManifestSidecar.select( + existing, existingMeta, null, null, type, buckets) + .blocks()) + .isEmpty(); + } + } + + private byte[] sidecarWithoutBuckets(boolean partitioned) throws IOException { + byte[] a = partitioned ? partition(7, "left") : null; + byte[] b = partitioned ? partition(9, null) : null; + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, false); + builder.beginBlock(header.length, 100, 3); + builder.add(0L, 10, a); + builder.add(5L, 5, a); + builder.add(20L, 5, b); + builder.endBlock(); + builder.beginBlock(header.length + 100, 200, 2); + builder.add((1L << 32) - 2, 5, b); + builder.add(8254058425445L, 1, a); + builder.endBlock(); + builder.beginBlock(header.length + 300, 100, 2); + builder.add(20L, 5, a); + builder.add(Long.MAX_VALUE, 1, b); + builder.endBlock(); + return builder.serialize(header.length + 400, 7); + } + + @Test + void partitionTuplesNullsAndDerivedOrdinals() throws Exception { + byte[] data = testSidecar(); + ManifestFileMeta meta = testMeta(); + PartitionPredicate filter = spy(part(7)); + assertThat(ManifestSidecar.select(data, meta, query(20), filter, type).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 5L); + verify(filter, times(2)).test(any(BinaryRow.class)); + PartitionPredicate nullFilter = + PartitionPredicate.fromPredicate(type, new PredicateBuilder(type).isNull(1)); + assertThat(ManifestSidecar.select(data, meta, null, nullFilter, type).blocks()).hasSize(3); + assertThat(ManifestSidecar.select(data, meta, null, part(99), type).blocks()).isEmpty(); + // Missing partition payloads cannot be pruned by dictionary misses. + assertThat( + ManifestSidecar.select( + sidecarWithoutBuckets(false), meta, null, part(99), type) + .blocks()) + .hasSize(3); + } + + @Test + void unavailableDimensionsAreIndependentAndDoNotPoisonLaterBlocks() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 1); + builder.add(null, 10, partition(7, "left")); + builder.endBlock(); + builder.beginBlock(header.length + 100, 100, 1); + builder.add(200L, 10, null); + builder.endBlock(); + builder.beginBlock(header.length + 200, 100, 1); + builder.add(300L, 10, partition(7, "left")); // an existing dictionary ID remains usable + builder.endBlock(); + byte[] data = builder.serialize(header.length + 300, 3); + ManifestFileMeta meta = meta("m", header.length + 300, 3); + assertThat(ManifestSidecar.select(data, meta, null, part(9), type).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(1L); + assertThat(ManifestSidecar.select(data, meta, query(999), part(7), type).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L); + assertThat(ManifestSidecar.select(data, meta, query(200), part(9), type).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(1L); + } + + private int partitionCount(byte[] data) throws IOException { + ByteBuffer in = ByteBuffer.wrap(data); + in.position(4); + VarLengthIntUtils.decodeLong(in); + int header = (int) VarLengthIntUtils.decodeLong(in); + in.position(in.position() + header); + return (int) VarLengthIntUtils.decodeLong(in); + } + + private List positions(byte[] data) throws IOException { + ByteBuffer in = ByteBuffer.wrap(data); + in.position(4); + VarLengthIntUtils.decodeLong(in); + int header = (int) VarLengthIntUtils.decodeLong(in); + in.position(in.position() + header); + int partitions = (int) VarLengthIntUtils.decodeLong(in); + for (int i = 0; i < partitions; i++) { + int length = (int) VarLengthIntUtils.decodeLong(in); + in.position(in.position() + length); + } + int blocks = (int) VarLengthIntUtils.decodeLong(in); + List result = new ArrayList<>(); + for (int i = 0; i < blocks; i++) { + int block = in.position(); + VarLengthIntUtils.decodeLong(in); + VarLengthIntUtils.decodeLong(in); + VarLengthIntUtils.decodeLong(in); + int partition = skipPayload(in); + int row = skipPayload(in); + int bucket = skipPayload(in); + result.add(new int[] {block, partition, row, bucket}); + } + return result; + } + + private int skipPayload(ByteBuffer in) throws IOException { + int position = in.position(); + if (in.get() != 0) { + int length = (int) VarLengthIntUtils.decodeLong(in); + in.position(in.position() + length); + } + return position; + } + + private byte[] checksum(byte[] data) { + int limit = data.length - Integer.BYTES; + CRC32 crc = new CRC32(); + crc.update(data, 0, limit); + ByteBuffer.wrap(data, limit, Integer.BYTES).putInt((int) crc.getValue()); + return data; + } + + @Test + void absentRowOrBucketFiltersKeepRemainingDimensions() throws Exception { + byte[] data = testSidecar(); + ManifestFileMeta meta = meta("manifest-golden", header().length + 400, 7); + assertThat( + ManifestSidecar.select(data, meta, null, part(7), type, bucketFilter(1)) + .blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L); + assertThat(ManifestSidecar.select(data, meta, query(20), part(7), type, null).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 5L); + assertThat(ManifestSidecar.select(data, meta, null, null, type, null).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 3L, 5L); + } + + @Test + void bucketPayloadAndTotalBucketsArePreserved() throws Exception { + byte[] header = header(); + byte[] data = testSidecar(); + ManifestFileMeta meta = meta("manifest-golden", header.length + 400, 7); + BiPredicate bucket = bucketFilter(1); + assertThat(ManifestSidecar.select(data, meta, null, null, type, bucket).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L); + // Existing row-id and partition coverage remains independently usable. + assertThat( + ManifestSidecar.select(data, meta, query(0), part(7), type, bucketFilter(2)) + .blocks()) + .isEmpty(); + BiPredicate filter = (bucketId, total) -> bucketId == 2 && total == 8; + assertThat(ManifestSidecar.select(data, meta, null, null, type, filter).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(3L); + // The caller can omit a bucket filter which requires an actual entry partition. + assertThat(ManifestSidecar.select(data, meta, null, null, type, null).blocks()).hasSize(3); + for (boolean partitioned : new boolean[] {false, true}) { + assertThat( + ManifestSidecar.select( + sidecarWithoutBuckets(partitioned), + meta, + null, + null, + type, + bucketFilter(99)) + .blocks()) + .hasSize(3); + } + } + + @Test + void splitBucketArraysPreserveDecreasingTotalsAndRescaling() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, false, true); + builder.beginBlock(header.length, 100, 5); + for (int[] pair : new int[][] {{3, 4}, {2, 8}, {0, Integer.MAX_VALUE}, {2, 4}, {2, 4}}) { + builder.add(null, 0, partition(7, "left"), pair[0], pair[1]); + } + builder.endBlock(); + byte[] data = builder.serialize(header.length + 100, 5); + ManifestFileMeta meta = meta("m", header.length + 100, 5); + List pairs = new ArrayList<>(); + assertThat( + ManifestSidecar.select( + data, + meta, + null, + null, + type, + (bucket, total) -> { + pairs.add(bucket + ":" + total); + return false; + }) + .blocks()) + .isEmpty(); + assertThat(pairs).containsExactly("0:2147483647", "2:4", "2:8", "3:4"); + assertThat( + ManifestSidecar.select( + data, + meta, + null, + null, + type, + (bucket, total) -> bucket == 2 && total == 4) + .blocks()) + .hasSize(1); + assertThat( + ManifestSidecar.select( + data, + meta, + null, + null, + type, + (bucket, total) -> bucket == 3 && total == 8) + .blocks()) + .isEmpty(); + } + + @Test + void mismatchedBucketCountsAndNegativeTotalsAreRejected() throws Exception { + for (byte[] payload : + Arrays.asList( + new byte[] {2, 1, 0, 1, 8, 8}, // Two buckets, one total, extra bytes. + bucketPayload(2, new long[] {1, 0}, 1, 0), // ZigZag -1 from zero. + bucketPayload(2, new long[] {1, 0}, 8, 0))) { // Repeated pair. + byte[] data = replacePayload(testSidecar(), 0, 3, payload); + assertThatThrownBy( + () -> + ManifestSidecar.select( + data, testMeta(), null, null, type, bucketFilter(99))) + .isInstanceOf(IOException.class); + } + } + + @Test + void unknownOrInvalidBucketPayloadIsUnavailable() throws Exception { + byte[] header = header(); + for (Integer[] pair : + Arrays.asList( + new Integer[] {null, null}, + new Integer[] {-1, 4}, + new Integer[] {4, 4}, + new Integer[] {0, 0})) { + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 2); + builder.add(100L, 10, partition(7, "left"), 1, 4); + builder.add(200L, 10, partition(7, "left"), pair[0], pair[1]); + builder.endBlock(); + builder.beginBlock(header.length + 100, 100, 1); + builder.add(300L, 10, partition(7, "left"), 1, 4); + builder.endBlock(); + byte[] data = builder.serialize(header.length + 200, 3); + int bucket = positions(data).get(0)[3]; + assertThat(data[bucket]).isZero(); + assertThat(positions(data).get(1)[0]).isEqualTo(bucket + 1); + ManifestFileMeta meta = meta("m", header.length + 200, 3); + assertThat( + ManifestSidecar.select(data, meta, null, null, type, bucketFilter(99)) + .blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L); + assertThat( + ManifestSidecar.select( + data, meta, query(999), null, type, bucketFilter(99)) + .blocks()) + .isEmpty(); + } + } + + @Test + void largePayloadsKeepExactCoverage() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + int blocks = 33; + int entriesPerBlock = 4097; + int entries = blocks * entriesPerBlock; + int blockBytes = 1024 * 1024; + for (int block = 0; block < blocks; block++) { + builder.beginBlock( + header.length + (long) block * blockBytes, blockBytes, entriesPerBlock); + for (int i = 0; i < entriesPerBlock; i++) { + int entry = block * entriesPerBlock + i; + builder.add(entry * 2L, 1, partition(entry, null), i, entriesPerBlock + 1); + } + builder.endBlock(); + } + long fileSize = header.length + (long) blocks * blockBytes; + byte[] data = builder.serialize(fileSize, entries); + ManifestFileMeta meta = meta("m", fileSize, entries); + long last = (entries - 1L) * 2; + assertThat(ManifestSidecar.select(data, meta, query(last)).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly((blocks - 1L) * entriesPerBlock); + assertThat(ManifestSidecar.select(data, meta, query(last - 1)).blocks()).isEmpty(); + assertThat(ManifestSidecar.select(data, meta, null, part(entries), type).blocks()) + .isEmpty(); + assertThat( + ManifestSidecar.select( + data, meta, null, null, type, bucketFilter(entriesPerBlock)) + .blocks()) + .isEmpty(); + } + + @Test + void absentPayloadsOmitLengthFieldsForEveryDimensionCombination() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + for (int mask = 0; mask < 8; mask++) { + builder.beginBlock(header.length + mask * 100L, 100, 1); + builder.add( + (mask & 2) == 0 ? null : 100L + mask, + 1, + (mask & 1) == 0 ? null : partition(7, "left"), + (mask & 4) == 0 ? null : 0, + (mask & 4) == 0 ? null : 1); + builder.endBlock(); + } + byte[] data = builder.serialize(header.length + 800, 8); + List positions = positions(data); + int[] presentSizes = {4, 19, 6}; + for (int mask = 0; mask < 8; mask++) { + for (int dimension = 0; dimension < 3; dimension++) { + int start = positions.get(mask)[dimension + 1]; + int end = + dimension < 2 + ? positions.get(mask)[dimension + 2] + : mask < 7 + ? positions.get(mask + 1)[0] + : data.length - Integer.BYTES; + boolean present = (mask & (1 << dimension)) != 0; + assertThat(data[start]).isEqualTo((byte) (present ? 1 : 0)); + assertThat(end - start).isEqualTo(present ? presentSizes[dimension] : 1); + } + } + ManifestFileMeta meta = meta("m", header.length + 800, 8); + assertThat(ManifestSidecar.select(data, meta, null, part(99), type).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 2L, 4L, 6L); + assertThat(ManifestSidecar.select(data, meta, query(999)).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 1L, 4L, 5L); + BiPredicate buckets = bucketFilter(99); + assertThat(ManifestSidecar.select(data, meta, null, null, type, buckets).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 1L, 2L, 3L); + assertThat(ManifestSidecar.select(data, meta, query(999), part(99), type, buckets).blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L); + } + + @Test + void deltaVarintsCompressSortedPayloadsWithoutCoarseningRowIds() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + int count = 10000; + builder.beginBlock(header.length, 100, count); + for (int i = count - 1; i >= 0; i--) { + builder.add(i * 4L, 3, partition(i, null), i, count); + } + builder.endBlock(); + byte[] data = builder.serialize(header.length + 100, count); + int[] block = positions(data).get(0); + assertThat(payloadLength(data, block[1])).isEqualTo(2 + count); + assertThat(payloadLength(data, block[2])).isEqualTo(16 + 3 + 2 * (count - 1)); + assertThat(payloadLength(data, block[3])).isEqualTo(2 * count + 6); + ManifestFileMeta meta = meta("m", header.length + 100, count); + assertThat(ManifestSidecar.select(data, meta, query(3)).blocks()).isEmpty(); + assertThat( + ManifestSidecar.select( + data, meta, query(4), part(9999), type, bucketFilter(9999)) + .blocks()) + .hasSize(1); + } + + @Test + void unpartitionedTablesStillRecordTheEmptyPartition() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, false, false); + builder.beginBlock(header.length, 100, 1); + builder.add(null, 0, SerializationUtils.serializeBinaryRow(BinaryRow.EMPTY_ROW)); + builder.endBlock(); + byte[] data = builder.serialize(header.length + 100, 1); + assertThat(partitionCount(data)).isEqualTo(1); + int[] block = positions(data).get(0); + assertThat(data[block[1]]).isEqualTo((byte) 1); + assertThat(data[block[2]]).isZero(); + assertThat(data[block[3]]).isZero(); + assertThat( + ManifestSidecar.select( + data, + meta("m", header.length + 100, 1), + null, + null, + RowType.of()) + .blocks()) + .hasSize(1); + } + + @Test + void rowMissSkipsPartitionAndBucketDecoding() throws Exception { + byte[] data = replacePayload(testSidecar(), 0, 1, compressedPayload(2, 999, 1)); + data = replacePayload(data, 0, 3, bucketPayload(2, new long[] {0, 0}, 0, 0)); + BiPredicate buckets = mock(BiPredicate.class); + assertThat( + ManifestSidecar.select(data, testMeta(), query(15), part(7), type, buckets) + .blocks()) + .isEmpty(); + verifyNoInteractions(buckets); + } + + @Test + void partitionMissSkipsBucketDecodingWithOrWithoutRowQuery() throws Exception { + byte[] data = + replacePayload(testSidecar(), 0, 3, bucketPayload(2, new long[] {0, 0}, 0, 0)); + for (RowRangeIndex rows : Arrays.asList(null, query(0))) { + BiPredicate buckets = mock(BiPredicate.class); + assertThat( + ManifestSidecar.select(data, testMeta(), rows, part(99), type, buckets) + .blocks()) + .isEmpty(); + verifyNoInteractions(buckets); + } + } + + @Test + void absentPartitionFilterDoesNotDecodePartitionIds() throws Exception { + byte[] data = replacePayload(testSidecar(), 0, 1, compressedPayload(2, 999, 1)); + BiPredicate buckets = spy(bucketFilter(1)); + assertThat( + ManifestSidecar.select(data, testMeta(), query(20), null, type, buckets) + .blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L); + verify(buckets).test(1, 4); + verify(buckets).test(0, 1); + verify(buckets).test(3, 4); + verifyNoMoreInteractions(buckets); + } + + @Test + void matchesSkipUnusedDeltas() throws Exception { + byte[] partitions = replacePayload(testSidecar(), 0, 1, compressedPayload(2, 0, 999)); + assertThat(ManifestSidecar.select(partitions, testMeta(), query(0), part(7), type).blocks()) + .hasSize(1); + assertThatThrownBy( + () -> + ManifestSidecar.select( + partitions, testMeta(), query(0), part(99), type)) + .isInstanceOf(IOException.class); + byte[] buckets = + replacePayload( + testSidecar(), + 0, + 3, + bucketPayload(2, new long[] {1, 0}, 8, Long.MAX_VALUE)); + assertThat( + ManifestSidecar.select( + buckets, testMeta(), query(0), null, type, bucketFilter(1)) + .blocks()) + .hasSize(1); + assertThatThrownBy( + () -> + ManifestSidecar.select( + buckets, + testMeta(), + query(0), + null, + type, + bucketFilter(99))) + .isInstanceOf(IOException.class); + + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 3); + for (long first : new long[] {0, 20, 40}) { + builder.add(first, 10); + } + builder.endBlock(); + byte[] rows = + replacePayload( + builder.serialize(header.length + 100, 3), + 0, + 2, + rowPayload(3, 0, 49, 9, 11, 9, 99)); + ManifestFileMeta meta = meta("m", header.length + 100, 3); + assertThat(ManifestSidecar.select(rows, meta, query(0)).blocks()).hasSize(1); + assertThat(ManifestSidecar.select(rows, meta, query(20)).blocks()).hasSize(1); + assertThat(ManifestSidecar.select(rows, meta, query(100)).blocks()).isEmpty(); + assertThatThrownBy(() -> ManifestSidecar.select(rows, meta, query(35))) + .isInstanceOf(IOException.class); + } + + @Test + void malformedCompressedPayloadsFailWhenConsumed() throws Exception { + List badRows = + Arrays.asList( + rowPayload(2, 0, 24, 9), // Missing an endpoint. + rowPayload(2, 0, 24, 9, 0), // Overlapping intervals. + rowPayload(2, 0, 24, Long.MAX_VALUE, 0), // Exceeds the envelope. + rowPayload(2, 0, 24, 9, 11, 0), // More values than declared. + rowPayload(1, Long.MAX_VALUE, 1), // Reversed envelope. + rowPayload(1, -1, 24), + rowPayload(1, 0, -1), + Arrays.copyOf(rowPayload(1, 0, 24), 16), // Missing endpoint count. + rowPayload(1, 0, 24, 0)); // Unexpected value for a single interval. + for (byte[] payload : badRows) { + byte[] data = replacePayload(testSidecar(), 0, 2, payload); + assertThatThrownBy(() -> ManifestSidecar.select(data, testMeta(), query(15))) + .isInstanceOf(IOException.class); + } + for (byte[] payload : + Arrays.asList(compressedPayload(2, 999, 0), compressedPayload(2, 0, 0))) { + byte[] data = replacePayload(testSidecar(), 0, 1, payload); + assertThatThrownBy( + () -> + ManifestSidecar.select( + data, testMeta(), query(0), part(99), type)) + .isInstanceOf(IOException.class); + } + for (byte[] payload : + Arrays.asList( + bucketPayload(2, new long[] {0, 4}, 2, 0), + bucketPayload(2, new long[] {1L << 31, 4}, 8, 0), + bucketPayload(2, new long[] {0, 0}, 2, 0))) { + byte[] data = replacePayload(testSidecar(), 0, 3, payload); + assertThatThrownBy( + () -> + ManifestSidecar.select( + data, + testMeta(), + query(0), + null, + type, + bucketFilter(99))) + .isInstanceOf(IOException.class); + } + } + + @Test + void malformedDeltaVarintsFailWhenConsumed() throws Exception { + byte[] overlong = new byte[10]; + Arrays.fill(overlong, (byte) 0x80); + for (byte[] deltas : + Arrays.asList( + new byte[] {(byte) 0x80}, + new byte[] {(byte) 0x80, (byte) 0x80}, // Truncated delta. + new byte[] {(byte) 0x81, 0, 0}, // Noncanonical delta. + overlong)) { + for (int dimension = 1; dimension <= 3; dimension++) { + ByteArrayOutputStream payload = new ByteArrayOutputStream(); + payload.write( + dimension == 2 + ? rowPayload(2, 0, 24) + : dimension == 3 + ? bucketPayload(2, new long[] {1, 0}) + : compressedPayload(2)); + payload.write(deltas); + byte[] data = replacePayload(testSidecar(), 0, dimension, payload.toByteArray()); + int dim = dimension; + assertThatThrownBy( + () -> + ManifestSidecar.select( + data, + testMeta(), + query(0), + dim == 1 ? part(99) : null, + type, + dim == 3 ? bucketFilter(99) : null)) + .isInstanceOf(IOException.class); + } + } + } + + @Test + void invalidCountFramingAndDirectoryFailEvenWhenFiltersMiss() throws Exception { + List bad = + Arrays.asList( + new byte[0], + new byte[] {(byte) 0x80}, // Incomplete varint count. + compressedPayload(-1, 1, 1), + compressedPayload(0, 1, 1), + compressedPayload(4, 1, 1), + compressedPayload(Integer.MAX_VALUE, 1, 1), + compressedPayload(1), // Count without payload data. + compressedPayload(2, 1)); + for (int dimension = 1; dimension <= 3; dimension++) { + for (byte[] payload : bad) { + byte[] data = replacePayload(testSidecar(), 0, dimension, payload); + assertThatThrownBy(() -> ManifestSidecar.select(data, testMeta(), query(999))) + .isInstanceOf(IOException.class); + } + byte[] data = testSidecar(); + int position = positions(data).get(0)[dimension]; + Arrays.fill(data, position + 1, position + 6, (byte) 0xff); + checksum(data); + assertThatThrownBy(() -> ManifestSidecar.select(data, testMeta(), query(999))) + .isInstanceOf(IOException.class); + } + byte[] missingEnvelope = + replacePayload(testSidecar(), 0, 2, Arrays.copyOf(rowPayload(1, 0, 24), 16)); + assertThatThrownBy(() -> ManifestSidecar.select(missingEnvelope, testMeta(), null)) + .isInstanceOf(IOException.class); + byte[] data = testSidecar(); + int block = positions(data).get(0)[0]; + ByteBuffer directory = ByteBuffer.wrap(data); + directory.position(block); + VarLengthIntUtils.decodeLong(directory); + VarLengthIntUtils.decodeLong(directory); + data[directory.position()] = 2; + checksum(data); + assertThatThrownBy(() -> ManifestSidecar.select(data, testMeta(), query(999))) + .isInstanceOf(IOException.class); + } + + @Test + void unknownEncodingsRemainIndependent() throws Exception { + for (int dimension = 1; dimension <= 3; dimension++) { + byte[] data = replacePayload(testSidecar(), 0, dimension, new byte[] {(byte) 0x80}); + data[positions(data).get(0)[dimension]] = (byte) 202; + checksum(data); + assertThat( + ManifestSidecar.select( + data, + testMeta(), + query(dimension == 2 ? 999 : 0), + part(dimension == 1 ? 99 : 7), + type, + bucketFilter(dimension == 3 ? 99 : 1)) + .blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L); + } + } + + private ManifestFileMeta testMeta() throws Exception { + return meta("manifest-golden", header().length + 400, 7); + } + + private byte[] replacePayload(byte[] data, int block, int dimension, byte[] payload) + throws Exception { + int start = positions(data).get(block)[dimension]; + ByteBuffer in = ByteBuffer.wrap(data); + in.position(start); + skipPayload(in); + int end = in.position(); + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(buffer); + out.write(data, 0, start); + out.writeByte(1); + VarLengthIntUtils.encodeLong(out, payload.length); + out.write(payload); + out.write(data, end, data.length - end); + return checksum(buffer.toByteArray()); + } + + private static byte[] compressedPayload(int count, long... values) throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(buffer); + out.write(deltaBytes(count)); + out.write(deltaBytes(values)); + return buffer.toByteArray(); + } + + private static byte[] rowPayload(int count, long min, long max, long... values) + throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(buffer); + out.writeLong(min); + out.writeLong(max); + out.write(deltaBytes(2L * (count - 1))); + out.write(deltaBytes(values)); + return buffer.toByteArray(); + } + + private static int payloadLength(byte[] data, int position) throws IOException { + ByteBuffer buffer = ByteBuffer.wrap(data); + buffer.position(position + 1); + return (int) VarLengthIntUtils.decodeLong(buffer); + } + + private static byte[] bucketPayload(int count, long[] bucketDeltas, long... totalDeltas) + throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + buffer.write(compressedPayload(count, bucketDeltas)); + buffer.write(compressedPayload(count, totalDeltas)); + return buffer.toByteArray(); + } + + private static byte[] deltaBytes(long... values) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + for (long value : values) { + while ((value & ~0x7fL) != 0) { + out.write((int) (value & 0x7f) | 0x80); + value >>>= 7; + } + out.write((int) value); + } + return out.toByteArray(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java new file mode 100644 index 000000000000..b4a7b3e29bf3 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestSidecarTest.java @@ -0,0 +1,1121 @@ +/* + * 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.data.BinaryRow; +import org.apache.paimon.data.BinaryRowWriter; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.SingleSegments; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.avro.AvroFileFormat; +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.manifest.ManifestSidecar.ManifestSidecarSegment; +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.CompatibilityUtils; +import org.apache.paimon.utils.IOUtils; +import org.apache.paimon.utils.PathFactory; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RowRangeIndex; +import org.apache.paimon.utils.SegmentsCache; +import org.apache.paimon.utils.SerializationUtils; + +import org.apache.avro.io.BinaryEncoder; +import org.apache.avro.io.EncoderFactory; +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.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CancellationException; +import java.util.zip.CRC32; + +import static org.apache.paimon.TestKeyValueGenerator.DEFAULT_PART_TYPE; +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.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** Golden file format, physical block positions, completeness and allocation bounds. */ +class ManifestSidecarTest { + + private static final String GENERATE_GOLDEN_FILES_PROPERTY = + "generateManifestSidecarGoldenFiles"; + + @TempDir java.nio.file.Path temp; + + 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 + ManifestSidecar.SUFFIX)); + when(meta.numAddedFiles()).thenReturn(entries); + return meta; + } + + static byte[] readGoldenFile() throws IOException { + String resource = "/compatibility/manifest-sidecar-v1"; + try (InputStream input = ManifestSidecarTest.class.getResourceAsStream(resource)) { + assertThat(input).as("Golden file %s", resource).isNotNull(); + return IOUtils.readFully(input, false); + } + } + + static byte[] verifyGoldenFile(byte[] current) throws IOException { + if (Boolean.parseBoolean( + System.getProperties().getProperty(GENERATE_GOLDEN_FILES_PROPERTY))) { + CompatibilityUtils.writeCompatibilityFile("manifest-sidecar-v1", current); + return current; + } + byte[] golden = readGoldenFile(); + assertThat(current).isEqualTo(golden); + return golden; + } + + static byte[] header() throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(output, null); + encoder.writeFixed(new byte[] {'O', 'b', 'j', 1}); + // Fix both metadata order and sync marker for deterministic golden files. + encoder.writeMapStart(); + encoder.setItemCount(2); + encoder.startItem(); + encoder.writeString("avro.codec"); + encoder.writeBytes("null".getBytes(StandardCharsets.UTF_8)); + encoder.startItem(); + encoder.writeString("avro.schema"); + encoder.writeBytes("\"long\"".getBytes(StandardCharsets.UTF_8)); + encoder.writeMapEnd(); + encoder.writeFixed(new byte[16]); + encoder.flush(); + return output.toByteArray(); + } + + static byte[] partition(int p, String q) { + BinaryRow row = new BinaryRow(2); + BinaryRowWriter writer = new BinaryRowWriter(row); + writer.writeInt(0, p); + if (q == null) { + writer.setNullAt(1); + } else { + writer.writeString(1, BinaryString.fromString(q)); + } + writer.complete(); + return SerializationUtils.serializeBinaryRow(row); + } + + static byte[] testSidecar() throws IOException { + byte[] header = header(); + byte[] a = partition(7, "left"); + byte[] b = partition(9, null); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 3); + builder.add(0L, 10, a, 1, 4); + builder.add(5L, 5, a, 1, 4); + builder.add(20L, 5, b, 1, 8); + builder.endBlock(); + builder.beginBlock(header.length + 100, 200, 2); + builder.add((1L << 32) - 2, 5, b, 2, 4); + builder.add(8254058425445L, 1, a, 2, 8); + builder.endBlock(); + builder.beginBlock(header.length + 300, 100, 2); + builder.add(20L, 5, a, 0, 1); + builder.add(Long.MAX_VALUE, 1, b, 3, 4); + builder.endBlock(); + return builder.serialize(header.length + 400, 7); + } + + private ManifestFileMeta testMeta() throws IOException { + return meta("manifest-golden", header().length + 400, 7); + } + + @Test + void emptyManifestHasACompleteSidecar() throws Exception { + byte[] header = header(); + byte[] data = new ManifestSidecar.Builder(header, true, true).serialize(header.length, 0); + assertThat(ByteBuffer.wrap(data).getInt()).isEqualTo(0x504d5343); + assertThat(ManifestSidecar.select(data, meta("empty", header.length, 0), null).blocks()) + .isEmpty(); + } + + @Test + void crc32FooterAndTruncatedSidecars() throws Exception { + byte[] good = testSidecar(); + ManifestFileMeta meta = testMeta(); + int limit = good.length - Integer.BYTES; + // Independently generated CRC32, stored in big-endian byte order. + assertThat(ByteBuffer.wrap(good, limit, Integer.BYTES).getInt()).isEqualTo(0xdf82cd30); + assertThat(ManifestSidecar.select(good, meta, null).blocks()).hasSize(3); + for (int position = limit; position < good.length; position++) { + byte[] bad = good.clone(); + bad[position] ^= 1; + assertThatThrownBy(() -> ManifestSidecar.select(bad, meta, null)) + .isInstanceOf(IOException.class); + } + for (int length = 0; length < good.length; length++) { + byte[] truncated = Arrays.copyOf(good, length); + assertThatThrownBy(() -> ManifestSidecar.select(truncated, meta, null)) + .isInstanceOf(IOException.class); + } + } + + @Test + void buildAndReadSelectedBlocksFromPhysicalManifests() throws Exception { + FileIO io = LocalFileIO.create(); + ManifestTestDataGenerator generator = ManifestTestDataGenerator.builder().build(); + List entries = new ArrayList<>(); + for (int i = 0; i < 4000; i++) { + ManifestEntry entry = generator.next(); + entries.add( + ManifestEntry.create( + i % 2 == 0 ? FileKind.ADD : FileKind.DELETE, + entry.partition(), + 1, + 4, + entry.file().newFirstRowId(i * 1000000L))); + } + Path sourcePath = new Path(temp.toString(), "manifest-source"); + ManifestFileMeta source; + try (ManifestAvroWriter writer = writer(io, sourcePath)) { + writer.write(entries); + writer.close(); + source = writer.result().get(0); + } + Path rewrittenPath = new Path(temp.toString(), "manifest-rewritten"); + ManifestFileMeta rewritten; + try (ManifestAvroWriter writer = writer(io, rewrittenPath); + ManifestAvroReader reader = new ManifestAvroReader(io.newInputStream(sourcePath))) { + writer.writeEncodedManifest(reader, source); + writer.close(); + rewritten = writer.result().get(0); + } + for (ManifestFileMeta meta : Arrays.asList(source, rewritten)) { + Path path = new Path(temp.toString(), meta.fileName()); + byte[] data = + ManifestSidecar.build(io, path, meta.fileSize(), entries.size(), true, true); + ManifestSidecar.Selection selected = + ManifestSidecar.select( + data, + meta, + RowRangeIndex.create( + Arrays.asList( + new Range(1000000000L, 1000000000L), + new Range(3000000000L, 3000000000L))), + null, + DEFAULT_PART_TYPE, + (bucket, totalBuckets) -> bucket == 1 && totalBuckets == 4); + assertThat(selected.blocks()).hasSize(2); + List expected = new ArrayList<>(); + for (ManifestSidecar.Block block : selected.blocks()) { + expected.addAll( + entries.subList( + (int) block.firstRecord, + (int) (block.firstRecord + block.recordCount))); + } + List actual = new ArrayList<>(); + try (ManifestAvroReader reader = + new ManifestAvroReader( + ManifestSidecar.openManifest(io, path, selected)); + CloseableIterator rows = + reader.read(ManifestEntry.MANIFEST_ROW_TYPE, null, null)) { + ManifestEntrySerializer serializer = new ManifestEntrySerializer(); + while (rows.hasNext()) { + actual.add(serializer.fromRow(rows.next())); + } + } + assertThat(actual).containsExactlyElementsOf(expected); + } + } + + private ManifestAvroWriter writer(FileIO io, Path path) { + PathFactory paths = mock(PathFactory.class); + when(paths.newPath()).thenReturn(path); + return new ManifestAvroWriter( + io, + new FileSystemSchemaManager(io, new Path(temp.toUri())), + DEFAULT_PART_TYPE, + (AvroFileFormat) FileFormat.fromIdentifier("avro", new Options()), + new ManifestEntrySerializer(), + "zstd", + paths, + Long.MAX_VALUE); + } + + @Test + void rowIdCoverageAndBlockOrdinals() throws Exception { + byte[] header = header(); + byte[] data = testSidecar(); + ManifestFileMeta meta = testMeta(); + 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(); + } + ManifestSidecar.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); + + ManifestSidecar.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(ManifestSidecar.select(data, meta, query).blocks()).isEmpty(); + assertThat(query.ranges()).containsExactly(new Range(10, 19), new Range(25, 40)); + } + + @Test + void minMaxSkipsExactIntersectionChecksAndHandlesOneInterval() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + 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(header.length + 300, 5); + ManifestFileMeta meta = meta("m", header.length + 300, 5); + RowRangeIndex outside = + spy(RowRangeIndex.create(Collections.singletonList(new Range(50, 59)))); + ManifestSidecar.Selection none = ManifestSidecar.select(data, meta, outside); + 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)))); + ManifestSidecar.Selection hit = ManifestSidecar.select(data, meta, one); + 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 singleIntervalHandlesBoundariesAndAbsentQueries() throws Exception { + byte[] header = header(); + for (Range range : + Arrays.asList( + new Range(0, 0), + new Range(42, 51), + new Range(Long.MAX_VALUE, Long.MAX_VALUE))) { + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 1); + builder.add(range.from, range.to - range.from + 1); + builder.endBlock(); + byte[] data = builder.serialize(header.length + 100, 1); + ManifestFileMeta meta = meta("m", header.length + 100, 1); + assertThat(ManifestSidecar.select(data, meta, null).blocks()).hasSize(1); + assertThat( + ManifestSidecar.select( + data, + meta, + RowRangeIndex.create(Collections.emptyList())) + .blocks()) + .isEmpty(); + for (long point : new long[] {range.from, range.to}) { + RowRangeIndex query = + spy( + RowRangeIndex.create( + Collections.singletonList(new Range(point, point)))); + assertThat(ManifestSidecar.select(data, meta, query).blocks()).hasSize(1); + verify(query).intersects(range.from, range.to); + } + long missing = range.from > 0 ? range.from - 1 : range.to + 1; + assertThat(select(data, meta, missing).blocks()).isEmpty(); + } + } + + @Test + void hugeRangesAreNotExpandedAndInvalidCoverageRetainsBlocks() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 2); + builder.add(0L, Long.MAX_VALUE); + builder.add(Long.MAX_VALUE, 1); + builder.endBlock(); + byte[] data = builder.serialize(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 ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 1); + builder.add(first, 2); + builder.endBlock(); + assertThat( + select( + builder.serialize(header.length + 100, 1), + meta("m", header.length + 100, 1), + 100) + .blocks()) + .hasSize(1); + } + for (long count : new long[] {0, -1}) { + builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 1); + builder.add(0L, count); + builder.endBlock(); + assertThat( + select( + builder.serialize(header.length + 100, 1), + meta("m", header.length + 100, 1), + 100) + .blocks()) + .hasSize(1); + } + builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 64); + for (int i = 0; i < 64; i++) { + builder.add(i * 10L, 1); + } + builder.endBlock(); + assertThat( + select( + builder.serialize(header.length + 100, 64), + meta("m", header.length + 100, 64), + 5) + .blocks()) + .isEmpty(); + } + + @Test + void cacheRespectsElementThreshold() throws Exception { + byte[] data = testSidecar(); + Path path = new Path(temp.toString(), "manifest-golden"); + Path sidecar = ManifestSidecar.path(path); + Files.write(temp.resolve(sidecar.getName()), data); + ManifestFileMeta meta = testMeta(); + FileIO io = spy(LocalFileIO.create()); + SegmentsCache tooSmall = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), data.length - 1L, null, false); + assertThat(readCached(io, path, meta, tooSmall).blocks()).hasSize(2); + assertThat(readCached(io, path, meta, tooSmall).blocks()).hasSize(2); + assertThat(tooSmall.getIfPresents(sidecar)).isNull(); + verify(io, times(2)).newInputStream(sidecar); + + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), data.length, null, false); + assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); + assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); + verify(io, times(3)).newInputStream(sidecar); + } + + @Test + void cachedSegmentsRequireSidecarType() throws Exception { + byte[] data = testSidecar(); + Path path = new Path(temp.toString(), "manifest-golden"); + Path sidecar = ManifestSidecar.path(path); + Files.write(temp.resolve(sidecar.getName()), data); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); + cache.put(sidecar, new SingleSegments(MemorySegment.wrap(data), data.length)); + FileIO io = spy(LocalFileIO.create()); + + assertThat(readCached(io, path, testMeta(), cache).blocks()).hasSize(2); + assertThat(cache.getIfPresents(sidecar)).isInstanceOf(ManifestSidecarSegment.class); + ManifestSidecarSegment cached = (ManifestSidecarSegment) cache.getIfPresents(sidecar); + assertThat(cached.bytes()).containsExactly(data); + assertThat(cached.totalMemorySize()).isEqualTo(data.length); + assertThat(readCached(io, path, testMeta(), cache).blocks()).hasSize(2); + verify(io, times(1)).newInputStream(sidecar); + } + + @Test + void missingAndInvalidSidecarsAreNotCached() throws Exception { + byte[] data = testSidecar(); + Path path = new Path(temp.toString(), "manifest-golden"); + Path sidecar = ManifestSidecar.path(path); + ManifestFileMeta meta = testMeta(); + FileIO io = spy(LocalFileIO.create()); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); + assertThat(readCached(io, path, meta, cache)).isNull(); + assertThat(cache.getIfPresents(sidecar)).isNull(); + byte[] corrupt = data.clone(); + corrupt[0] ^= 1; + Files.write(temp.resolve(sidecar.getName()), corrupt); + assertThat(readCached(io, path, meta, cache)).isNull(); + assertThat(cache.getIfPresents(sidecar)).isNull(); + Files.write(temp.resolve(sidecar.getName()), data); + assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); + assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); + verify(io, times(3)).newInputStream(sidecar); + } + + @Test + void cachedBytesPreservePerQueryCancellation() throws Exception { + Path path = new Path(temp.toString(), "manifest-golden"); + Path sidecar = ManifestSidecar.path(path); + Files.write(temp.resolve(sidecar.getName()), testSidecar()); + ManifestFileMeta meta = testMeta(); + FileIO io = spy(LocalFileIO.create()); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), Long.MAX_VALUE, null, false); + RowRangeIndex cancelled = mock(RowRangeIndex.class); + when(cancelled.intersects(anyLong(), anyLong())) + .thenThrow(new CancellationException("cancelled")); + assertThatThrownBy( + () -> + ManifestSidecar.read( + io, path, meta, cancelled, null, null, null, cache)) + .isInstanceOf(CancellationException.class); + assertThat(cache.getIfPresents(sidecar)).isNull(); + + assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); + assertThatThrownBy( + () -> + ManifestSidecar.read( + io, path, meta, cancelled, null, null, null, cache)) + .isInstanceOf(CancellationException.class); + assertThat(readCached(io, path, meta, cache).blocks()).hasSize(2); + verify(io, times(2)).newInputStream(sidecar); + } + + private ManifestSidecar.Selection readCached( + FileIO io, Path path, ManifestFileMeta meta, SegmentsCache cache) { + return ManifestSidecar.read( + io, + path, + meta, + RowRangeIndex.create(Collections.singletonList(new Range(20, 20))), + null, + null, + null, + cache); + } + + @Test + void corruptMissingIncompleteAndMismatchedIndexesFallback() throws Exception { + Path manifest = new Path(temp.toString(), "manifest-golden"); + java.nio.file.Path index = temp.resolve("manifest-golden" + ManifestSidecar.SUFFIX); + ManifestFileMeta meta = testMeta(); + RowRangeIndex query = RowRangeIndex.create(Collections.singletonList(new Range(11, 11))); + + assertThat(ManifestSidecar.read(LocalFileIO.create(), manifest, meta, query)).isNull(); + byte[] good = testSidecar(); + 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(ManifestSidecar.read(LocalFileIO.create(), manifest, meta, query)).isNull(); + } + // A valid checksum cannot make an unsupported container version readable. + for (int version : new int[] {0, 2, 99}) { + byte[] bad = good.clone(); + bad[4] = (byte) version; + int limit = bad.length - Integer.BYTES; + CRC32 crc = new CRC32(); + crc.update(bad, 0, limit); + ByteBuffer.wrap(bad, limit, Integer.BYTES).putInt((int) crc.getValue()); + assertThatThrownBy(() -> ManifestSidecar.select(bad, meta, query)) + .isInstanceOf(IOException.class); + } + Files.write(index, Arrays.copyOf(good, good.length - 1)); + assertThat(ManifestSidecar.read(LocalFileIO.create(), manifest, meta, query)).isNull(); + Files.write(index, good); + assertThat(ManifestSidecar.read(LocalFileIO.create(), manifest, meta, query).blocks()) + .isEmpty(); + // The sidecar is bound to physical coverage, not to a particular file name. + assertThat( + ManifestSidecar.select( + good, + meta("renamed", meta.fileSize(), 7), + RowRangeIndex.create( + Collections.singletonList(new Range(20, 20)))) + .blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 5L); + assertThatThrownBy( + () -> + ManifestSidecar.select( + good, meta("renamed", meta.fileSize() + 1, 7), query)) + .isInstanceOf(IOException.class); + assertThatThrownBy( + () -> + ManifestSidecar.select( + good, meta("renamed", meta.fileSize(), 8), query)) + .isInstanceOf(IOException.class); + } + + @Test + void ioFailuresFallBackWithoutInspectingNestedExceptions() throws Exception { + Path path = new Path(temp.toString(), "m"); + IOException suppressed = new IOException("read failed"); + suppressed.addSuppressed(new CancellationException("cancelled during close")); + for (IOException failure : + Arrays.asList( + new IOException("read failed"), + new java.net.SocketTimeoutException("timeout"), + new java.io.InterruptedIOException("no thread interruption flag"), + new IOException("wrapped", new InterruptedException("interrupted")), + suppressed)) { + FileIO fileIO = mock(FileIO.class); + when(fileIO.newInputStream(ManifestSidecar.path(path))).thenThrow(failure); + assertThat(ManifestSidecar.read(fileIO, path, meta("m", 1, 1), null)).isNull(); + assertThat(Thread.currentThread().isInterrupted()).isFalse(); + } + } + + @Test + void interruptedThreadDoesNotFallBackOnIoFailure() throws Exception { + Path path = new Path(temp.toString(), "m"); + IOException failure = new IOException("read failed"); + FileIO fileIO = mock(FileIO.class); + when(fileIO.newInputStream(ManifestSidecar.path(path))).thenThrow(failure); + try { + Thread.currentThread().interrupt(); + assertThatThrownBy(() -> ManifestSidecar.read(fileIO, path, meta("m", 1, 1), null)) + .isInstanceOf(java.io.UncheckedIOException.class) + .hasCauseReference(failure); + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + } finally { + Thread.interrupted(); + } + } + + @Test + void uncheckedFailuresPropagateUnchanged() throws Exception { + Path path = new Path(temp.toString(), "m"); + for (Throwable failure : + Arrays.asList( + new IllegalStateException("unexpected failure"), + new java.io.UncheckedIOException(new IOException("wrapped I/O")), + new CancellationException("cancelled"), + new AssertionError("error"))) { + FileIO fileIO = mock(FileIO.class); + when(fileIO.newInputStream(ManifestSidecar.path(path))).thenThrow(failure); + assertThatThrownBy(() -> ManifestSidecar.read(fileIO, path, meta("m", 1, 1), null)) + .isSameAs(failure); + } + } + + @Test + void indexReadsUseBoundedBulkRequests() throws Exception { + byte[] header = header(); + for (int blockCount : new int[] {5000, 25000, 131073}) { + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + 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(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(ManifestSidecar.path(path))).thenReturn(stream); + ManifestSidecar.Selection actual = + ManifestSidecar.read( + io, + path, + meta, + RowRangeIndex.create(Collections.singletonList(new Range(0, 0)))); + 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 sidecarsLargerThanTheFormerDefaultLimitAreReadCompletely() throws Exception { + byte[] header = header(); + BinaryRow partition = new BinaryRow(1); + BinaryRowWriter rowWriter = new BinaryRowWriter(partition); + byte[] value = new byte[17 * 1024 * 1024]; + rowWriter.writeBinary(0, value, 0, value.length); + rowWriter.complete(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, 100, 1); + builder.add(0L, 1, SerializationUtils.serializeBinaryRow(partition)); + builder.endBlock(); + byte[] data = builder.serialize(header.length + 100, 1); + assertThat(data.length).isGreaterThan(16 * 1024 * 1024); + Path path = new Path(temp.toString(), "manifest-large"); + CountingInput stream = new CountingInput(data, Integer.MAX_VALUE); + FileIO io = mock(FileIO.class); + when(io.newInputStream(ManifestSidecar.path(path))).thenReturn(stream); + assertThat( + ManifestSidecar.read( + io, + path, + meta("manifest-large", header.length + 100, 1), + null) + .blocks()) + .hasSize(1); + assertThat(stream.readLengths.stream().mapToInt(Integer::intValue).sum()) + .isEqualTo(data.length); + assertThat(stream.requests).allMatch(request -> request <= 1 << 20); + assertThat(stream.closed).isTrue(); + } + + @Test + void indexShortReadsReadTheWholeFile() throws Exception { + byte[] data = testSidecar(); + 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(ManifestSidecar.path(path))).thenReturn(stream); + ManifestSidecar.Selection actual = + ManifestSidecar.read( + io, + path, + testMeta(), + RowRangeIndex.create(Collections.singletonList(new Range(20, 20)))); + assertThat(actual.blocks()) + .extracting(block -> block.firstRecord) + .containsExactly(0L, 5L); + 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; + } + ManifestSidecar.Selection selected = + ManifestSidecar.select( + testSidecar(), + testMeta(), + RowRangeIndex.create( + Arrays.asList( + new Range(0, 0), + new Range(8254058425445L, 8254058425445L)))); + 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 = ManifestSidecar.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 = + ManifestSidecar.openManifest( + io, path, select(testSidecar(), testMeta(), 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(); + verifyNoInteractions(io); + } + assertThat(stream.closed).isEqualTo(point == 20); + } + } + + @Test + void cachedBlocksAreSharedByDifferentSelectionsWithoutOpeningTheManifest() throws Exception { + byte[] header = header(); + byte[] manifest = Arrays.copyOf(header, header.length + 400); + for (int i = header.length; i < manifest.length; i++) { + manifest[i] = (byte) i; + } + Path path = new Path(temp.toString(), "manifest-golden"); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), 400, null, false); + cache.put( + new ManifestSidecar.BlockCacheKey(path, header.length, 100), + new SingleSegments(MemorySegment.wrap(new byte[100]), 100)); + FileIO io = mock(FileIO.class); + CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); + when(io.newInputStream(path)).thenReturn(stream); + ManifestSidecar.Selection all = + ManifestSidecar.select( + testSidecar(), + testMeta(), + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); + } + assertThat(stream.readLengths).containsExactly(400); + assertThat(stream.seeks).containsExactly((long) header.length); + assertThat(cache.estimatedSize()).isEqualTo(3); + assertThat(cache.getIfPresents(path)).isNull(); + when(io.newInputStream(path)).thenThrow(new IOException("Must use cached blocks")); + for (long point : new long[] {20, 8254058425445L, 0}) { + ManifestSidecar.Selection selected = select(testSidecar(), testMeta(), point); + ByteArrayOutputStream expected = new ByteArrayOutputStream(); + expected.write(header); + for (ManifestSidecar.Block block : selected.blocks()) { + expected.write(manifest, (int) block.offset, (int) block.length); + } + try (InputStream in = ManifestSidecar.openManifest(io, path, selected, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(expected.toByteArray()); + } + } + verify(io, times(1)).newInputStream(path); + + Path other = new Path(temp.toString(), "other/manifest-golden"); + byte[] otherBytes = manifest.clone(); + otherBytes[header.length] ^= 1; + when(io.newInputStream(other)).thenReturn(new CountingInput(otherBytes, Integer.MAX_VALUE)); + try (InputStream in = + ManifestSidecar.openManifest( + io, other, select(testSidecar(), testMeta(), 0), cache)) { + assertThat(IOUtils.readFully(in, false)) + .isEqualTo(Arrays.copyOf(otherBytes, header.length + 100)); + } + verify(io).newInputStream(other); + } + + @Test + void mixedHitsAndMissesReadOnlyUncachedBlocks() throws Exception { + byte[] header = header(); + byte[] manifest = Arrays.copyOf(header, header.length + 400); + FileIO io = mock(FileIO.class); + Path path = new Path(temp.toString(), "manifest-golden"); + CountingInput cold = new CountingInput(manifest, Integer.MAX_VALUE); + CountingInput mixed = new CountingInput(manifest, Integer.MAX_VALUE); + when(io.newInputStream(path)).thenReturn(cold, mixed); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), 400, null, false); + try (InputStream in = + ManifestSidecar.openManifest( + io, path, select(testSidecar(), testMeta(), 8254058425445L), cache)) { + IOUtils.readFully(in, false); + } + ManifestSidecar.Selection all = + ManifestSidecar.select( + testSidecar(), + testMeta(), + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); + } + assertThat(mixed.readLengths).containsExactly(100, 100); + assertThat(mixed.seeks).containsExactly((long) header.length, header.length + 300L); + try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); + } + verify(io, times(2)).newInputStream(path); + } + + @Test + void truncatedCoalescedReadsDoNotPopulateTheBlockCache() throws Exception { + byte[] header = header(); + byte[] manifest = Arrays.copyOf(header, header.length + 400); + FileIO io = mock(FileIO.class); + Path path = new Path(temp.toString(), "manifest-golden"); + CountingInput truncated = + new CountingInput(Arrays.copyOf(manifest, manifest.length - 1), 7); + CountingInput complete = new CountingInput(manifest, 7); + when(io.newInputStream(path)).thenReturn(truncated, complete); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(1), 400, null, false); + ManifestSidecar.Selection all = + ManifestSidecar.select( + testSidecar(), + testMeta(), + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { + assertThatThrownBy(() -> IOUtils.readFully(in, false)).isInstanceOf(EOFException.class); + } + assertThat(truncated.closed).isTrue(); + assertThat(cache.estimatedSize()).isZero(); + try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); + } + assertThat(complete.closed).isTrue(); + assertThat(cache.estimatedSize()).isEqualTo(3); + } + + @Test + void evictedBlocksAreReadAgainWithinTheSharedBudget() throws Exception { + byte[] header = header(); + byte[] manifest = Arrays.copyOf(header, header.length + 400); + Path path = new Path(temp.toString(), "manifest-golden"); + FileIO io = mock(FileIO.class); + when(io.newInputStream(path)) + .thenAnswer(ignored -> new CountingInput(manifest, Integer.MAX_VALUE)); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofBytes(1300), 400, null, false); + ManifestSidecar.Selection all = + ManifestSidecar.select( + testSidecar(), + testMeta(), + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + for (int round = 0; round < 2; round++) { + try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); + } + assertThat(cache.totalCacheBytes()).isLessThanOrEqualTo(1300); + assertThat(cache.estimatedSize()).isEqualTo(1); + } + verify(io, times(2)).newInputStream(path); + } + + @Test + void oversizedBlocksUseBoundedReadsWithoutModifyingPreviouslyCachedBytes() throws Exception { + byte[] header = header(); + int cachedLength = (4 << 20) + 17; + int uncachedLength = 2 * (4 << 20) + 31; + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + builder.beginBlock(header.length, cachedLength, 1); + builder.add(0L, 1); + builder.endBlock(); + builder.beginBlock(header.length + cachedLength, uncachedLength, 1); + builder.add(100L, 1); + builder.endBlock(); + byte[] manifest = Arrays.copyOf(header, header.length + cachedLength + uncachedLength); + Arrays.fill(manifest, header.length, header.length + cachedLength, (byte) 7); + Arrays.fill(manifest, header.length + cachedLength, manifest.length, (byte) 9); + byte[] data = builder.serialize(manifest.length, 2); + ManifestFileMeta meta = meta("large", manifest.length, 2); + Path path = new Path(temp.toString(), "large"); + FileIO io = mock(FileIO.class); + CountingInput cold = new CountingInput(manifest, Integer.MAX_VALUE); + CountingInput mixed = new CountingInput(manifest, Integer.MAX_VALUE); + when(io.newInputStream(path)).thenReturn(cold, mixed); + SegmentsCache cache = + new SegmentsCache<>(1024, MemorySize.ofMebiBytes(8), cachedLength, null, false); + ManifestSidecar.Selection first = select(data, meta, 0); + byte[] expected = Arrays.copyOf(manifest, header.length + cachedLength); + try (InputStream in = ManifestSidecar.openManifest(io, path, first, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(expected); + } + assertThat(cold.readLengths).containsExactly(4 << 20, 17); + ManifestSidecar.Selection all = + ManifestSidecar.select( + data, + meta, + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + try (InputStream in = ManifestSidecar.openManifest(io, path, all, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(manifest); + } + assertThat(mixed.readLengths).containsExactly(4 << 20, 4 << 20, 31); + assertThat(cache.estimatedSize()).isEqualTo(1); + assertThat( + cache.getIfPresents( + new ManifestSidecar.BlockCacheKey( + path, header.length + cachedLength, uncachedLength))) + .isNull(); + try (InputStream in = ManifestSidecar.openManifest(io, path, first, cache)) { + assertThat(IOUtils.readFully(in, false)).isEqualTo(expected); + } + verify(io, times(2)).newInputStream(path); + } + + @Test + void largeBlockSpansUseBoundedReads() throws Exception { + byte[] header = header(); + ManifestSidecar.Builder builder = new ManifestSidecar.Builder(header, true, true); + long offset = header.length; + for (int length : new int[] {2 << 20, 2 << 20, 2 << 20, 2 << 20, 1 << 20}) { + builder.beginBlock(offset, length, 1); + builder.add(20L, 1); + builder.endBlock(); + offset += length; + } + byte[] data = builder.serialize(offset, 5); + byte[] manifest = Arrays.copyOf(header, (int) offset); + Path path = new Path(temp.toString(), "manifest-large"); + for (boolean withCache : new boolean[] {false, true}) { + CountingInput stream = new CountingInput(manifest, Integer.MAX_VALUE); + FileIO io = mock(FileIO.class); + when(io.newInputStream(path)).thenReturn(stream); + SegmentsCache cache = + withCache + ? new SegmentsCache<>( + 1024, MemorySize.ofMebiBytes(16), 4 << 20, null, false) + : null; + try (InputStream input = + ManifestSidecar.openManifest( + io, path, select(data, meta("manifest-large", offset, 5), 20), cache)) { + assertThat(IOUtils.readFully(input, false)).isEqualTo(manifest); + } + assertThat(stream.readLengths).containsExactly(4 << 20, 4 << 20, 1 << 20); + if (withCache) { + assertThat(stream.seeks) + .containsExactly( + (long) header.length, + header.length + (4L << 20), + header.length + (8L << 20)); + assertThat(cache.estimatedSize()).isEqualTo(5); + } else { + 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"); + ManifestSidecar.Selection selected = + ManifestSidecar.select( + testSidecar(), + testMeta(), + RowRangeIndex.create( + Collections.singletonList(new Range(0, Long.MAX_VALUE)))); + 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 = ManifestSidecar.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 ManifestSidecar.Selection select(byte[] data, ManifestFileMeta meta, long point) + throws IOException { + return ManifestSidecar.select( + data, + meta, + RowRangeIndex.create(Collections.singletonList(new Range(point, point)))); + } +} diff --git a/paimon-core/src/test/resources/compatibility/manifest-sidecar-v1 b/paimon-core/src/test/resources/compatibility/manifest-sidecar-v1 new file mode 100644 index 000000000000..53bf4c421e3f Binary files /dev/null and b/paimon-core/src/test/resources/compatibility/manifest-sidecar-v1 differ