@@ -147,7 +147,7 @@ jobs:
               python -c "import rosbags; print('rosbags installed')"
             fi
             if [[ "${{ matrix.python-version }}" == "3.10" ]]; then
-              python -m pip install './paimon-python[lerobot]'
+              python -m pip install './paimon-python[lerobot,video]'
               python -c "import datasets, lerobot; print('datasets', datasets.__version__, 'lerobot', lerobot.__version__)"
             fi
 
 
@@ -71,6 +71,33 @@ frames.add_videos([
 ])
 ```
 
+To encode an ordered sequence of image bytes or Pillow images before storing
+it, use Python 3.10 or newer, install the optional video dependencies, and call
+`add_images_as_video`:
+
+```shell
+pip install 'pypaimon[video]'
+```
+
+```python
+frames.add_images_as_video(
+    image_frames,
+    episode_44_rows,
+    fps=30,
+    codec="libx264",
+    pixel_format="yuv420p",
+    gop_size=2,
+    codec_options={"crf": "18", "preset": "fast"},
+)
+```
+
+The image and row counts must match, and every image must have the same
+dimensions. The container is MP4; codec-specific options are passed to PyAV.
+Encoding may be lossy and the original image payloads are not retained. A
+smaller GOP improves random frame access but usually increases file size.
+After encoding, Paimon copies the MP4 bytes into `.video` storage without
+transcoding them again.
+
 Each item may also be `(video, frame_rows, first_frame)`. Frame ordinals are
 generated consecutively from `first_frame`. If application semantics require
 PTS, wall-clock time, or a non-unit sampling map, retain that value in a normal
 
@@ -133,6 +133,54 @@ def add_video(
             [(video, frames, first_frame)], video_column=video_column
         )
 
+    def add_images_as_video(
+            self,
+            images,
+            frames,
+            *,
+            fps,
+            codec,
+            pixel_format,
+            gop_size,
+            codec_options=None,
+            video_column=None):
+        """Encode ordered images as one MP4 and append its logical frame rows.
+
+        Images may be encoded image bytes or Pillow images. ``frames`` supplies
+        every table column except ``video_column`` and must contain one row per
+        image. Encoding finishes before the Paimon write starts.
+        """
+        import os
+        import tempfile
+
+        from pypaimon.multimodal.video import _encode_images_to_video
+
+        column = self._resolve_video_frame_column(video_column)
+        target_schema = _target_schema(self.raw_table)
+        non_video_schema = pa.schema([
+            field for field in target_schema if field.name != column
+        ])
+        frame_table = _to_arrow_table(frames, non_video_schema)
+        with tempfile.TemporaryDirectory(prefix="pypaimon_video_") as directory:
+            video_path = os.path.join(directory, "video.mp4")
+            image_count = _encode_images_to_video(
+                images,
+                video_path,
+                fps=fps,
+                codec=codec,
+                pixel_format=pixel_format,
+                gop_size=gop_size,
+                codec_options=codec_options,
+            )
+            if image_count != frame_table.num_rows:
+                raise ValueError(
+                    "Image count %d does not match frame row count %d."
+                    % (image_count, frame_table.num_rows)
+                )
+            return self.add_video(
+                video_path, frame_table, video_column=column
+            )
+
     def add_videos(self, videos, *, video_column=None):
         """Append several encoded videos with one writer and one commit.
 
 
@@ -15,15 +15,109 @@
 # specific language governing permissions and limitations
 # under the License.
 
-"""PyTorch DataLoader helpers for descriptor-backed video frame rows."""
+"""Helpers for descriptor-backed video frame rows."""
 
+import io
 import os
 from collections import OrderedDict
 from collections.abc import Mapping
+from fractions import Fraction
+from itertools import chain
 
 from pypaimon.table.row.blob import Blob, VideoFrameDescriptor
 
 
+def _encode_images_to_video(
+        images,
+        output_path,
+        *,
+        fps,
+        codec,
+        pixel_format,
+        gop_size,
+        codec_options=None):
+    if isinstance(fps, bool) or not isinstance(fps, int) or fps <= 0:
+        raise ValueError("fps must be a positive int.")
+    if not isinstance(codec, str) or not codec:
+        raise ValueError("codec must be a non-empty string.")
+    if not isinstance(pixel_format, str) or not pixel_format:
+        raise ValueError("pixel_format must be a non-empty string.")
+    if isinstance(gop_size, bool) \
+            or not isinstance(gop_size, int) or gop_size <= 0:
+        raise ValueError("gop_size must be a positive int.")
+    if codec_options is not None and not isinstance(codec_options, Mapping):
+        raise ValueError("codec_options must be a mapping or None.")
+
+    try:
+        import av
+        from PIL import Image
+    except ImportError as error:
+        raise ImportError(
+            "Image-to-video encoding requires PyAV and Pillow; install "
+            "pypaimon[video]."
+        ) from error
+
+    iterator = iter(images)
+    try:
+        first = next(iterator)
+    except StopIteration as error:
+        raise ValueError("images must contain at least one frame.") from error
+
+    def open_image(value):
+        if isinstance(value, (bytes, bytearray, memoryview)):
+            image = Image.open(io.BytesIO(bytes(value)))
+        elif isinstance(value, Image.Image):
+            image = value
+        else:
+            raise TypeError(
+                "images must contain encoded image bytes or Pillow images."
+            )
+        image.load()
+        return image.convert("RGB")
+
+    first = open_image(first)
+    width, height = first.size
+    time_base = Fraction(1, fps)
+    count = 0
+    try:
+        with av.open(output_path, mode="w", format="mp4") as container:
+            stream = container.add_stream(
+                codec,
+                rate=fps,
+                options={
+                    str(key): str(value)
+                    for key, value in (codec_options or {}).items()
+                },
+            )
+            stream.width = width
+            stream.height = height
+            stream.pix_fmt = pixel_format
+            stream.gop_size = gop_size
+            stream.time_base = time_base
+            decoded = chain(
+                (first,), (open_image(value) for value in iterator)
+            )
+            for count, image in enumerate(decoded, start=1):
+                if image.size != (width, height):
+                    raise ValueError(
+                        "All images must have the same dimensions."
+                    )
+                frame = av.VideoFrame.from_image(image)
+                frame.pts = count - 1
+                frame.time_base = time_base
+                for packet in stream.encode(frame):
+                    container.mux(packet)
+            for packet in stream.encode():
+                container.mux(packet)
+    except BaseException:
+        try:
+            os.remove(output_path)
+        except FileNotFoundError:
+            pass
+        raise
+    return count
+
+
 class VideoFrameCollator:
     """Decode frame rows in a DataLoader worker while reusing video sessions.
 
 
@@ -161,6 +161,104 @@ def test_add_video_embeds_frame_ordinals_outside_normal_data(self):
         _, bodies = table.scan().read_blobs("video", parallelism=2)
         self.assertEqual([video_bytes] * 3, bodies["video"])
 
+    def test_add_images_as_video_encodes_and_stores_frame_rows(self):
+        table = self.conn.create_table(
+            "image_video_frames",
+            schema=_schema({
+                "episode_id": pa.int64(),
+                "video": pa.large_binary(),
+            }),
+            options=dict(_PARQUET_OPTIONS, **{
+                "video-frame-field": "video",
+                "blob-as-descriptor": "true",
+            }),
+        )
+        calls = []
+
+        def encode(images, output_path, **options):
+            images = list(images)
+            calls.append((images, options))
+            with open(output_path, "wb") as output:
+                output.write(b"encoded-video")
+            return len(images)
+
+        with patch(
+                "pypaimon.multimodal.video._encode_images_to_video",
+                encode):
+            table.add_images_as_video(
+                [b"frame-0", b"frame-1"],
+                [{"episode_id": 42}, {"episode_id": 42}],
+                fps=30,
+                codec="mpeg4",
+                pixel_format="yuv420p",
+                gop_size=2,
+                codec_options={"qscale": "3"},
+            )
+
+        self.assertEqual(
+            [
+                (
+                    [b"frame-0", b"frame-1"],
+                    {
+                        "fps": 30,
+                        "codec": "mpeg4",
+                        "pixel_format": "yuv420p",
+                        "gop_size": 2,
+                        "codec_options": {"qscale": "3"},
+                    },
+                )
+            ],
+            calls,
+        )
+        rows = table.scan().select(["episode_id", "video"]).to_list()
+        descriptors = [
+            pmm.VideoFrameDescriptor.deserialize(row["video"])
+            for row in rows
+        ]
+        self.assertEqual([0, 1], [value.frame_index for value in descriptors])
+        _, bodies = table.scan().read_blobs("video")
+        self.assertEqual([b"encoded-video"] * 2, bodies["video"])
+
+    def test_add_images_as_video_rejects_frame_count_mismatch(self):
+        table = self.conn.create_table(
+            "mismatched_image_video_frames",
+            schema=_schema({
+                "episode_id": pa.int64(),
+                "video": pa.large_binary(),
+            }),
+            options=dict(_PARQUET_OPTIONS, **{
+                "video-frame-field": "video",
+                "blob-as-descriptor": "true",
+            }),
+        )
+        output_paths = []
+
+        def encode(unused_images, output_path, **unused_options):
+            output_paths.append(output_path)
+            with open(output_path, "wb") as output:
+                output.write(b"one-frame-video")
+            return 1
+
+        with patch(
+                "pypaimon.multimodal.video._encode_images_to_video",
+                encode):
+            with self.assertRaisesRegex(
+                    ValueError,
+                    "Image count 1 does not match frame row count 2"):
+                table.add_images_as_video(
+                    [b"frame-0"],
+                    [{"episode_id": 42}, {"episode_id": 42}],
+                    fps=30,
+                    codec="mpeg4",
+                    pixel_format="yuv420p",
+                    gop_size=2,
+                )
+
+        self.assertFalse(os.path.exists(output_paths[0]))
+        self.assertIsNone(
+            table.raw_table.snapshot_manager().get_latest_snapshot()
+        )
+
     def test_add_videos_packs_multiple_videos_in_one_commit(self):
         from pypaimon.table.row.blob import Blob, VideoFrameDescriptor