From 0d519ceff997e3f2633b03634027244f55957a72 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 01:46:44 -0700 Subject: [PATCH 01/10] [format][python] Keep large video keyframe indexes compact --- docs/docs/concepts/spec/fileformat.md | 3 +- .../format/blob/VideoKeyframeIndex.java | 3 +- .../format/blob/VideoFileFormatTest.java | 28 ++++ .../python/write/video_format_writer.py | 9 +- .../pypaimon/multimodal/lerobot/dataset.py | 3 +- .../table/row/video_keyframe_index.py | 145 +++++++++++------- .../pypaimon/tests/video_format_test.py | 18 ++- .../tests/video_keyframe_index_test.py | 72 +++++++-- 8 files changed, 203 insertions(+), 78 deletions(-) diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 18c1bb59d969..97828172a732 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -445,7 +445,8 @@ A keyframe-index block contains a 17-byte header (version `1`: uint8; magic `0x5 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 +and 16 MiB of decoded keyframe entries (699,050 entries), 16 MiB per stored block, +64 MiB per file. Older readers limited to 65,536 entries cannot read larger indexes. The index covers the first video stream; its time base stays in the video. An Arrow/data-file cell stores a separately versioned, little-endian `VideoFrameDescriptor`: 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..1004170a39c5 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 @@ -32,7 +32,8 @@ final class VideoKeyframeIndex { private static final int METADATA_RANGE_LENGTH = Long.BYTES * 2; private static final int ENTRY_LENGTH = Long.BYTES * 3; static final long MAX_METADATA_RANGE_COUNT = 64 * 1024; - static final long MAX_KEYFRAME_COUNT = 64 * 1024; + static final long MAX_DECODED_KEYFRAME_BYTES = 16 * 1024 * 1024; + static final long MAX_KEYFRAME_COUNT = MAX_DECODED_KEYFRAME_BYTES / ENTRY_LENGTH; private VideoKeyframeIndex() {} 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..b00e9bb34cd0 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; @@ -55,6 +56,7 @@ import java.nio.file.Files; import java.util.ArrayList; import java.util.List; +import java.util.zip.DeflaterOutputStream; import static org.apache.paimon.utils.StreamUtils.intToLittleEndian; import static org.assertj.core.api.Assertions.assertThat; @@ -217,6 +219,32 @@ public void testRejectInvalidKeyframeIndex() throws IOException { .hasMessageContaining("Invalid video keyframe index"); } + @Test + public void testKeyframeIndexAboveOldLimit() throws IOException { + int count = 65537; + ByteArrayOutputStream compressed = new ByteArrayOutputStream(); + try (DeflaterOutputStream output = new DeflaterOutputStream(compressed)) { + ByteBuffer entry = ByteBuffer.allocate(24).order(ByteOrder.LITTLE_ENDIAN); + for (int i = 0; i < count; i++) { + entry.clear(); + entry.putLong(i).putLong(i).putLong(i); + output.write(entry.array()); + } + } + byte[] mapping = + ByteBuffer.allocate(17 + compressed.size()) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) + .putLong(0x564944454F4B4649L) + .putInt(0) + .putInt(count) + .put(compressed.toByteArray()) + .array(); + VideoKeyframeIndex.validate(mapping, count); + assertThat(VideoKeyframeIndex.MAX_KEYFRAME_COUNT * 24) + .isLessThanOrEqualTo(VideoKeyframeIndex.MAX_DECODED_KEYFRAME_BYTES); + } + @Test public void testRejectTooManyKeyframes() { byte[] mapping = 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..2771402c3506 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,7 @@ # specific language governing permissions and limitations # under the License. +import logging import struct import sys @@ -25,7 +26,7 @@ BlobRef, VideoFrameDescriptor, ) -from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex +from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex, VideoKeyframeIndexLimitError from pypaimon.execution.python.write.blob_format_writer import BlobFormatWriter @@ -185,7 +186,11 @@ def _keyframe_index(self, blob, frame): stream.seek(position) mapping = VideoKeyframeIndex.inspect( stream, payload_length).serialize() - except _video_index_fallback_errors(): + except _video_index_fallback_errors() as error: + if isinstance(error, VideoKeyframeIndexLimitError): + logging.getLogger(__name__).warning( + "Video seek index exceeds its memory budget; storing video without " + "a seek index. Random reads may require scanning: %s", error) return b'' finally: stream.close() diff --git a/paimon-python/pypaimon/multimodal/lerobot/dataset.py b/paimon-python/pypaimon/multimodal/lerobot/dataset.py index e1f55123600e..90eb8e25687d 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -26,6 +26,7 @@ import sys import zlib from abc import ABC, abstractmethod +from array import array from collections import OrderedDict from collections.abc import Mapping from concurrent.futures import ThreadPoolExecutor @@ -2006,7 +2007,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] + array("q", (entry[0] for entry in keyframe_index.keyframes)) if keyframe_index is not None else []) self._container = None self._stream = None diff --git a/paimon-python/pypaimon/table/row/video_keyframe_index.py b/paimon-python/pypaimon/table/row/video_keyframe_index.py index acabf9ce4acb..8dd5fc488b47 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -20,6 +20,44 @@ import operator import struct import zlib +from array import array +from collections.abc import Sequence + + +class _PackedKeyframes(Sequence): + """Immutable int64 triples, unpacked only as individual entries are read.""" + + ENTRY = struct.Struct('= len(self): + raise IndexError("Keyframe index out of range.") + return self.ENTRY.unpack_from(self.data, index * self.ENTRY.size) + + def __iter__(self): + return self.ENTRY.iter_unpack(self.data) + + def __eq__(self, other): + if isinstance(other, _PackedKeyframes): + return self.data == other.data + if not isinstance(other, Sequence): + return NotImplemented + return len(self) == len(other) and all(a == b for a, b in zip(self, other)) + + +class VideoKeyframeIndexLimitError(ValueError): + """A video index exceeds a bounded resource budget.""" class VideoKeyframeIndex: @@ -31,7 +69,8 @@ class VideoKeyframeIndex: METADATA_RANGE = struct.Struct('= 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.keyframes = keyframes previous_end = 0 for offset, length in self.metadata_ranges: if offset < previous_end or length <= 0: @@ -92,18 +120,21 @@ def __init__(self, metadata_ranges, keyframes): @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 + data = bytearray() + for entry in entries: + if len(data) // self.ENTRY.size >= self.MAX_KEYFRAME_COUNT: + raise VideoKeyframeIndexLimitError( + "Video keyframe index exceeds the %s-entry limit." + % self.MAX_KEYFRAME_COUNT) + data.extend(self.ENTRY.pack(*(operator.index(value) for value in entry))) + self._keyframes = _PackedKeyframes(bytes(data)) def serialize(self): - entries = b''.join(self.ENTRY.pack(*entry) for entry in self.keyframes) + entries = self._keyframes.data return self.HEADER.pack( self.VERSION, self.MAGIC, @@ -124,6 +155,8 @@ def inspect(cls, source, payload_length): "Video seek-index generation requires PyAV." ) from error + import numpy as np + metadata_ranges = cls._iso_bmff_metadata_ranges( source, payload_length) source.seek(0) @@ -131,7 +164,7 @@ def inspect(cls, source, payload_length): if not container.streams.video: raise ValueError("Video has no video stream.") stream = container.streams.video[0] - packet_positions = {} + packet_data = array("q") 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 @@ -142,46 +175,41 @@ def inspect(cls, source, payload_length): 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( + if len(packet_data) // 2 >= cls.MAX_KEYFRAME_COUNT: + raise VideoKeyframeIndexLimitError( "Video keyframe index exceeds the %s-entry limit." % cls.MAX_KEYFRAME_COUNT ) - packet_positions[pts] = int(packet.pos) + packet_data.extend((pts, int(packet.pos))) + + packets = np.asarray(packet_data, dtype=np.int64).reshape(-1, 2) + order = np.argsort(packets[:, 0]) + packet_pts = packets[order, 0] + packet_positions = packets[order, 1] + del packet_data, packets, order + if np.any(packet_pts[1:] == packet_pts[:-1]): + raise ValueError("Video has duplicate keyframe timestamps.") 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) + + 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 = np.searchsorted(packet_pts, pts) + if match == len(packet_pts) or packet_pts[match] != pts: + raise ValueError("Video keyframe has no packet byte position.") + yield ordinal, pts, int(packet_positions[match]) + previous_pts = pts + return cls(metadata_ranges, entries()) @staticmethod def _stream_pts(pts, time_base, stream_time_base): @@ -259,7 +287,7 @@ def deserialize(cls, data, payload_length=None): else: metadata_ranges.append((start, length)) - # The header caps this buffer at 1.5 MiB. One extra byte detects + # The header caps this buffer at MAX_DECODED_KEYFRAME_BYTES. One extra byte detects # excess output without expanding an untrusted compressed stream. expected_size = keyframe_count * cls.ENTRY.size decoder = zlib.decompressobj() @@ -292,11 +320,10 @@ def deserialize(cls, data, payload_length=None): "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. + # Keep entries compact even after a decoder starts consuming them. result = cls.__new__(cls) result.metadata_ranges = tuple(metadata_ranges) - result._keyframes = raw + result._keyframes = _PackedKeyframes(raw) return result @classmethod @@ -317,7 +344,7 @@ def _read_header(cls, data): % cls.MAX_METADATA_RANGE_COUNT ) if keyframe_count > cls.MAX_KEYFRAME_COUNT: - raise ValueError( + raise VideoKeyframeIndexLimitError( "Video keyframe index exceeds the %s-entry limit." % cls.MAX_KEYFRAME_COUNT ) diff --git a/paimon-python/pypaimon/tests/video_format_test.py b/paimon-python/pypaimon/tests/video_format_test.py index bd17f887bef2..f5864d5ad9f8 100644 --- a/paimon-python/pypaimon/tests/video_format_test.py +++ b/paimon-python/pypaimon/tests/video_format_test.py @@ -405,6 +405,22 @@ def test_ffmpeg_index_errors_preserve_video_payload(self): self.assertEqual(payload, output.getvalue()) self.assertEqual([b""], writer._keyframe_indexes) + def test_index_memory_limit_warns_and_preserves_payload(self): + from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndexLimitError + payload = b"video" + 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) + output = io.BytesIO() + writer = VideoFormatWriter(output) + with mock.patch.object(VideoKeyframeIndex, "inspect", side_effect=VideoKeyframeIndexLimitError("limit")): + with self.assertLogs("pypaimon.execution.python.write.video_format_writer", level="WARNING") as logs: + writer.add_element(GenericRow([blob], [self.field], RowKind.INSERT)) + self.assertIn("without a seek index", logs.output[0]) + self.assertEqual(payload, output.getvalue()) + self.assertEqual([b""], writer._keyframe_indexes) + def test_rejects_invalid_keyframe_index(self): video = b"video" mapping = b"mapping" @@ -501,7 +517,7 @@ def test_rejects_compressed_invalid_index_without_expanding_it(self): try: with self.assertRaisesRegex( ValueError, - "entry limit"): + "ordinals must be strictly increasing"): writer.add_element( GenericRow([blob], [self.field], RowKind.INSERT) ) diff --git a/paimon-python/pypaimon/tests/video_keyframe_index_test.py b/paimon-python/pypaimon/tests/video_keyframe_index_test.py index 4a5c6a3206d8..8eec9cc213a2 100644 --- a/paimon-python/pypaimon/tests/video_keyframe_index_test.py +++ b/paimon-python/pypaimon/tests/video_keyframe_index_test.py @@ -31,22 +31,68 @@ class VideoKeyframeIndexTest(unittest.TestCase): - def test_deserialize_materializes_keyframes_only_when_used(self): - index = VideoKeyframeIndex([(0, 10)], [(0, -100, 20), (12, 120, 200)]) + def test_keyframes_stay_compact_when_read_and_pickled(self): + values = ((0, -100, 20), (12, 120, 200)) + index = VideoKeyframeIndex([(0, 10)], values) data = index.serialize() - entry = VideoKeyframeIndex.ENTRY - with mock.patch.object(VideoKeyframeIndex, 'ENTRY', mock.Mock( - size=entry.size, iter_unpack=mock.Mock(wraps=entry.iter_unpack))): - restored = VideoKeyframeIndex.deserialize(data, 1000) - self.assertEqual(index.metadata_ranges, restored.metadata_ranges) - restored = pickle.loads(pickle.dumps(restored)) - VideoKeyframeIndex.ENTRY.iter_unpack.assert_not_called() - - self.assertEqual(index.keyframes, restored.keyframes) - self.assertIs(restored.keyframes, restored.keyframes) - VideoKeyframeIndex.ENTRY.iter_unpack.assert_called_once() + restored = pickle.loads(pickle.dumps(VideoKeyframeIndex.deserialize(data, 1000))) + packed = restored.keyframes.data + self.assertEqual(values, restored.keyframes) + self.assertEqual(values[-1], restored.keyframes[-1]) + self.assertEqual(values[::-1], restored.keyframes[::-1]) + self.assertEqual(list(values), list(restored.keyframes)) + self.assertIs(packed, restored.keyframes.data) + self.assertEqual(2 * VideoKeyframeIndex.ENTRY.size, len(packed)) + with self.assertRaises(IndexError): + _ = restored.keyframes[2] + with self.assertRaises(IndexError): + _ = restored.keyframes[-3] self.assertEqual(data, restored.serialize()) + def test_real_video_above_old_keyframe_limit(self): + import tempfile + from pathlib import Path + import pytest + av = pytest.importorskip("av") + np = pytest.importorskip("numpy") + pytest.importorskip("torch") + from pypaimon.multimodal.lerobot.dataset import _PyAVVideoDecoder + + count = 65537 + with tempfile.TemporaryDirectory() as directory: + path = str(Path(directory) / "all-keyframes.mp4") + with av.open(path, "w") as output: + stream = output.add_stream("libx264", rate=30, options={"preset": "ultrafast", "crf": "0"}) + stream.width = stream.height = 16 + stream.pix_fmt = "yuv420p" + stream.gop_size = 1 + stream.codec_context.max_b_frames = 0 + for ordinal in range(count): + frame = av.VideoFrame.from_ndarray( + np.full((16, 16, 3), ordinal % 200, dtype=np.uint8), format="rgb24") + frame.pts = ordinal + for packet in stream.encode(frame): + output.mux(packet) + for packet in stream.encode(): + output.mux(packet) + with open(path, "rb") as source: + index = VideoKeyframeIndex.inspect(source, Path(path).stat().st_size) + self.assertEqual(count, len(index.keyframes)) + restored = VideoKeyframeIndex.deserialize(index.serialize()) + ordinals = [count - 1, 0, 32768] + with av.open(path) as container: + expected = {i: frame.to_ndarray(format="rgb24") + for i, frame in enumerate(container.decode(video=0)) if i in ordinals} + with open(path, "rb") as source: + decoder = _PyAVVideoDecoder(source, keyframe_index=restored, thread_count=1) + try: + actual = decoder.get_frames_at(indices=ordinals).data.numpy() + for ordinal, pixels in zip(ordinals, actual): + np.testing.assert_array_equal(expected[ordinal], pixels.transpose(1, 2, 0)) + finally: + decoder.close() + self.assertEqual(count * 24, len(restored.keyframes.data)) + def test_replace_unmaterialized_keyframes(self): index = VideoKeyframeIndex([(0, 10)], [(0, 0, 20), (12, 120, 200)]) restored = VideoKeyframeIndex.deserialize(index.serialize(), 1000) From 815dc7cd64e0697917d4d6376ed6ca67bc072b9e Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 08:16:52 -0700 Subject: [PATCH 02/10] [format][python] Page video indexes instead of limiting keyframe counts --- docs/docs/concepts/spec/fileformat.md | 32 +- docs/docs/pypaimon/video.md | 4 +- .../paimon/format/blob/VideoFileMeta.java | 6 - .../paimon/format/blob/VideoFormatWriter.java | 186 ++++-- .../format/blob/VideoKeyframeIndex.java | 190 +++--- .../format/blob/VideoFileFormatTest.java | 134 +++-- .../paimon/format/blob/video-paged-index.hex | 1 + .../python/read/reader/video_format_reader.py | 4 - .../python/write/video_format_writer.py | 167 +++--- .../pypaimon/multimodal/lerobot/dataset.py | 3 +- paimon-python/pypaimon/multimodal/video.py | 18 +- .../table/row/video_keyframe_index.py | 547 ++++++++++-------- .../pypaimon/tests/video_format_test.py | 140 +++-- .../tests/video_keyframe_index_test.py | 117 ++-- 14 files changed, 889 insertions(+), 660 deletions(-) create mode 100644 paimon-format/src/test/resources/org/apache/paimon/format/blob/video-paged-index.hex diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 97828172a732..73698b129081 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,13 +441,29 @@ 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 16 MiB of decoded keyframe entries (699,050 entries), 16 MiB per stored block, -64 MiB per file. Older readers limited to 65,536 entries cannot read larger indexes. The index covers the first video -stream; its time base stays in the video. +Keyframe-index version `2` uses this layout. Numeric fields are little endian; +byte offsets are relative to the embedded video payload. + +| Field | Size | +| --- | ---: | +| Version (`2`) | 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. There is no keyframe-count +or aggregate index-byte budget. Metadata ranges remain limited to 65,536. PyPaimon +caches up to 1 MiB of entry pages per index and spools generated indexes to temporary disk. +Version `1` remains readable: its 17-byte header has two uint32 counts and its entries +form one zlib stream. Version-1-only readers must be upgraded before reading version `2`. 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..ad8cd605f539 100644 --- a/docs/docs/pypaimon/video.md +++ b/docs/docs/pypaimon/video.md @@ -116,7 +116,9 @@ 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. +as MP4); other decodable videos use the scan fallback. Indexes are paged, so long +videos retain seek indexes without loading all keyframes into memory. Generation +uses temporary disk space; readers cache up to 1 MiB of entry pages per video. 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-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 1004170a39c5..9397f64e6905 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,102 +18,130 @@ 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.CRC32; import java.util.zip.DataFormatException; import java.util.zip.Inflater; -/** Validates persisted video metadata ranges and sparse keyframe entries. */ +/** Streaming validation for legacy and 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_DECODED_KEYFRAME_BYTES = 16 * 1024 * 1024; - static final long MAX_KEYFRAME_COUNT = MAX_DECODED_KEYFRAME_BYTES / ENTRY_LENGTH; 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 < 17 || 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 && version != 2) || magic != MAGIC) { + throw invalid("version or magic"); } - if (metadataRangeCount > MAX_METADATA_RANGE_COUNT) { + long count = + version == 1 + ? Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())) + : Long.reverseBytes(in.readLong()); + long offset = (version == 1 ? 17 : 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"); - } - - long entriesOffset = HEADER_LENGTH + metadataRangeCount * METADATA_RANGE_LENGTH; - if (entriesOffset >= data.length) { - throw invalid("metadata ranges"); + if (count <= 0 || offset >= length) { + throw invalid("count or 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; } + EntryValidator validator = new EntryValidator(payloadLength); + if (version == 2) { + long pages = (count - 1) / PAGE_ENTRIES + 1; + if (count > (length - offset) / ENTRY_LENGTH + || pages * 4 != length - offset - count * ENTRY_LENGTH) { + throw invalid("entries length"); + } + 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++) { + validator.accept(values); + } + done += entries; + } + } else { + validateLegacy(in, length - offset, count, validator); + } + } + private static void validateLegacy( + InputStream in, long remaining, long count, EntryValidator validator) + throws IOException { Inflater inflater = new Inflater(); - inflater.setInput(data, (int) entriesOffset, data.length - (int) entriesOffset); + byte[] input = new byte[64 * 1024]; byte[] entry = new byte[ENTRY_LENGTH]; - long previousOrdinal = -1; - long previousPts = 0; - long previousPacketPosition = -1; + int filled = 0; + long decoded = 0; 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"); + while (!inflater.finished()) { + if (inflater.needsInput() && remaining > 0) { + int n = in.read(input, 0, (int) Math.min(input.length, remaining)); + if (n <= 0) { + throw invalid("truncated compressed entries"); + } + remaining -= n; + inflater.setInput(input, 0, n); } - if (i > 0 && pts <= previousPts) { - throw invalid("keyframe timestamps must be strictly increasing"); + int n = inflater.inflate(entry, filled, ENTRY_LENGTH - filled); + filled += n; + if (filled == ENTRY_LENGTH) { + if (decoded++ >= count) { + throw invalid("entry count"); + } + validator.accept(ByteBuffer.wrap(entry).order(ByteOrder.LITTLE_ENDIAN)); + filled = 0; } - if (packetPosition <= previousPacketPosition || packetPosition >= payloadLength) { - throw invalid( - "keyframe packet positions must be within the video payload and strictly increasing"); + if (n == 0 + && !inflater.finished() + && (inflater.needsDictionary() + || !inflater.needsInput() + || remaining == 0)) { + throw invalid("compressed entries"); } - previousOrdinal = ordinal; - previousPts = pts; - previousPacketPosition = packetPosition; } - - if (inflater.inflate(entry, 0, 1) != 0 - || !inflater.finished() - || inflater.getRemaining() != 0) { - throw invalid("entry count"); + if (filled != 0 || decoded != count || remaining != 0 || inflater.getRemaining() != 0) { + throw invalid("entry count or trailing data"); } } catch (DataFormatException e) { throw new IllegalArgumentException("Invalid video keyframe index payload.", e); @@ -122,14 +150,30 @@ static void validate(byte[] data, long payloadLength) { } } - 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"); + private static final class EntryValidator { + private final long payloadLength; + private long ordinal = -1; + private long pts; + private long position = -1; + + private EntryValidator(long payloadLength) { + this.payloadLength = payloadLength; + } + + private void accept(ByteBuffer values) { + 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"); } - offset += length; + ordinal = nextOrdinal; + pts = nextPts; + position = nextPosition; } } 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 b00e9bb34cd0..9a6807ed0a67 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 @@ -132,8 +132,19 @@ public void testPackRawVideosAndMapFrameRuns() throws IOException { @Test public void testCrossLanguageV1Fixture() throws IOException { byte[] fixture = fixture("video-v1.hex"); + checkVideoFixture(fixture, fromHex(KEYFRAME_INDEX_HEX)); + } + + @Test + public void testCrossLanguagePagedIndexFixture() throws IOException { + byte[] fixture = fixture("video-paged-index.hex"); + byte[] mapping = java.util.Arrays.copyOfRange(fixture, 7, 7 + 89); + VideoKeyframeIndex.validate(mapping, 3); + checkVideoFixture(fixture, mapping); + } + + 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); @@ -221,7 +232,7 @@ public void testRejectInvalidKeyframeIndex() throws IOException { @Test public void testKeyframeIndexAboveOldLimit() throws IOException { - int count = 65537; + int count = 700001; ByteArrayOutputStream compressed = new ByteArrayOutputStream(); try (DeflaterOutputStream output = new DeflaterOutputStream(compressed)) { ByteBuffer entry = ByteBuffer.allocate(24).order(ByteOrder.LITTLE_ENDIAN); @@ -241,25 +252,86 @@ public void testKeyframeIndexAboveOldLimit() throws IOException { .put(compressed.toByteArray()) .array(); VideoKeyframeIndex.validate(mapping, count); - assertThat(VideoKeyframeIndex.MAX_KEYFRAME_COUNT * 24) - .isLessThanOrEqualTo(VideoKeyframeIndex.MAX_DECODED_KEYFRAME_BYTES); } @Test - public void testRejectTooManyKeyframes() { + public void testRejectForgedPagedCount() { byte[] mapping = - ByteBuffer.allocate(18) + ByteBuffer.allocate(21) .order(ByteOrder.LITTLE_ENDIAN) - .put((byte) 1) + .put((byte) 2) .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) 2) + .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 @@ -279,33 +351,6 @@ 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); @@ -492,8 +537,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} }; @@ -526,14 +571,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-paged-index.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-paged-index.hex new file mode 100644 index 000000000000..dc9e5f593637 --- /dev/null +++ b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-paged-index.hex @@ -0,0 +1 @@ +6162635758595a0249464b4f45444956010000000200000000000000000000000000000001000000000000000000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000dfd9164b0602b201b101040100020100010106010403000e0602000000040000000500000005000000050000004944454f01 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 2771402c3506..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,7 +15,9 @@ # specific language governing permissions and limitations # under the License. -import logging +import errno +import shutil +import tempfile import struct import sys @@ -26,7 +28,7 @@ BlobRef, VideoFrameDescriptor, ) -from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex, VideoKeyframeIndexLimitError +from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex from pypaimon.execution.python.write.blob_format_writer import BlobFormatWriter @@ -45,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, @@ -61,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 = [] @@ -105,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 @@ -135,91 +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 90eb8e25687d..c13aaf1d1c39 100644 --- a/paimon-python/pypaimon/multimodal/lerobot/dataset.py +++ b/paimon-python/pypaimon/multimodal/lerobot/dataset.py @@ -26,7 +26,6 @@ import sys import zlib from abc import ABC, abstractmethod -from array import array from collections import OrderedDict from collections.abc import Mapping from concurrent.futures import ThreadPoolExecutor @@ -2007,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 = ( - array("q", (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 8dd5fc488b47..4dac2cb0a74c 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -15,25 +15,72 @@ # 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 array import array +from collections import OrderedDict from collections.abc import Sequence +from contextlib import closing -class _PackedKeyframes(Sequence): - """Immutable int64 triples, unpacked only as individual entries are read.""" +class _RangeView: + def __init__(self, read_at, length): + self.read_at, self.length = read_at, length + def __len__(self): + return self.length + + def __getitem__(self, key): + start, end, step = key.indices(self.length) + if step != 1: + raise ValueError("Non-contiguous index read.") + return self.read_at(start, end - start) + + +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): @@ -41,82 +88,87 @@ def __getitem__(self, index): index = operator.index(index) if index < 0: index += len(self) - if index < 0 or index >= len(self): + if not 0 <= index < len(self): raise IndexError("Keyframe index out of range.") - return self.ENTRY.unpack_from(self.data, index * self.ENTRY.size) + page, offset = divmod(index, self.PAGE_ENTRIES) + return self.ENTRY.unpack_from(self._page(page), offset * self.ENTRY.size) def __iter__(self): - return self.ENTRY.iter_unpack(self.data) + 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 isinstance(other, _PackedKeyframes): - return self.data == other.data 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 VideoKeyframeIndexLimitError(ValueError): - """A video index exceeds a bounded resource budget.""" +class _Ordinals(Sequence): + def __init__(self, keyframes): + self.keyframes = keyframes -class VideoKeyframeIndex: - """Persist video metadata ranges and presentation-order 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.") - VERSION = 1 - MAGIC = 0x564944454F4B4649 # "VIDEOKFI" + +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 # Legacy zlib stream. + PAGED_VERSION = 2 + MAGIC = 0x564944454F4B4649 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) + 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): @@ -124,92 +176,198 @@ def keyframes(self): @keyframes.setter def keyframes(self, entries): - data = bytearray() - for entry in entries: - if len(data) // self.ENTRY.size >= self.MAX_KEYFRAME_COUNT: - raise VideoKeyframeIndexLimitError( - "Video keyframe index exceeds the %s-entry limit." - % self.MAX_KEYFRAME_COUNT) - data.extend(self.ENTRY.pack(*(operator.index(value) for value in entry))) - self._keyframes = _PackedKeyframes(bytes(data)) + 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: + header = cls.HEADER + elif version == cls.PAGED_VERSION: + header = cls.PAGED_HEADER + else: + raise ValueError("Invalid video keyframe index version.") + _, 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) + if version == cls.VERSION: + return cls(ranges, cls._iter_keyframes(_RangeView(read, length), offset, count, 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 - - import numpy as np - - 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_data = array("q") - 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 len(packet_data) // 2 >= cls.MAX_KEYFRAME_COUNT: - raise VideoKeyframeIndexLimitError( - "Video keyframe index exceeds the %s-entry limit." - % cls.MAX_KEYFRAME_COUNT - ) - packet_data.extend((pts, int(packet.pos))) - - packets = np.asarray(packet_data, dtype=np.int64).reshape(-1, 2) - order = np.argsort(packets[:, 0]) - packet_pts = packets[order, 0] - packet_positions = packets[order, 1] - del packet_data, packets, order - if np.any(packet_pts[1:] == packet_pts[:-1]): - raise ValueError("Video has duplicate keyframe timestamps.") - - 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 = np.searchsorted(packet_pts, pts) - if match == len(packet_pts) or packet_pts[match] != pts: - raise ValueError("Video keyframe has no packet byte position.") - yield ordinal, pts, int(packet_positions[match]) - previous_pts = pts - return cls(metadata_ranges, entries()) + 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): @@ -262,117 +420,6 @@ def _iso_bmff_metadata_ranges(cls, source, payload_length): 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 MAX_DECODED_KEYFRAME_BYTES. 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." - ) - - # Keep entries compact even after a decoder starts consuming them. - result = cls.__new__(cls) - result.metadata_ranges = tuple(metadata_ranges) - result._keyframes = _PackedKeyframes(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 VideoKeyframeIndexLimitError( - "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: diff --git a/paimon-python/pypaimon/tests/video_format_test.py b/paimon-python/pypaimon/tests/video_format_test.py index f5864d5ad9f8..d228e0c384f0 100644 --- a/paimon-python/pypaimon/tests/video_format_test.py +++ b/paimon-python/pypaimon/tests/video_format_test.py @@ -159,9 +159,18 @@ 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() + encoded = VideoKeyframeIndex.HEADER.pack(1, VideoKeyframeIndex.MAGIC, 1, 2) + encoded += struct.pack(' Date: Fri, 30 Oct 2026 09:19:04 -0700 Subject: [PATCH 03/10] [format][python] Define paged video indexes as version 1 --- docs/docs/concepts/spec/fileformat.md | 6 +- .../apache/paimon/append/BlobTableTest.java | 5 +- .../format/blob/VideoKeyframeIndex.java | 100 ++++-------------- .../format/blob/VideoFileFormatTest.java | 54 ++-------- .../paimon/format/blob/video-paged-index.hex | 1 - .../apache/paimon/format/blob/video-v1.hex | 2 +- .../table/row/video_keyframe_index.py | 97 ++--------------- .../pypaimon/tests/video_format_test.py | 11 +- .../tests/video_keyframe_index_test.py | 89 +++------------- 9 files changed, 65 insertions(+), 300 deletions(-) delete mode 100644 paimon-format/src/test/resources/org/apache/paimon/format/blob/video-paged-index.hex diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 73698b129081..ec85e8c59892 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -441,12 +441,12 @@ 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 version `2` uses this layout. Numeric fields are little endian; +Keyframe-index version `1` uses this layout. Numeric fields are little endian; byte offsets are relative to the embedded video payload. | Field | Size | | --- | ---: | -| Version (`2`) | 1 byte | +| Version (`1`) | 1 byte | | Magic (`0x564944454F4B4649`) | 8 bytes | | Metadata-range count | 4 bytes | | Keyframe count | 8 bytes | @@ -462,8 +462,6 @@ The index covers the first video stream; its time base remains in the video. Readers load pages on demand; writers validate all pages. There is no keyframe-count or aggregate index-byte budget. Metadata ranges remain limited to 65,536. PyPaimon caches up to 1 MiB of entry pages per index and spools generated indexes to temporary disk. -Version `1` remains readable: its 17-byte header has two uint32 counts and its entries -form one zlib stream. Version-1-only readers must be upgraded before reading version `2`. An Arrow/data-file cell stores a separately versioned, little-endian `VideoFrameDescriptor`: 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/VideoKeyframeIndex.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/VideoKeyframeIndex.java index 9397f64e6905..6e495a89f119 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 @@ -25,10 +25,8 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.zip.CRC32; -import java.util.zip.DataFormatException; -import java.util.zip.Inflater; -/** Streaming validation for legacy and paged video seek indexes. */ +/** Streaming validation for paged video seek indexes. */ final class VideoKeyframeIndex { private static final long MAGIC = 0x564944454F4B4649L; private static final int ENTRY_LENGTH = 24; @@ -46,21 +44,18 @@ static void validate(byte[] data, long payloadLength) { } static void validate(InputStream source, long length, long payloadLength) throws IOException { - if (length < 17 || payloadLength <= 0) { + if (length < 21 || payloadLength <= 0) { throw invalid("header or empty video payload"); } DataInputStream in = new DataInputStream(source); int version = in.readUnsignedByte(); long magic = Long.reverseBytes(in.readLong()); long metadataCount = Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())); - if ((version != 1 && version != 2) || magic != MAGIC) { + if (version != 1 || magic != MAGIC) { throw invalid("version or magic"); } - long count = - version == 1 - ? Integer.toUnsignedLong(Integer.reverseBytes(in.readInt())) - : Long.reverseBytes(in.readLong()); - long offset = (version == 1 ? 17 : 21) + metadataCount * 16; + long count = Long.reverseBytes(in.readLong()); + long offset = 21 + metadataCount * 16; if (metadataCount > MAX_METADATA_RANGE_COUNT) { throw invalid("metadata range count exceeds limit"); } @@ -77,76 +72,27 @@ static void validate(InputStream source, long length, long payloadLength) throws previousEnd = start + size; } EntryValidator validator = new EntryValidator(payloadLength); - if (version == 2) { - long pages = (count - 1) / PAGE_ENTRIES + 1; - if (count > (length - offset) / ENTRY_LENGTH - || pages * 4 != length - offset - count * ENTRY_LENGTH) { - throw invalid("entries length"); - } - 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++) { - validator.accept(values); - } - done += entries; - } - } else { - validateLegacy(in, length - offset, count, validator); + 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 validateLegacy( - InputStream in, long remaining, long count, EntryValidator validator) - throws IOException { - Inflater inflater = new Inflater(); - byte[] input = new byte[64 * 1024]; - byte[] entry = new byte[ENTRY_LENGTH]; - int filled = 0; - long decoded = 0; - try { - while (!inflater.finished()) { - if (inflater.needsInput() && remaining > 0) { - int n = in.read(input, 0, (int) Math.min(input.length, remaining)); - if (n <= 0) { - throw invalid("truncated compressed entries"); - } - remaining -= n; - inflater.setInput(input, 0, n); - } - int n = inflater.inflate(entry, filled, ENTRY_LENGTH - filled); - filled += n; - if (filled == ENTRY_LENGTH) { - if (decoded++ >= count) { - throw invalid("entry count"); - } - validator.accept(ByteBuffer.wrap(entry).order(ByteOrder.LITTLE_ENDIAN)); - filled = 0; - } - if (n == 0 - && !inflater.finished() - && (inflater.needsDictionary() - || !inflater.needsInput() - || remaining == 0)) { - throw invalid("compressed entries"); - } + 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"); } - if (filled != 0 || decoded != count || remaining != 0 || inflater.getRemaining() != 0) { - throw invalid("entry count or trailing data"); + ByteBuffer values = ByteBuffer.wrap(page, 0, bytes).order(ByteOrder.LITTLE_ENDIAN); + for (int i = 0; i < entries; i++) { + validator.accept(values); } - } catch (DataFormatException e) { - throw new IllegalArgumentException("Invalid video keyframe index payload.", e); - } finally { - inflater.end(); + 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 9a6807ed0a67..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 @@ -56,7 +56,6 @@ import java.nio.file.Files; import java.util.ArrayList; import java.util.List; -import java.util.zip.DeflaterOutputStream; import static org.apache.paimon.utils.StreamUtils.intToLittleEndian; import static org.assertj.core.api.Assertions.assertThat; @@ -66,8 +65,9 @@ public class VideoFileFormatTest { private static final String KEYFRAME_INDEX_HEX = - "0149464b4f45444956010000000200000000000000000000000100000000000000" - + "789c6360c00e78a0f4821e08cd04e503001394013b"; + "0149464b4f4544495601000000020000000000000000000000000000000100000000000000000000" + + "0000000000000000000000000000000000000000000c00000000000000a08c000000000000020000" + + "0000000000dfd9164b"; @TempDir java.nio.file.Path tempPath; @@ -135,14 +135,6 @@ public void testCrossLanguageV1Fixture() throws IOException { checkVideoFixture(fixture, fromHex(KEYFRAME_INDEX_HEX)); } - @Test - public void testCrossLanguagePagedIndexFixture() throws IOException { - byte[] fixture = fixture("video-paged-index.hex"); - byte[] mapping = java.util.Arrays.copyOfRange(fixture, 7, 7 + 89); - VideoKeyframeIndex.validate(mapping, 3); - checkVideoFixture(fixture, mapping); - } - private void checkVideoFixture(byte[] fixture, byte[] mapping) throws IOException { byte[] video = "abc".getBytes(StandardCharsets.UTF_8); java.nio.file.Path source = tempPath.resolve("indexed.mp4"); @@ -230,36 +222,12 @@ public void testRejectInvalidKeyframeIndex() throws IOException { .hasMessageContaining("Invalid video keyframe index"); } - @Test - public void testKeyframeIndexAboveOldLimit() throws IOException { - int count = 700001; - ByteArrayOutputStream compressed = new ByteArrayOutputStream(); - try (DeflaterOutputStream output = new DeflaterOutputStream(compressed)) { - ByteBuffer entry = ByteBuffer.allocate(24).order(ByteOrder.LITTLE_ENDIAN); - for (int i = 0; i < count; i++) { - entry.clear(); - entry.putLong(i).putLong(i).putLong(i); - output.write(entry.array()); - } - } - byte[] mapping = - ByteBuffer.allocate(17 + compressed.size()) - .order(ByteOrder.LITTLE_ENDIAN) - .put((byte) 1) - .putLong(0x564944454F4B4649L) - .putInt(0) - .putInt(count) - .put(compressed.toByteArray()) - .array(); - VideoKeyframeIndex.validate(mapping, count); - } - @Test public void testRejectForgedPagedCount() { byte[] mapping = ByteBuffer.allocate(21) .order(ByteOrder.LITTLE_ENDIAN) - .put((byte) 2) + .put((byte) 1) .putLong(0x564944454F4B4649L) .putInt(0) .putLong(Long.MAX_VALUE) @@ -314,7 +282,7 @@ private static byte[] pagedIndex(int count) throws IOException { out.write( ByteBuffer.allocate(21) .order(ByteOrder.LITTLE_ENDIAN) - .put((byte) 2) + .put((byte) 1) .putLong(0x564944454F4B4649L) .putInt(0) .putLong(count) @@ -337,12 +305,12 @@ private static byte[] pagedIndex(int count) throws IOException { @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(); @@ -356,8 +324,8 @@ 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); @@ -385,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); diff --git a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-paged-index.hex b/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-paged-index.hex deleted file mode 100644 index dc9e5f593637..000000000000 --- a/paimon-format/src/test/resources/org/apache/paimon/format/blob/video-paged-index.hex +++ /dev/null @@ -1 +0,0 @@ -6162635758595a0249464b4f45444956010000000200000000000000000000000000000001000000000000000000000000000000000000000000000000000000000000000c00000000000000a08c0000000000000200000000000000dfd9164b0602b201b101040100020100010106010403000e0602000000040000000500000005000000050000004944454f01 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/table/row/video_keyframe_index.py b/paimon-python/pypaimon/table/row/video_keyframe_index.py index 4dac2cb0a74c..550eb1555950 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -29,20 +29,6 @@ from contextlib import closing -class _RangeView: - def __init__(self, read_at, length): - self.read_at, self.length = read_at, length - - def __len__(self): - return self.length - - def __getitem__(self, key): - start, end, step = key.indices(self.length) - if step != 1: - raise ValueError("Non-contiguous index read.") - return self.read_at(start, end - start) - - class _PagedKeyframes(Sequence): ENTRY = struct.Struct('= 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 d228e0c384f0..6665fd9ee0c7 100644 --- a/paimon-python/pypaimon/tests/video_format_test.py +++ b/paimon-python/pypaimon/tests/video_format_test.py @@ -159,16 +159,9 @@ 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.HEADER.pack(1, VideoKeyframeIndex.MAGIC, 1, 2) - encoded += struct.pack(' Date: Fri, 30 Oct 2026 09:20:26 -0700 Subject: [PATCH 04/10] [docs] Simplify paged video index documentation --- docs/docs/concepts/spec/fileformat.md | 5 ++--- docs/docs/pypaimon/video.md | 7 +++---- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index ec85e8c59892..d3e70895b737 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -459,9 +459,8 @@ last page may be partial. Page `p` starts at `21 + 16 * metadata_count + p * 983 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. There is no keyframe-count -or aggregate index-byte budget. Metadata ranges remain limited to 65,536. PyPaimon -caches up to 1 MiB of entry pages per index and spools generated indexes to temporary disk. +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 ad8cd605f539..3f50c40f96c2 100644 --- a/docs/docs/pypaimon/video.md +++ b/docs/docs/pypaimon/video.md @@ -115,10 +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. Indexes are paged, so long -videos retain seek indexes without loading all keyframes into memory. Generation -uses temporary disk space; readers cache up to 1 MiB of entry pages per video. +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 From c619318c9063cc9551b5ddc3f3fed9d4732ea4de Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 18:33:12 -0700 Subject: [PATCH 05/10] [format] Inline video keyframe entry validation --- .../format/blob/VideoKeyframeIndex.java | 45 +++++++------------ 1 file changed, 16 insertions(+), 29 deletions(-) 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 6e495a89f119..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 @@ -71,12 +71,14 @@ static void validate(InputStream source, long length, long payloadLength) throws } previousEnd = start + size; } - EntryValidator validator = new EntryValidator(payloadLength); long pages = (count - 1) / PAGE_ENTRIES + 1; if (count > (length - offset) / ENTRY_LENGTH || pages * 4 != length - offset - count * ENTRY_LENGTH) { throw invalid("entries length"); } + long ordinal = -1; + long pts = 0; + long position = -1; byte[] page = new byte[PAGE_ENTRIES * ENTRY_LENGTH]; for (long done = 0; done < count; ) { int entries = (int) Math.min(PAGE_ENTRIES, count - done); @@ -90,39 +92,24 @@ static void validate(InputStream source, long length, long payloadLength) throws } ByteBuffer values = ByteBuffer.wrap(page, 0, bytes).order(ByteOrder.LITTLE_ENDIAN); for (int i = 0; i < entries; i++) { - validator.accept(values); + 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; } done += entries; } } - private static final class EntryValidator { - private final long payloadLength; - private long ordinal = -1; - private long pts; - private long position = -1; - - private EntryValidator(long payloadLength) { - this.payloadLength = payloadLength; - } - - private void accept(ByteBuffer values) { - 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; - } - } - private static IllegalArgumentException invalid(String reason) { return new IllegalArgumentException("Invalid video keyframe index: " + reason + '.'); } From 2711658083db157f23c0b5e5e83383850de4c063 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 18:45:09 -0700 Subject: [PATCH 06/10] [python] Validate uncached video index page boundaries --- docs/docs/concepts/spec/fileformat.md | 4 +- .../table/row/video_keyframe_index.py | 45 ++++++++++--------- .../tests/video_keyframe_index_test.py | 37 +++++++++++++-- 3 files changed, 61 insertions(+), 25 deletions(-) diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index d3e70895b737..03f0abf63243 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -459,8 +459,8 @@ last page may be partial. Page `p` starts at `21 + 16 * metadata_count + p * 983 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. +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. An Arrow/data-file cell stores a separately versioned, little-endian `VideoFrameDescriptor`: diff --git a/paimon-python/pypaimon/table/row/video_keyframe_index.py b/paimon-python/pypaimon/table/row/video_keyframe_index.py index 550eb1555950..82b12aa26a98 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -44,28 +44,33 @@ def __init__(self, read_at, count, payload_length=None, owner=None): def __len__(self): return self.count + def _load_page(self, page): + data = self.cache.pop(page, None) + if data is None: + count = min(self.PAGE_ENTRIES, self.count - page * self.PAGE_ENTRIES) + size = count * self.ENTRY.size + block = self.read_at(page * (self.PAGE_BYTES + 4), size + 4) + if len(block) != size + 4 or zlib.crc32(block[:size]) != struct.unpack(' max(1, self.CACHE_BYTES // self.PAGE_BYTES): + self.cache.popitem(last=False) + return data + def _page(self, page): with self.lock: - data = self.cache.pop(page, None) - if data is None: - count = min(self.PAGE_ENTRIES, self.count - page * self.PAGE_ENTRIES) - size = count * self.ENTRY.size - block = self.read_at(page * (self.PAGE_BYTES + 4), size + 4) - if len(block) != size + 4 or zlib.crc32(block[:size]) != struct.unpack(' max(1, self.CACHE_BYTES // self.PAGE_BYTES): - self.cache.popitem(last=False) + data = self._load_page(page) + # Check both boundaries even when adjacent pages were not cached. + if page > 0: + previous = self._load_page(page - 1) + _validate_boundary(self.ENTRY.unpack_from(previous, len(previous) - self.ENTRY.size), + self.ENTRY.unpack_from(data)) + if (page + 1) * self.PAGE_ENTRIES < self.count: + following = self._load_page(page + 1) + _validate_boundary(self.ENTRY.unpack_from(data, len(data) - self.ENTRY.size), + self.ENTRY.unpack_from(following)) return data def __getitem__(self, index): diff --git a/paimon-python/pypaimon/tests/video_keyframe_index_test.py b/paimon-python/pypaimon/tests/video_keyframe_index_test.py index 946bd66ada6b..8f36ba6c53b5 100644 --- a/paimon-python/pypaimon/tests/video_keyframe_index_test.py +++ b/paimon-python/pypaimon/tests/video_keyframe_index_test.py @@ -241,9 +241,14 @@ def read_at(offset, size): with VideoKeyframeIndex.open(read_at, length, count * 2) as restored: self.assertEqual(count, len(restored.keyframes)) - self.assertLess(sum(size for _, size in reads), 100_000) + self.assertLess(sum(size for _, size in reads), 200_000) for ordinal in (count - 1, 32768, 0, 499999): + before = len(reads) self.assertEqual((ordinal, ordinal - 100, ordinal * 2), restored.keyframes[ordinal]) + self.assertLessEqual(len(reads) - before, 3) + before = len(reads) + self.assertEqual((ordinal, ordinal - 100, ordinal * 2), restored.keyframes[ordinal]) + self.assertEqual(before, len(reads)) self.assertEqual(ordinal, bisect.bisect_left(restored.ordinals, ordinal)) self.assertLessEqual(max(size for _, size in reads), 4096 * 24 + 25) self.assertLessEqual(sum(map(len, restored.keyframes.cache.values())), 1024 * 1024) @@ -255,8 +260,8 @@ def test_paged_corruption_and_boundary_validation(self): second = VideoKeyframeIndex.HEADER.size + 4096 * 24 + 4 corrupt = bytearray(data) corrupt[second] ^= 1 - with VideoKeyframeIndex.open(lambda o, n: corrupt[o:o + n], len(corrupt)) as index: - with self.assertRaisesRegex(ValueError, 'checksum'): + with self.assertRaisesRegex(ValueError, 'checksum'): + with VideoKeyframeIndex.open(lambda o, n: corrupt[o:o + n], len(corrupt)) as index: _ = index.keyframes[4096] # A valid page checksum must not hide a descending page boundary. corrupt = bytearray(data) @@ -267,6 +272,32 @@ def test_paged_corruption_and_boundary_validation(self): with self.assertRaises(ValueError): VideoKeyframeIndex.validate(invalid) + def test_lazy_reads_validate_uncached_page_boundaries(self): + page_entries = 4096 + stride = page_entries * 24 + 4 + with VideoKeyframeIndex([], ((i, i, i) for i in range(16 * page_entries))) as index: + original = index.serialize() + for column in range(3): + corrupt = bytearray(original) + offset = VideoKeyframeIndex.HEADER.size + 12 * stride + page = bytearray(corrupt[offset:offset + page_entries * 24]) + for entry in range(page_entries): + struct.pack_into(' Date: Fri, 30 Oct 2026 18:52:53 -0700 Subject: [PATCH 07/10] [python] Coalesce adjacent video index page reads --- .../table/row/video_keyframe_index.py | 63 +++++++++++-------- .../tests/video_keyframe_index_test.py | 32 +++++++++- 2 files changed, 66 insertions(+), 29 deletions(-) diff --git a/paimon-python/pypaimon/table/row/video_keyframe_index.py b/paimon-python/pypaimon/table/row/video_keyframe_index.py index 82b12aa26a98..d5ef04c0b3c3 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -44,34 +44,42 @@ def __init__(self, read_at, count, payload_length=None, owner=None): def __len__(self): return self.count - def _load_page(self, page): - data = self.cache.pop(page, None) - if data is None: - count = min(self.PAGE_ENTRIES, self.count - page * self.PAGE_ENTRIES) - size = count * self.ENTRY.size - block = self.read_at(page * (self.PAGE_BYTES + 4), size + 4) - if len(block) != size + 4 or zlib.crc32(block[:size]) != struct.unpack(' max(1, self.CACHE_BYTES // self.PAGE_BYTES): - self.cache.popitem(last=False) - return data - def _page(self, page): with self.lock: - data = self._load_page(page) - # Check both boundaries even when adjacent pages were not cached. - if page > 0: - previous = self._load_page(page - 1) - _validate_boundary(self.ENTRY.unpack_from(previous, len(previous) - self.ENTRY.size), - self.ENTRY.unpack_from(data)) - if (page + 1) * self.PAGE_ENTRIES < self.count: - following = self._load_page(page + 1) - _validate_boundary(self.ENTRY.unpack_from(data, len(data) - self.ENTRY.size), - self.ENTRY.unpack_from(following)) - return data + pages = range(max(0, page - 1), min((self.count - 1) // self.PAGE_ENTRIES, page + 1) + 1) + for p in pages: + if p in self.cache: + self.cache.move_to_end(p) + missing = [p for p in pages if p not in self.cache] + if missing: + first, last = missing[0], missing[-1] + stride = self.PAGE_BYTES + 4 + last_size = min(self.PAGE_ENTRIES, self.count - last * self.PAGE_ENTRIES) * self.ENTRY.size + size = (last - first) * stride + last_size + 4 + # Fetch missing neighbors together instead of serial range requests. + block = self.read_at(first * stride, size) + if len(block) != size: + raise ValueError("Invalid video keyframe index page length.") + for p in range(first, last + 1): + start = (p - first) * stride + size = min(self.PAGE_ENTRIES, self.count - p * self.PAGE_ENTRIES) * self.ENTRY.size + data = block[start:start + size] + if zlib.crc32(data) != struct.unpack_from(' max(1, self.CACHE_BYTES // self.PAGE_BYTES): + self.cache.popitem(last=False) + previous = None + for p in pages: + data = self.cache[p] + if previous is not None: + _validate_boundary(self.ENTRY.unpack_from(previous, len(previous) - self.ENTRY.size), + self.ENTRY.unpack_from(data)) + previous = data + self.cache.move_to_end(p) + self.cache.move_to_end(page) + return self.cache[page] def __getitem__(self, index): if isinstance(index, slice): @@ -248,7 +256,8 @@ def read(offset, size): return b'' if offset + size <= len(prefix): return prefix[offset:offset + size] - data = read_at(offset, size) + cached = prefix[offset:min(offset + size, len(prefix))] + data = cached + read_at(offset + len(cached), size - len(cached)) if len(data) != size: raise ValueError("Invalid video keyframe index: truncated read.") return data diff --git a/paimon-python/pypaimon/tests/video_keyframe_index_test.py b/paimon-python/pypaimon/tests/video_keyframe_index_test.py index 8f36ba6c53b5..6280e05646bc 100644 --- a/paimon-python/pypaimon/tests/video_keyframe_index_test.py +++ b/paimon-python/pypaimon/tests/video_keyframe_index_test.py @@ -245,15 +245,43 @@ def read_at(offset, size): for ordinal in (count - 1, 32768, 0, 499999): before = len(reads) self.assertEqual((ordinal, ordinal - 100, ordinal * 2), restored.keyframes[ordinal]) - self.assertLessEqual(len(reads) - before, 3) + self.assertLessEqual(len(reads) - before, 1) before = len(reads) self.assertEqual((ordinal, ordinal - 100, ordinal * 2), restored.keyframes[ordinal]) self.assertEqual(before, len(reads)) self.assertEqual(ordinal, bisect.bisect_left(restored.ordinals, ordinal)) - self.assertLessEqual(max(size for _, size in reads), 4096 * 24 + 25) + self.assertLessEqual(max(size for _, size in reads), 3 * (4096 * 24 + 4)) self.assertLessEqual(sum(map(len, restored.keyframes.cache.values())), 1024 * 1024) VideoKeyframeIndex.validate_stream(stored, length, count * 2) + def test_binary_search_coalesces_adjacent_page_reads(self): + import bisect + import tempfile + count = 1_000_000 + with VideoKeyframeIndex([], ((i, i, i) for i in range(count))) as original: + with tempfile.TemporaryFile() as stored: + original.write_to(stored) + length = stored.tell() + reads = [] + + def read_at(offset, size): + reads.append((offset, size)) + stored.seek(offset) + return stored.read(size) + + for target in (count - 1, 123456, 500000): + with self.subTest(target=target): + reads.clear() + with VideoKeyframeIndex.open(read_at, length, count) as index: + reads.clear() + self.assertEqual(target, bisect.bisect_left(index.ordinals, target)) + self.assertLessEqual(len(reads), 9) + self.assertLessEqual(max(size for _, size in reads), 3 * (4096 * 24 + 4)) + before = len(reads) + self.assertEqual((target, target, target), index.keyframes[target]) + self.assertEqual(before, len(reads)) + self.assertLessEqual(sum(map(len, index.keyframes.cache.values())), 1024 * 1024) + def test_paged_corruption_and_boundary_validation(self): with VideoKeyframeIndex([], ((i, i, i) for i in range(8193))) as index: data = index.serialize() From ed4eb6c928586ba1e55a0172a8e093249a112270 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 19:58:27 -0700 Subject: [PATCH 08/10] [docs] Shorten paged video index descriptions --- docs/docs/concepts/spec/fileformat.md | 9 ++++----- docs/docs/pypaimon/video.md | 6 +++--- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 03f0abf63243..29c3dc0c24dd 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/docs/docs/concepts/spec/fileformat.md @@ -441,8 +441,7 @@ 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 version `1` uses this layout. Numeric fields are little endian; -byte offsets are relative to the embedded video payload. +Keyframe-index v1 is little endian; byte offsets are relative to the video payload. | Field | Size | | --- | ---: | @@ -453,9 +452,9 @@ byte offsets are relative to the embedded video payload. | 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`. +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. diff --git a/docs/docs/pypaimon/video.md b/docs/docs/pypaimon/video.md index 3f50c40f96c2..97705b635b26 100644 --- a/docs/docs/pypaimon/video.md +++ b/docs/docs/pypaimon/video.md @@ -115,9 +115,9 @@ video = pm.BlobDescriptor( frames.add_video(video, episode_43_rows) ``` -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. +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. The writer deduplicates exact payload descriptor identity inside each `.video` file. Its video grouping policy coordinates normal, BLOB, and vector rolling From 40e60200909eb05cf1092641e53f4f432fbe8abf Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 20:20:52 -0700 Subject: [PATCH 09/10] [python] Reuse paged video index validation and ordinal probes --- docs/docs/pypaimon/video.md | 2 +- .../table/row/video_keyframe_index.py | 22 ++++++++++-- .../tests/video_keyframe_index_test.py | 34 +++++++++++++++++++ 3 files changed, 55 insertions(+), 3 deletions(-) diff --git a/docs/docs/pypaimon/video.md b/docs/docs/pypaimon/video.md index 97705b635b26..873d1ebf1031 100644 --- a/docs/docs/pypaimon/video.md +++ b/docs/docs/pypaimon/video.md @@ -117,7 +117,7 @@ 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. +entry pages per index and 4,096 ordinal lookups per search view. 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-python/pypaimon/table/row/video_keyframe_index.py b/paimon-python/pypaimon/table/row/video_keyframe_index.py index d5ef04c0b3c3..11c5c973f5f1 100644 --- a/paimon-python/pypaimon/table/row/video_keyframe_index.py +++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py @@ -39,6 +39,7 @@ def __init__(self, read_at, count, payload_length=None, owner=None): self.read_at, self.count = read_at, count self.payload_length, self.owner = payload_length, owner self.cache = OrderedDict() + self.validated = set() self.lock = threading.RLock() def __len__(self): @@ -46,6 +47,9 @@ def __len__(self): def _page(self, page): with self.lock: + if page in self.validated: + self.cache.move_to_end(page) + return self.cache[page] pages = range(max(0, page - 1), min((self.count - 1) // self.PAGE_ENTRIES, page + 1) + 1) for p in pages: if p in self.cache: @@ -69,7 +73,8 @@ def _page(self, page): _validate_entries(data, self.payload_length, initial=p == 0) self.cache[p] = data while len(self.cache) > max(1, self.CACHE_BYTES // self.PAGE_BYTES): - self.cache.popitem(last=False) + evicted, _ = self.cache.popitem(last=False) + self.validated.discard(evicted) previous = None for p in pages: data = self.cache[p] @@ -78,6 +83,7 @@ def _page(self, page): self.ENTRY.unpack_from(data)) previous = data self.cache.move_to_end(p) + self.validated.add(page) self.cache.move_to_end(page) return self.cache[page] @@ -105,17 +111,29 @@ def close(self): if self.owner is not None: self.owner.close() self.cache.clear() + self.validated.clear() class _Ordinals(Sequence): + CACHE_ENTRIES = 4096 + def __init__(self, keyframes): self.keyframes = keyframes + self.cache = OrderedDict() def __len__(self): return len(self.keyframes) def __getitem__(self, index): - return self.keyframes[index][0] + with self.keyframes.lock: + if index in self.cache: + self.cache.move_to_end(index) + return self.cache[index] + value = self.keyframes[index][0] + self.cache[index] = value + if len(self.cache) > self.CACHE_ENTRIES: + self.cache.popitem(last=False) + return value def _validate_boundary(previous, current): diff --git a/paimon-python/pypaimon/tests/video_keyframe_index_test.py b/paimon-python/pypaimon/tests/video_keyframe_index_test.py index 6280e05646bc..5e33e7db98a9 100644 --- a/paimon-python/pypaimon/tests/video_keyframe_index_test.py +++ b/paimon-python/pypaimon/tests/video_keyframe_index_test.py @@ -282,6 +282,40 @@ def read_at(offset, size): self.assertEqual(before, len(reads)) self.assertLessEqual(sum(map(len, index.keyframes.cache.values())), 1024 * 1024) + def test_validated_page_does_not_reload_evicted_neighbors(self): + with VideoKeyframeIndex([], ((i, i, i) for i in range(16 * 4096))) as original: + data = original.serialize() + reads = [] + + def read_at(offset, size): + reads.append((offset, size)) + return data[offset:offset + size] + + with VideoKeyframeIndex.open(read_at, len(data)) as index: + self.assertEqual((0, 0, 0), index.keyframes[0]) + for page in range(3, 16): + index.keyframes[page * 4096] + before = len(reads) + self.assertEqual((0, 0, 0), index.keyframes[0]) + self.assertEqual(before, len(reads)) + index.keyframes.cache.pop(1, None) + index.keyframes.validated.discard(1) + before = len(reads) + self.assertEqual((0, 0, 0), index.keyframes[0]) + self.assertEqual(before, len(reads)) + self.assertLessEqual(len(index.keyframes.validated), len(index.keyframes.cache)) + + def test_ordinal_probe_cache_is_bounded(self): + with VideoKeyframeIndex([], ((i, i, i) for i in range(8192))) as original: + data = original.serialize() + with VideoKeyframeIndex.deserialize(data) as index: + ordinals = index.ordinals + for i in range(5000): + self.assertEqual(i, ordinals[i]) + self.assertEqual(4096, len(ordinals.cache)) + with mock.patch.object(index.keyframes, '_page', side_effect=AssertionError): + self.assertEqual(4999, ordinals[4999]) + def test_paged_corruption_and_boundary_validation(self): with VideoKeyframeIndex([], ((i, i, i) for i in range(8193))) as index: data = index.serialize() From f7a59baa6927af99437041b7da61d417e96ea6fc Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 20:30:58 -0700 Subject: [PATCH 10/10] [format] Propagate video index spool write failures --- .../paimon/format/blob/VideoFormatWriter.java | 50 +++++++++++++----- .../format/blob/VideoFileFormatTest.java | 52 +++++++++++++++++++ 2 files changed, 88 insertions(+), 14 deletions(-) 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 6de398d3ab06..ff3d9b97c247 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 @@ -289,26 +289,48 @@ private boolean readIndex( long start = spool.getFilePointer(); boolean complete = false; try { - try (InputStream in = index.newInputStream()) { + InputStream source; + try { + source = index.newInputStream(); + } catch (IOException | RuntimeException e) { + if (handleFetchFailure(e, index)) { + return false; + } + throw e; + } + try (InputStream in = source) { 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."); + while (true) { + int n; + try { + n = + in.read( + buffer, + 0, + remaining == 0 + ? 1 + : (int) Math.min(remaining, buffer.length)); + if (remaining == 0) { + if (n != -1) { + throw new IOException( + "Video keyframe index has trailing data."); + } + break; + } + if (n <= 0) { + throw new java.io.EOFException( + "Video keyframe index length changed while reading."); + } + } catch (IOException | RuntimeException e) { + if (handleFetchFailure(e, index)) { + return false; + } + throw e; } 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; } spool.seek(start); VideoKeyframeIndex.validate( 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 dd4565c8c36d..1bd58e5472cf 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 @@ -20,6 +20,7 @@ import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobData; +import org.apache.paimon.data.BlobFetchMetricReporter; import org.apache.paimon.data.BlobPlaceholder; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; @@ -544,6 +545,57 @@ public void testReaderLargeKeyframeIndexOffsets() throws IOException { } } + @Test + public void testIndexSpoolWriteFailureDoesNotWriteNull() throws Exception { + byte[] mapping = fromHex(KEYFRAME_INDEX_HEX); + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + bytes.write(new byte[] {1, 2, 3}); + bytes.write(mapping); + java.nio.file.Path source = tempPath.resolve("spool-source.mp4"); + Files.write(source, bytes.toByteArray()); + Blob frame = + Blob.fromDescriptor( + org.apache.paimon.utils.UriReader.fromFile(fileIO), + new VideoFrameDescriptor( + source.toUri().toString(), 0, 3, 0, 3, mapping.length)); + for (boolean fallback : new boolean[] {false, true}) { + try (PositionOutputStream out = fileIO.newOutputStream(file, true); + RandomAccessFile spool = + new RandomAccessFile(tempPath.resolve("index.tmp").toFile(), "rw") { + @Override + public void write(byte[] b, int offset, int length) + throws IOException { + throw new IOException("No space left on device"); + } + }) { + VideoFormatWriter writer = + new VideoFormatWriter( + out, + rowType, + false, + fallback, + BlobFetchMetricReporter.NOOP, + BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + writer.setFile(file); + java.lang.reflect.Field field = + VideoFormatWriter.class.getDeclaredField("indexSpool"); + field.setAccessible(true); + field.set(writer, spool); + try { + assertThatThrownBy(() -> writer.addElement(GenericRow.of(frame))) + .isInstanceOf(IOException.class) + .hasMessage("No space left on device"); + assertThat(out.getPos()).isZero(); + assertThat(writer.physicalVideoCount()).isZero(); + assertThat(writer.runCount()).isZero(); + assertThat(spool.length()).isZero(); + } finally { + writer.close(); + } + } + } + } + @Test public void testIndexFetchNullPolicyBeforePayloadWrite() throws IOException { for (int status : new int[] {404, 416, 503, 0}) {