[format][python] Compress video index pages and reduce seek I/O

docs/docs/concepts/spec/fileformat.md

@@ -441,25 +441,46 @@ Run arrays have equal lengths. Non-negative references select a video; `-1` mean
 means a data-evolution placeholder. Row `r` in a run starting at `s` maps to frame
 `run_first_frame + r - s`; gaps start new runs.
 
-Keyframe-index v1 is little endian; byte offsets are relative to the video payload.
+Each physical video has its own keyframe index; v1 and v2 may coexist in one `.video`
+file. Fields are little endian; metadata and packet offsets refer to the video payload.
+
+Released v1 (Paimon 2.1.0) stores `uint8 version=1`, `uint64 magic=0x564944454F4B4649`,
+`uint32 metadata count`, and `uint32 keyframe count`, followed by `(offset, length)`
+int64 pairs and one zlib stream of `(ordinal, PTS, packet position)` int64 triples.
+It permits at most 65,536 keyframes per physical video.
+
+Keyframe-index v2 adds a page directory and independently encoded pages:
 
 | Field | Size |
 | --- | ---: |
-| Version (`1`) | 1 byte |
+| Version (`2`) | 1 byte |
 | Magic (`0x564944454F4B4649`) | 8 bytes |
 | Metadata-range count | 4 bytes |
 | Keyframe count | 8 bytes |
+| Page codec (`0`: raw; `1`: column-delta zlib) | 1 byte |
 | Metadata `(offset, length)` pairs | 16 bytes each |
+| Page directory | 68 bytes per page |
 | Keyframe pages | Variable |
 
-Each page holds 4,096 uncompressed `(frame ordinal, PTS, packet position)` int64 triples
-and their 4-byte CRC32; the last page may be shorter. Page `p` starts at
-`21 + 16 * metadata_count + p * 98308`.
-Ordinals, PTS values and packet positions strictly increase; the first ordinal is zero.
-The index covers the first video stream; its time base remains in the video.
-
-Readers validate each accessed page and its adjacent page boundaries; writers validate all pages.
-Metadata ranges are limited to 65,536; keyframe count and total index size have no configured limit.
+Each page except the last holds exactly 4,096 `(frame ordinal, PTS, packet position)` int64 triples.
+Each directory record stores the index-relative page offset (8 bytes), encoded length (4),
+uncompressed CRC32 (4), first and last triples (48), and CRC32 of the preceding 64 bytes (4).
+Pages follow the directory contiguously and fill the remainder of the index.
+
+- Codec `0`: uncompressed row-major triples.
+- Codec `1`: independently zlib-compressed pages of three columns (ordinal, PTS, position).
+  Each column stores its first value and successive deltas as little-endian uint64 words,
+  using arithmetic modulo 2^64. Decode by cumulative addition modulo 2^64, then restore
+  signed int64 row-major triples for CRC32 and ordering checks.
+
+Decoded pages must contain exactly `entry_count * 24` bytes; encoded pages cannot exceed
+`4,096 * 24 + 64` bytes. Trailing or concatenated zlib streams are invalid.
+
+Ordinals, PTS and packet positions strictly increase from ordinal zero. The index covers the
+first video stream, using its time base. Metadata ranges are limited to 65,536.
+New writes use v2; copying or compaction preserves existing index bytes. Readers seek
+through either version. Unknown versions/codecs fall back to scanning and survive rewrite;
+corrupt indexes fail validation.
 
 An Arrow/data-file cell stores a separately versioned, little-endian `VideoFrameDescriptor`:
 

docs/docs/pypaimon/video.md

