[core][python] Support shared BLOB payloads across rows

docs/docs/concepts/spec/fileformat.md

@@ -930,4 +930,39 @@ Limitations:
 2. BLOB format does not support predicate pushdown.
 3. Statistics collection is not supported for BLOB columns.
 
+### Shared BLOB
+
+Shared BLOB is an independent, versioned format with the `.shared-blob` extension. It reuses the
+ordinary scalar BLOB entry encoding for its physical data region, but replaces the positional
+footer with two indexes:
+
+```
++-----------------------+
+| Physical Blob Entry 1 |  Ordinary scalar BLOB entry encoding
++-----------------------+
+| Physical Blob Entry 2 |
++-----------------------+
+| ...                   |
++-----------------------+
+| Physical Length Index |  Delta-Varint compressed record lengths
++-----------------------+
+| Row Reference Index   |  Delta-Varint compressed physical ordinals
++-----------------------+
+| Physical Index Length |  4 bytes (Little Endian)
+| Row Index Length      |  4 bytes (Little Endian)
+| Magic Number          |  4 bytes (0x4C424853, Little Endian)
+| Version               |  1 byte
++-----------------------+
+```
+
+Each non-negative row reference is an ordinal in the physical length index. Multiple logical rows
+may contain the same ordinal and therefore resolve to the same physical payload. `-1` represents a
+null field and `-2` represents a data-evolution placeholder. Readers validate footer bounds,
+physical record lengths, the complete data-region length, and every row-reference ordinal before
+returning records.
+
+Shared BLOB currently supports one scalar BLOB field per file. Physical sharing uses exact input
+`BlobDescriptor` identity and is file-local; the format does not contain cross-file references.
+Ordinary `.blob` files keep their existing version and layout.
+
 For usage details, configuration options, and examples, see [Blob Type](../../multimodal-table/blob).

docs/docs/multimodal-table/blob.mdx

@@ -49,6 +49,12 @@ When you define a table with a Blob column, Paimon automatically separates the s
 1. **Normal Data Files** (e.g., `.parquet`, `.orc`): Store regular columns (INT, STRING, etc.)
 2. **Blob Data Files** (`.blob`): Store the actual blob data
 
