[python] Import videos without explicit frame metadata

docs/docs/pypaimon/video.md

@@ -24,7 +24,48 @@ under the License.
 
 # Video Frames
 
-Store complete encoded videos and read their logical frame rows for training. First create a multimodal `conn` as shown in [Multimodal Tables](./multimodal-tables#connect). Provide your own video files and frame metadata for the examples below.
+Store encoded videos in `.video` and read individual frames for training.
+
+## Import a Video
+
+With Python 3.10 or newer, install `pypaimon[video]`:
+
+```shell
+pip install 'pypaimon[video]'
+```
+
+```python
+import pyarrow as pa
+import pypaimon.multimodal as pm
+
+conn = pm.connect(options={"warehouse": "file:///tmp/video_lake"})
+frames = conn.create_table(
+    "auto_frames",
+    schema=pa.schema([
+        pa.field("frame_index", pa.int64()),
+        pa.field("timestamp", pa.float64()),
+        pa.field("episode_id", pa.int64()),
+        pa.field("video", pa.large_binary()),
+    ]),
+    options={"video-frame-field": "video"},
+)
+frames.add_video("/data/example.mp4")
+frames.add_video("oss://source-bucket/episode-42.mp4", metadata={"episode_id": 42})
+# Raw HEVC: explicitly supply a fixed frame rate.
+frames.add_video("oss://source-bucket/episode-42.h265", source_fps=30)
+```
+
+Generates zero-based `frame_index` and `timestamp` in seconds from the first
+frame. `metadata` supplies per-video constants. Containerized videos retain
+original bytes and variable frame timing; require one decodable video stream
+with strictly increasing PTS.
+
+Raw HEVC is remuxed to MP4 without re-encoding. `source_fps` generates fixed-rate
+timestamps, not original capture times. Temporary raw/MP4 files are cleaned up
+after import, including failures. Omit `source_fps` for containerized videos.
+
+Configure remote credentials on the connection and keep the source unchanged
+during import. To supply frame metadata yourself, pass `frames` as shown below.
 
 ## Video Frame Storage
 

paimon-python/pypaimon/common/file_io.py

@@ -31,6 +31,15 @@
 _LOG = logging.getLogger(__name__)
 
 
+def _is_local_path(path):
+    from urllib.parse import urlparse
+
+    uri = urlparse(path)
+    # urlparse treats a Windows drive letter as a URI scheme.
+    return not uri.scheme or uri.scheme == "file" or (
+        os.name == "nt" and len(uri.scheme) == 1 and not uri.netloc)
+
+
 def normalize_naive_datetimes(value):
     """Recursively attach UTC to naive datetimes, including ones nested inside
     ROW (dict) and ARRAY/MAP (list/tuple) values.
@@ -670,9 +679,7 @@ def get(path: str, catalog_options: Optional[Options] = None) -> 'FileIO':
         uri = urlparse(path)
         scheme = uri.scheme
 
-        # urlparse treats a Windows drive letter as a URI scheme.
-        if not scheme or scheme == "file" or (
-                _os.name == "nt" and len(scheme) == 1 and not uri.netloc):
+        if _is_local_path(path):
             from pypaimon.filesystem.local_file_io import LocalFileIO
             return LocalFileIO(path, catalog_options)
 

paimon-python/pypaimon/filesystem/local_file_io.py

@@ -69,7 +69,8 @@ def __init__(self, path: str = None, catalog_options: Optional[Options] = None):
     def create():
         return LocalFileIO()
 
-    def _to_file(self, path: str) -> Path:
+    @staticmethod
+    def _to_file(path: str) -> Path:
         parsed = urlparse(path)
 
         if parsed.scheme and len(parsed.scheme) == 1 and not parsed.netloc:

paimon-python/pypaimon/multimodal/table.py

@@ -124,17 +124,116 @@ def add(self, data):
     def add_video(
             self,
             video,
-            frames,
+            frames=None,
             *,
             video_column=None,
-            first_frame=0):
-        """Append logical frame rows backed by one complete encoded video.
-
-        ``frames`` supplies every table column except the configured
-        ``video-frame-field``. Frame ordinals are generated from
-        ``first_frame`` and stored in the video descriptor, not in the normal
-        data file.
+            first_frame=0,
+            metadata=None,
+            source_fps=None):
+        """Append frame rows sharing one encoded video.
+
+        ``frames`` supplies non-video columns; descriptor ordinals start at
+        ``first_frame``. Omit it to generate ``frame_index`` (int64) and
+        ``timestamp`` (float64 seconds relative to the first frame).
+        ``metadata`` supplies constant fields. Auto import requires PyAV
+        and one video stream. For raw HEVC, ``source_fps`` sets fixed-rate
+        timing and remuxes to MP4 without re-encoding.
         """