@@ -157,8 +157,12 @@ frames.add_video(video, episode_43_rows)
 ```
 
 PyAV indexes supported ISO BMFF videos (e.g. MP4); other decodable videos fall back
-to scanning. Index generation uses temporary disk; readers cache up to 1 MiB of
-entry pages per index and 4,096 ordinal lookups per search view.
+to scanning. Index generation uses temporary disk.
+
+New keyframe indexes use v2. Existing Paimon 2.1.0 v1 indexes remain readable through
+the index and keep their version when copied or compacted. `video.index.compression`
+selects `delta-zlib` (default) or `none` for newly generated v2 indexes. Readers
+auto-detect the index version and encoding. Compression does not change video bytes.
 
 The writer deduplicates exact payload descriptor identity inside each `.video`
 file. Its video grouping policy coordinates normal, BLOB, and vector rolling

paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java

@@ -1819,9 +1819,7 @@ public void testVideoRollingByBlobTargetSize() throws Exception {
         byte[] secondBytes = "second-video".getBytes();
         byte[] keyframeIndex =
                 fromHex(
-                        "0149464b4f4544495601000000020000000000000000000000000000000100000000000000000000"
-                                + "0000000000000000000000000000000000000000000c00000000000000a08c000000000000020000"
-                                + "0000000000dfd9164b");
+                        "0249464b4f4544495601000000020000000000000000000000000000000001000000000000006a0000000000000030000000dfd9164b0000000000000000000000000000000000000000000000000c00000000000000a08c000000000000020000000000000023e343cf0000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000");
         java.nio.file.Path firstSource = tempPath.resolve("size-first-source.mp4");
         java.nio.file.Path secondSource = tempPath.resolve("size-second-source.mp4");
         java.nio.file.Files.write(firstSource, concat("first-video".getBytes(), keyframeIndex));

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java

@@ -1263,7 +1263,8 @@ public void testUpdateRawBlobColumnSplitsFilesAndKeepsUnchangedPayload() throws
                         Collections.emptyList(),
                         Collections.emptyList(),
                         Collections.emptyMap()));
-        insertInto("BLOB_SPLIT_UPDATE_S", "(1, X'02')");
+        // Two updated payloads force rolling regardless of the initial row-ID order.
+        insertInto("BLOB_SPLIT_UPDATE_S", "(1, X'02')", "(3, X'04')");
 
         builder(warehouse, database, "BLOB_SPLIT_UPDATE_T")
                 .withMergeCondition("BLOB_SPLIT_UPDATE_T.id=BLOB_SPLIT_UPDATE_S.id")
@@ -1278,7 +1279,7 @@ public void testUpdateRawBlobColumnSplitsFilesAndKeepsUnchangedPayload() throws
                 Arrays.asList(
                         changelogRow("+I", 1, new byte[] {2}),
                         changelogRow("+I", 2, large),
-                        changelogRow("+I", 3, new byte[] {3})));
+                        changelogRow("+I", 3, new byte[] {4})));
 
         List<Row> newBlobFiles = new ArrayList<>();
         boolean sawOriginalLargeFile = false;

paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java

@@ -28,10 +28,14 @@
 import org.apache.paimon.format.FormatWriter;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.DeltaVarintCompressor;
 import org.apache.paimon.utils.LongArrayList;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.RandomAccessFile;
@@ -52,6 +56,8 @@
  */
 public class VideoFormatWriter implements FileAwareFormatWriter {
 
+    private static final Logger LOG = LoggerFactory.getLogger(VideoFormatWriter.class);
+
     public static final byte VERSION = 1;
     public static final int MAGIC_NUMBER = 0x4F454449; // "IDEO" in little endian
     public static final long NULL_REFERENCE = -1L;
@@ -333,20 +339,38 @@ private boolean readIndex(
                     }
                 }
                 spool.seek(start);
-                VideoKeyframeIndex.validate(
-                        new InputStream() {
-                            @Override
-                            public int read() throws IOException {
-                                return spool.read();
-                            }
+                try {
+                    VideoKeyframeIndex.validate(
+                            new SeekableInputStream() {
+                                @Override
+                                public void seek(long position) throws IOException {
+                                    spool.seek(position);
+                                }
 
-                            @Override
-                            public int read(byte[] b, int offset, int length) throws IOException {
-                                return spool.read(b, offset, length);
-                            }
-                        },
-                        expectedLength,
-                        payloadLength);
+                                @Override
+                                public long getPos() throws IOException {
+                                    return spool.getFilePointer();
+                                }
+
+                                @Override
+                                public void close() {}
+
+                                @Override
+                                public int read() throws IOException {
+                                    return spool.read();
+                                }
+
+                                @Override
+                                public int read(byte[] b, int offset, int length)
+                                        throws IOException {
+                                    return spool.read(b, offset, length);
+                                }
+                            },
+                            expectedLength,
+                            payloadLength);
+                } catch (VideoKeyframeIndex.UnsupportedIndexException e) {
+                    LOG.warn("Preserving unsupported video index: {}", e.getMessage());
+                }
                 spool.seek(start + expectedLength);
                 recordSuccess(expectedLength);
                 complete = true;