+For append-only workloads in which many rows refer to the same large object, a scalar BLOB field
+can instead use **shared BLOB files** (`.shared-blob`). A shared file stores one physical payload
+for each exact `BlobDescriptor` and a compact row-reference index. For example, video-frame rows
+can all refer to one MP4 payload while `frame_index` or `timestamp` in the normal data file tells
+the decoder which frame to retrieve.
+
 For example, given a table with schema `(id INT, name STRING, picture BLOB)`:
 
 ```
@@ -104,6 +110,13 @@ Flink, Spark, and Python.
       </tr>
     </thead>
     <tbody>
+    <tr>
+      <td><h5>blob-shared-field</h5></td>
+      <td>No</td>
+      <td style={{wordWrap: "break-word"}}>-</td>
+      <td>String</td>
+      <td>Names one scalar BLOB field whose logical rows may share payloads in <code>.shared-blob</code> files. This first version supports append-only tables and exact descriptor-backed input only.</td>
+    </tr>
     <tr>
       <td><h5>blob-as-descriptor</h5></td>
       <td>No</td>
@@ -267,6 +280,65 @@ schema = Schema.from_pyarrow_schema(
 
 </Tabs>
 
+## Sharing One Video Across Frame Rows
+
+Use `blob-shared-field` when the table has one row per frame, but the source and storage unit is a
+whole video. The option also marks a SQL `BYTES` or `BINARY` column as a BLOB column, so a new
+comment directive and a change to the ordinary `.blob` format are not required.
+
+```sql
+CREATE TABLE video_frames (
+    episode_id BIGINT,
+    frame_index INT,
+    frame_timestamp DOUBLE,
+    video BYTES
+) WITH (
+    'row-tracking.enabled' = 'true',
+    'data-evolution.enabled' = 'true',
+    'blob-shared-field' = 'video'
+);
+
+-- Every row for this video uses the same serialized BlobDescriptor.
+INSERT INTO video_frames
+SELECT episode_id, frame_index, frame_timestamp,
+       sys.path_to_descriptor('file:///data/episode-42.mp4')
+FROM frame_metadata;
+```
+
+The normal data file holds `episode_id`, `frame_index`, and `frame_timestamp`. Its aligned
+`.shared-blob` file has two indexes: physical record lengths and logical-row-to-physical-record
+references. Several logical positions can therefore resolve to the same `(file, offset, length)`
+descriptor. Paimon does not parse MP4 containers or store an MP4 frame offset itself; the frame
+columns are interpreted by the application or video decoder.
+
+Deduplication uses exact descriptor identity (descriptor version, URI, offset, and length), not a
+content hash. It is local to each `.shared-blob` file: files are self-contained and never point at
+payloads owned by another data file. After a size or row-count target is reached, the rolling
+boundary is delayed until a contiguous descriptor group ends, so one video's consecutive frame
+rows stay together. This makes the target size soft for large videos. If the same descriptor is
+written again after a file has already rolled or a commit/checkpoint has closed the writer, another
+physical copy may be stored.
+
+Compaction reads descriptors, writes a new self-contained `.shared-blob` file, and rebuilds both
+indexes, so surviving rows again share one physical payload per exact descriptor in the output.
+The ordinary `.blob` format and its behavior are unchanged, and ordinary BLOB fields can coexist
+with the shared field in the same table.
+
+Current restrictions:
+
+- Append-only tables only; primary-key tables continue to use managed BLOB storage.
+- At most one `blob-shared-field` per table.
+- The field must be a scalar `BLOB`; `ARRAY<BLOB>` and `MAP<K, BLOB>` continue to use `.blob`.
+- Non-null writes must be exact descriptor-backed `BlobRef` values. Inline `BlobData` is rejected,
+  because it has no stable identity for sharing.
+- A `BlobConsumer` callback is not supported for the shared field.
+
+Set `blob-as-descriptor=true` when the consumer can decode the referenced MP4 itself. This avoids
+materializing the same large payload separately for every selected frame row.
+
+For Python ingestion and PyTorch `DataLoader` usage, including decoder-session reuse in workers,
+see [PyPaimon Multimodal API: Video Frames Sharing One MP4](../pypaimon/multimodal-api#video-frames-sharing-one-mp4).
+
 The comment directive format is `__DIRECTIVE; optional user comment`. Paimon converts `BYTES`/`BINARY` to `BLOB`, `ARRAY<BYTES>`/`ARRAY<BINARY>` to `ARRAY<BLOB>`, and `MAP<K, BYTES>`/`MAP<K, BINARY>` to `MAP<K, BLOB>`. It registers the field in the corresponding option and stores the text after `;` as the column's real comment.
 
 Supported directives:

docs/docs/primary-key-table/blob-storage.md

@@ -37,6 +37,11 @@ This mode stores:
 
 For general BLOB concepts and read options, see [BLOB Storage](../multimodal-table/blob).
 
+The append-only `blob-shared-field` mode is deliberately separate from primary-key managed BLOB
+storage. It writes self-contained `.shared-blob` files whose logical rows can share a physical
+payload, and is not supported on primary-key tables. Enabling it does not change `.managed.blob`
+packs, `.blobref` ownership, or their garbage-collection behavior.
+
 ## Create a Table
 
 Use `blob-field` to mark scalar, array, or map fields whose payloads should be stored in managed BLOB files.

docs/docs/pypaimon/blob.md

@@ -76,6 +76,12 @@ write_builder.new_commit().commit(writer.prepare_commit())
 writer.close()
 ```
 
