[format][python] Support large video indexes with paged reads

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`:
 

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

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));

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) {

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<byte[]> 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(