JingsongLi commented on code in PR #9831:
URL: https://github.com/apache/paimon/pull/9831#discussion_r4105580907
##########
paimon-python/pypaimon/multimodal/lerobot/dataset.py:
##########
@@ -1598,6 +1789,141 @@ def __getitem__(self, index):
) from error
raise IndexError("Video frame index %d is out of range." % index)
+ def get_frames_at(self, *, indices):
+ import torch
+ frames = (
+ self._read_indexed(indices)
+ if self._keyframe_index is not None
+ else [self[index] for index in indices]
+ )
+ return SimpleNamespace(data=torch.stack(frames))
+
+ def _read_indexed(self, indices):
+ requested = [operator.index(index) for index in indices]
+ if any(index < 0 for index in requested):
+ raise IndexError("Video frame index is out of range.")
+ missing = sorted(set(requested) - set(self._cache))
+ decoded = self._decode_indexed(missing) if missing else {}
+ try:
+ return [
+ self._cache[index] if index in self._cache else decoded[index]
+ for index in requested
+ ]
+ except KeyError as error:
+ raise IndexError(
+ "Video frame index %d is out of range." % error.args[0]
+ ) from error
+
+ def _decode_indexed(self, indices):
+ import av
+ groups, ranges = self._indexed_plan(indices)
+ decoded = {}
+ source = _RangeBackedVideo(
+ self._video_length(), self._read_video_ranges)
+ try:
+ source.prefetch(_merge_video_ranges(
Review Comment:
[P1] Bound batch range prefetch by bytes
This prefetches the metadata plus every GOP window for all missing indices
in one call, and `_RangeBackedVideo._ensure` keeps every returned body in
`_segments` until the whole batch finishes. A shuffled batch that selects
frames from many distant GOPs in one large, valid video can therefore
materialize the sum of those windows, potentially approaching the whole video;
multiple DataLoader workers multiply the peak. The 16/64 MiB limits only bound
index blocks, not video ranges. Please decode groups in byte-bounded chunks or
evict range segments instead of retaining the full batch working set.
##########
paimon-format/src/main/java/org/apache/paimon/format/blob/VideoFormatWriter.java:
##########
@@ -121,23 +131,45 @@ public void addElement(InternalRow element) throws
IOException {
"Video fields require an exact BlobRef containing a
VideoFrameDescriptor.");
BlobDescriptor payload = frame.payloadDescriptor();
+ BlobDescriptor keyframeIndexDescriptor =
frame.keyframeIndexDescriptor();
Integer ordinal = physicalVideos.get(payload);
if (ordinal == null) {
+ long keyframeIndexLength =
+ keyframeIndexDescriptor == null ? 0 :
keyframeIndexDescriptor.length();
+ checkKeyframeIndexSize(keyframeIndexLength, keyframeIndexBytes);
long length = payloadWriter.write(element);
if (length == BlobFormatWriter.NULL_LENGTH) {
append(NULL_REFERENCE, 0);
return;
}
ordinal = physicalVideoLengths.size();
physicalVideoLengths.add(length);
+ Blob keyframeIndex = VideoFrameDescriptor.keyframeIndexBlob(blob);
+ byte[] mapping = keyframeIndex == null ? new byte[0] :
keyframeIndex.toData();
Review Comment:
[P2] Apply the configured NULL policy to the index fetch
`payloadWriter.write(element)` honors `blob-write-null-on-missing-file` and
`blob-write-null-on-fetch-failure`, but this independent `toData()` fetch
bypasses both policies and the fetch metrics. If the payload range succeeds and
the index range then returns 404/416, is truncated, or fails transiently, the
writer aborts even when NULL fallback is enabled; the payload bytes have
already been written by that point. Please fetch and validate the index through
the same policy before writing the payload, append NULL when the configured
fallback applies, and cover payload-success/index-failure for both options.
##########
paimon-python/pypaimon/read/reader/video_format_reader.py:
##########
@@ -72,22 +75,49 @@ def __init__(self, stream, file_size: int):
indexes.append(DeltaVarintCompressor.decompress(raw))
offset += length
- physical_lengths, run_lengths, references, first_frames = indexes
+ (
+ physical_lengths,
+ keyframe_index_lengths,
+ run_lengths,
+ references,
+ first_frames,
+ ) = indexes
+ if len(keyframe_index_lengths) != len(physical_lengths):
+ raise IOError(
+ "Corrupt video file: physical video and keyframe index "
+ "indexes have different counts."
+ )
+ if any(length < 0 for length in keyframe_index_lengths):
+ raise IOError(
+ "Corrupt video file: negative keyframe index length."
+ )
+
+ keyframe_index_size = sum(keyframe_index_lengths)
Review Comment:
[P2] Enforce the documented index-size limits while reading
The format specification limits each keyframe-index block to 16 MiB and all
blocks in one file to 64 MiB, but this reader only rejects negative lengths or
totals outside the file. `VideoFrameCollator` later passes the declared length
to `read_file_range` and materializes the entire block before
`VideoKeyframeIndex.deserialize` can validate it. A corrupt or forged `.video`
file can therefore force an arbitrarily large range read and allocation. Please
reject per-block and cumulative lengths above the documented limits in
`VideoFileMeta` before exposing these descriptors, and mirror the check in the
Java reader.
##########
paimon-python/pypaimon/write/video_format_writer.py:
##########
@@ -115,27 +141,73 @@ def close(self) -> None:
self._run_first_frames
)
for index in (
- physical_index,
- run_length_index,
- run_reference_index,
- first_frame_index,
- ):
+ physical_index,
+ keyframe_length_index,
+ run_length_index,
+ run_reference_index,
+ first_frame_index):
self.output_stream.write(index)
- self.output_stream.write(struct.pack(
- '<IIIIIB',
+ footer = struct.pack(
+ '<IIIIIIB',
len(physical_index),
+ len(keyframe_length_index),
len(run_length_index),
len(run_reference_index),
len(first_frame_index),
self.FOOTER_MAGIC_NUMBER,
self.VERSION,
- ))
+ )
+ self.output_stream.write(footer)
if hasattr(self.output_stream, 'flush'):
self.output_stream.flush()
if hasattr(self.output_stream, 'close'):
self.output_stream.close()
self._closed = True
+ def _keyframe_index(self, blob, frame):
+ descriptor = frame.keyframe_index_descriptor
+ if descriptor is None:
+ stream = blob.new_input_stream()
+ try:
+ try:
+ payload_length = frame.payload_descriptor.length
+ if payload_length < 0:
+ position = stream.tell()
+ payload_length = stream.seek(0, 2)
+ stream.seek(position)
+ mapping = VideoKeyframeIndex.inspect(
+ stream, payload_length).serialize()
+ except (ImportError, OSError, ValueError, EOFError):
Review Comment:
[P2] Include PyAV FFmpeg errors in the scan fallback
The documented behavior is that videos which cannot be indexed use the scan
fallback, but this handler does not cover every `av.error.FFmpegError`. For
example, decoder-not-found and other FFmpeg failures may inherit `LookupError`
or only `FFmpegError`, so they escape this tuple and abort ingestion instead of
returning an empty index. The reader in this PR already adds
`av.error.FFmpegError` dynamically in `_video_decoder_fallback_errors`; please
apply the same exception family here and add a real PyAV regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