+For frame tables in which many logical rows reference one MP4, configure
+`blob-shared-field` and write descriptor-backed `Blob` or `BlobDescriptor`
+values instead of raw bytes. The high-level multimodal API also provides
+single-commit `add_batches` ingestion. See
+[Video Frames Sharing One MP4](./multimodal-api#video-frames-sharing-one-mp4).
+
 ## Reading Blob Data
 
 ### Batch reading (recommended)
@@ -169,3 +175,5 @@ header). This mirrors Java's `Blob.fromBytes(...)`.
   SQL/Java API
 - [Data Evolution](./data-evolution) — required for
   blob tables
+- [Multimodal video frames](./multimodal-api#video-frames-sharing-one-mp4) —
+  shared MP4 writing and PyTorch DataLoader decoding

docs/docs/pypaimon/multimodal-api.mdx

@@ -179,7 +179,10 @@ vector or blob types.
 
 `add` accepts `pyarrow.Table`, `pyarrow.RecordBatch`, a list of dictionaries, a
 dictionary of arrays, or a pandas DataFrame. Input columns are aligned and cast
-to the Paimon table schema before writing.
+to the Paimon table schema before writing. For a BLOB column, list, dictionary,
+and pandas inputs may also contain `Blob` or `BlobDescriptor` objects. A
+descriptor-backed `Blob` is copied through its stream; it is not first loaded
+into Python memory.
 
 ```python
 docs.add([
@@ -192,6 +195,189 @@ docs.add([
 ])
 ```
 
+## Video Frames Sharing One MP4
+
+For a frame table, keep the frame mapping in ordinary columns and store one
+shared BLOB descriptor in every row for the video. Set `blob-shared-field` to
+the video column when creating the table:
+
+```python
+import pyarrow as pa
+import pypaimon.multimodal as pm
+
+frames = conn.create_table(
+    "video_frames",
+    schema=pa.schema([
+        pa.field("episode_id", pa.int64()),
+        pa.field("frame_index", pa.int32()),
+        pa.field("frame_timestamp", pa.float64()),
+        pa.field("video", pa.large_binary()),
+    ]),
+    options={
+        "blob-shared-field": "video",
+    },
+)
+```
+
+Pass the same descriptor-backed `Blob` for consecutive rows belonging to one
+video. `Blob.from_local` creates a reference; `add` serializes that reference
+and the shared writer copies the MP4 once for the output file:
+
+```python
+video = pm.Blob.from_local("/data/episode-42.mp4")
+fps = 30.0
+
+frames.add([
+    {
+        "episode_id": 42,
+        "frame_index": frame_index,
+        "frame_timestamp": frame_index / fps,
+        "video": video,
+    }
+    for frame_index in range(1800)
+])
+```
+
+For larger ingestion jobs, `add_batches` accepts an iterable of any supported
+input form and keeps one writer plus one commit for the whole iterable:
+
+```python
+frames.add_batches(frame_batches)
+```
+
+This is the preferred path when one video's contiguous rows cross Arrow batch
+boundaries. An empty iterable is a no-op and does not create a snapshot. If an
+error occurs before commit begins, the open writer is aborted; a commit error
+has the usual unknown-result semantics and is not retried.
+
+Remote sources can use an explicit descriptor. Its URI must be readable with
+the table's configured `FileIO` credentials:
+
+```python
+video = pm.BlobDescriptor(
+    "oss://source-bucket/episode-43.mp4",
+    offset=0,
+    length=video_size,
+)
+frames.add([
+    {
+        "episode_id": 43,
+        "frame_index": frame_index,
+        "frame_timestamp": frame_index / fps,
+        "video": video,
+    }
+    for frame_index in range(frame_count)
+])
+```
+
+Keep each video's rows contiguous in the input. Deduplication is by exact
+descriptor identity within one `.shared-blob` file, and the writer delays a
+pending roll until the current contiguous descriptor group ends. A later call
+to `add`, a checkpoint, or a prior file roll can therefore create another
+physical copy of the same source video.
+
+### Read with PyTorch DataLoader
+
+Install the PyTorch extra, then convert the multimodal scan directly. Unlike a
+regular BLOB materialization, `ScanQuery.to_torch` always returns serialized
+descriptors for BLOB columns. DataLoader workers receive Paimon splits and open
+the selected video ranges themselves:
+
+```shell
+pip install 'pypaimon[torch]'
+```
+
+```python
+from torch.utils.data import DataLoader
+
+dataset = (
+    frames.scan()
+    .select(["episode_id", "frame_index", "frame_timestamp", "video"])
+    .to_torch(streaming=True)
+)
+
+loader = DataLoader(
+    dataset,
+    batch_size=32,
+    num_workers=4,
+    shuffle=False,
+)
+```
+
+The `video` value above is descriptor bytes, not repeated MP4 bytes. If the
+training step needs decoded frames, use `VideoFrameCollator`. It owns a bounded,
+process-local LRU cache keyed by exact descriptor, so consecutive frame rows
+reuse one decoder session in each DataLoader worker. The decoder is an adapter:
+`decoder_factory(stream)` opens any codec library, and
+`decode_fn(decoder, row)` locates the frame from `frame_index` or
+`frame_timestamp`.
+
+```python
+import av
+import torch
+import pypaimon.multimodal as pm
+from torch.utils.data import DataLoader
+
+
+class SequentialPyAvDecoder:
+    def __init__(self, stream):
+        self.container = av.open(stream)
+        self._reset()
+
+    def _reset(self):
+        self.frames = iter(self.container.decode(video=0))
+        self.next_index = 0
+
+    def frame(self, index):
+        if index < self.next_index:
+            self.container.seek(0)
+            self._reset()
+        while self.next_index <= index:
+            frame = next(self.frames)
+            self.next_index += 1
+        array = frame.to_ndarray(format="rgb24")
+        return torch.from_numpy(array).permute(2, 0, 1)
+
+    def close(self):
+        self.container.close()
+
+
+def decode_frame(decoder, row):
+    return decoder.frame(row["frame_index"])
+
+
+collator = pm.VideoFrameCollator(
+    frames,
+    video_column="video",
+    decoder_factory=SequentialPyAvDecoder,
+    decode_fn=decode_frame,
+    output_column="frame",
+    max_open_videos=4,
+)
+
+loader = DataLoader(
+    dataset,
+    batch_size=32,
+    num_workers=4,
+    shuffle=False,
+    collate_fn=collator,
+    persistent_workers=True,
+)
+
+for batch in loader:
+    # batch["frame"] is [N, C, H, W].
+    train(batch["frame"], batch["episode_id"])
+```
+
+The example decoder additionally requires `pip install av`; PyAV is not a
+PyPaimon dependency.
+
+Define the decoder class and function at module scope when DataLoader uses the
+`spawn` multiprocessing method. For sequential video decoding, keep
+`shuffle=False`; row-buffer shuffle can turn monotonic frame access into seeks
+and require more open decoder sessions. Paimon's worker sharding is split-level,
+so it does not duplicate a split across workers.
+
 ## Load HDF5
 
 `MultimodalConnection.load_from_hdf5` streams one or more local or remote HDF5

docs/docs/pypaimon/pytorch.md

@@ -88,6 +88,25 @@ disable a second batching step. Batch streaming does not support `shuffle=True`.
 Numeric tensors may share read-only Arrow buffers; clone them before in-place
 mutation. Batch formats currently require `prefetch_concurrency=1`.
 
+## Descriptor-backed video frames
+
+For a multimodal frame table, use the higher-level scan API:
+
+```python
+dataset = (
+    frames.scan()
+    .select(["episode_id", "frame_index", "video"])
+    .to_torch(streaming=True)
+)
+```
+
+This path forces BLOB columns to remain serialized descriptors instead of
+materializing the referenced MP4 once per frame row. Use
+`pypaimon.multimodal.VideoFrameCollator` as the DataLoader `collate_fn` to open
+descriptor ranges and cache decoder sessions per worker. See
+[Multimodal API: Video Frames Sharing One MP4](multimodal-api#video-frames-sharing-one-mp4)
+for the write path and a complete decoder example.
+
 ## File Format Metadata Cache
 
 Reusable PyArrow Dataset metadata is cached across reads. Configure its estimated