sundapeng commented on code in PR #10905:
URL: https://github.com/apache/paimon/pull/10905#discussion_r4430786028
##########
docs/docs/concepts/spec/fileformat.md:
##########
@@ -441,25 +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.
-Keyframe-index v1 is little endian; byte offsets are relative to the video
payload.
+Keyframe-index v1 is little endian. Metadata and packet offsets are relative
to the video payload.
| Field | Size |
| --- | ---: |
| Version (`1`) | 1 byte |
| Magic (`0x564944454F4B4649`) | 8 bytes |
| Metadata-range count | 4 bytes |
| Keyframe count | 8 bytes |
+| Page codec (`0`: uncompressed, `1`: zlib) | 1 byte |
| Metadata `(offset, length)` pairs | 16 bytes each |
+| Page directory | 68 bytes per page |
| Keyframe pages | Variable |
-Each page holds 4,096 uncompressed `(frame ordinal, PTS, packet position)`
int64 triples
-and their 4-byte CRC32; the last page may be shorter. Page `p` starts at
-`21 + 16 * metadata_count + p * 98308`.
+Each page holds up to 4,096 `(frame ordinal, PTS, packet position)` int64
triples,
Review Comment:
This says "up to 4,096", but both readers assume every page except the last
has exactly 4,096 entries: they compute the entry count as `min(4096, count - p
* 4096)` and locate the page by `divmod(index, 4096)`. So a writer which
follows the spec and writes variable-size pages (e.g. a future Rust writer)
will be rejected.
Maybe say "exactly 4,096 except the last page" here, or store `page_entries`
in the header. The directory already has the first ordinal of each page, so
variable-size pages are not hard to support.
##########
paimon-python/pypaimon/table/row/video_keyframe_index.py:
##########
@@ -244,34 +384,55 @@ def __exit__(self, *args):
def __reduce__(self):
return (type(self).deserialize, (self.serialize(),))
- def write_to(self, output):
- output.write(self.HEADER.pack(
- self.VERSION, self.MAGIC, len(self.metadata_ranges),
len(self.keyframes)))
- for value in self.metadata_ranges:
- output.write(self.METADATA_RANGE.pack(*value))
- for page in range((len(self.keyframes) + _PagedKeyframes.PAGE_ENTRIES
- 1) // _PagedKeyframes.PAGE_ENTRIES):
- data = self.keyframes._page(page)
- output.write(data)
- output.write(struct.pack('<I', zlib.crc32(data)))
+ def preceding(self, ordinal):
+ if isinstance(self.keyframes, _IndexedKeyframes):
+ return self.keyframes.preceding(ordinal)
+ return bisect.bisect_right(self.ordinals, ordinal) - 1
+
+ def write_to(self, output, compression='zlib'):
+ if compression not in ('none', 'zlib'):
+ raise ValueError("Video index compression must be 'none' or
'zlib'.")
+ codec = int(compression == 'zlib')
+ count = len(self.keyframes)
+ pages = (count - 1) // _PagedKeyframes.PAGE_ENTRIES + 1
+ directory = _IndexedKeyframes.DIRECTORY
+ offset = self.HEADER.size + len(self.metadata_ranges) * 16 + pages *
directory.size
+ # Spool variable-size pages; neither the directory nor encoded data
grows in memory.
+ with tempfile.TemporaryFile() as records, tempfile.TemporaryFile() as
payload:
+ for page in range(pages):
+ raw = self.keyframes._page(page)
+ data = zlib.compress(raw, 1) if codec else raw
+ record = directory.pack(offset, len(data), zlib.crc32(raw),
+ *self.ENTRY.unpack_from(raw),
+ *self.ENTRY.unpack_from(raw, len(raw)
- self.ENTRY.size), 0)
+ records.write(record[:-4] + struct.pack('<I',
zlib.crc32(record[:-4])))
+ payload.write(data)
+ offset += len(data)
+ output.write(self.HEADER.pack(self.VERSION, self.MAGIC,
len(self.metadata_ranges), count, codec))
+ for value in self.metadata_ranges:
+ output.write(self.METADATA_RANGE.pack(*value))
+ for source in (records, payload):
+ source.seek(0)
+ shutil.copyfileobj(source, output, 64 * 1024)
- def serialize(self):
+ def serialize(self, compression='zlib'):
output = io.BytesIO()
- self.write_to(output)
+ self.write_to(output, compression)
return output.getvalue()
@classmethod
def open(cls, read_at, length, payload_length=None):
if length < cls.HEADER.size:
raise ValueError("Invalid video keyframe index length.")
- prefix = read_at(0, min(length, cls.HEADER.size +
_PagedKeyframes.PAGE_BYTES + 4))
- if len(prefix) != min(length, cls.HEADER.size +
_PagedKeyframes.PAGE_BYTES + 4):
+ # Small indexes use one request; larger indexes retain only a bounded
prefix.
+ prefix_size = length if length <= _IndexedKeyframes.EAGER_BYTES else
64 * 1024
Review Comment:
Here the whole index is read when the encoded length <= 2 MiB, but
`load_small()` only goes eager when the decoded size `count * 24` <= 2 MiB. For
compressed indexes in between, the whole encoded index stays in `prefix`, and
pages are still decoded into the 1 MiB page cache. With 200K keyframes I see
1.58 MB held + 0.97 MB page cache, and it's not eager.
So the memory per open index goes from ~1.1 MB to ~2.5 MB, multiplied by
`max_open_videos` (16 by default) and the number of DataLoader workers. Maybe
use one threshold on the decoded size, or document this case.
##########
paimon-format/src/main/java/org/apache/paimon/format/blob/VideoKeyframeIndex.java:
##########
@@ -71,42 +78,130 @@ static void validate(InputStream source, long length, long
payloadLength) throws
}
previousEnd = start + size;
}
- 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);
- 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");
+ Path path = Files.createTempFile("paimon-video-directory-", ".tmp");
Review Comment:
This creates a temp file for every index only to read the directory twice.
The caller's spool in `VideoFormatWriter` is already a `RandomAccessFile`, so
maybe we can pass a seekable input here, or just keep the directory in memory.
It is bounded by the index length, and only about 16.6 KB for 1M keyframes.
--
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]