+        if frames is None:
+            import os
+            import stat
+            import tempfile
+            from contextlib import ExitStack
+            from urllib.parse import urlparse
+
+            from pypaimon.table.row.blob import Blob, BlobDescriptor
+            from pypaimon.common.file_io import _is_local_path
+            from pypaimon.filesystem.local_file_io import LocalFileIO
+            from pypaimon.multimodal.video import (
+                _inspect_video_frames, _require_video_av, _raw_video_time_base, _remux_raw_hevc,
+            )
+
+            time_base = _raw_video_time_base(source_fps) if source_fps is not None else None
+
+            if isinstance(first_frame, bool) or not isinstance(first_frame, int) or first_frame != 0:
+                raise ValueError("first_frame must be 0 when frames is omitted.")
+            column = self._resolve_video_frame_column(video_column)
+            schema = _target_schema(self.raw_table)
+            generated = {"frame_index": pa.int64(), "timestamp": pa.float64()}
+            for name, dtype in generated.items():
+                if name == column or name not in schema.names or schema.field(name).type != dtype:
+                    raise ValueError(
+                        "Automatic video import requires column %r of type %s."
+                        % (name, dtype))
+            if metadata is not None and not isinstance(metadata, Mapping):
+                raise ValueError("metadata must be a mapping or None.")
+            metadata = dict(metadata or {})
+            allowed = set(schema.names) - set(generated) - set(self.raw_table.options.video_frame_fields())
+            if set(metadata) - allowed:
+                raise ValueError("metadata must contain only non-video, non-generated columns.")
+            fields = [field for field in schema if field.name not in generated and field.name != column]
+            constants = _to_arrow_table([metadata], pa.schema(fields)) if fields else None
+            for field in fields:
+                if not field.nullable and constants[field.name].null_count:
+                    raise ValueError("Missing required metadata: %s." % field.name)
+            if isinstance(video, os.PathLike):
+                video = os.fspath(video)
+            if isinstance(video, str):
+                if not urlparse(video).scheme:
+                    video = os.path.abspath(video)
+                video = BlobDescriptor(video, 0, -1)
+            if not isinstance(video, BlobDescriptor):
+                raise ValueError("Automatic video import requires a path, URI, or BlobDescriptor.")
+            _require_video_av()
+            local_path = LocalFileIO._to_file(video.uri) if _is_local_path(video.uri) else None
+
+            def source_state():
+                if local_path is None:
+                    return None
+                info = os.stat(local_path)
+                if not stat.S_ISREG(info.st_mode):
+                    raise ValueError("Automatic video import requires a regular local file.")
+                return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
+
+            with ExitStack() as stack:
+                original = source_state()
+                # Reuse the writer's source credentials.
+                reader = self.raw_table.file_io.uri_reader_factory.create(video.uri)
+                with Blob.from_descriptor(reader, video).new_input_stream() as stream:
+                    if time_base is not None:
+                        directory = stack.enter_context(tempfile.TemporaryDirectory(prefix="paimon_hevc_"))
+                        path = os.path.join(directory, "video.mp4")
+                        count = _remux_raw_hevc(stream, path, time_base)
+                        video = BlobDescriptor(path, 0, os.path.getsize(path))
+                        timestamps = _inspect_video_frames(path)
+                        if len(timestamps) != count:
+                            raise ValueError("HEVC frame count changed during remux.")
+                    else:
+                        timestamps = _inspect_video_frames(stream)
+
+                def check_source():
+                    if source_state() != original:
+                        raise ValueError("Video source changed during import.")
+
+                check_source()
+
+                def batches():
+                    for start in range(0, len(timestamps), 1024):
+                        end = min(start + 1024, len(timestamps))
+                        values = {
+                            "frame_index": pa.array(range(start, end), type=pa.int64()),
+                            "timestamp": pa.array(timestamps[start:end], type=pa.float64()),
+                        }
+                        if constants is not None:
+                            repeated = constants.take(pa.array([0] * (end - start), type=pa.int32()))
+                            values.update({name: repeated[name] for name in repeated.column_names})
+                        yield _video_frame_table(video, pa.table(values), column, start, schema)
+                    # Check again before commit.
+                    check_source()
+
+                return self.add_batches(batches())
+        elif metadata is not None or source_fps is not None:
+            raise ValueError("metadata and source_fps are only supported when frames is omitted.")
         return self.add_videos(
             [(video, frames, first_frame)], video_column=video_column
         )

paimon-python/pypaimon/multimodal/video.py

@@ -30,6 +30,131 @@
 from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex
 
 
