diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 18c1bb59d969..d3e70895b737 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -413,7 +413,7 @@ trailers, or per-entry CRC. On-disk order: +----------------------------+ | ... | +----------------------------+ -| Keyframe Index 1 | Video metadata ranges and compressed keyframe entries +| Keyframe Index 1 | Video metadata ranges and keyframe entries +----------------------------+ | Keyframe Index 2 | +----------------------------+ @@ -441,12 +441,26 @@ 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. -A keyframe-index block contains a 17-byte header (version `1`: uint8; magic `0x564944454F4B4649`: -uint64; metadata-range and keyframe counts: uint32 each), metadata `(offset, length)` pairs -(int64 each), and zlib-compressed `(frame ordinal, PTS, packet position)` entries (int64 each). -Numeric fields are little endian; offsets are payload-relative. Limits: 65,536 metadata ranges -and 65,536 keyframes, 16 MiB per block, 64 MiB per file. The index covers the first video -stream; its time base stays in the video. +Keyframe-index version `1` uses this layout. Numeric fields are little endian; +byte offsets are relative to the embedded video payload. + +| Field | Size | +| --- | ---: | +| Version (`1`) | 1 byte | +| Magic (`0x564944454F4B4649`) | 8 bytes | +| Metadata-range count | 4 bytes | +| Keyframe count | 8 bytes | +| Metadata `(offset, length)` pairs | 16 bytes each | +| Keyframe pages | Variable | + +Each page contains up to 4,096 uncompressed `(frame ordinal, PTS, packet position)` +entries (three int64 values), followed by a 4-byte CRC32 of its entries. Only the +last page may be partial. 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 load pages on demand; writers validate all pages. Metadata ranges are +limited to 65,536; keyframe count and total index size have no configured limit. 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 30b347755014..3f50c40f96c2 100644 --- a/docs/docs/pypaimon/video.md +++ b/docs/docs/pypaimon/video.md @@ -115,8 +115,9 @@ video = pm.BlobDescriptor( frames.add_video(video, episode_43_rows) ``` -With PyAV, PyPaimon builds keyframe indexes for supported ISO BMFF videos (such -as MP4); other decodable videos use the scan fallback. +PyAV builds seek indexes for supported ISO BMFF videos (such as MP4); other +decodable videos use scan fallback. Index generation uses temporary disk; reads +load pages on demand with a 1 MiB entry-page cache per index. 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 f67f3fc58234..e8b6a0b51d30 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,8 +1819,9 @@ public void testVideoRollingByBlobTargetSize() throws Exception { byte[] secondBytes = "second-video".getBytes(); byte[] keyframeIndex = fromHex( - "0149464b4f45444956010000000200000000000000000000000100000000000000" - + "789c6360c00e78a0f4821e08cd04e503001394013b"); + "0149464b4f4544495601000000020000000000000000000000000000000100000000000000000000" + + "0000000000000000000000000000000000000000000c00000000000000a08c000000000000020000" + + "0000000000dfd9164b"); 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-format/src/main/java/org/apache/paimon/format/blob/VideoFileMeta.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFileMeta.java index 08d7d5dff415..6fceaa83f245 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFileMeta.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFileMeta.java @@ -109,12 +109,6 @@ public VideoFileMeta(SeekableInputStream in, long fileSize, @Nullable RoaringBit throw corrupt("invalid keyframe index length %s.", length); } keyframeIndexSize += length; - if (length > VideoFormatWriter.MAX_KEYFRAME_INDEX_BYTES) { - throw corrupt("keyframe index exceeds the 16 MiB limit."); - } - if (keyframeIndexSize > VideoFormatWriter.MAX_TOTAL_KEYFRAME_INDEX_BYTES) { - throw corrupt("keyframe indexes exceed the 64 MiB limit."); - } } long keyframeIndexStart = indexStart - keyframeIndexSize; if (keyframeIndexStart < 0) { 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 a2a92e5e0b81..6de398d3ab06 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 @@ -33,9 +33,10 @@ import org.apache.paimon.utils.LongArrayList; import java.io.IOException; -import java.util.ArrayList; +import java.io.InputStream; +import java.io.RandomAccessFile; +import java.nio.file.Files; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.Objects; @@ -56,13 +57,13 @@ public class VideoFormatWriter implements FileAwareFormatWriter { public static final long NULL_REFERENCE = -1L; public static final long PLACEHOLDER_REFERENCE = -2L; public static final int FILE_FOOTER_LENGTH = Integer.BYTES * 6 + Byte.BYTES; - static final long MAX_KEYFRAME_INDEX_BYTES = 16L * 1024 * 1024; - static final long MAX_TOTAL_KEYFRAME_INDEX_BYTES = 64L * 1024 * 1024; private final PositionOutputStream out; private final RawVideoPayloadWriter payloadWriter; private final LongArrayList physicalVideoLengths; - private final List keyframeIndexes; + private final LongArrayList keyframeIndexLengths; + private java.nio.file.Path indexSpoolPath; + private RandomAccessFile indexSpool; private final LongArrayList runLengths; private final LongArrayList runReferences; private final LongArrayList runFirstFrames; @@ -94,7 +95,7 @@ public VideoFormatWriter( blobFetchMetricReporter, copyBufferSize); this.physicalVideoLengths = new LongArrayList(16); - this.keyframeIndexes = new ArrayList<>(); + this.keyframeIndexLengths = new LongArrayList(16); this.runLengths = new LongArrayList(16); this.runReferences = new LongArrayList(16); this.runFirstFrames = new LongArrayList(16); @@ -136,25 +137,41 @@ public void addElement(InternalRow element) throws IOException { if (ordinal == null) { long keyframeIndexLength = keyframeIndexDescriptor == null ? 0 : keyframeIndexDescriptor.length(); - checkKeyframeIndexSize(keyframeIndexLength, keyframeIndexBytes); + if (keyframeIndexLength < 0) { + throw new IllegalArgumentException("Negative video keyframe index length."); + } Blob keyframeIndex = VideoFrameDescriptor.keyframeIndexBlob(blob); - byte[] mapping = payloadWriter.readIndex(keyframeIndex, keyframeIndexLength); - if (mapping == null) { + if (keyframeIndexLength > 0 && indexSpool == null) { + indexSpoolPath = Files.createTempFile("paimon-video-index-", ".tmp"); + indexSpool = new RandomAccessFile(indexSpoolPath.toFile(), "rw"); + } + if (!payloadWriter.readIndex( + keyframeIndex, keyframeIndexLength, indexSpool, payload.length())) { append(NULL_REFERENCE, 0); return; } - if (mapping.length > 0) { - VideoKeyframeIndex.validate(mapping, payload.length()); + long length; + try { + length = payloadWriter.write(element); + } catch (IOException | RuntimeException e) { + if (indexSpool != null) { + indexSpool.setLength(keyframeIndexBytes); + indexSpool.seek(keyframeIndexBytes); + } + throw e; } - long length = payloadWriter.write(element); if (length == BlobFormatWriter.NULL_LENGTH) { + if (indexSpool != null) { + indexSpool.setLength(keyframeIndexBytes); + indexSpool.seek(keyframeIndexBytes); + } append(NULL_REFERENCE, 0); return; } ordinal = physicalVideoLengths.size(); physicalVideoLengths.add(length); - keyframeIndexes.add(mapping); - keyframeIndexBytes += mapping.length; + keyframeIndexLengths.add(keyframeIndexLength); + keyframeIndexBytes += keyframeIndexLength; physicalVideos.put(payload, ordinal); physicalVideoKeyframeIndexes.put(payload, keyframeIndexDescriptor); } else { @@ -176,33 +193,50 @@ public void close() throws IOException { if (closed) { return; } - flushRun(); - payloadWriter.close(); + try { + flushRun(); + payloadWriter.close(); + + if (indexSpool != null) { + indexSpool.seek(0); + byte[] buffer = new byte[64 * 1024]; + int n; + while ((n = indexSpool.read(buffer)) != -1) { + out.write(buffer, 0, n); + } + } + byte[] physicalIndex = + DeltaVarintCompressor.compressLongArrayList(physicalVideoLengths); + byte[] keyframeLengthIndex = + DeltaVarintCompressor.compressLongArrayList(keyframeIndexLengths); + byte[] runLengthIndex = DeltaVarintCompressor.compressLongArrayList(runLengths); + byte[] runReferenceIndex = DeltaVarintCompressor.compressLongArrayList(runReferences); + byte[] firstFrameIndex = DeltaVarintCompressor.compressLongArrayList(runFirstFrames); + out.write(physicalIndex); + out.write(keyframeLengthIndex); + out.write(runLengthIndex); + out.write(runReferenceIndex); + out.write(firstFrameIndex); + out.write(intToLittleEndian(physicalIndex.length)); + out.write(intToLittleEndian(keyframeLengthIndex.length)); + out.write(intToLittleEndian(runLengthIndex.length)); + out.write(intToLittleEndian(runReferenceIndex.length)); + out.write(intToLittleEndian(firstFrameIndex.length)); + out.write(intToLittleEndian(MAGIC_NUMBER)); + out.write(VERSION); - for (byte[] mapping : keyframeIndexes) { - out.write(mapping); + } finally { + closed = true; + try { + if (indexSpool != null) { + indexSpool.close(); + } + } finally { + if (indexSpoolPath != null) { + Files.deleteIfExists(indexSpoolPath); + } + } } - byte[] physicalIndex = DeltaVarintCompressor.compressLongArrayList(physicalVideoLengths); - LongArrayList keyframeIndexLengths = new LongArrayList(keyframeIndexes.size()); - keyframeIndexes.forEach(mapping -> keyframeIndexLengths.add(mapping.length)); - byte[] keyframeLengthIndex = - DeltaVarintCompressor.compressLongArrayList(keyframeIndexLengths); - byte[] runLengthIndex = DeltaVarintCompressor.compressLongArrayList(runLengths); - byte[] runReferenceIndex = DeltaVarintCompressor.compressLongArrayList(runReferences); - byte[] firstFrameIndex = DeltaVarintCompressor.compressLongArrayList(runFirstFrames); - out.write(physicalIndex); - out.write(keyframeLengthIndex); - out.write(runLengthIndex); - out.write(runReferenceIndex); - out.write(firstFrameIndex); - out.write(intToLittleEndian(physicalIndex.length)); - out.write(intToLittleEndian(keyframeLengthIndex.length)); - out.write(intToLittleEndian(runLengthIndex.length)); - out.write(intToLittleEndian(runReferenceIndex.length)); - out.write(intToLittleEndian(firstFrameIndex.length)); - out.write(intToLittleEndian(MAGIC_NUMBER)); - out.write(VERSION); - closed = true; } int physicalVideoCount() { @@ -213,18 +247,6 @@ int runCount() { return runLengths.size() + (currentRunLength == 0 ? 0 : 1); } - static void checkKeyframeIndexSize(long length, long bufferedBytes) { - checkArgument( - length >= 0 && length <= MAX_KEYFRAME_INDEX_BYTES, - "Video keyframe index length %s exceeds the %s-byte limit.", - length, - MAX_KEYFRAME_INDEX_BYTES); - checkArgument( - bufferedBytes <= MAX_TOTAL_KEYFRAME_INDEX_BYTES - length, - "Buffered video keyframe indexes exceed the %s-byte limit.", - MAX_TOTAL_KEYFRAME_INDEX_BYTES); - } - private void append(long reference, long frameIndex) { if (canExtend(reference, frameIndex)) { currentRunLength++; @@ -258,25 +280,61 @@ private void flushRun() { /** Copies raw video bytes without the ordinary BLOB record header and trailer. */ private static class RawVideoPayloadWriter extends AbstractBlobElementWriter { - private byte[] readIndex(Blob index, long expectedLength) throws IOException { + private boolean readIndex( + Blob index, long expectedLength, RandomAccessFile spool, long payloadLength) + throws IOException { if (index == null) { - return new byte[0]; + return true; } - byte[] bytes; + long start = spool.getFilePointer(); + boolean complete = false; try { - bytes = index.toData(); - if (bytes.length != expectedLength) { - throw new java.io.EOFException( - "Video keyframe index length changed while reading."); + try (InputStream in = index.newInputStream()) { + byte[] buffer = new byte[64 * 1024]; + long remaining = expectedLength; + while (remaining > 0) { + int n = in.read(buffer, 0, (int) Math.min(remaining, buffer.length)); + if (n <= 0) { + throw new java.io.EOFException( + "Video keyframe index length changed while reading."); + } + spool.write(buffer, 0, n); + remaining -= n; + } + if (in.read() != -1) { + throw new IOException("Video keyframe index has trailing data."); + } + } catch (IOException | RuntimeException e) { + if (handleFetchFailure(e, index)) { + return false; + } + throw e; } - } catch (IOException | RuntimeException e) { - if (handleFetchFailure(e, index)) { - return null; + spool.seek(start); + VideoKeyframeIndex.validate( + new InputStream() { + @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); + spool.seek(start + expectedLength); + recordSuccess(expectedLength); + complete = true; + return true; + } finally { + if (!complete) { + spool.setLength(start); + spool.seek(start); } - throw e; } - recordSuccess(bytes.length); - return bytes; } private RawVideoPayloadWriter( 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 34cfc0e59010..307aac4cb3b7 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,117 +18,95 @@ package org.apache.paimon.format.blob; +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.ByteOrder; -import java.util.zip.DataFormatException; -import java.util.zip.Inflater; +import java.util.zip.CRC32; -/** Validates persisted video metadata ranges and sparse keyframe entries. */ +/** Streaming validation for paged video seek indexes. */ final class VideoKeyframeIndex { - - private static final int VERSION = 1; - private static final long MAGIC = 0x564944454F4B4649L; // "VIDEOKFI" - private static final int HEADER_LENGTH = Byte.BYTES + Long.BYTES + Integer.BYTES * 2; - private static final int METADATA_RANGE_LENGTH = Long.BYTES * 2; - private static final int ENTRY_LENGTH = Long.BYTES * 3; + private static final long MAGIC = 0x564944454F4B4649L; + private static final int ENTRY_LENGTH = 24; + private static final int PAGE_ENTRIES = 4096; static final long MAX_METADATA_RANGE_COUNT = 64 * 1024; - static final long MAX_KEYFRAME_COUNT = 64 * 1024; private VideoKeyframeIndex() {} static void validate(byte[] data, long payloadLength) { - if (data == null || data.length <= HEADER_LENGTH) { - throw invalid("too short"); - } - if (payloadLength <= 0) { - throw invalid("empty video payload"); + try { + validate(new ByteArrayInputStream(data), data.length, payloadLength); + } catch (IOException e) { + throw new IllegalArgumentException("Invalid video keyframe index.", e); } + } - ByteBuffer header = ByteBuffer.wrap(data, 0, HEADER_LENGTH).order(ByteOrder.LITTLE_ENDIAN); - int version = Byte.toUnsignedInt(header.get()); - long magic = header.getLong(); - long metadataRangeCount = Integer.toUnsignedLong(header.getInt()); - long keyframeCount = Integer.toUnsignedLong(header.getInt()); - if (version != VERSION || magic != MAGIC) { - throw invalid("version or magic"); + static void validate(InputStream source, long length, long payloadLength) throws IOException { + if (length < 21 || payloadLength <= 0) { + throw invalid("header or empty video payload"); } - if (keyframeCount == 0) { - throw invalid("empty"); + 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 (metadataRangeCount > MAX_METADATA_RANGE_COUNT) { + long count = Long.reverseBytes(in.readLong()); + long offset = 21 + metadataCount * 16; + if (metadataCount > MAX_METADATA_RANGE_COUNT) { throw invalid("metadata range count exceeds limit"); } - if (keyframeCount > MAX_KEYFRAME_COUNT) { - throw invalid("keyframe count exceeds limit"); + if (count <= 0 || offset >= length) { + throw invalid("count or metadata ranges"); } - - long entriesOffset = HEADER_LENGTH + metadataRangeCount * METADATA_RANGE_LENGTH; - if (entriesOffset >= data.length) { - throw invalid("metadata ranges"); - } - ByteBuffer ranges = ByteBuffer.wrap(data).order(ByteOrder.LITTLE_ENDIAN); - ranges.position(HEADER_LENGTH); long previousEnd = 0; - for (long i = 0; i < metadataRangeCount; i++) { - long offset = ranges.getLong(); - long length = ranges.getLong(); - if (offset < previousEnd || length <= 0 || offset > Long.MAX_VALUE - length) { - throw invalid("metadata ranges"); - } - previousEnd = offset + length; - if (previousEnd > payloadLength) { + for (long i = 0; i < metadataCount; i++) { + long start = Long.reverseBytes(in.readLong()); + long size = Long.reverseBytes(in.readLong()); + if (start < previousEnd || size <= 0 || start > payloadLength - size) { throw invalid("metadata range outside video payload"); } + previousEnd = start + size; } - - Inflater inflater = new Inflater(); - inflater.setInput(data, (int) entriesOffset, data.length - (int) entriesOffset); - byte[] entry = new byte[ENTRY_LENGTH]; - long previousOrdinal = -1; - long previousPts = 0; - long previousPacketPosition = -1; - try { - for (long i = 0; i < keyframeCount; i++) { - inflateEntry(inflater, entry); - ByteBuffer values = ByteBuffer.wrap(entry).order(ByteOrder.LITTLE_ENDIAN); - long ordinal = values.getLong(); - long pts = values.getLong(); - long packetPosition = values.getLong(); - if ((i == 0 && ordinal != 0) || ordinal <= previousOrdinal) { - throw invalid("keyframe ordinals must be strictly increasing from zero"); - } - if (i > 0 && pts <= previousPts) { - throw invalid("keyframe timestamps must be strictly increasing"); - } - if (packetPosition <= previousPacketPosition || packetPosition >= payloadLength) { - throw invalid( - "keyframe packet positions must be within the video payload and strictly increasing"); - } - previousOrdinal = ordinal; - previousPts = pts; - previousPacketPosition = packetPosition; - } - - if (inflater.inflate(entry, 0, 1) != 0 - || !inflater.finished() - || inflater.getRemaining() != 0) { - throw invalid("entry count"); - } - } catch (DataFormatException e) { - throw new IllegalArgumentException("Invalid video keyframe index payload.", e); - } finally { - inflater.end(); + long pages = (count - 1) / PAGE_ENTRIES + 1; + if (count > (length - offset) / ENTRY_LENGTH + || pages * 4 != length - offset - count * ENTRY_LENGTH) { + throw invalid("entries length"); } - } - - private static void inflateEntry(Inflater inflater, byte[] entry) throws DataFormatException { - int offset = 0; - while (offset < entry.length) { - int length = inflater.inflate(entry, offset, entry.length - offset); - if (length == 0) { - throw invalid("entries"); + 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"); + } + ByteBuffer values = ByteBuffer.wrap(page, 0, bytes).order(ByteOrder.LITTLE_ENDIAN); + 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"); + } + ordinal = nextOrdinal; + pts = nextPts; + position = nextPosition; } - offset += length; + 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 59f89147402e..dd4565c8c36d 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 @@ -47,6 +47,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.RandomAccessFile; import java.nio.ByteBuffer; @@ -64,8 +65,9 @@ public class VideoFileFormatTest { private static final String KEYFRAME_INDEX_HEX = - "0149464b4f45444956010000000200000000000000000000000100000000000000" - + "789c6360c00e78a0f4821e08cd04e503001394013b"; + "0149464b4f4544495601000000020000000000000000000000000000000100000000000000000000" + + "0000000000000000000000000000000000000000000c00000000000000a08c000000000000020000" + + "0000000000dfd9164b"; @TempDir java.nio.file.Path tempPath; @@ -130,8 +132,11 @@ public void testPackRawVideosAndMapFrameRuns() throws IOException { @Test public void testCrossLanguageV1Fixture() throws IOException { byte[] fixture = fixture("video-v1.hex"); + checkVideoFixture(fixture, fromHex(KEYFRAME_INDEX_HEX)); + } + + private void checkVideoFixture(byte[] fixture, byte[] mapping) throws IOException { byte[] video = "abc".getBytes(StandardCharsets.UTF_8); - byte[] mapping = fromHex(KEYFRAME_INDEX_HEX); java.nio.file.Path source = tempPath.resolve("indexed.mp4"); byte[] sourceBytes = new byte[video.length + mapping.length]; System.arraycopy(video, 0, sourceBytes, 0, video.length); @@ -218,31 +223,94 @@ public void testRejectInvalidKeyframeIndex() throws IOException { } @Test - public void testRejectTooManyKeyframes() { + public void testRejectForgedPagedCount() { byte[] mapping = - ByteBuffer.allocate(18) + ByteBuffer.allocate(21) .order(ByteOrder.LITTLE_ENDIAN) .put((byte) 1) .putLong(0x564944454F4B4649L) .putInt(0) - .putInt((int) VideoKeyframeIndex.MAX_KEYFRAME_COUNT + 1) - .put((byte) 0) + .putLong(Long.MAX_VALUE) .array(); - assertThatThrownBy(() -> VideoKeyframeIndex.validate(mapping, 1)) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + public void testPagedIndexValidation() throws IOException { + int count = 750000; + byte[] mapping = pagedIndex(count); + assertThat(mapping.length).isGreaterThan(16 * 1024 * 1024); + VideoKeyframeIndex.validate(mapping, count); + java.nio.file.Path source = tempPath.resolve("large-index.mp4"); + try (java.io.OutputStream out = Files.newOutputStream(source)) { + out.write(new byte[count]); + out.write(mapping); + } + Blob frame = + Blob.fromDescriptor( + org.apache.paimon.utils.UriReader.fromFile(fileIO), + new VideoFrameDescriptor( + new Path(source.toUri()).toString(), + 0, + count, + 0, + count, + mapping.length)); + write(frame); + VideoFrameDescriptor restored = descriptor(read(null).get(0)); + assertThat( + VideoFrameDescriptor.keyframeIndexBlob( + Blob.fromDescriptor( + org.apache.paimon.utils.UriReader.fromFile(fileIO), + restored)) + .toData()) + .isEqualTo(mapping); + mapping[21 + 4096 * 24 + 4] ^= 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); + assertThatThrownBy(() -> VideoKeyframeIndex.validate(descending, 8192)) .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("keyframe count exceeds limit"); + .hasMessageContaining("ordinals"); + } + + private static byte[] pagedIndex(int count) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + out.write( + ByteBuffer.allocate(21) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putLong(0x564944454F4B4649L) + .putInt(0) + .putLong(count) + .array()); + for (int start = 0; start < count; start += 4096) { + int end = Math.min(count, start + 4096); + ByteBuffer page = + ByteBuffer.allocate((end - start) * 24).order(ByteOrder.LITTLE_ENDIAN); + for (int i = start; i < end; i++) { + page.putLong(i).putLong(i).putLong(i); + } + java.util.zip.CRC32 crc = new java.util.zip.CRC32(); + crc.update(page.array()); + out.write(page.array()); + out.write(intToLittleEndian((int) crc.getValue())); + } + return out.toByteArray(); } @Test public void testRejectTooManyMetadataRanges() { byte[] mapping = - ByteBuffer.allocate(18) + ByteBuffer.allocate(22) .order(ByteOrder.LITTLE_ENDIAN) .put((byte) 1) .putLong(0x564944454F4B4649L) .putInt((int) VideoKeyframeIndex.MAX_METADATA_RANGE_COUNT + 1) - .putInt(1) + .putLong(1) .put((byte) 0) .array(); @@ -251,40 +319,13 @@ public void testRejectTooManyMetadataRanges() { .hasMessageContaining("metadata range count exceeds limit"); } - @Test - public void testRejectOversizedKeyframeIndexesBeforeFetch() throws IOException { - String missing = new Path(tempPath.resolve("missing.mp4").toUri()).toString(); - Blob frame = - Blob.fromDescriptor( - org.apache.paimon.utils.UriReader.fromFile(fileIO), - new VideoFrameDescriptor( - missing, - 0, - 1, - 0, - 1, - VideoFormatWriter.MAX_KEYFRAME_INDEX_BYTES + 1)); - - assertThatThrownBy(() -> write(frame)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("keyframe index length") - .hasMessageContaining("limit"); - assertThatThrownBy( - () -> - VideoFormatWriter.checkKeyframeIndexSize( - 1, VideoFormatWriter.MAX_TOTAL_KEYFRAME_INDEX_BYTES)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Buffered video keyframe indexes") - .hasMessageContaining("limit"); - } - @Test public void testRejectSeekOffsetsOutsideVideoPayload() throws IOException { byte[] video = "video".getBytes(StandardCharsets.UTF_8); byte[] mapping = fromHex( - "0149464b4f45444956010000000100000000000000000000000600000000000000" - + "789c6360c00e0000180001"); + "0149464b4f4544495601000000010000000000000000000000000000000600000000000000000000" + + "00000000000000000000000000000000000000000020cac1a3"); 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); @@ -312,8 +353,8 @@ public void testRejectInconsistentKeyframeIndexesForSamePayload() throws IOExcep byte[] firstIndex = fromHex(KEYFRAME_INDEX_HEX); byte[] secondIndex = fromHex( - "0149464b4f454449560000000002000000789c6360c00ed8a074801b846682f201" - + "09ea009f"); + "0149464b4f4544495600000000020000000000000000000000000000000000000000000000000000" + + "00000000000600000000000000504600000000000002000000000000005381c820"); 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); @@ -464,8 +505,8 @@ public void testRejectCorruptRunReference() throws IOException { } @Test - public void testReaderKeyframeIndexSizeLimits() throws IOException { - long limit = VideoFormatWriter.MAX_KEYFRAME_INDEX_BYTES; + public void testReaderLargeKeyframeIndexOffsets() throws IOException { + long limit = 16 * 1024 * 1024; long[][] cases = { {limit + 1}, {limit, limit, limit, limit, 1}, {limit, limit, limit, limit}, {0} }; @@ -498,14 +539,7 @@ public void testReaderKeyframeIndexSizeLimits() throws IOException { out.write(VideoFormatWriter.VERSION); } try (SeekableInputStream in = fileIO.newInputStream(file)) { - if (i < 2) { - String message = i == 0 ? "16 MiB" : "64 MiB"; - assertThatThrownBy(() -> new VideoFileMeta(in, fileIO.getFileSize(file), null)) - .isInstanceOf(IOException.class) - .hasMessageContaining(message); - } else { - new VideoFileMeta(in, fileIO.getFileSize(file), null); - } + new VideoFileMeta(in, fileIO.getFileSize(file), null); } } } 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 93f047531e7b..257cddbd1844 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. -6162635758595a0149464b4f45444956010000000200000000000000000000000100000000000000789c6360c00e78a0f4821e08cd04e503001394013b06026c6b040100020100010106010403000e0602000000020000000500000005000000050000004944454f01 +6162635758595a0149464b4f45444956010000000200000000000000000000000000000001000000000000000000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000dfd9164b0602b201b101040100020100010106010403000e0602000000040000000500000005000000050000004944454f01 diff --git a/paimon-python/pypaimon/execution/python/read/reader/video_format_reader.py b/paimon-python/pypaimon/execution/python/read/reader/video_format_reader.py index fa66128952d1..04ac0be4b545 100644 --- a/paimon-python/pypaimon/execution/python/read/reader/video_format_reader.py +++ b/paimon-python/pypaimon/execution/python/read/reader/video_format_reader.py @@ -93,10 +93,6 @@ def __init__(self, stream, file_size: int): ) keyframe_index_size = sum(keyframe_index_lengths) - if any(length > 16 * 1024 * 1024 for length in keyframe_index_lengths): - raise IOError("Corrupt video file: keyframe index exceeds the 16 MiB limit.") - if keyframe_index_size > 64 * 1024 * 1024: - raise IOError("Corrupt video file: keyframe indexes exceed the 64 MiB limit.") mapping_start = index_start - keyframe_index_size if mapping_start < 0: raise IOError( 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 859214c308f6..3fc5ab39f4e9 100644 --- a/paimon-python/pypaimon/execution/python/write/video_format_writer.py +++ b/paimon-python/pypaimon/execution/python/write/video_format_writer.py @@ -15,6 +15,9 @@ # specific language governing permissions and limitations # under the License. +import errno +import shutil +import tempfile import struct import sys @@ -44,8 +47,6 @@ class VideoFormatWriter(BlobFormatWriter): FOOTER_SIZE = 25 NULL_REFERENCE = -1 PLACE_HOLDER_REFERENCE = -2 - MAX_KEYFRAME_INDEX_BYTES = 16 * 1024 * 1024 - MAX_TOTAL_KEYFRAME_INDEX_BYTES = 64 * 1024 * 1024 def __init__( self, @@ -60,6 +61,7 @@ def __init__( ) self._physical_lengths = [] self._keyframe_indexes = [] + self._index_spool = tempfile.SpooledTemporaryFile(max_size=1024 * 1024) self._physical_videos = {} self._physical_video_keyframe_indexes = {} self._run_lengths = [] @@ -104,11 +106,16 @@ def add_element(self, row) -> None: ordinal = self._physical_videos.get(payload) if ordinal is None: keyframe_index = self._keyframe_index(value, frame) - length = self._write_video_payload(value) + try: + length = self._write_video_payload(value) + except BaseException: + self._index_spool.seek(self._keyframe_index_bytes) + self._index_spool.truncate() + raise ordinal = len(self._physical_lengths) self._physical_lengths.append(length) self._keyframe_indexes.append(keyframe_index) - self._keyframe_index_bytes += len(keyframe_index) + self._keyframe_index_bytes += keyframe_index self._physical_videos[payload] = ordinal self._physical_video_keyframe_indexes[ payload] = keyframe_index_descriptor @@ -134,87 +141,80 @@ def run_count(self) -> int: def close(self) -> None: if self._closed: return - self._flush_run() - for keyframe_index in self._keyframe_indexes: - self.output_stream.write(keyframe_index) - physical_index = DeltaVarintCompressor.compress(self._physical_lengths) - keyframe_length_index = DeltaVarintCompressor.compress( - [len(mapping) for mapping in self._keyframe_indexes] - ) - run_length_index = DeltaVarintCompressor.compress(self._run_lengths) - run_reference_index = DeltaVarintCompressor.compress( - self._run_references - ) - first_frame_index = DeltaVarintCompressor.compress( - self._run_first_frames - ) - for index in ( - physical_index, - keyframe_length_index, - run_length_index, - run_reference_index, - first_frame_index): - self.output_stream.write(index) - footer = struct.pack( - ' self.MAX_KEYFRAME_INDEX_BYTES: - raise ValueError( - "Video keyframe index length %s exceeds the %s-byte limit." - % (length, self.MAX_KEYFRAME_INDEX_BYTES) - ) - if (self._keyframe_index_bytes - > self.MAX_TOTAL_KEYFRAME_INDEX_BYTES - length): - raise ValueError( - "Buffered video keyframe indexes exceed the %s-byte limit." - % self.MAX_TOTAL_KEYFRAME_INDEX_BYTES - ) + stream.seek(0, 2) + payload_length = stream.tell() + stream.seek(0) + try: + mapping = VideoKeyframeIndex.inspect(stream, payload_length) + except _video_index_fallback_errors() as error: + if isinstance(error, OSError) and error.errno in (errno.ENOSPC, errno.EDQUOT, errno.EIO): + raise + return 0 + with mapping: + mapping.write_to(self._index_spool) + 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) + mapping.seek(0) + shutil.copyfileobj(mapping, self._index_spool, self.copy_buffer_size) + return self._index_spool.tell() - start + except BaseException: + self._index_spool.seek(start) + self._index_spool.truncate() + raise def _write_video_payload(self, blob: BlobRef) -> int: start = self.position diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index e1f55123600e..c13aaf1d1c39 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -2006,7 +2006,7 @@ def __init__(self, stream, keyframe_index=None, thread_count=0): self._keyframe_index = keyframe_index self._thread_count = thread_count self._anchor_ordinals = ( - [entry[0] for entry in keyframe_index.keyframes] + keyframe_index.ordinals if keyframe_index is not None else []) self._container = None self._stream = None diff --git a/paimon-python/pypaimon/multimodal/video.py b/paimon-python/pypaimon/multimodal/video.py index a45b01b0a0cf..dfa1d1a93b7f 100644 --- a/paimon-python/pypaimon/multimodal/video.py +++ b/paimon-python/pypaimon/multimodal/video.py @@ -383,13 +383,11 @@ def _decoder(self, descriptor, keyframe_index_descriptor=None, replica=0): ).new_input_stream() try: if keyframe_index_descriptor is not None: - mapping = self.file_io.read_file_range( - keyframe_index_descriptor.uri, - keyframe_index_descriptor.offset, - keyframe_index_descriptor.length, - ) - stream.video_keyframe_index = ( - VideoKeyframeIndex.deserialize(mapping, descriptor.length)) + 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) stream.video_length = descriptor.length stream.video_read_ranges = lambda ranges: ( self.file_io.read_ranges_coalesced( @@ -403,6 +401,9 @@ def _decoder(self, descriptor, keyframe_index_descriptor=None, replica=0): ) decoder = self.decoder_factory(stream) except Exception: + mapping = getattr(stream, "video_keyframe_index", None) + if mapping is not None: + mapping.close() stream.close() raise resource = (decoder, stream) @@ -443,4 +444,7 @@ def _close_resource(resource): if close is not None: close() finally: + mapping = getattr(stream, "video_keyframe_index", None) + if mapping is not None: + mapping.close() stream.close() diff --git a/paimon-python/pypaimon/table/row/video_keyframe_index.py b/paimon-python/pypaimon/table/row/video_keyframe_index.py index acabf9ce4acb..550eb1555950 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -15,173 +15,337 @@ # specific language governing permissions and limitations # under the License. -"""Sparse seek metadata for encoded video payloads.""" +"""Paged seek metadata for encoded video payloads.""" +import io import operator +import sqlite3 import struct +import tempfile +import threading import zlib +from collections import OrderedDict +from collections.abc import Sequence +from contextlib import closing -class VideoKeyframeIndex: - """Persist video metadata ranges and presentation-order keyframes.""" +class _PagedKeyframes(Sequence): + ENTRY = struct.Struct(' max(1, self.CACHE_BYTES // self.PAGE_BYTES): + self.cache.popitem(last=False) + return data + + def __getitem__(self, index): + if isinstance(index, slice): + return tuple(self[i] for i in range(*index.indices(len(self)))) + index = operator.index(index) + if index < 0: + index += len(self) + if not 0 <= index < len(self): + raise IndexError("Keyframe index out of range.") + page, offset = divmod(index, self.PAGE_ENTRIES) + return self.ENTRY.unpack_from(self._page(page), offset * self.ENTRY.size) + + def __iter__(self): + for page in range((self.count + self.PAGE_ENTRIES - 1) // self.PAGE_ENTRIES): + yield from self.ENTRY.iter_unpack(self._page(page)) + + def __eq__(self, other): + if not isinstance(other, Sequence): + return NotImplemented + return len(self) == len(other) and all(a == b for a, b in zip(self, other)) + + def close(self): + if self.owner is not None: + self.owner.close() + self.cache.clear() + + +class _Ordinals(Sequence): + def __init__(self, keyframes): + self.keyframes = keyframes + + def __len__(self): + return len(self.keyframes) + def __getitem__(self, index): + return self.keyframes[index][0] + + +def _validate_boundary(previous, current): + if any(a >= b for a, b in zip(previous, current)): + raise ValueError("Keyframe ordinals, timestamps and packet positions must be strictly increasing.") + + +def _validate_entries(data, payload_length, initial=False): + import numpy as np + + entries = np.frombuffer(data, dtype='= payload_length): + raise ValueError("Keyframe packet positions must be within the video payload.") + + +class VideoKeyframeIndex: VERSION = 1 - MAGIC = 0x564944454F4B4649 # "VIDEOKFI" - HEADER = struct.Struct('= self.MAX_METADATA_RANGE_COUNT: - raise ValueError( - "Video keyframe index exceeds the %s-metadata-range " - "limit." % self.MAX_METADATA_RANGE_COUNT - ) - metadata_range_count += 1 - offset, length = operator.index(offset), operator.index(length) - if normalized_metadata_ranges: - previous_offset, previous_length = ( - normalized_metadata_ranges[-1]) - if previous_offset + previous_length == offset: - normalized_metadata_ranges[-1] = ( - previous_offset, previous_length + length) - continue - normalized_metadata_ranges.append((offset, length)) - self.metadata_ranges = tuple(normalized_metadata_ranges) - normalized_keyframes = [] - for ordinal, pts, position in keyframes: - if len(normalized_keyframes) >= self.MAX_KEYFRAME_COUNT: - raise ValueError( - "Video keyframe index exceeds the %s-entry limit." - % self.MAX_KEYFRAME_COUNT - ) - normalized_keyframes.append(( - operator.index(ordinal), operator.index(pts), - operator.index(position), - )) - self.keyframes = tuple(normalized_keyframes) + self.metadata_ranges = self._metadata(metadata_ranges) + self.keyframes = keyframes + + @classmethod + def _metadata(cls, ranges, payload_length=None): + result = [] previous_end = 0 - for offset, length in self.metadata_ranges: - if offset < previous_end or length <= 0: - raise ValueError( - "Video metadata ranges must be ordered, non-overlapping, " - "and non-empty." - ) - previous_end = offset + length - if not self.keyframes or self.keyframes[0][0] != 0: - raise ValueError("Video keyframe index requires an initial keyframe.") - previous_ordinal, previous_pts, previous_position = -1, None, -1 - for ordinal, pts, position in self.keyframes: - if ordinal <= previous_ordinal: - raise ValueError("Keyframe ordinals must be strictly increasing.") - if previous_pts is not None and pts <= previous_pts: - raise ValueError("Keyframe timestamps must be strictly increasing.") - if position <= previous_position: - raise ValueError( - "Keyframe packet positions must be non-negative and " - "strictly increasing." - ) - previous_ordinal, previous_pts, previous_position = ( - ordinal, pts, position) + for i, (start, length) in enumerate(ranges): + if i >= cls.MAX_METADATA_RANGE_COUNT: + raise ValueError("Video keyframe index exceeds the metadata-range limit.") + start, length = operator.index(start), operator.index(length) + if start < previous_end or length <= 0: + raise ValueError("Invalid video keyframe index metadata ranges.") + if payload_length is not None and start + length > payload_length: + raise ValueError("Video metadata range is outside the video payload.") + cls.METADATA_RANGE.pack(start, length) + if result and start == previous_end: + result[-1] = (result[-1][0], result[-1][1] + length) + else: + result.append((start, length)) + previous_end = start + length + return tuple(result) @property def keyframes(self): - entries = self._keyframes - if isinstance(entries, bytes): - entries = tuple(self.ENTRY.iter_unpack(entries)) - self._keyframes = entries - return entries + return self._keyframes @keyframes.setter def keyframes(self, entries): - self._keyframes = entries + storage = tempfile.TemporaryFile() + count, previous = 0, None + page = bytearray() + try: + for entry in entries: + entry = tuple(operator.index(v) for v in entry) + packed = self.ENTRY.pack(*entry) + if previous is None: + if entry[0] != 0 or entry[2] < 0: + raise ValueError("Video keyframe index requires an initial keyframe.") + else: + _validate_boundary(previous, entry) + page.extend(packed) + count += 1 + previous = entry + if len(page) == _PagedKeyframes.PAGE_BYTES: + storage.write(page) + storage.write(struct.pack(' length: + raise ValueError("Invalid video keyframe index length.") + if size == 0: + return b'' + if offset + size <= len(prefix): + return prefix[offset:offset + size] + data = read_at(offset, size) + if len(data) != size: + raise ValueError("Invalid video keyframe index: truncated read.") + return data + + version = read(0, 1)[0] + if version != cls.VERSION: + raise ValueError("Invalid video keyframe index version.") + header = cls.HEADER + _, magic, metadata_count, count = header.unpack(read(0, header.size)) + if magic != cls.MAGIC or count == 0: + raise ValueError("Invalid video keyframe index header.") + if metadata_count > 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: + raise ValueError("Invalid video keyframe index entries length.") + result = cls.__new__(cls) + result.metadata_ranges = ranges + result._keyframes = _PagedKeyframes(lambda start, size: read(offset + start, size), count, payload_length) + result.keyframes[0] + return result + + def _validate_all(self): + previous = None + for page in range((len(self.keyframes) + _PagedKeyframes.PAGE_ENTRIES - 1) // _PagedKeyframes.PAGE_ENTRIES): + data = self.keyframes._page(page) + if previous is not None: + _validate_boundary(previous, self.ENTRY.unpack_from(data)) + previous = self.ENTRY.unpack_from(data, len(data) - self.ENTRY.size) + + @classmethod + def validate_stream(cls, source, length, payload_length=None): + def read_at(offset, size): + source.seek(offset) + return source.read(size) + with cls.open(read_at, length, payload_length) as index: + index._validate_all() + + @classmethod + def validate(cls, data, payload_length=None): + cls.validate_stream(io.BytesIO(data), len(data), payload_length) + + @classmethod + def deserialize(cls, data, payload_length=None): + if not isinstance(data, (bytes, bytearray)): + raise TypeError("VideoKeyframeIndex expects bytes.") + index = cls.open(lambda start, size: data[start:start + size], len(data), payload_length) + try: + index._validate_all() + return index + except BaseException: + index.close() + raise @classmethod def inspect(cls, source, payload_length): - """Build an index for the first video stream of an ISO BMFF file.""" + """Build an index using a disk-backed packet lookup.""" try: import av except ImportError as error: - raise ImportError( - "Video seek-index generation requires PyAV." - ) from error - - metadata_ranges = cls._iso_bmff_metadata_ranges( - source, payload_length) - source.seek(0) - with av.open(source) as container: - if not container.streams.video: - raise ValueError("Video has no video stream.") - stream = container.streams.video[0] - packet_positions = {} - for packet in container.demux(stream): - if (not packet.is_keyframe or packet.pts is None - or packet.pos is None or packet.pos < 0 - or getattr(packet, "is_discard", False)): - continue - pts = cls._stream_pts( - packet.pts, - packet.time_base or stream.time_base, - stream.time_base, - ) - if pts in packet_positions: - raise ValueError("Video has duplicate keyframe timestamps.") - if len(packet_positions) >= cls.MAX_KEYFRAME_COUNT: - raise ValueError( - "Video keyframe index exceeds the %s-entry limit." - % cls.MAX_KEYFRAME_COUNT - ) - packet_positions[pts] = int(packet.pos) - - source.seek(0) - with av.open(source) as container: - stream = container.streams.video[0] - keyframes = [] - previous_pts = None - for ordinal, frame in enumerate(container.decode(stream)): - if frame.pts is None: - raise ValueError("Video frame has no presentation timestamp.") - pts = cls._stream_pts( - frame.pts, - frame.time_base or stream.time_base, - stream.time_base, - ) - if previous_pts is not None and pts <= previous_pts: - raise ValueError( - "Video frame timestamps must be strictly increasing." - ) - if frame.key_frame: - position = packet_positions.get(pts) - if position is None: - raise ValueError( - "Video keyframe has no packet byte position." - ) - if len(keyframes) >= cls.MAX_KEYFRAME_COUNT: - raise ValueError( - "Video keyframe index exceeds the %s-entry limit." - % cls.MAX_KEYFRAME_COUNT - ) - keyframes.append((ordinal, pts, position)) - previous_pts = pts - return cls(metadata_ranges, keyframes) + raise ImportError("Video seek-index generation requires PyAV.") from error + metadata_ranges = cls._iso_bmff_metadata_ranges(source, payload_length) + with tempfile.TemporaryDirectory(prefix="paimon_keyframes_") as directory: + with closing(sqlite3.connect(directory + "/packets.db")) as packets: + packets.execute("PRAGMA cache_size = -2048") + packets.execute("CREATE TABLE packets (pts INTEGER PRIMARY KEY, position INTEGER)") + source.seek(0) + with av.open(source) as container: + if not container.streams.video: + raise ValueError("Video has no video stream.") + stream = container.streams.video[0] + for packet in container.demux(stream): + if (not packet.is_keyframe or packet.pts is None or packet.pos is None + or packet.pos < 0 or getattr(packet, 'is_discard', False)): + continue + pts = cls._stream_pts(packet.pts, packet.time_base or stream.time_base, stream.time_base) + try: + packets.execute("INSERT INTO packets VALUES (?, ?)", (pts, packet.pos)) + except sqlite3.IntegrityError as error: + raise ValueError("Video has duplicate keyframe timestamps.") from error + source.seek(0) + with av.open(source) as container: + stream = container.streams.video[0] + + def entries(): + previous_pts = None + for ordinal, frame in enumerate(container.decode(stream)): + if frame.pts is None: + raise ValueError("Video frame has no presentation timestamp.") + pts = cls._stream_pts(frame.pts, frame.time_base or stream.time_base, stream.time_base) + if previous_pts is not None and pts <= previous_pts: + raise ValueError("Video frame timestamps must be strictly increasing.") + if frame.key_frame: + match = packets.execute("SELECT position FROM packets WHERE pts = ?", (pts,)).fetchone() + if match is None: + raise ValueError("Video keyframe has no packet byte position.") + yield ordinal, pts, match[0] + previous_pts = pts + return cls(metadata_ranges, entries()) @staticmethod def _stream_pts(pts, time_base, stream_time_base): @@ -233,174 +397,3 @@ def _iso_bmff_metadata_ranges(cls, source, payload_length): if offset != payload_length or not has_moov: raise ValueError("Video is not a supported ISO BMFF file.") return ranges - - @classmethod - def validate(cls, data, payload_length=None): - metadata_count, keyframe_count, entries_offset = cls._read_header(data) - for _ in cls._iter_metadata_ranges( - data, metadata_count, payload_length): - pass - for _ in cls._iter_keyframes( - data, entries_offset, keyframe_count, payload_length): - pass - - @classmethod - def deserialize(cls, data, payload_length=None): - import numpy as np - - metadata_count, keyframe_count, entries_offset = cls._read_header(data) - metadata_ranges = [] - for start, length in cls._iter_metadata_ranges( - data, metadata_count, payload_length): - if (metadata_ranges - and sum(metadata_ranges[-1]) == start): - previous_start, previous_length = metadata_ranges[-1] - metadata_ranges[-1] = (previous_start, previous_length + length) - else: - metadata_ranges.append((start, length)) - - # The header caps this buffer at 1.5 MiB. One extra byte detects - # excess output without expanding an untrusted compressed stream. - expected_size = keyframe_count * cls.ENTRY.size - decoder = zlib.decompressobj() - try: - raw = decoder.decompress( - memoryview(data)[entries_offset:], expected_size + 1) - except zlib.error as error: - raise ValueError( - "Invalid video keyframe index payload." - ) from error - if (len(raw) != expected_size or not decoder.eof - or decoder.unused_data or decoder.unconsumed_tail): - raise ValueError("Invalid video keyframe index entries.") - - # ENTRY '= payload_length)): - raise ValueError( - "Keyframe packet positions must be within the " - "video payload and strictly increasing." - ) - - # Validation is eager, but decoders that do not consume keyframes need - # not allocate Python tuples and integers for every entry. - result = cls.__new__(cls) - result.metadata_ranges = tuple(metadata_ranges) - result._keyframes = raw - return result - - @classmethod - def _read_header(cls, data): - if not isinstance(data, (bytes, bytearray)): - raise TypeError("VideoKeyframeIndex expects bytes.") - if len(data) <= cls.HEADER.size: - raise ValueError("Invalid video keyframe index: too short.") - version, magic, metadata_count, keyframe_count = ( - cls.HEADER.unpack(data[:cls.HEADER.size])) - if version != cls.VERSION or magic != cls.MAGIC: - raise ValueError("Invalid video keyframe index version or magic.") - if keyframe_count == 0: - raise ValueError("Invalid video keyframe index header.") - if metadata_count > cls.MAX_METADATA_RANGE_COUNT: - raise ValueError( - "Video keyframe index exceeds the %s-metadata-range limit." - % cls.MAX_METADATA_RANGE_COUNT - ) - if keyframe_count > cls.MAX_KEYFRAME_COUNT: - raise ValueError( - "Video keyframe index exceeds the %s-entry limit." - % cls.MAX_KEYFRAME_COUNT - ) - entries_offset = ( - cls.HEADER.size + metadata_count * cls.METADATA_RANGE.size) - if entries_offset >= len(data): - raise ValueError("Invalid video keyframe index metadata ranges.") - return metadata_count, keyframe_count, entries_offset - - @classmethod - def _iter_metadata_ranges(cls, data, count, payload_length): - previous_end = 0 - offset = cls.HEADER.size - for _ in range(count): - start, length = cls.METADATA_RANGE.unpack_from(data, offset) - end = start + length - if start < previous_end or length <= 0: - raise ValueError( - "Invalid video keyframe index metadata ranges." - ) - if payload_length is not None and end > payload_length: - raise ValueError( - "Video metadata range is outside the video payload." - ) - previous_end = end - offset += cls.METADATA_RANGE.size - yield start, length - - @classmethod - def _iter_keyframes(cls, data, input_offset, count, payload_length): - try: - decoder = zlib.decompressobj() - pending = b'' - remainder = b'' - decoded_count = 0 - previous_ordinal, previous_pts, previous_position = -1, None, -1 - while not decoder.eof: - if not pending and input_offset < len(data): - end = min(input_offset + cls._CHUNK_SIZE, len(data)) - pending = data[input_offset:end] - input_offset = end - output = decoder.decompress(pending, cls._CHUNK_SIZE) - pending = decoder.unconsumed_tail - entries = remainder + output - complete = len(entries) - len(entries) % cls.ENTRY.size - remainder = entries[complete:] - for ordinal, pts, position in cls.ENTRY.iter_unpack( - entries[:complete]): - if decoded_count >= count: - raise ValueError( - "Invalid video keyframe index entries." - ) - if decoded_count == 0 and ordinal != 0: - raise ValueError( - "Video keyframe index requires an initial " - "keyframe." - ) - if ordinal <= previous_ordinal: - raise ValueError( - "Keyframe ordinals must be strictly increasing." - ) - if previous_pts is not None and pts <= previous_pts: - raise ValueError( - "Keyframe timestamps must be strictly increasing." - ) - if (position <= previous_position - or (payload_length is not None - and position >= payload_length)): - raise ValueError( - "Keyframe packet positions must be within the " - "video payload and strictly increasing." - ) - decoded_count += 1 - previous_ordinal, previous_pts, previous_position = ( - ordinal, pts, position) - yield ordinal, pts, position - if not output and not pending and input_offset == len(data): - break - if (not decoder.eof or decoder.unused_data - or pending or input_offset != len(data) - or remainder or decoded_count != count): - raise ValueError("Invalid video keyframe index entries.") - except (struct.error, zlib.error) as error: - raise ValueError( - "Invalid video keyframe index payload." - ) from error diff --git a/paimon-python/pypaimon/tests/video_format_test.py b/paimon-python/pypaimon/tests/video_format_test.py index bd17f887bef2..6665fd9ee0c7 100644 --- a/paimon-python/pypaimon/tests/video_format_test.py +++ b/paimon-python/pypaimon/tests/video_format_test.py @@ -159,9 +159,11 @@ 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) - encoded = VideoKeyframeIndex( - [(0, 1)], [(0, 0, 0), (12, 36000, 2)] - ).serialize() + with VideoKeyframeIndex([(0, 1)], [(0, 0, 0), (12, 36000, 2)]) as index: + encoded = index.serialize() + self._check_video_fixture(fixture, encoded) + + def _check_video_fixture(self, fixture, encoded): video = b"abc" source_path = self.root / "indexed.mp4" source_path.write_bytes(video + encoded) @@ -403,7 +405,54 @@ def test_ffmpeg_index_errors_preserve_video_payload(self): side_effect=error_type(1, "Cannot index video")): writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) self.assertEqual(payload, output.getvalue()) - self.assertEqual([b""], writer._keyframe_indexes) + self.assertEqual([0], writer._keyframe_indexes) + + def test_generated_large_index_is_kept(self): + count = 700_000 + payload = b'x' * count + source = self.root / "large-index.mp4" + source.write_bytes(payload) + 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)) + 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)) + self.assertGreater(writer._keyframe_indexes[0], 16 * 1024 * 1024) + self.assertTrue(writer._index_spool._rolled) + writer.close() + result = VideoFrameDescriptor.deserialize(self._read(target)[0]) + self.assertGreater(result.keyframe_index_descriptor.length, 16 * 1024 * 1024) + + def test_index_spool_failure_does_not_silently_drop_index(self): + import errno + blob = self._source_frame("spool-failure.mp4", b"video", 0) + writer = VideoFormatWriter(io.BytesIO()) + for error in (OSError(errno.ENOSPC, "Disk full"), OSError(errno.EDQUOT, "Quota")): + with mock.patch.object(VideoKeyframeIndex, "inspect", side_effect=error): + with self.assertRaises(OSError): + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + with VideoKeyframeIndex([], [(0, 0, 0)]) as index: + with mock.patch.object(VideoKeyframeIndex, "inspect", return_value=index), mock.patch.object( + index, "write_to", side_effect=OSError("Spool write failed")): + with self.assertRaises(OSError): + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + self.assertEqual(0, writer._index_spool.tell()) + self.assertEqual([], writer._keyframe_indexes) + writer.close() + + def test_payload_failure_discards_staged_index(self): + with VideoKeyframeIndex([], [(0, 0, 0)]) as index: + mapping = index.serialize() + blob = self._indexed_frame("payload-failure.mp4", b"video", mapping) + writer = VideoFormatWriter(io.BytesIO()) + with mock.patch.object(writer, "_write_video_payload", side_effect=IOError("Failed payload")): + with self.assertRaises(IOError): + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + self.assertEqual(0, writer._index_spool.seek(0, 2)) + self.assertEqual([], writer._keyframe_indexes) + writer.close() def test_rejects_invalid_keyframe_index(self): video = b"video" @@ -423,42 +472,30 @@ def test_rejects_invalid_keyframe_index(self): GenericRow([blob], [self.field], RowKind.INSERT) ) - def test_rejects_oversized_keyframe_indexes_before_fetch(self): - descriptor = VideoFrameDescriptor( - (self.root / "missing.mp4").as_uri(), - 0, - 1, - 0, - 1, - VideoFormatWriter.MAX_KEYFRAME_INDEX_BYTES + 1, - ) - blob = Blob.from_descriptor( - self.file_io.uri_reader_factory.create(descriptor.uri), descriptor - ) - - with self.assertRaisesRegex(ValueError, "keyframe index length.*limit"): - VideoFormatWriter(io.BytesIO()).add_element( - GenericRow([blob], [self.field], RowKind.INSERT) - ) - - def test_rejects_cumulative_keyframe_indexes_before_fetch(self): - mapping = VideoKeyframeIndex([], [(0, 0, 0)]).serialize() - first = self._indexed_frame("first-indexed.mp4", b"a", mapping) - second = self._indexed_frame("second-indexed.mp4", b"b", mapping) - writer = VideoFormatWriter(io.BytesIO()) - - with mock.patch.object( - VideoFormatWriter, - 'MAX_TOTAL_KEYFRAME_INDEX_BYTES', - len(mapping)): - writer.add_element( - GenericRow([first], [self.field], RowKind.INSERT) - ) - with self.assertRaisesRegex( - ValueError, "Buffered video keyframe indexes.*limit"): - writer.add_element( - GenericRow([second], [self.field], RowKind.INSERT) - ) + 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() + video = b'x' * count + target = (self.root / "many-indexes.video").as_uri() + writer = VideoFormatWriter(self.file_io.new_output_stream(target)) + for i in range(4): + frame = self._indexed_frame("large-%s.mp4" % i, video, mapping) + writer.add_element(GenericRow([frame], [self.field], RowKind.INSERT)) + self.assertGreater(writer._keyframe_index_bytes, 64 * 1024 * 1024) + self.assertTrue(writer._index_spool._rolled) + writer.close() + values = self._read(target) + rewritten = (self.root / "large-rewritten.video").as_uri() + writer = VideoFormatWriter(self.file_io.new_output_stream(rewritten)) + for value in values: + writer.add_element(GenericRow([Blob.from_bytes(value, file_io=self.file_io)], + [self.field], RowKind.INSERT)) + writer.close() + for value in self._read(rewritten): + descriptor = VideoFrameDescriptor.deserialize(value).keyframe_index_descriptor + self.assertEqual(mapping, Blob.from_file(self.file_io, descriptor.uri, + descriptor.offset, descriptor.length).to_data()) def test_rejects_seek_offsets_outside_video_payload(self): video = b"video" @@ -501,7 +538,7 @@ def test_rejects_compressed_invalid_index_without_expanding_it(self): try: with self.assertRaisesRegex( ValueError, - "entry limit"): + "entries length"): writer.add_element( GenericRow([blob], [self.field], RowKind.INSERT) ) @@ -644,11 +681,9 @@ def test_rejects_out_of_range_run_reference(self): blob_as_descriptor=True, ) - def test_reader_keyframe_index_size_limits(self): + def test_reader_large_keyframe_index_offsets(self): limit = 16 * 1024 * 1024 - for lengths, error in (([limit + 1], "16 MiB"), - ([limit] * 4 + [1], "64 MiB"), - ([limit] * 4, None), ([0], None)): + for lengths in ([limit + 1], [limit] * 4 + [1], [limit] * 4, [0]): with self.subTest(lengths=lengths): indexes = [DeltaVarintCompressor.compress(values) for values in ([1] * len(lengths), lengths, [], [], [])] @@ -661,11 +696,7 @@ def test_reader_keyframe_index_size_limits(self): '