This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e20de7d2ac [python] Speed up video keyframe index deserialization
(#10794)
e20de7d2ac is described below
commit e20de7d2aca62c81bae1dfd289d74970e52b0575
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Oct 25 10:45:29 2026 +0800
[python] Speed up video keyframe index deserialization (#10794)
---
.../pypaimon/table/row/video_keyframe_index.py | 56 ++++++++++++--
.../pypaimon/tests/video_keyframe_index_test.py | 86 ++++++++++++++++++++++
paimon-python/pyproject.toml | 1 +
3 files changed, 138 insertions(+), 5 deletions(-)
diff --git a/paimon-python/pypaimon/table/row/video_keyframe_index.py
b/paimon-python/pypaimon/table/row/video_keyframe_index.py
index 9d8f34e158..2c38322706 100644
--- a/paimon-python/pypaimon/table/row/video_keyframe_index.py
+++ b/paimon-python/pypaimon/table/row/video_keyframe_index.py
@@ -234,12 +234,58 @@ class VideoKeyframeIndex:
@classmethod
def deserialize(cls, data, payload_length=None):
+ import numpy as np
+
metadata_count, keyframe_count, entries_offset = cls._read_header(data)
- return cls(
- cls._iter_metadata_ranges(data, metadata_count, payload_length),
- cls._iter_keyframes(
- data, entries_offset, keyframe_count, payload_length),
- )
+ metadata_ranges = []
+ for start, length in cls._iter_metadata_ranges(
+ data, metadata_count, payload_length):
+ if (metadata_ranges
+ and sum(metadata_ranges[-1]) == start):
+ previous_start, previous_length = metadata_ranges[-1]
+ metadata_ranges[-1] = (previous_start, previous_length +
length)
+ else:
+ metadata_ranges.append((start, length))
+
+ # The header caps this buffer at 1.5 MiB. One extra byte detects
+ # excess output without expanding an untrusted compressed stream.
+ expected_size = keyframe_count * cls.ENTRY.size
+ decoder = zlib.decompressobj()
+ try:
+ raw = decoder.decompress(
+ memoryview(data)[entries_offset:], expected_size + 1)
+ except zlib.error as error:
+ raise ValueError(
+ "Invalid video keyframe index payload."
+ ) from error
+ if (len(raw) != expected_size or not decoder.eof
+ or decoder.unused_data or decoder.unconsumed_tail):
+ raise ValueError("Invalid video keyframe index entries.")
+
+ # ENTRY '<qqq': ordinal, PTS and packet position are little-endian
int64.
+ entries = np.frombuffer(raw, dtype='<i8').reshape(keyframe_count, 3)
+ if entries[0, 0] != 0:
+ raise ValueError("Video keyframe index requires an initial
keyframe.")
+ # Compare directly: subtraction can overflow for signed PTS values.
+ if np.any(entries[1:, 0] <= entries[:-1, 0]):
+ raise ValueError("Keyframe ordinals must be strictly increasing.")
+ if np.any(entries[1:, 1] <= entries[:-1, 1]):
+ raise ValueError("Keyframe timestamps must be strictly
increasing.")
+ if (entries[0, 2] < 0
+ or np.any(entries[1:, 2] <= entries[:-1, 2])
+ or (payload_length is not None
+ and int(entries[-1, 2]) >= payload_length)):
+ raise ValueError(
+ "Keyframe packet positions must be within the "
+ "video payload and strictly increasing."
+ )
+
+ # Both sections have already been validated. Preserve Python tuples
+ # without repeating per-entry normalization and validation in __init__.
+ result = cls.__new__(cls)
+ result.metadata_ranges = tuple(metadata_ranges)
+ result.keyframes = tuple(cls.ENTRY.iter_unpack(raw))
+ return result
@classmethod
def _read_header(cls, data):
diff --git a/paimon-python/pypaimon/tests/video_keyframe_index_test.py
b/paimon-python/pypaimon/tests/video_keyframe_index_test.py
index 9488432632..b4738d136d 100644
--- a/paimon-python/pypaimon/tests/video_keyframe_index_test.py
+++ b/paimon-python/pypaimon/tests/video_keyframe_index_test.py
@@ -97,6 +97,92 @@ class VideoKeyframeIndexTest(unittest.TestCase):
VideoKeyframeIndex.deserialize(
header + zlib.compress(struct.pack('<qqq', 0, 1, 2)))
+ def test_deserialize_matches_streaming_validation(self):
+ def encode(entries, count=None, compressed=None):
+ return VideoKeyframeIndex.HEADER.pack(
+ VideoKeyframeIndex.VERSION, VideoKeyframeIndex.MAGIC,
+ 0, len(entries) if count is None else count,
+ ) + (zlib.compress(b''.join(
+ VideoKeyframeIndex.ENTRY.pack(*entry) for entry in entries
+ )) if compressed is None else compressed)
+
+ valid = [(0, -(2 ** 63), 0), (2 ** 63 - 1, 2 ** 63 - 1, 99)]
+ for entries in ([(0, -10, 0)], valid):
+ data = encode(entries)
+ VideoKeyframeIndex.validate(data, 100)
+ restored = VideoKeyframeIndex.deserialize(data, 100)
+ self.assertEqual(tuple(entries), restored.keyframes)
+ self.assertTrue(all(type(value) is int
+ for entry in restored.keyframes
+ for value in entry))
+
+ invalid = [
+ encode([(1, 0, 0)]),
+ encode([(0, 0, -1)]),
+ encode([(0, 0, 0), (0, 1, 1)]),
+ encode([(0, 0, 0), (-1, 1, 1)]),
+ encode([(0, 1, 0), (1, 1, 1)]),
+ encode([(0, 1, 0), (1, -1, 1)]),
+ encode([(0, 0, 1), (1, 1, 1)]),
+ encode([(0, 0, 1), (1, 1, 0)]),
+ encode([(0, 0, 100)]),
+ encode(valid, count=1),
+ encode(valid, count=3),
+ encode(valid)[:-1],
+ encode(valid) + b'trailing',
+ encode(valid) + zlib.compress(b''),
+ encode(valid, compressed=b'not zlib'),
+ encode(valid, compressed=zlib.compress(b'partial entry')),
+ ]
+ for data in invalid:
+ with self.subTest(data=data):
+ for read in (VideoKeyframeIndex.validate,
+ VideoKeyframeIndex.deserialize):
+ with self.assertRaises(ValueError):
+ read(data, 100)
+
+ def test_deserialize_preserves_metadata_normalization(self):
+ data = VideoKeyframeIndex.HEADER.pack(
+ VideoKeyframeIndex.VERSION, VideoKeyframeIndex.MAGIC, 3, 1
+ ) + b''.join(VideoKeyframeIndex.METADATA_RANGE.pack(*entry)
+ for entry in [(0, 2), (2, 3), (8, 2)])
+ data += zlib.compress(VideoKeyframeIndex.ENTRY.pack(0, -1, 10))
+ restored = VideoKeyframeIndex.deserialize(data, 11)
+ self.assertEqual(((0, 5), (8, 2)), restored.metadata_ranges)
+ self.assertEqual(((0, -1, 10),), restored.keyframes)
+
+ def test_deserialize_bounds_output_before_materializing_entries(self):
+ data = VideoKeyframeIndex.HEADER.pack(
+ VideoKeyframeIndex.VERSION, VideoKeyframeIndex.MAGIC, 0, 1
+ ) + zlib.compress(b'\0' * (4 * 1024 * 1024))
+ real_decompressobj = zlib.decompressobj
+ output_sizes = []
+
+ class TrackingDecompressor:
+ def __init__(self):
+ self._delegate = real_decompressobj()
+
+ def decompress(self, value, max_length=0):
+ output = self._delegate.decompress(value, max_length)
+ output_sizes.append(len(output))
+ return output
+
+ def __getattr__(self, name):
+ return getattr(self._delegate, name)
+
+ with mock.patch.object(keyframe_index_module.zlib, 'decompressobj',
+ side_effect=TrackingDecompressor):
+ with self.assertRaises(ValueError):
+ VideoKeyframeIndex.deserialize(data)
+ self.assertEqual([VideoKeyframeIndex.ENTRY.size + 1], output_sizes)
+
+ def test_deserialize_at_entry_limit(self):
+ entries = tuple((value, value - 100, value * 2)
+ for value in
range(VideoKeyframeIndex.MAX_KEYFRAME_COUNT))
+ index = VideoKeyframeIndex([], entries)
+ restored = VideoKeyframeIndex.deserialize(index.serialize())
+ self.assertEqual(entries, restored.keyframes)
+
def test_validate_rejects_ranges_outside_video_payload(self):
metadata_outside = VideoKeyframeIndex(
[(8, 4)], [(0, 0, 1)]).serialize()
diff --git a/paimon-python/pyproject.toml b/paimon-python/pyproject.toml
index db1f3ab767..e5c4f3e24a 100644
--- a/paimon-python/pyproject.toml
+++ b/paimon-python/pyproject.toml
@@ -54,6 +54,7 @@ dependencies = [
"isal>=1.8,<2; python_version >= '3.9' and (platform_machine == 'x86_64'
or platform_machine == 'AMD64' or platform_machine == 'aarch64' or
platform_machine == 'arm64')",
"fsspec>=2021.10,<2026; python_version < '3.8'",
"fsspec>=2023,<2026; python_version >= '3.8'",
+ "numpy>=1.19.5,<3",
"packaging>=21",
"pandas>=1.1,<2; python_version < '3.7'",
"pandas>=1.3,<3; python_version >= '3.7' and python_version < '3.9'",
