[format][python] Add video keyframe index format for efficient random reads

docs/docs/concepts/spec/fileformat.md

@@ -414,15 +414,22 @@ the ordinary BLOB entry header, length trailer, or per-entry CRC:
 +----------------------------+
 | ...                        |
 +----------------------------+
+| Keyframe Index 1           |  Video metadata ranges and compressed keyframe entries
++----------------------------+
+| Keyframe Index 2           |
++----------------------------+
 | Physical Length Index      |  Delta-Varint video lengths
 +----------------------------+
+| Keyframe-Index Length Index |  Delta-Varint keyframe-index lengths
++----------------------------+
 | Run Length Index           |  Delta-Varint logical row counts
 +----------------------------+
 | Run Reference Index        |  Delta-Varint physical video ordinals
 +----------------------------+
 | Run First-Frame Index      |  Delta-Varint frame ordinals
 +----------------------------+
 | Physical Index Length      |  4 bytes (Little Endian)
+| Keyframe Length-Index Size |  4 bytes (Little Endian)
 | Run-Length Index Length    |  4 bytes (Little Endian)
 | Run-Reference Index Length |  4 bytes (Little Endian)
 | First-Frame Index Length   |  4 bytes (Little Endian)
@@ -435,20 +442,38 @@ The run arrays have equal element counts. A non-negative run reference is an ord
 physical length index. For logical row `r` in a run beginning at logical row `s`, the returned
 `VideoFrameDescriptor` identifies the referenced raw video range and frame ordinal
 `run_first_frame + (r - s)`. `-1` is a NULL run and `-2` is a data-evolution placeholder run.
-Non-negative runs have fixed frame stride one in version 1; a discontinuity starts another run.
+Non-negative runs have fixed frame stride one; a discontinuity starts another run. Version 2 may
+store one sparse keyframe index per physical video. Version 1 remains readable and omits the
+keyframe-index blocks and their length index.
+
+A supplied keyframe-index block has a 17-byte little-endian header: version (`1`, uint8), magic
+(`0x564944454F4B4649`, uint64), metadata-range count (uint32), and keyframe count (uint32).
+The header is followed by metadata ranges as byte offset and length pairs (two int64 values), then
+zlib-compressed keyframe entries containing presentation-order frame ordinal, PTS, and packet byte
+position (three int64 values). Byte offsets are relative to the first byte of the encoded-video
+payload. For MP4, a metadata range locates `moov`, while a packet position locates the start of a
+keyframe's GOP. Writers reject ranges and packet positions outside the video payload.
+
+The index describes the first video stream. PTS uses that stream's time base, read from the encoded
+video rather than stored again. Only keyframes have entries. An indexed decoder can fetch video
+metadata and the target GOP range, seek to the keyframe PTS, and decode forward in presentation
+order; it does not infer timestamps from average FPS. The reader may extend the range through the
+following GOP when reordered frames cross a GOP boundary.
 
 The serialized `VideoFrameDescriptor` stored in an Arrow/data-file cell has its own versioned
 wire layout. All numeric values are little endian:
 
 | Field | Size | Description |
 | --- | ---: | --- |
-| Version | 1 byte | Descriptor version, currently `1` |
+| Version | 1 byte | Descriptor version, currently `2` |
 | Magic | 8 bytes | `0x564944454F46524D` (`VIDEOFRM`) |
 | URI length | 4 bytes | UTF-8 URI byte length |
 | URI | variable | URI of the containing `.video` file |
 | Offset | 8 bytes | Start of the complete encoded-video payload |
 | Length | 8 bytes | Encoded-video payload length |
 | Frame index | 8 bytes | Zero-based presentation-order frame ordinal |
+| Keyframe-index offset | 8 bytes | Version 2 only: index offset in the `.video` file |
+| Keyframe-index length | 8 bytes | Version 2 only: index length |
 
 Descriptor bytes are independently versioned from the `.video` container. Java and Python share
 canonical descriptor and container fixtures to keep both implementations byte-compatible.

docs/docs/multimodal-table/video.mdx

@@ -83,10 +83,31 @@ to write complete videos and their logical frame rows.
 
 Each logical value is a `VideoFrameDescriptor`: its URI range identifies one complete encoded
 video and its frame ordinal selects a frame inside that video. On write, a `.video` data region
