JingsongLi commented on code in PR #9831:
URL: https://github.com/apache/paimon/pull/9831#discussion_r4055764965
##########
paimon-python/pypaimon/write/video_format_writer.py:
##########
@@ -106,36 +127,65 @@ def close(self) -> None:
if self._closed:
return
self._flush_run()
+ version = (
+ self.VERSION if any(self._keyframe_indexes) else self.V1_VERSION)
+ if version >= 2:
+ 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,
- run_length_index,
- run_reference_index,
- first_frame_index,
- ):
+ indexes = [physical_index]
+ if version >= 2:
+ indexes.append(keyframe_length_index)
+ indexes.extend((run_length_index, run_reference_index,
first_frame_index))
+ for index in indexes:
self.output_stream.write(index)
- self.output_stream.write(struct.pack(
- '<IIIIIB',
- len(physical_index),
- len(run_length_index),
- len(run_reference_index),
- len(first_frame_index),
- self.FOOTER_MAGIC_NUMBER,
- self.VERSION,
- ))
+ if version == self.V1_VERSION:
+ footer = struct.pack(
+ '<IIIIIB',
+ len(physical_index),
+ len(run_length_index),
+ len(run_reference_index),
+ len(first_frame_index),
+ self.FOOTER_MAGIC_NUMBER,
+ version,
+ )
+ else:
+ 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,
+ 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
+ @staticmethod
+ def _keyframe_index(blob, frame):
+ descriptor = frame.keyframe_index_descriptor
+ if descriptor is None:
+ return b''
+ mapping = Blob.from_descriptor(blob.uri_reader, descriptor).to_data()
Review Comment:
[P1] Avoid materializing unbounded keyframe indexes
This materializes the entire compressed index before the chunked validator
runs, and `_keyframe_indexes` retains every block until the writer closes.
`BlobRef.to_data()` issues a single read for the descriptor length, while
rolling is checked only after `add_element`; therefore a valid or crafted large
index can allocate beyond the worker heap despite the 64 KiB decompression
chunks. The Java path has the same behavior in `VideoFormatWriter.java`. Please
validate from a bounded stream into spillable storage, or enforce explicit
per-index and cumulative limits, instead of materializing unbounded bytes.
--
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]
