diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 29c3dc0c24dd..2945dd829c35 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -441,25 +441,46 @@ Run arrays have equal lengths. Non-negative references select a video; `-1` mean means a data-evolution placeholder. Row `r` in a run starting at `s` maps to frame `run_first_frame + r - s`; gaps start new runs. -Keyframe-index v1 is little endian; byte offsets are relative to the video payload. +Each physical video has its own keyframe index; v1 and v2 may coexist in one `.video` +file. Fields are little endian; metadata and packet offsets refer to the video payload. + +Released v1 (Paimon 2.1.0) stores `uint8 version=1`, `uint64 magic=0x564944454F4B4649`, +`uint32 metadata count`, and `uint32 keyframe count`, followed by `(offset, length)` +int64 pairs and one zlib stream of `(ordinal, PTS, packet position)` int64 triples. +It permits at most 65,536 keyframes per physical video. + +Keyframe-index v2 adds a page directory and independently encoded pages: | Field | Size | | --- | ---: | -| Version (`1`) | 1 byte | +| Version (`2`) | 1 byte | | Magic (`0x564944454F4B4649`) | 8 bytes | | Metadata-range count | 4 bytes | | Keyframe count | 8 bytes | +| Page codec (`0`: raw; `1`: column-delta zlib) | 1 byte | | Metadata `(offset, length)` pairs | 16 bytes each | +| Page directory | 68 bytes per page | | Keyframe pages | Variable | -Each page holds 4,096 uncompressed `(frame ordinal, PTS, packet position)` int64 triples -and their 4-byte CRC32; the last page may be shorter. Page `p` starts at -`21 + 16 * metadata_count + p * 98308`. -Ordinals, PTS values and packet positions strictly increase; the first ordinal is zero. -The index covers the first video stream; its time base remains in the video. - -Readers validate each accessed page and its adjacent page boundaries; writers validate all pages. -Metadata ranges are limited to 65,536; keyframe count and total index size have no configured limit. +Each page except the last holds exactly 4,096 `(frame ordinal, PTS, packet position)` int64 triples. +Each directory record stores the index-relative page offset (8 bytes), encoded length (4), +uncompressed CRC32 (4), first and last triples (48), and CRC32 of the preceding 64 bytes (4). +Pages follow the directory contiguously and fill the remainder of the index. + +- Codec `0`: uncompressed row-major triples. +- Codec `1`: independently zlib-compressed pages of three columns (ordinal, PTS, position). + Each column stores its first value and successive deltas as little-endian uint64 words, + using arithmetic modulo 2^64. Decode by cumulative addition modulo 2^64, then restore + signed int64 row-major triples for CRC32 and ordering checks. + +Decoded pages must contain exactly `entry_count * 24` bytes; encoded pages cannot exceed +`4,096 * 24 + 64` bytes. Trailing or concatenated zlib streams are invalid. + +Ordinals, PTS and packet positions strictly increase from ordinal zero. The index covers the +first video stream, using its time base. Metadata ranges are limited to 65,536. +New writes use v2; copying or compaction preserves existing index bytes. Readers seek +through either version. Unknown versions/codecs fall back to scanning and survive rewrite; +corrupt indexes fail validation. An Arrow/data-file cell stores a separately versioned, little-endian `VideoFrameDescriptor`: diff --git a/docs/docs/pypaimon/video.md b/docs/docs/pypaimon/video.md index 4ca27adc9621..5c08ff9a7086 100644 --- a/docs/docs/pypaimon/video.md +++ b/docs/docs/pypaimon/video.md @@ -157,8 +157,12 @@ frames.add_video(video, episode_43_rows) ``` PyAV indexes supported ISO BMFF videos (e.g. MP4); other decodable videos fall back -to scanning. Index generation uses temporary disk; readers cache up to 1 MiB of -entry pages per index and 4,096 ordinal lookups per search view. +to scanning. Index generation uses temporary disk. + +New keyframe indexes use v2. Existing Paimon 2.1.0 v1 indexes remain readable through +the index and keep their version when copied or compacted. `video.index.compression` +selects `delta-zlib` (default) or `none` for newly generated v2 indexes. Readers +auto-detect the index version and encoding. Compression does not change video bytes. The writer deduplicates exact payload descriptor identity inside each `.video` file. Its video grouping policy coordinates normal, BLOB, and vector rolling diff --git a/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java b/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java index e8b6a0b51d30..f4afe7f34b25 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java @@ -1819,9 +1819,7 @@ public void testVideoRollingByBlobTargetSize() throws Exception { byte[] secondBytes = "second-video".getBytes(); byte[] keyframeIndex = fromHex( - "0149464b4f4544495601000000020000000000000000000000000000000100000000000000000000" - + "0000000000000000000000000000000000000000000c00000000000000a08c000000000000020000" - + "0000000000dfd9164b"); + "0249464b4f4544495601000000020000000000000000000000000000000001000000000000006a0000000000000030000000dfd9164b0000000000000000000000000000000000000000000000000c00000000000000a08c000000000000020000000000000023e343cf0000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000"); java.nio.file.Path firstSource = tempPath.resolve("size-first-source.mp4"); java.nio.file.Path secondSource = tempPath.resolve("size-second-source.mp4"); java.nio.file.Files.write(firstSource, concat("first-video".getBytes(), keyframeIndex)); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java index ff1480cd0c56..2aba8475c68c 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java @@ -1263,7 +1263,8 @@ public void testUpdateRawBlobColumnSplitsFilesAndKeepsUnchangedPayload() throws Collections.emptyList(), Collections.emptyList(), Collections.emptyMap())); - insertInto("BLOB_SPLIT_UPDATE_S", "(1, X'02')"); + // Two updated payloads force rolling regardless of the initial row-ID order. + insertInto("BLOB_SPLIT_UPDATE_S", "(1, X'02')", "(3, X'04')"); builder(warehouse, database, "BLOB_SPLIT_UPDATE_T") .withMergeCondition("BLOB_SPLIT_UPDATE_T.id=BLOB_SPLIT_UPDATE_S.id") @@ -1278,7 +1279,7 @@ public void testUpdateRawBlobColumnSplitsFilesAndKeepsUnchangedPayload() throws Arrays.asList( changelogRow("+I", 1, new byte[] {2}), changelogRow("+I", 2, large), - changelogRow("+I", 3, new byte[] {3}))); + changelogRow("+I", 3, new byte[] {4}))); List newBlobFiles = new ArrayList<>(); boolean sawOriginalLargeFile = false; diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java index ff3d9b97c247..23abd7e562fc 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java @@ -28,10 +28,14 @@ import org.apache.paimon.format.FormatWriter; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.DeltaVarintCompressor; import org.apache.paimon.utils.LongArrayList; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.io.IOException; import java.io.InputStream; import java.io.RandomAccessFile; @@ -52,6 +56,8 @@ */ public class VideoFormatWriter implements FileAwareFormatWriter { + private static final Logger LOG = LoggerFactory.getLogger(VideoFormatWriter.class); + public static final byte VERSION = 1; public static final int MAGIC_NUMBER = 0x4F454449; // "IDEO" in little endian public static final long NULL_REFERENCE = -1L; @@ -333,20 +339,38 @@ private boolean readIndex( } } spool.seek(start); - VideoKeyframeIndex.validate( - new InputStream() { - @Override - public int read() throws IOException { - return spool.read(); - } + try { + VideoKeyframeIndex.validate( + new SeekableInputStream() { + @Override + public void seek(long position) throws IOException { + spool.seek(position); + } - @Override - public int read(byte[] b, int offset, int length) throws IOException { - return spool.read(b, offset, length); - } - }, - expectedLength, - payloadLength); + @Override + public long getPos() throws IOException { + return spool.getFilePointer(); + } + + @Override + public void close() {} + + @Override + public int read() throws IOException { + return spool.read(); + } + + @Override + public int read(byte[] b, int offset, int length) + throws IOException { + return spool.read(b, offset, length); + } + }, + expectedLength, + payloadLength); + } catch (VideoKeyframeIndex.UnsupportedIndexException e) { + LOG.warn("Preserving unsupported video index: {}", e.getMessage()); + } spool.seek(start + expectedLength); recordSuccess(expectedLength); complete = true; diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoKeyframeIndex.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoKeyframeIndex.java index 307aac4cb3b7..dee7487dbac5 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoKeyframeIndex.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoKeyframeIndex.java @@ -18,15 +18,18 @@ package org.apache.paimon.format.blob; -import java.io.ByteArrayInputStream; +import org.apache.paimon.fs.ByteArraySeekableStream; +import org.apache.paimon.fs.SeekableInputStream; + import java.io.DataInputStream; import java.io.IOException; -import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.zip.CRC32; +import java.util.zip.DataFormatException; +import java.util.zip.Inflater; -/** Streaming validation for paged video seek indexes. */ +/** Validation for versioned video seek indexes. */ final class VideoKeyframeIndex { private static final long MAGIC = 0x564944454F4B4649L; private static final int ENTRY_LENGTH = 24; @@ -37,31 +40,175 @@ private VideoKeyframeIndex() {} static void validate(byte[] data, long payloadLength) { try { - validate(new ByteArrayInputStream(data), data.length, payloadLength); + validate(new ByteArraySeekableStream(data), data.length, payloadLength); } catch (IOException e) { throw new IllegalArgumentException("Invalid video keyframe index.", e); } } - static void validate(InputStream source, long length, long payloadLength) throws IOException { - if (length < 21 || payloadLength <= 0) { + static void validate(SeekableInputStream source, long length, long payloadLength) + throws IOException { + long base = source.getPos(); + if (length < 9 || payloadLength <= 0) { throw invalid("header or empty video payload"); } DataInputStream in = new DataInputStream(source); int version = in.readUnsignedByte(); long magic = Long.reverseBytes(in.readLong()); - long metadataCount = Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())); - if (version != 1 || magic != MAGIC) { - throw invalid("version or magic"); + if (magic != MAGIC) { + throw invalid("magic"); + } + if (version == 1) { + validateV1(source, length, payloadLength); + return; + } + if (version != 2) { + throw new UnsupportedIndexException("version " + version); } + if (length < 22) { + throw invalid("header"); + } + long metadataCount = Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())); long count = Long.reverseBytes(in.readLong()); - long offset = 21 + metadataCount * 16; + int codec = in.readUnsignedByte(); + if (codec > 1) { + throw new UnsupportedIndexException("codec " + codec); + } + long offset = 22 + metadataCount * 16; + if (metadataCount > MAX_METADATA_RANGE_COUNT) { + throw invalid("metadata range count exceeds limit"); + } + long pages = count > 0 ? (count - 1) / PAGE_ENTRIES + 1 : 0; + if (count <= 0 || offset >= length || pages > (length - offset) / 69) { + throw invalid("entries length"); + } + validateMetadataRanges(in, metadataCount, payloadLength); + byte[] record = new byte[68]; + long nextOffset = offset + pages * record.length; + long[] previous = null; + for (long page = 0; page < pages; page++) { + in.readFully(record); + ByteBuffer values = ByteBuffer.wrap(record).order(ByteOrder.LITTLE_ENDIAN); + long start = values.getLong(); + long size = Integer.toUnsignedLong(values.getInt()); + values.getInt(); + long[] first = entry(values); + long[] last = entry(values); + if (checksum(record, 64) != Integer.toUnsignedLong(values.getInt())) { + throw invalid("directory checksum"); + } + int entries = (int) Math.min(PAGE_ENTRIES, count - page * PAGE_ENTRIES); + if (start != nextOffset + || size <= 0 + || size > PAGE_ENTRIES * ENTRY_LENGTH + 64 + || start > length - size + || (codec == 0 && size != entries * ENTRY_LENGTH)) { + throw invalid("page offsets or length"); + } + validateEntry(first, previous, page == 0, payloadLength); + if (entries == 1) { + if (!java.util.Arrays.equals(first, last)) { + throw invalid("page endpoints"); + } + } else { + validateEntry(last, first, false, payloadLength); + } + previous = last; + nextOffset = start + size; + } + if (nextOffset != length) { + throw invalid("entries length"); + } + byte[] decoded = new byte[PAGE_ENTRIES * ENTRY_LENGTH]; + byte[] encoded = new byte[PAGE_ENTRIES * ENTRY_LENGTH + 64]; + for (long page = 0; page < pages; page++) { + source.seek(base + offset + page * record.length); + in.readFully(record); + ByteBuffer descriptor = ByteBuffer.wrap(record).order(ByteOrder.LITTLE_ENDIAN); + long start = descriptor.getLong(); + int size = descriptor.getInt(); + long crc = Integer.toUnsignedLong(descriptor.getInt()); + long[] first = entry(descriptor); + long[] last = entry(descriptor); + int entries = (int) Math.min(PAGE_ENTRIES, count - page * PAGE_ENTRIES); + int bytes = entries * ENTRY_LENGTH; + source.seek(base + start); + if (codec == 0) { + in.readFully(decoded, 0, bytes); + } else { + in.readFully(encoded, 0, size); + decodePage(encoded, size, decoded, entries); + } + if (checksum(decoded, bytes) != crc) { + throw invalid("page checksum"); + } + ByteBuffer values = ByteBuffer.wrap(decoded, 0, bytes).order(ByteOrder.LITTLE_ENDIAN); + previous = null; + for (int i = 0; i < entries; i++) { + long[] value = entry(values); + validateEntry(value, previous, page == 0 && i == 0, payloadLength); + if ((i == 0 && !java.util.Arrays.equals(value, first)) + || (i == entries - 1 && !java.util.Arrays.equals(value, last))) { + throw invalid("page endpoints"); + } + previous = value; + } + } + } + + private static void validateV1(SeekableInputStream source, long length, long payloadLength) + throws IOException { + if (length < 17) { + throw invalid("header"); + } + DataInputStream in = new DataInputStream(source); + long metadataCount = Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())); if (metadataCount > MAX_METADATA_RANGE_COUNT) { throw invalid("metadata range count exceeds limit"); } - if (count <= 0 || offset >= length) { - throw invalid("count or metadata ranges"); + long count = Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())); + long offset = 17 + metadataCount * 16; + if (count <= 0 + || count > 64 * 1024 + || offset >= length + || length - offset > count * ENTRY_LENGTH + 64 * 1024) { + throw invalid("compressed entries length"); + } + validateMetadataRanges(in, metadataCount, payloadLength); + byte[] encoded = new byte[(int) (length - offset)]; + in.readFully(encoded); + int expected = (int) count * ENTRY_LENGTH; + byte[] decoded = new byte[expected + 1]; + Inflater inflater = new Inflater(); + try { + inflater.setInput(encoded); + int total = 0; + while (!inflater.finished() && total < decoded.length) { + int n = inflater.inflate(decoded, total, decoded.length - total); + if (n == 0) { + break; + } + total += n; + } + if (total != expected || !inflater.finished() || inflater.getRemaining() != 0) { + throw invalid("compressed entries length"); + } + } catch (DataFormatException e) { + throw invalid("compressed entries"); + } finally { + inflater.end(); + } + ByteBuffer values = ByteBuffer.wrap(decoded, 0, expected).order(ByteOrder.LITTLE_ENDIAN); + long[] previous = null; + for (int i = 0; i < count; i++) { + long[] value = entry(values); + validateEntry(value, previous, i == 0, payloadLength); + previous = value; } + } + + private static void validateMetadataRanges( + DataInputStream in, long metadataCount, long payloadLength) throws IOException { long previousEnd = 0; for (long i = 0; i < metadataCount; i++) { long start = Long.reverseBytes(in.readLong()); @@ -71,42 +218,73 @@ static void validate(InputStream source, long length, long payloadLength) throws } previousEnd = start + size; } - long pages = (count - 1) / PAGE_ENTRIES + 1; - if (count > (length - offset) / ENTRY_LENGTH - || pages * 4 != length - offset - count * ENTRY_LENGTH) { - throw invalid("entries length"); - } - long ordinal = -1; - long pts = 0; - long position = -1; - byte[] page = new byte[PAGE_ENTRIES * ENTRY_LENGTH]; - for (long done = 0; done < count; ) { - int entries = (int) Math.min(PAGE_ENTRIES, count - done); - int bytes = entries * ENTRY_LENGTH; - in.readFully(page, 0, bytes); - long expected = Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())); - CRC32 crc = new CRC32(); - crc.update(page, 0, bytes); - if (crc.getValue() != expected) { - throw invalid("page checksum"); + } + + private static void decodePage(byte[] encoded, int size, byte[] decoded, int entries) { + int bytes = entries * ENTRY_LENGTH; + byte[] deltas = new byte[bytes + 1]; + Inflater inflater = new Inflater(); + try { + inflater.setInput(encoded, 0, size); + int total = 0; + while (!inflater.finished() && total < deltas.length) { + int n = inflater.inflate(deltas, total, deltas.length - total); + if (n == 0) { + break; + } + total += n; + } + if (total != bytes || !inflater.finished() || inflater.getRemaining() != 0) { + throw invalid("compressed page length"); } - ByteBuffer values = ByteBuffer.wrap(page, 0, bytes).order(ByteOrder.LITTLE_ENDIAN); + } catch (DataFormatException e) { + throw invalid("compressed page"); + } finally { + inflater.end(); + } + ByteBuffer input = ByteBuffer.wrap(deltas).order(ByteOrder.LITTLE_ENDIAN); + ByteBuffer output = ByteBuffer.wrap(decoded).order(ByteOrder.LITTLE_ENDIAN); + for (int column = 0; column < 3; column++) { + long value = 0; for (int i = 0; i < entries; i++) { - long nextOrdinal = values.getLong(); - long nextPts = values.getLong(); - long nextPosition = values.getLong(); - if ((ordinal == -1 && nextOrdinal != 0) - || nextOrdinal <= ordinal - || (ordinal != -1 && nextPts <= pts) - || nextPosition <= position - || nextPosition >= payloadLength) { - throw invalid("ordinals, timestamps or packet positions outside video payload"); + // Java long addition supplies the format's modulo-2^64 arithmetic. + value += input.getLong(); + output.putLong(i * ENTRY_LENGTH + column * Long.BYTES, value); + } + } + } + + static final class UnsupportedIndexException extends IllegalArgumentException { + UnsupportedIndexException(String reason) { + super("Unsupported video keyframe index " + reason + '.'); + } + } + + private static long[] entry(ByteBuffer values) { + return new long[] {values.getLong(), values.getLong(), values.getLong()}; + } + + private static long checksum(byte[] bytes, int length) { + CRC32 crc = new CRC32(); + crc.update(bytes, 0, length); + return crc.getValue(); + } + + private static void validateEntry( + long[] value, long[] previous, boolean initial, long payloadLength) { + if (value[0] < 0 + || (initial && value[0] != 0) + || value[2] < 0 + || value[2] >= payloadLength) { + throw invalid("ordinals or packet positions outside video payload"); + } + if (previous != null) { + for (int i = 0; i < 3; i++) { + if (value[i] <= previous[i]) { + throw invalid( + "ordinals, timestamps and packet positions must be strictly increasing"); } - ordinal = nextOrdinal; - pts = nextPts; - position = nextPosition; } - done += entries; } } diff --git a/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java index 1bd58e5472cf..fe8e326424b4 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/blob/VideoFileFormatTest.java @@ -56,7 +56,10 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; +import java.util.zip.CRC32; +import java.util.zip.Deflater; import static org.apache.paimon.utils.StreamUtils.intToLittleEndian; import static org.assertj.core.api.Assertions.assertThat; @@ -66,9 +69,7 @@ public class VideoFileFormatTest { private static final String KEYFRAME_INDEX_HEX = - "0149464b4f4544495601000000020000000000000000000000000000000100000000000000000000" - + "0000000000000000000000000000000000000000000c00000000000000a08c000000000000020000" - + "0000000000dfd9164b"; + "0249464b4f4544495601000000020000000000000000000000000000000001000000000000006a0000000000000030000000dfd9164b0000000000000000000000000000000000000000000000000c00000000000000a08c000000000000020000000000000023e343cf0000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000"; @TempDir java.nio.file.Path tempPath; @@ -136,6 +137,83 @@ public void testCrossLanguageV1Fixture() throws IOException { checkVideoFixture(fixture, fromHex(KEYFRAME_INDEX_HEX)); } + @Test + public void testPublishedV1IndexFixture() throws IOException { + byte[] mapping = fixture("video-index-v1-zlib.hex"); + VideoKeyframeIndex.validate(mapping, 3); + checkVideoFixture(fixture("video-v1-index-v1.hex"), mapping); + byte[] truncated = Arrays.copyOf(mapping, mapping.length - 1); + assertThatThrownBy(() -> VideoKeyframeIndex.validate(truncated, 3)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testCompressedCrossLanguageFixtures() throws IOException { + byte[] mapping = fixture("video-index-delta-zlib.hex"); + VideoKeyframeIndex.validate(mapping, 3); + checkVideoFixture(fixture("video-v1-delta-zlib.hex"), mapping); + VideoKeyframeIndex.validate(fixture("video-index-delta-zlib-boundary.hex"), 4097); + } + + @Test + public void testRejectMalformedCompressedStreams() throws IOException { + byte[] mapping = fixture("video-index-delta-zlib.hex"); + int directory = 38; + int page = directory + 68; + byte[] payload = Arrays.copyOfRange(mapping, page, mapping.length); + Deflater compressor = new Deflater(); + byte[] expanded = new byte[1024 * 1024]; + byte[] compressed = new byte[4096]; + compressor.setInput(expanded); + compressor.finish(); + int size = compressor.deflate(compressed); + compressor.end(); + for (byte[] invalid : + new byte[][] { + Arrays.copyOf(payload, payload.length - 1), + Arrays.copyOf(payload, payload.length + 1), + Arrays.copyOf(compressed, size), + new byte[] {1, 2, 3} + }) { + byte[] damaged = Arrays.copyOf(mapping, page + invalid.length); + System.arraycopy(invalid, 0, damaged, page, invalid.length); + ByteBuffer buffer = ByteBuffer.wrap(damaged).order(ByteOrder.LITTLE_ENDIAN); + buffer.putInt(directory + 8, invalid.length); + CRC32 crc = new CRC32(); + crc.update(damaged, directory, 64); + buffer.putInt(directory + 64, (int) crc.getValue()); + assertThatThrownBy(() -> VideoKeyframeIndex.validate(damaged, 3)) + .isInstanceOf(IllegalArgumentException.class); + } + } + + @Test + public void testPreserveUnsupportedIndexes() throws IOException { + for (int offset : new int[] {0, 21}) { + file = new Path(tempPath.resolve("unsupported-" + offset + ".video").toUri()); + byte[] mapping = fromHex(KEYFRAME_INDEX_HEX); + byte[] expected = fixture("video-v1.hex"); + mapping[offset] = (byte) (offset == 0 ? 3 : 2); + expected[7 + offset] = mapping[offset]; + assertThatThrownBy(() -> VideoKeyframeIndex.validate(mapping, 3)) + .isInstanceOf(VideoKeyframeIndex.UnsupportedIndexException.class); + checkVideoFixture(expected, mapping); + } + } + + @Test + public void testPageCorruption() { + byte[] mapping = fromHex(KEYFRAME_INDEX_HEX); + mapping[mapping.length - 1] ^= 1; + assertThatThrownBy(() -> VideoKeyframeIndex.validate(mapping, 3)) + .isInstanceOf(IllegalArgumentException.class); + byte[] codec = fromHex(KEYFRAME_INDEX_HEX); + codec[21] = 2; + assertThatThrownBy(() -> VideoKeyframeIndex.validate(codec, 3)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("codec"); + } + private void checkVideoFixture(byte[] fixture, byte[] mapping) throws IOException { byte[] video = "abc".getBytes(StandardCharsets.UTF_8); java.nio.file.Path source = tempPath.resolve("indexed.mp4"); @@ -226,12 +304,13 @@ public void testRejectInvalidKeyframeIndex() throws IOException { @Test public void testRejectForgedPagedCount() { byte[] mapping = - ByteBuffer.allocate(21) + ByteBuffer.allocate(22) .order(ByteOrder.LITTLE_ENDIAN) - .put((byte) 1) + .put((byte) 2) .putLong(0x564944454F4B4649L) .putInt(0) .putLong(Long.MAX_VALUE) + .put((byte) 0) .array(); assertThatThrownBy(() -> VideoKeyframeIndex.validate(mapping, 1)) .isInstanceOf(IllegalArgumentException.class); @@ -267,12 +346,18 @@ public void testPagedIndexValidation() throws IOException { restored)) .toData()) .isEqualTo(mapping); - mapping[21 + 4096 * 24 + 4] ^= 1; + mapping[mapping.length - 1] ^= 1; assertThatThrownBy(() -> VideoKeyframeIndex.validate(mapping, count)) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("checksum"); byte[] descending = pagedIndex(8192); - System.arraycopy(descending, 21, descending, 21 + 4096 * 24 + 4, 4096 * 24 + 4); + int directory = 22 + 68; + ByteBuffer.wrap(descending).order(ByteOrder.LITTLE_ENDIAN).putLong(directory + 16, 0); + java.util.zip.CRC32 checksum = new java.util.zip.CRC32(); + checksum.update(descending, directory, 64); + ByteBuffer.wrap(descending) + .order(ByteOrder.LITTLE_ENDIAN) + .putInt(directory + 64, (int) checksum.getValue()); assertThatThrownBy(() -> VideoKeyframeIndex.validate(descending, 8192)) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("ordinals"); @@ -280,14 +365,17 @@ public void testPagedIndexValidation() throws IOException { private static byte[] pagedIndex(int count) throws IOException { ByteArrayOutputStream out = new ByteArrayOutputStream(); + ByteArrayOutputStream data = new ByteArrayOutputStream(); out.write( - ByteBuffer.allocate(21) + ByteBuffer.allocate(22) .order(ByteOrder.LITTLE_ENDIAN) - .put((byte) 1) + .put((byte) 2) .putLong(0x564944454F4B4649L) .putInt(0) .putLong(count) + .put((byte) 0) .array()); + long offset = 22 + ((count - 1L) / 4096 + 1) * 68; for (int start = 0; start < count; start += 4096) { int end = Math.min(count, start + 4096); ByteBuffer page = @@ -297,18 +385,28 @@ private static byte[] pagedIndex(int count) throws IOException { } java.util.zip.CRC32 crc = new java.util.zip.CRC32(); crc.update(page.array()); - out.write(page.array()); - out.write(intToLittleEndian((int) crc.getValue())); + ByteBuffer record = ByteBuffer.allocate(68).order(ByteOrder.LITTLE_ENDIAN); + byte[] encoded = page.array(); + record.putLong(offset).putInt(encoded.length).putInt((int) crc.getValue()); + record.putLong(start).putLong(start).putLong(start); + record.putLong(end - 1).putLong(end - 1).putLong(end - 1); + crc.reset(); + crc.update(record.array(), 0, 64); + record.putInt((int) crc.getValue()); + out.write(record.array()); + data.write(encoded); + offset += encoded.length; } + data.writeTo(out); return out.toByteArray(); } @Test public void testRejectTooManyMetadataRanges() { byte[] mapping = - ByteBuffer.allocate(22) + ByteBuffer.allocate(23) .order(ByteOrder.LITTLE_ENDIAN) - .put((byte) 1) + .put((byte) 2) .putLong(0x564944454F4B4649L) .putInt((int) VideoKeyframeIndex.MAX_METADATA_RANGE_COUNT + 1) .putLong(1) @@ -325,8 +423,7 @@ public void testRejectSeekOffsetsOutsideVideoPayload() throws IOException { byte[] video = "video".getBytes(StandardCharsets.UTF_8); byte[] mapping = fromHex( - "0149464b4f4544495601000000010000000000000000000000000000000600000000000000000000" - + "00000000000000000000000000000000000000000020cac1a3"); + "0249464b4f4544495601000000010000000000000000000000000000000006000000000000006a000000000000001800000020cac1a30000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001f780023000000000000000000000000000000000000000000000000"); java.nio.file.Path source = tempPath.resolve("out-of-range-index.mp4"); byte[] sourceBytes = new byte[video.length + mapping.length]; int offset = put(sourceBytes, 0, video); @@ -354,8 +451,7 @@ public void testRejectInconsistentKeyframeIndexesForSamePayload() throws IOExcep byte[] firstIndex = fromHex(KEYFRAME_INDEX_HEX); byte[] secondIndex = fromHex( - "0149464b4f4544495600000000020000000000000000000000000000000000000000000000000000" - + "00000000000600000000000000504600000000000002000000000000005381c820"); + "0249464b4f45444956000000000200000000000000005a00000000000000300000005381c820000000000000000000000000000000000000000000000000060000000000000050460000000000000200000000000000b9762612000000000000000000000000000000000000000000000000060000000000000050460000000000000200000000000000"); java.nio.file.Path source = tempPath.resolve("inconsistent-index.mp4"); byte[] sourceBytes = new byte[video.length + firstIndex.length + secondIndex.length]; int offset = put(sourceBytes, 0, video); diff --git a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-delta-zlib-boundary.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-delta-zlib-boundary.hex new file mode 100644 index 000000000000..cdfa698bf0d9 --- /dev/null +++ b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-delta-zlib-boundary.hex @@ -0,0 +1,18 @@ +# 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. + +0249464b4f45444956000000000110000000000000019e0000000000000044020000bb611eab000000000000000000000000000000800000000000000000ff0f000000000000feffffffffffff7fff0f000000000000a2ae5b20e20200000000000013000000d75d56340010000000000000ffffffffffffff7f00100000000000000010000000000000ffffffffffffff7f001000000000000055b1376b7801edd4b10d00210c0441d33925d301083ba104240fc9c76fdd4e44bde19b1770077bb817b0033bb0031d7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0817e0ed4ea63ea5ffffaefd7bfee75af7bdd7380031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0817f1c88b5f3e95297bafca74b3dea518f7ae4000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e708003fd1ca8d5f7fb6fdef18e77bae7000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce0000738c0010e7080031ce000075e070e51fb376778016310600083ff1050cf00e5030061140799 diff --git a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-delta-zlib.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-delta-zlib.hex new file mode 100644 index 000000000000..ac687637b645 --- /dev/null +++ b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-delta-zlib.hex @@ -0,0 +1,18 @@ +# 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. + +0249464b4f4544495601000000020000000000000001000000000000000001000000000000006a0000000000000018000000dfd9164b0000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000f90bab277801636080001e280da316f4c058109a09ca05001db4013b diff --git a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-v1-zlib.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-v1-zlib.hex new file mode 100644 index 000000000000..7ba8f971da77 --- /dev/null +++ b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-index-v1-zlib.hex @@ -0,0 +1,18 @@ +# 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. + +0149464b4f45444956010000000200000000000000000000000100000000000000789c6360c00e78a0f4821e08cd04e503001394013b diff --git a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1-delta-zlib.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1-delta-zlib.hex new file mode 100644 index 000000000000..d918a7a3a86c --- /dev/null +++ b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1-delta-zlib.hex @@ -0,0 +1,18 @@ +# 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. + +6162635758595a0249464b4f4544495601000000020000000000000001000000000000000001000000000000006a0000000000000018000000dfd9164b0000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000f90bab277801636080001e280da316f4c058109a09ca05001db4013b060284028302040100020100010106010403000e0602000000040000000500000005000000050000004944454f01 diff --git a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1-index-v1.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1-index-v1.hex new file mode 100644 index 000000000000..93f047531e7b --- /dev/null +++ b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1-index-v1.hex @@ -0,0 +1,18 @@ +# 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. + +6162635758595a0149464b4f45444956010000000200000000000000000000000100000000000000789c6360c00e78a0f4821e08cd04e503001394013b06026c6b040100020100010106010403000e0602000000020000000500000005000000050000004944454f01 diff --git a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1.hex index 257cddbd1844..aa49cc734b4a 100644 --- a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1.hex +++ b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-v1.hex @@ -15,4 +15,4 @@ # specific language governing permissions and limitations # under the License. -6162635758595a0149464b4f45444956010000000200000000000000000000000000000001000000000000000000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000dfd9164b0602b201b101040100020100010106010403000e0602000000040000000500000005000000050000004944454f01 +6162635758595a0249464b4f4544495601000000020000000000000000000000000000000001000000000000006a0000000000000030000000dfd9164b0000000000000000000000000000000000000000000000000c00000000000000a08c000000000000020000000000000023e343cf0000000000000000000000000000000000000000000000000c00000000000000a08c00000000000002000000000000000602b402b302040100020100010106010403000e0602000000040000000500000005000000050000004944454f01 diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 8c5214f52a2e..d1a639a1ab7e 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1204,6 +1204,14 @@ class CoreOptions: .with_description("Read batch size for any file format if it supports.") ) + VIDEO_INDEX_COMPRESSION: ConfigOption[str] = ( + ConfigOptions.key("video.index.compression") + .string_type() + .default_value("delta-zlib") + .with_description("Compression for newly generated video keyframe indexes: delta-zlib or none. " + "Copied indexes retain their encoding.") + ) + READ_VIDEO_MAX_OPEN_DECODERS: ConfigOption[int] = ( ConfigOptions.key("read.video.max-open-decoders") .int_type() @@ -2021,6 +2029,12 @@ def local_cache_whitelist(self) -> str: def read_batch_size(self, default=None) -> int: return self.options.get(CoreOptions.READ_BATCH_SIZE, default or 1024) + def video_index_compression(self) -> str: + value = self.options.get(CoreOptions.VIDEO_INDEX_COMPRESSION) + if value not in ('none', 'delta-zlib'): + raise ValueError("video.index.compression must be 'none' or 'delta-zlib'.") + return value + def read_video_max_open_decoders(self) -> int: return self._positive_video_option(CoreOptions.READ_VIDEO_MAX_OPEN_DECODERS) diff --git a/paimon-python/pypaimon/execution/python/write/video_format_writer.py b/paimon-python/pypaimon/execution/python/write/video_format_writer.py index 3fc5ab39f4e9..7fc967b8a1a1 100644 --- a/paimon-python/pypaimon/execution/python/write/video_format_writer.py +++ b/paimon-python/pypaimon/execution/python/write/video_format_writer.py @@ -16,6 +16,7 @@ # under the License. import errno +import logging import shutil import tempfile import struct @@ -28,7 +29,7 @@ BlobRef, VideoFrameDescriptor, ) -from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex +from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex, UnsupportedVideoIndex from pypaimon.execution.python.write.blob_format_writer import BlobFormatWriter @@ -52,13 +53,17 @@ def __init__( self, output_stream, file_path=None, - copy_buffer_size=BlobFormatWriter.BUFFER_SIZE): + copy_buffer_size=BlobFormatWriter.BUFFER_SIZE, + index_compression="delta-zlib"): super().__init__( output_stream, blob_consumer=None, file_path=file_path, copy_buffer_size=copy_buffer_size, ) + if index_compression not in ('none', 'delta-zlib'): + raise ValueError("Video index compression must be 'none' or 'delta-zlib'.") + self._index_compression = index_compression self._physical_lengths = [] self._keyframe_indexes = [] self._index_spool = tempfile.SpooledTemporaryFile(max_size=1024 * 1024) @@ -200,14 +205,17 @@ def _keyframe_index(self, blob, frame): raise return 0 with mapping: - mapping.write_to(self._index_spool) + mapping.write_to(self._index_spool, self._index_compression) else: with tempfile.TemporaryFile() as mapping: with Blob.from_descriptor(blob.uri_reader, descriptor).new_input_stream() as source: shutil.copyfileobj(source, mapping, self.copy_buffer_size) if mapping.tell() != descriptor.length: raise ValueError("Video keyframe index length changed while reading.") - VideoKeyframeIndex.validate_stream(mapping, descriptor.length, frame.payload_descriptor.length) + try: + VideoKeyframeIndex.validate_stream(mapping, descriptor.length, frame.payload_descriptor.length) + except UnsupportedVideoIndex as error: + logging.getLogger(__name__).warning("Preserving unsupported video index: %s", error) mapping.seek(0) shutil.copyfileobj(mapping, self._index_spool, self.copy_buffer_size) return self._index_spool.tell() - start diff --git a/paimon-python/pypaimon/execution/python/write/writer/blob_file_writer.py b/paimon-python/pypaimon/execution/python/write/writer/blob_file_writer.py index ce54bb47b837..29f487564719 100644 --- a/paimon-python/pypaimon/execution/python/write/writer/blob_file_writer.py +++ b/paimon-python/pypaimon/execution/python/write/writer/blob_file_writer.py @@ -45,7 +45,7 @@ class BlobFileWriter: def __init__(self, file_io, file_path: Path, blob_consumer: Optional[BlobConsumer] = None, copy_buffer_size: int = BlobFormatWriter.BUFFER_SIZE, - video: bool = False, uri_reader_factory=None): + video: bool = False, uri_reader_factory=None, video_index_compression="delta-zlib"): self.file_io = file_io self.file_path = file_path self._uri_reader_factory = uri_reader_factory @@ -58,6 +58,7 @@ def __init__(self, file_io, file_path: Path, blob_consumer: Optional[BlobConsume self.writer = VideoFormatWriter( self.output_stream, file_path=str(file_path), + index_compression=video_index_compression, copy_buffer_size=copy_buffer_size, ) else: diff --git a/paimon-python/pypaimon/execution/python/write/writer/blob_writer.py b/paimon-python/pypaimon/execution/python/write/writer/blob_writer.py index 65ea655e5a4d..630d83c2c590 100644 --- a/paimon-python/pypaimon/execution/python/write/writer/blob_writer.py +++ b/paimon-python/pypaimon/execution/python/write/writer/blob_writer.py @@ -144,6 +144,7 @@ def open_current_writer(self): blob_consumer=self._blob_consumer, copy_buffer_size=self.blob_copy_buffer_size, video=self.video, + video_index_compression=self.options.video_index_compression(), uri_reader_factory=self._uri_reader_factory, ) diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index c13aaf1d1c39..46acfffbb023 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -2005,9 +2005,6 @@ def __init__(self, stream, keyframe_index=None, thread_count=0): self._source = stream self._keyframe_index = keyframe_index self._thread_count = thread_count - self._anchor_ordinals = ( - keyframe_index.ordinals - if keyframe_index is not None else []) self._container = None self._stream = None self._next_index = 0 @@ -2134,8 +2131,7 @@ def _indexed_plan(self, indices): groups = {} ranges = [] for index in indices: - preceding = bisect.bisect_right( - self._anchor_ordinals, index) - 1 + preceding = mapping.preceding(index) if preceding < 0: raise IndexError( "Video frame index %d is out of range." % index) diff --git a/paimon-python/pypaimon/multimodal/video.py b/paimon-python/pypaimon/multimodal/video.py index bb83636ad1fc..a2443ff95740 100644 --- a/paimon-python/pypaimon/multimodal/video.py +++ b/paimon-python/pypaimon/multimodal/video.py @@ -18,6 +18,7 @@ """Helpers for descriptor-backed video frame rows.""" import io +import logging import os from collections import OrderedDict, deque from collections.abc import Mapping @@ -27,7 +28,7 @@ from itertools import chain from pypaimon.table.row.blob import Blob, VideoFrameDescriptor -from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex +from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex, UnsupportedVideoIndex def _require_video_av(): @@ -509,10 +510,13 @@ def _decoder(self, descriptor, keyframe_index_descriptor=None, replica=0): try: if keyframe_index_descriptor is not None: index = keyframe_index_descriptor - stream.video_keyframe_index = VideoKeyframeIndex.open( - lambda offset, length: self.file_io.read_file_range( - index.uri, index.offset + offset, length), - index.length, descriptor.length) + try: + stream.video_keyframe_index = VideoKeyframeIndex.open( + lambda offset, length: self.file_io.read_file_range( + index.uri, index.offset + offset, length), + index.length, descriptor.length) + except UnsupportedVideoIndex as error: + logging.getLogger(__name__).warning("Scanning video without index: %s", error) stream.video_length = descriptor.length stream.video_read_ranges = lambda ranges: ( self.file_io.read_ranges_coalesced( diff --git a/paimon-python/pypaimon/table/row/video_keyframe_index.py b/paimon-python/pypaimon/table/row/video_keyframe_index.py index 11c5c973f5f1..899b846dba66 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -17,6 +17,7 @@ """Paged seek metadata for encoded video payloads.""" +import bisect import io import operator import sqlite3 @@ -114,6 +115,153 @@ def close(self): self.validated.clear() +class _IndexedKeyframes(_PagedKeyframes): + # A directory record contains offset, length, CRC, and page endpoints. + DIRECTORY = struct.Struct(' 2: + self.directory_cache.popitem(last=False) + self.directory_cache.move_to_end(block) + return self.directory_cache[block], local * self.DIRECTORY.size + + def _validate_record(self, page, record): + offset, size = record[:2] + first, last = record[3:6], record[6:9] + entries = min(self.PAGE_ENTRIES, self.count - page * self.PAGE_ENTRIES) + if (size <= 0 or size > self.PAGE_BYTES + 64 or offset > self.length - size + or offset < self.directory_offset + self.pages * self.DIRECTORY.size + or (self.codec == 0 and size != entries * self.ENTRY.size)): + raise ValueError("Invalid video keyframe index page length.") + if first[0] < 0 or (page == 0 and first[0] != 0): + raise ValueError("Keyframe ordinals must be strictly increasing from zero.") + if first[2] < 0 or (self.payload_length is not None and last[2] >= self.payload_length): + raise ValueError("Keyframe packet positions must be within the video payload.") + if entries == 1: + if first != last: + raise ValueError("Invalid video keyframe index page endpoints.") + else: + _validate_boundary(first, last) + if page == 0 and offset != self.directory_offset + self.pages * self.DIRECTORY.size: + raise ValueError("Invalid video keyframe index page offset.") + if page == self.pages - 1 and offset + size != self.length: + raise ValueError("Invalid video keyframe index entries length.") + return record + + def _descriptor(self, page): + record = self._record(page) + for left, right in ((page - 1, page), (page, page + 1)): + if left >= 0 and right < self.pages: + a = record if left == page else self._record(left) + b = record if right == page else self._record(right) + if a[0] + a[1] != b[0]: + raise ValueError("Invalid video keyframe index page offsets.") + _validate_boundary(a[6:9], b[3:6]) + return record + + def _page(self, page): + with self.lock: + if self.eager is not None: + start = page * self.PAGE_BYTES + return memoryview(self.eager)[start:start + self.PAGE_BYTES] + if page not in self.cache: + record = (self.DIRECTORY.unpack_from(self.directory, page * self.DIRECTORY.size) + if self.directory is not None else self._descriptor(page)) + offset, size, crc = record[:3] + data = self.read_at(offset, size) + expected = min(self.PAGE_ENTRIES, self.count - page * self.PAGE_ENTRIES) * self.ENTRY.size + if self.codec == 1: + data = _decode_page(data, expected) + if len(data) != expected or zlib.crc32(data) != crc: + raise ValueError("Invalid video keyframe index page checksum or length.") + _validate_entries(data, self.payload_length, initial=page == 0) + if (self.ENTRY.unpack_from(data) != record[3:6] + or self.ENTRY.unpack_from(data, len(data) - self.ENTRY.size) != record[6:9]): + raise ValueError("Invalid video keyframe index page endpoints.") + self.cache[page] = data + while len(self.cache) > max(1, self.CACHE_BYTES // self.PAGE_BYTES): + self.cache.popitem(last=False) + self.cache.move_to_end(page) + return self.cache[page] + + def load_small(self): + if max(self.length, self.count * self.ENTRY.size) <= self.EAGER_BYTES: + data = bytearray(self.count * self.ENTRY.size) + for page in range(self.pages): + start = page * self.PAGE_BYTES + block = self._page(page) + data[start:start + len(block)] = block + self.eager = data + self.cache.clear() + self.directory_cache.clear() + self.directory = None + self.read_at = None + + def preceding(self, ordinal): + import numpy as np + + with self.lock: + if self.eager is not None: + values = np.frombuffer(self.eager, dtype=' length: raise ValueError("Invalid video keyframe index length.") - if size == 0: - return b'' if offset + size <= len(prefix): return prefix[offset:offset + size] cached = prefix[offset:min(offset + size, len(prefix))] @@ -280,25 +480,76 @@ def read(offset, size): raise ValueError("Invalid video keyframe index: truncated read.") return data - version = read(0, 1)[0] + version, magic = struct.unpack_from(' cls.MAX_METADATA_RANGE_COUNT: raise ValueError("Video keyframe index exceeds the metadata-range limit.") - offset = header.size + metadata_count * cls.METADATA_RANGE.size - ranges = cls._metadata(cls.METADATA_RANGE.iter_unpack( - read(header.size, metadata_count * 16)), payload_length) - pages = (count + _PagedKeyframes.PAGE_ENTRIES - 1) // _PagedKeyframes.PAGE_ENTRIES - if offset + count * 24 + pages * 4 != length: + offset = cls.HEADER.size + metadata_count * cls.METADATA_RANGE.size + pages = (count - 1) // _PagedKeyframes.PAGE_ENTRIES + 1 + if offset + pages * (_IndexedKeyframes.DIRECTORY.size + 1) > length: raise ValueError("Invalid video keyframe index entries length.") + if max(length, count * cls.ENTRY.size) <= _IndexedKeyframes.EAGER_BYTES: + prefix += read(len(prefix), length - len(prefix)) + ranges = cls._metadata(cls.METADATA_RANGE.iter_unpack( + read(cls.HEADER.size, metadata_count * 16)), payload_length) result = cls.__new__(cls) result.metadata_ranges = ranges - result._keyframes = _PagedKeyframes(lambda start, size: read(offset + start, size), count, payload_length) + result._keyframes = _IndexedKeyframes(read, count, offset, length, codec, payload_length) + result.keyframes._descriptor(pages - 1) result.keyframes[0] + result.keyframes.load_small() + return result + + @classmethod + def _open_v1(cls, read, length, payload_length, prefix): + if length < cls.V1_HEADER.size: + raise ValueError("Invalid video keyframe index length.") + _, _, metadata_count, count = cls.V1_HEADER.unpack_from(prefix) + if metadata_count > cls.MAX_METADATA_RANGE_COUNT: + raise ValueError("Video keyframe index exceeds the metadata-range limit.") + offset = cls.V1_HEADER.size + metadata_count * cls.METADATA_RANGE.size + if not count or count > 64 * 1024 or offset >= length: + raise ValueError("Invalid video keyframe index entries length.") + expected = count * cls.ENTRY.size + if length - offset > expected + 64 * 1024: + raise ValueError("Invalid video keyframe index compressed length.") + ranges = cls._metadata(cls.METADATA_RANGE.iter_unpack( + read(cls.V1_HEADER.size, metadata_count * 16)), payload_length) + inflater = zlib.decompressobj() + chunks = [] + decoded = 0 + try: + while offset < length: + size = min(64 * 1024, length - offset) + block = inflater.decompress(read(offset, size), expected - decoded + 1) + chunks.append(block) + decoded += len(block) + offset += size + if decoded > expected or inflater.unused_data or inflater.unconsumed_tail: + raise ValueError("Invalid video keyframe index compressed length.") + if inflater.eof: + break + except zlib.error as error: + raise ValueError("Invalid video keyframe index compressed page.") from error + if decoded != expected or not inflater.eof or offset != length: + raise ValueError("Invalid video keyframe index compressed length.") + raw = b''.join(chunks) + _validate_entries(raw, payload_length, initial=True) + result = cls.__new__(cls) + result.metadata_ranges = ranges + result.keyframes = cls.ENTRY.iter_unpack(raw) return result def _validate_all(self): diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py b/paimon-python/pypaimon/tests/multimodal_table_test.py index be94381585a6..0af5da35f0bd 100644 --- a/paimon-python/pypaimon/tests/multimodal_table_test.py +++ b/paimon-python/pypaimon/tests/multimodal_table_test.py @@ -322,6 +322,32 @@ def test_auto_video_real_timestamps_and_payload(self): for expected, frame in zip(pixels, container.decode(video=0)): np.testing.assert_array_equal(expected, frame.to_ndarray(format="rgb24")) + def test_video_index_compression_table_option(self): + from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex + path = self._auto_video_file() + for compression, codec in (("none", 0), ("delta-zlib", 1)): + table = self.conn.create_table( + "codec_" + str(codec), + schema=pa.schema([ + pa.field("frame_index", pa.int64()), + pa.field("timestamp", pa.float64()), + pa.field("video", pa.large_binary()), + ]), + options=dict(_PARQUET_OPTIONS, **{ + "video-frame-field": "video", "blob-as-descriptor": "true", + "video.index.compression": compression, + }), + ) + table.add_video(path) + descriptor = pmm.VideoFrameDescriptor.deserialize(table.scan().to_list()[0]["video"]) + mapping = descriptor.keyframe_index_descriptor + with table.raw_table.file_io.new_input_stream(mapping.uri) as stream: + stream.seek(mapping.offset) + data = stream.read(mapping.length) + self.assertEqual(codec, data[21]) + with VideoKeyframeIndex.deserialize(data) as index: + self.assertEqual(0, index.keyframes[0][0]) + def _raw_hevc_file(self, b_frames=3): av = pytest.importorskip("av") import numpy as np diff --git a/paimon-python/pypaimon/tests/multimodal_video_test.py b/paimon-python/pypaimon/tests/multimodal_video_test.py index 8c7ff6fad948..41d2148b5f3a 100644 --- a/paimon-python/pypaimon/tests/multimodal_video_test.py +++ b/paimon-python/pypaimon/tests/multimodal_video_test.py @@ -253,6 +253,42 @@ def test_rejects_missing_or_invalid_decode_callbacks(self): **kwargs, ) + def test_unknown_index_falls_back_but_corruption_fails(self): + from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex + + with VideoKeyframeIndex([], [(0, 0, 0)]) as index: + valid = index.serialize() + for offset in (0, 21, len(valid) - 1): + with self.subTest(offset=offset): + mapping = bytearray(valid) + mapping[offset] ^= 2 + path = os.path.join(self.temp_dir.name, "indexed.mp4") + with open(path, 'wb') as out: + out.write(b'video' + mapping) + descriptor = VideoFrameDescriptor(path, 0, 5, 0, 5, len(mapping)) + calls = [] + + def factory(stream): + self.assertIsNone(getattr(stream, 'video_keyframe_index', None)) + calls.append('open') + return _Decoder(stream, calls) + + collator = VideoFrameCollator( + self.table, video_column='video', decoder_factory=factory, + decode_fn=lambda decoder, frame, row: decoder.decode(frame), + collate_fn=lambda rows: rows) + try: + if offset == len(valid) - 1: + with self.assertRaisesRegex(ValueError, 'checksum|compressed page'): + collator([{'video': descriptor.serialize()}]) + self.assertFalse(calls) + else: + with self.assertLogs('pypaimon.multimodal.video', level='WARNING'): + result = collator([{'video': descriptor.serialize()}]) + self.assertEqual((b'video', 0), result[0]['frame']) + finally: + collator.close() + def test_reuses_decoder_for_rows_with_same_descriptor(self): descriptors = [ self._descriptor("episode-1.mp4", b"video-one", frame) diff --git a/paimon-python/pypaimon/tests/video_format_test.py b/paimon-python/pypaimon/tests/video_format_test.py index 6665fd9ee0c7..9fcdacc2cf0d 100644 --- a/paimon-python/pypaimon/tests/video_format_test.py +++ b/paimon-python/pypaimon/tests/video_format_test.py @@ -159,9 +159,94 @@ def test_pack_raw_videos_and_map_frame_runs(self): def test_cross_language_video_v1_fixture(self): fixture = self._fixture_bytes(self.VIDEO_FIXTURE) + with VideoKeyframeIndex([(0, 1)], [(0, 0, 0), (12, 36000, 2)]) as index: + encoded = index.serialize('none') + self._check_video_fixture(fixture, encoded) + + def test_cross_language_compressed_video_fixture(self): + fixture = self._fixture_bytes(self.VIDEO_FIXTURE.with_name('video-v1-delta-zlib.hex')) with VideoKeyframeIndex([(0, 1)], [(0, 0, 0), (12, 36000, 2)]) as index: encoded = index.serialize() + self.assertEqual(encoded, self._fixture_bytes( + self.VIDEO_FIXTURE.with_name('video-index-delta-zlib.hex'))) self._check_video_fixture(fixture, encoded) + data = self._fixture_bytes(self.VIDEO_FIXTURE.with_name('video-index-delta-zlib-boundary.hex')) + with VideoKeyframeIndex.deserialize(data, 4097) as index: + expected = [(i, -(1 << 63) + i if i < 2048 else (1 << 63) - 4097 + i, i) + for i in range(4097)] + self.assertEqual(expected, list(index.keyframes)) + + def test_generated_index_compression_option(self): + from pypaimon.common.options.core_options import CoreOptions + self.assertEqual('delta-zlib', CoreOptions(Options({})).video_index_compression()) + for compression, codec in (('none', 0), ('delta-zlib', 1)): + options = CoreOptions(Options({'video.index.compression': compression})) + blob = self._source_frame(compression + '.mp4', b'video', 0) + writer = VideoFormatWriter(io.BytesIO(), index_compression=options.video_index_compression()) + with mock.patch.object(VideoKeyframeIndex, 'inspect', return_value=VideoKeyframeIndex( + [], [(0, 0, 0), (1, 1, 1)])): + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + writer._index_spool.seek(0) + self.assertEqual(b'\x02', writer._index_spool.read(1)) + writer._index_spool.seek(21) + self.assertEqual(bytes([codec]), writer._index_spool.read(1)) + writer.close() + with self.assertRaises(ValueError): + CoreOptions(Options({'video.index.compression': 'unknown'})).video_index_compression() + + def test_mixed_v1_v2_indexes_survive_rewrite(self): + target = (self.root / 'mixed-versions.video').as_uri() + writer = VideoFormatWriter(self.file_io.new_output_stream(target), index_compression='none') + mappings = [self._fixture_bytes( + self.VIDEO_FIXTURE.with_name('video-index-v1-zlib.hex'))] + for compression in ('none', 'delta-zlib'): + with VideoKeyframeIndex([], [(0, 0, 0), (1, 1, 1)]) as index: + mappings.append(index.serialize(compression)) + for ordinal, mapping in enumerate(mappings): + blob = self._indexed_frame(str(ordinal) + '.mp4', b'video', mapping) + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + writer.close() + rewritten = (self.root / 'mixed-versions-rewritten.video').as_uri() + writer = VideoFormatWriter(self.file_io.new_output_stream(rewritten)) + for row, expected in zip(self._read(target), mappings): + descriptor = VideoFrameDescriptor.deserialize(row).keyframe_index_descriptor + actual = Blob.from_file(self.file_io, descriptor.uri, descriptor.offset, descriptor.length).to_data() + self.assertEqual(expected, actual) + with VideoKeyframeIndex.deserialize(actual, 5) as index: + self.assertEqual((0, 0, 0), index.keyframes[0]) + self.assertEqual(1 if actual[0] == 2 else 0, index.preceding(1)) + frame = VideoFrameDescriptor.deserialize(row) + blob = Blob.from_descriptor(self.file_io.uri_reader_factory.create(frame.uri), frame) + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + writer.close() + for row, expected in zip(self._read(rewritten), mappings): + descriptor = VideoFrameDescriptor.deserialize(row).keyframe_index_descriptor + actual = Blob.from_file(self.file_io, descriptor.uri, descriptor.offset, descriptor.length).to_data() + self.assertEqual(expected, actual) + + def test_unsupported_indexes_survive_rewrite(self): + target = (self.root / "mixed.video").as_uri() + writer = VideoFormatWriter(self.file_io.new_output_stream(target)) + mappings = [] + for codec in ('version', 'codec'): + with VideoKeyframeIndex([], [(0, 0, 0), (1, 1, 1)]) as index: + data = bytearray(index.serialize()) + data[0 if codec == 'version' else 21] = 3 if codec == 'version' else 2 + mappings.append(bytes(data)) + blob = self._indexed_frame(codec + '.mp4', b'video', mappings[-1]) + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + writer.close() + rewritten = (self.root / "mixed-rewrite.video").as_uri() + writer = VideoFormatWriter(self.file_io.new_output_stream(rewritten)) + for serialized, expected in zip(self._read(target), mappings): + descriptor = VideoFrameDescriptor.deserialize(serialized) + blob = Blob.from_descriptor(self.file_io.uri_reader_factory.create(descriptor.uri), descriptor) + mapping = descriptor.keyframe_index_descriptor + self.assertEqual(expected, Blob.from_file( + self.file_io, mapping.uri, mapping.offset, mapping.length).to_data()) + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + writer.close() + self.assertEqual((self.root / 'mixed.video').read_bytes(), (self.root / 'mixed-rewrite.video').read_bytes()) def _check_video_fixture(self, fixture, encoded): video = b"abc" @@ -415,7 +500,7 @@ def test_generated_large_index_is_kept(self): descriptor = VideoFrameDescriptor(source.as_uri(), 0, len(payload), 0, -1, 0) blob = Blob.from_descriptor(self.file_io.uri_reader_factory.create(descriptor.uri), descriptor) target = (self.root / "large-index.video").as_uri() - writer = VideoFormatWriter(self.file_io.new_output_stream(target)) + writer = VideoFormatWriter(self.file_io.new_output_stream(target), index_compression="none") with mock.patch.object(VideoKeyframeIndex, "inspect", return_value=VideoKeyframeIndex([], ((i, i, i) for i in range(count)))): writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) @@ -475,7 +560,7 @@ def test_rejects_invalid_keyframe_index(self): def test_large_indexes_survive_rewrite(self): count = 750_000 with VideoKeyframeIndex([], ((i, i, i) for i in range(count))) as index: - mapping = index.serialize() + mapping = index.serialize("none") video = b'x' * count target = (self.root / "many-indexes.video").as_uri() writer = VideoFormatWriter(self.file_io.new_output_stream(target)) @@ -516,11 +601,11 @@ def test_rejects_seek_offsets_outside_video_payload(self): GenericRow([blob], [self.field], RowKind.INSERT) ) - def test_rejects_compressed_invalid_index_without_expanding_it(self): + def test_rejects_forged_count_without_expanding_it(self): count = 500_000 entry = VideoKeyframeIndex.ENTRY.pack(0, 0, 0) mapping = VideoKeyframeIndex.HEADER.pack( - VideoKeyframeIndex.VERSION, VideoKeyframeIndex.MAGIC, 0, count + VideoKeyframeIndex.VERSION, VideoKeyframeIndex.MAGIC, 0, count, 0 ) + zlib.compress(entry * count) video = b"video" source = self.root / "compressed-invalid-index.mp4" @@ -538,7 +623,7 @@ def test_rejects_compressed_invalid_index_without_expanding_it(self): try: with self.assertRaisesRegex( ValueError, - "entries length"): + "Invalid video keyframe index"): writer.add_element( GenericRow([blob], [self.field], RowKind.INSERT) ) diff --git a/paimon-python/pypaimon/tests/video_keyframe_index_test.py b/paimon-python/pypaimon/tests/video_keyframe_index_test.py index 5e33e7db98a9..4a67f1e0a794 100644 --- a/paimon-python/pypaimon/tests/video_keyframe_index_test.py +++ b/paimon-python/pypaimon/tests/video_keyframe_index_test.py @@ -18,6 +18,7 @@ import io import pickle from fractions import Fraction +from pathlib import Path import struct import sys from types import SimpleNamespace @@ -25,11 +26,56 @@ import zlib from unittest import mock -from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex +from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex, _IndexedKeyframes + + +def encode_index(entries, count=None, ranges=(), codec=0): + count = len(entries) if count is None else count + header = VideoKeyframeIndex.HEADER.pack( + VideoKeyframeIndex.VERSION, VideoKeyframeIndex.MAGIC, len(ranges), count, codec) + metadata = b''.join(struct.pack('