+def _require_video_av():
+    """Check the optional decoder before opening a video source."""
+    try:
+        import av
+    except ImportError as error:
+        raise ImportError(
+            "Automatic video import requires PyAV; install pypaimon[video]."
+        ) from error
+    return av
+
+
+def _raw_video_time_base(fps):
+    from numbers import Real
+
+    if isinstance(fps, bool) or not isinstance(fps, Real):
+        raise ValueError("source_fps must be a positive finite number.")
+    try:
+        rate = Fraction(str(fps))
+        if rate <= 0 or max(rate.numerator, rate.denominator) > 2147483647:
+            raise ValueError()
+    except (ValueError, OverflowError):
+        raise ValueError("source_fps must be a positive finite representable frame rate.")
+    return 1 / rate
+
+
+def _remux_raw_hevc(source, output_path, time_base):
+    """Preserve HEVC packets and assign constant-rate presentation timestamps."""
+    from array import array
+    import shutil
+    import tempfile
+
+    av = _require_video_av()
+    # Snapshot to disk for both passes.
+    with tempfile.TemporaryFile() as raw:
+        shutil.copyfileobj(source, raw, length=1024 * 1024)
+        raw.seek(0)
+        ordinals = array("q")
+        frame_count = 0
+        delay = 0
+        with av.open(raw) as container:
+            if container.format.name != "hevc" or len(container.streams) != 1:
+                raise ValueError("source_fps is only supported for a raw H.265/HEVC stream.")
+            for packet in container.demux(video=0):
+                if packet.size:
+                    # Track packet IDs through B-frame reordering.
+                    packet.pts = len(ordinals)
+                    packet.dts = None
+                    packet.time_base = time_base
+                    ordinals.append(-1)
+                for frame in packet.decode():
+                    identity = frame.pts
+                    if frame.is_corrupt or identity is None or not 0 <= identity < len(ordinals) \
+                            or ordinals[identity] != -1:
+                        raise ValueError("Cannot determine raw HEVC frame order.")
+                    ordinals[identity] = frame_count
+                    delay = max(delay, identity - frame_count)
+                    frame_count += 1
+        if not frame_count or frame_count != len(ordinals):
+            raise ValueError("Raw HEVC must contain one decodable frame per packet.")
+        raw.seek(0)
+        with av.open(raw) as container, av.open(output_path, "w", format="mp4") as output:
+            if hasattr(output, "add_stream_from_template"):
+                stream = output.add_stream_from_template(container.streams.video[0], opaque=True)
+            else:
+                stream = output.add_stream(template=container.streams.video[0])
+            stream.time_base = time_base
+            output.start_encoding()
+            if (time_base / stream.time_base).denominator != 1:
+                raise ValueError("MP4 cannot represent source_fps exactly.")
+            packet_time_bases = {}
+            count = 0
+            for packet in container.demux(video=0):
+                if not packet.size:
+                    continue
+                # Keep PyAV 12's read-only duration; bound the AVRational denominator.
+                duration = packet.duration
+                if duration <= 0:
+                    raise ValueError("Raw HEVC packet has no duration.")
+                if duration not in packet_time_bases:
+                    exact = time_base / duration
+                    bounded = exact.limit_denominator(2147483647)
+                    error = abs(bounded - exact) * duration * (frame_count + delay + 1)
+                    # Every PTS/DTS and duration must round to the same MP4 tick.
+                    if bounded <= 0 or bounded.numerator > 2147483647 or error >= stream.time_base / 2:
+                        raise ValueError("Cannot represent raw HEVC timestamps exactly in MP4.")
+                    packet_time_bases[duration] = bounded
+                packet.pts = ordinals[count] * duration
+                packet.dts = (count - delay) * duration
+                packet.time_base = packet_time_bases[duration]
+                packet.stream = stream
+                output.mux(packet)
+                count += 1
+            if count != frame_count:
+                raise ValueError("Raw HEVC packet count changed during remux.")
+        return frame_count
+
+
+def _inspect_video_frames(source):
+    """Read presentation timestamps without retaining decoded images."""
+    av = _require_video_av()
+    from array import array
+
+    timestamps = array("d")
+    first_time = None
+    previous_time = None
+    with av.open(source) as container:
+        if len(container.streams.video) != 1:
+            raise ValueError("Automatic video import requires one video stream.")
+        for frame in container.decode(container.streams.video[0]):
+            if frame.is_corrupt:
+                raise ValueError("Video contains a corrupt frame.")
+            if frame.pts is None or frame.time_base is None:
+                raise ValueError("Video frame is missing a presentation timestamp.")
+            current_time = frame.pts * frame.time_base
+            if previous_time is not None and current_time <= previous_time:
+                raise ValueError("Video presentation timestamps must be strictly increasing.")
+            if first_time is None:
+                first_time = current_time
+            timestamps.append(float(current_time - first_time))
+            previous_time = current_time
+    if not timestamps:
+        raise ValueError("Video contains no decodable frames.")
+    return timestamps
+
+
 class _VideoDecodePool:
     """One lazily created decode pool per dataset and worker process."""