-concatenates raw video payloads without ordinary BLOB entry wrappers. Four embedded delta-varint
-indexes record physical video lengths, logical run lengths, run-to-video references, and the first
-frame ordinal of each run. Consecutive frames therefore need one run entry rather than one index
-entry per row. NULL and data-evolution placeholders use negative run references.
+concatenates raw video payloads without ordinary BLOB entry wrappers. Four delta-varint indexes map
+rows to videos and frame ordinals; consecutive frames share a run, and negative references represent
+NULL or data-evolution placeholders. Version 2 carries optional sparse keyframe indexes.
+
+<pre>
+.video (version 2, in file order)
+├─ Complete encoded videos A, B, ...
+├─ <strong>NEW: Keyframe index blocks A, B, ...</strong>
+│  ├─ Video metadata ranges: offset and length within the encoded video
+│  └─ Compressed keyframe entries: frame ordinal, PTS, and packet byte position
+├─ Metadata indexes
+│  ├─ Video payload lengths
+│  ├─ <strong>NEW: Keyframe index block lengths</strong>
+│  └─ Run lengths, video references, and starting frame indices
+└─ Footer: five 4-byte index sizes, 4-byte magic, and 1-byte version
+</pre>
+
+The length index has one entry per video; zero selects the scan fallback. Metadata ranges let a
+cold reader fetch initialization data such as MP4 `moov` without probing the video. Keyframe entries
+locate the target GOP range and exact seek timestamp; readers may extend the range through the next
+GOP for reordered frames. All byte offsets are relative to the encoded-video payload. Writers
+validate and copy a supplied index without decoding an unindexed video. Stream identity and PTS
+time base remain in the encoded video. See the
+[file format specification](../concepts/spec/fileformat#video) for the binary layout.
+Writers reject an index larger than 16 MiB or more than 64 MiB of indexes in one file.
 
 ## Reuse, Rolling, and Compaction
 
@@ -102,9 +123,10 @@ together. The resulting file group remains row-aligned and a single large episod
 configured target.
 
 When BLOB compaction is enabled, it byte-copies the complete encoded-video ranges into a new self-contained `.video` pack
-and rebuilds the embedded indexes; it does not decode or re-encode frames. The aligned normal data
+and rebuilds the embedded indexes; it does not decode or re-encode frames. Persisted keyframe indexes
+are copied into the new pack. The aligned normal data
 file contains only application columns such as `episode_id`, state, and action. Paimon stores the
-frame mapping in the `.video` descriptor/index path.
+keyframe index in the `.video` descriptor/index path.
 `blob-compaction.enabled` defaults to `false`; ordinary normal-file compaction can
 leave the video packs unchanged. See [Data Evolution Maintenance](./data-evolution-maintenance#ordinary-compaction).
 
@@ -115,8 +137,7 @@ leave the video packs unchanged. See [Data Evolution Maintenance](./data-evoluti
   `.blob`.
 - Non-null writes must be exact descriptor-backed `BlobRef` values containing a
   `VideoFrameDescriptor`. Inline bytes and ordinary `BlobDescriptor` values are rejected.
-- Version 1 addresses frames by zero-based presentation-order ordinal with stride one. It does not
-  store PTS values or parse codec/container metadata.
+- Version 1 remains readable but has no persisted keyframe index.
 - A `BlobConsumer` callback is not supported for the video field.
 
 ## Read Frames

paimon-common/src/main/java/org/apache/paimon/data/VideoFrameDescriptor.java

@@ -35,17 +35,36 @@ public class VideoFrameDescriptor extends BlobDescriptor {
 
     private static final long serialVersionUID = 1L;
     private static final long MAGIC = 0x564944454F46524DL; // "VIDEOFRM"
-    private static final byte CURRENT_VERSION = 1;
-    private static final int FIXED_LENGTH =
+    private static final byte CURRENT_VERSION = 2;
+    private static final int V1_FIXED_LENGTH =
             Byte.BYTES + Long.BYTES + Integer.BYTES + 3 * Long.BYTES;
+    private static final int V2_FIXED_LENGTH = V1_FIXED_LENGTH + 2 * Long.BYTES;
 
     private final long frameIndex;
-
-    public VideoFrameDescriptor(String uri, long offset, long length, long frameIndex) {
+    private final long keyframeIndexOffset;
+    private final long keyframeIndexLength;
+    private byte version;
+
+    public VideoFrameDescriptor(
+            String uri,
+            long offset,
+            long length,
+            long frameIndex,
+            long keyframeIndexOffset,
+            long keyframeIndexLength) {
         super(uri, offset, length);
         checkArgument(
                 frameIndex >= 0, "Video frame index must be non-negative, but was %s.", frameIndex);
+        checkArgument(
+                keyframeIndexLength >= 0, "Video keyframe index length must be non-negative.");
+        checkArgument(
+                (keyframeIndexLength == 0 && keyframeIndexOffset == -1)
+                        || (keyframeIndexLength > 0 && keyframeIndexOffset >= 0),
+                "Invalid video keyframe index range.");
         this.frameIndex = frameIndex;
+        this.keyframeIndexOffset = keyframeIndexOffset;
+        this.keyframeIndexLength = keyframeIndexLength;
+        this.version = CURRENT_VERSION;
     }
 
     public long frameIndex() {
@@ -57,6 +76,22 @@ public BlobDescriptor payloadDescriptor() {
         return new BlobDescriptor(uri(), offset(), length());
     }
 
+    public @Nullable BlobDescriptor keyframeIndexDescriptor() {
+        return keyframeIndexLength == 0
+                ? null
+                : new BlobDescriptor(uri(), keyframeIndexOffset, keyframeIndexLength);
+    }
+
+    /** Returns the persisted keyframe index carried by an exact frame reference. */
+    public static @Nullable Blob keyframeIndexBlob(Blob blob) {
+        VideoFrameDescriptor frame = fromBlob(blob);
+        BlobDescriptor mapping = frame == null ? null : frame.keyframeIndexDescriptor();
+        if (mapping == null) {
+            return null;
+        }
+        return Blob.fromDescriptor(((BlobRef) blob).uriReader(), mapping);
+    }
+
     /** Returns the video frame carried by an exact lazy blob reference, or {@code null}. */
     public static @Nullable VideoFrameDescriptor fromBlob(@Nullable Blob blob) {
         if (blob == null || blob.getClass() != BlobRef.class) {
@@ -77,30 +112,35 @@ public BlobDescriptor payloadDescriptor() {
     @Override
     public byte[] serialize() {
         byte[] uriBytes = uri().getBytes(StandardCharsets.UTF_8);
+        int fixedLength = version == 1 ? V1_FIXED_LENGTH : V2_FIXED_LENGTH;
         ByteBuffer buffer =
-                ByteBuffer.allocate(FIXED_LENGTH + uriBytes.length).order(ByteOrder.LITTLE_ENDIAN);
-        buffer.put(CURRENT_VERSION);
+                ByteBuffer.allocate(fixedLength + uriBytes.length).order(ByteOrder.LITTLE_ENDIAN);
+        buffer.put(version);
         buffer.putLong(MAGIC);
         buffer.putInt(uriBytes.length);
         buffer.put(uriBytes);
         buffer.putLong(offset());
         buffer.putLong(length());
         buffer.putLong(frameIndex);
+        if (version >= 2) {
+            buffer.putLong(keyframeIndexOffset);
+            buffer.putLong(keyframeIndexLength);
+        }
         return buffer.array();
     }
 
     public static VideoFrameDescriptor deserialize(byte[] bytes) {
-        if (bytes == null || bytes.length < FIXED_LENGTH) {
+        if (bytes == null || bytes.length < V1_FIXED_LENGTH) {
             throw invalidPayload("too short");
         }
 
         ByteBuffer buffer = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN);
         byte version = buffer.get();
-        if (version != CURRENT_VERSION) {
+        if (version < 1 || version > CURRENT_VERSION) {
             throw new UnsupportedOperationException(
-                    "Expecting VideoFrameDescriptor version to be "
+                    "Expecting VideoFrameDescriptor version in [1, "
                             + CURRENT_VERSION
-                            + ", but found "
+                            + "], but found "
                             + version
                             + ".");
         }
@@ -117,7 +157,8 @@ public static VideoFrameDescriptor deserialize(byte[] bytes) {
         if (uriLength > buffer.remaining()) {
             throw invalidPayload("URI length exceeds data size");
         }
-        if (buffer.remaining() - uriLength < 3 * Long.BYTES) {
+        int trailingLength = version == 1 ? 3 * Long.BYTES : 5 * Long.BYTES;
+        if (buffer.remaining() - uriLength < trailingLength) {
             throw invalidPayload("missing offset/length/frame index");
         }
 
@@ -127,21 +168,32 @@ public static VideoFrameDescriptor deserialize(byte[] bytes) {
         long offset = buffer.getLong();
         long length = buffer.getLong();
         long frameIndex = buffer.getLong();
+        long keyframeIndexOffset = -1;
+        long keyframeIndexLength = 0;
+        if (version >= 2) {
+            keyframeIndexOffset = buffer.getLong();
+            keyframeIndexLength = buffer.getLong();
+        }
         if (buffer.hasRemaining()) {
             throw invalidPayload("trailing bytes");
         }
         if (frameIndex < 0) {
             throw invalidPayload("negative frame index: " + frameIndex);
         }
-        return new VideoFrameDescriptor(uri, offset, length, frameIndex);
+        VideoFrameDescriptor descriptor =
+                new VideoFrameDescriptor(
+                        uri, offset, length, frameIndex, keyframeIndexOffset, keyframeIndexLength);
+        descriptor.version = version;
+        return descriptor;
     }
 
     public static boolean isVideoFrameDescriptor(byte[] bytes) {
         if (bytes == null || bytes.length < Byte.BYTES + Long.BYTES) {
             return false;
         }
         ByteBuffer buffer = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN);
-        return buffer.get() == CURRENT_VERSION && buffer.getLong() == MAGIC;
+        byte version = buffer.get();
+        return version >= 1 && version <= CURRENT_VERSION && buffer.getLong() == MAGIC;
     }
 
     @Override
@@ -154,12 +206,13 @@ public boolean equals(Object o) {
         }
         VideoFrameDescriptor that = (VideoFrameDescriptor) o;
         return frameIndex == that.frameIndex
-                && payloadDescriptor().equals(that.payloadDescriptor());
+                && payloadDescriptor().equals(that.payloadDescriptor())
+                && Objects.equals(keyframeIndexDescriptor(), that.keyframeIndexDescriptor());
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(payloadDescriptor(), frameIndex);
+        return Objects.hash(payloadDescriptor(), frameIndex, keyframeIndexDescriptor());
     }
 
     @Override