This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 3aab881 Add multimodal SQL helper UDFs (#467)
3aab881 is described below
commit 3aab8812beeaa7d9100e1842b3d388c5a15160ea
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 7 14:20:09 2026 +0800
Add multimodal SQL helper UDFs (#467)
---
bindings/python/README.md | 6 +
bindings/python/project-description.md | 41 +++
bindings/python/python/pypaimon_rust/functions.py | 357 ++++++++++++++++++++++
bindings/python/src/context.rs | 91 +++++-
bindings/python/tests/test_datafusion.py | 138 ++++++++-
docs/src/sql.md | 51 ++++
6 files changed, 673 insertions(+), 11 deletions(-)
diff --git a/bindings/python/README.md b/bindings/python/README.md
index 1c908bc..6e67c8a 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -39,6 +39,12 @@ ctx.sql("INSERT INTO paimon.my_db.users VALUES (1, 'alice'),
(2, 'bob')")
# Query data
batches = ctx.sql("SELECT id, name FROM paimon.my_db.users ORDER BY id")
+# Inspect BLOB media or build thumbnails when installed with
pypaimon-rust[video]
+batches = ctx.sql(
+ "SELECT id, media_info(content), media_thumbnail(content, 160, 90) "
+ "FROM paimon.my_db.assets"
+)
+
# Register a temporary table from a PyArrow RecordBatch
batch = pa.record_batch([[1, 2], ["alice", "bob"]], names=["id", "name"])
ctx.register_batch("paimon.default.my_temp", batch)
diff --git a/bindings/python/project-description.md
b/bindings/python/project-description.md
index 0cab353..0db9978 100644
--- a/bindings/python/project-description.md
+++ b/bindings/python/project-description.md
@@ -53,6 +53,47 @@ ctx.sql("INSERT INTO paimon.my_db.users VALUES (1, 'alice'),
(2, 'bob')")
batches = ctx.sql("SELECT * FROM paimon.my_db.users")
```
+### Multimodal SQL Helpers
+
+`SQLContext` also registers Python scalar UDFs for common BLOB media and vector
+workflows. Install the optional media dependencies when you need image or video
+decoding:
+
+```shell
+pip install "pypaimon-rust[video]"
+```
+
+```python
+batches = ctx.sql("""
+ SELECT
+ id,
+ media_info(content) AS info_json,
+ media_thumbnail(content, 160, 90) AS preview_png,
+ video_snapshot(content, 1000) AS frame_png
+ FROM paimon.my_db.assets
+""")
+```
+
+For vector search, `vector_from_json` converts JSON-encoded embeddings into
+Arrow `List<Float32>` values that can be used from a temporary table or
subquery:
+
+```python
+batches = ctx.sql("""
+ WITH queries AS (
+ SELECT id, vector_from_json(embedding_json) AS embedding
+ FROM paimon.my_db.query_embeddings
+ )
+ SELECT q.id AS query_id, r.id AS result_id
+ FROM queries q
+ CROSS JOIN LATERAL vector_search(
+ 'paimon.my_db.items',
+ 'embedding',
+ q.embedding,
+ 10
+ ) AS r
+""")
+```
+
### Temporary Tables
You can register temporary in-memory tables programmatically. Names support
the same resolution rules as SQL: bare names use the current catalog and
database, partially qualified names use the current catalog, and fully
qualified names specify catalog.database.table.
diff --git a/bindings/python/python/pypaimon_rust/functions.py
b/bindings/python/python/pypaimon_rust/functions.py
index 2460a36..73d61c4 100644
--- a/bindings/python/python/pypaimon_rust/functions.py
+++ b/bindings/python/python/pypaimon_rust/functions.py
@@ -16,7 +16,9 @@
# under the License.
import io
+import json
import logging
+import math
import struct
from typing import Any, BinaryIO
@@ -126,11 +128,219 @@ def _encode_video_frame(frame, image_format: str) ->
bytes:
image = frame.to_image()
except ImportError as e:
raise ImportError("Pillow is required to encode video frame images")
from e
+ return _encode_image(image, image_format)
+
+
+def _encode_image(image, image_format: str) -> bytes:
output = io.BytesIO()
image.save(output, format=image_format)
return output.getvalue()
+def _rewind_stream(stream: BinaryIO) -> None:
+ try:
+ stream.seek(0)
+ except Exception:
+ pass
+
+
+def _json_dumps(value: Any) -> str:
+ return json.dumps(value, sort_keys=True, separators=(",", ":"))
+
+
+def _drop_none(value: dict[str, Any]) -> dict[str, Any]:
+ return {key: item for key, item in value.items() if item is not None}
+
+
+def _duration_millis(container, stream=None) -> int | None:
+ duration = getattr(container, "duration", None)
+ if duration is not None and duration >= 0:
+ return int(round(duration / 1000))
+
+ if stream is not None:
+ stream_duration = getattr(stream, "duration", None)
+ time_base = getattr(stream, "time_base", None)
+ if stream_duration is not None and stream_duration >= 0 and time_base
is not None:
+ return int(round(float(stream_duration * time_base) * 1000))
+ return None
+
+
+def _codec_name(stream) -> str | None:
+ codec_context = getattr(stream, "codec_context", None)
+ return getattr(codec_context, "name", None)
+
+
+def _average_rate(stream) -> float | None:
+ rate = getattr(stream, "average_rate", None)
+ if rate is None:
+ return None
+ try:
+ return float(rate)
+ except Exception:
+ return None
+
+
+def _frame_count(stream) -> int | None:
+ frames = getattr(stream, "frames", None)
+ if frames is None or frames <= 0:
+ return None
+ return int(frames)
+
+
+def _decode_image_info(stream: BinaryIO) -> dict[str, Any] | None:
+ try:
+ from PIL import Image
+ except ImportError as e:
+ raise ImportError("Pillow is required to inspect image media") from e
+
+ with Image.open(stream) as image:
+ image.load()
+ return _drop_none(
+ {
+ "media_type": "image",
+ "format": image.format.lower() if image.format else None,
+ "width": image.width,
+ "height": image.height,
+ "mode": image.mode,
+ }
+ )
+
+
+def _decode_av_media_info(stream: BinaryIO) -> dict[str, Any] | None:
+ try:
+ import av
+ except ImportError as e:
+ raise ImportError("PyAV is required to inspect video or audio media")
from e
+
+ with av.open(stream, mode="r") as container:
+ format_name = (container.format.name or "").lower() or None
+ format_names = set((container.format.name or "").split(","))
+ video_streams = list(container.streams.video)
+ audio_streams = list(container.streams.audio)
+
+ if video_streams:
+ stream0 = video_streams[0]
+ media_type = "image" if format_names & _STILL_IMAGE_FORMATS else
"video"
+ info = {
+ "media_type": media_type,
+ "format": format_name,
+ "duration_ms": _duration_millis(container, stream0),
+ "width": getattr(stream0, "width", None),
+ "height": getattr(stream0, "height", None),
+ "codec": _codec_name(stream0),
+ "frame_count": _frame_count(stream0),
+ "average_rate": _average_rate(stream0),
+ "has_audio": bool(audio_streams),
+ }
+ return _drop_none(info)
+
+ if audio_streams:
+ stream0 = audio_streams[0]
+ info = {
+ "media_type": "audio",
+ "format": format_name,
+ "duration_ms": _duration_millis(container, stream0),
+ "codec": _codec_name(stream0),
+ "has_audio": True,
+ }
+ return _drop_none(info)
+ return None
+
+
+def _decode_media_info(stream: BinaryIO) -> str | None:
+ try:
+ info = _decode_image_info(stream)
+ if info is not None:
+ return _json_dumps(info)
+ except ImportError:
+ raise
+ except Exception:
+ _rewind_stream(stream)
+
+ info = _decode_av_media_info(stream)
+ return _json_dumps(info) if info is not None else None
+
+
+def _positive_dimension(value: Any, default: int) -> int | None:
+ try:
+ dimension = default if value is None else int(value)
+ except Exception:
+ return None
+ return dimension if dimension > 0 else None
+
+
+def _thumbnail_image(image, image_format: str, max_width: int, max_height:
int) -> bytes:
+ try:
+ from PIL import Image
+ except ImportError as e:
+ raise ImportError("Pillow is required to encode media thumbnails")
from e
+
+ thumbnail = image.copy()
+ resampling = getattr(getattr(Image, "Resampling", Image), "LANCZOS", None)
+ if resampling is None:
+ thumbnail.thumbnail((max_width, max_height))
+ else:
+ thumbnail.thumbnail((max_width, max_height), resampling)
+ return _encode_image(thumbnail, image_format)
+
+
+def _decode_image_thumbnail(
+ stream: BinaryIO,
+ image_format: str,
+ max_width: int,
+ max_height: int,
+) -> bytes | None:
+ try:
+ from PIL import Image
+ except ImportError as e:
+ raise ImportError("Pillow is required to decode image thumbnails")
from e
+
+ with Image.open(stream) as image:
+ image.load()
+ return _thumbnail_image(image, image_format, max_width, max_height)
+
+
+def _decode_video_thumbnail(
+ stream: BinaryIO,
+ image_format: str,
+ max_width: int,
+ max_height: int,
+) -> bytes | None:
+ try:
+ import av
+ except ImportError as e:
+ raise ImportError("PyAV is required to decode video thumbnails") from e
+
+ with av.open(stream, mode="r") as container:
+ if not container.streams.video:
+ return None
+ for frame in container.decode(video=0):
+ try:
+ image = frame.to_image()
+ except ImportError as e:
+ raise ImportError("Pillow is required to encode video
thumbnails") from e
+ return _thumbnail_image(image, image_format, max_width, max_height)
+ return None
+
+
+def _decode_media_thumbnail(
+ stream: BinaryIO,
+ image_format: str,
+ max_width: int,
+ max_height: int,
+) -> bytes | None:
+ try:
+ thumbnail = _decode_image_thumbnail(stream, image_format, max_width,
max_height)
+ if thumbnail is not None:
+ return thumbnail
+ except ImportError:
+ raise
+ except Exception:
+ _rewind_stream(stream)
+
+ return _decode_video_thumbnail(stream, image_format, max_width, max_height)
+
+
def _decode_video_frame(
stream: BinaryIO,
image_format: str,
@@ -247,3 +457,150 @@ def _make_video_frame(image_format: str = "PNG",
blob_reader_registry=None):
return pa.array(frames, type=pa.binary())
return video_frame
+
+
+def _make_media_info(blob_reader_registry=None):
+ def media_info(values):
+ try:
+ import pyarrow as pa
+ except ImportError as e:
+ raise ImportError("pyarrow is required to return media_info
results") from e
+
+ infos = []
+ for raw_value in values.to_pylist():
+ if raw_value is None:
+ infos.append(None)
+ continue
+
+ try:
+ stream = open_blob_descriptor_stream(raw_value,
blob_reader_registry)
+ try:
+ infos.append(_decode_media_info(stream))
+ finally:
+ stream.close()
+ except ImportError:
+ raise
+ except Exception as e:
+ logger.warning("Failed to inspect media: %s", e)
+ infos.append(None)
+
+ return pa.array(infos, type=pa.string())
+
+ return media_info
+
+
+def _make_media_thumbnail(
+ image_format: str = "PNG",
+ blob_reader_registry=None,
+ default_max_width: int = 320,
+ default_max_height: int = 320,
+):
+ image_format = image_format.upper()
+
+ def media_thumbnail(values, max_widths=None, max_heights=None):
+ try:
+ import pyarrow as pa
+ except ImportError as e:
+ raise ImportError("pyarrow is required to return media_thumbnail
results") from e
+
+ thumbnails = []
+ raw_values = values.to_pylist()
+ if max_widths is None and max_heights is None:
+ width_values = [default_max_width] * len(raw_values)
+ height_values = [default_max_height] * len(raw_values)
+ elif max_widths is not None and max_heights is not None:
+ width_values = max_widths.to_pylist()
+ height_values = max_heights.to_pylist()
+ else:
+ raise ValueError("media_thumbnail requires both width and height
arguments")
+ if len(width_values) != len(raw_values) or len(height_values) !=
len(raw_values):
+ raise ValueError("media_thumbnail size arguments must have the
same row count")
+
+ for raw_value, max_width, max_height in zip(
+ raw_values, width_values, height_values
+ ):
+ if raw_value is None or max_width is None or max_height is None:
+ thumbnails.append(None)
+ continue
+
+ max_width = _positive_dimension(max_width, default_max_width)
+ max_height = _positive_dimension(max_height, default_max_height)
+ if max_width is None or max_height is None:
+ thumbnails.append(None)
+ continue
+
+ try:
+ stream = open_blob_descriptor_stream(raw_value,
blob_reader_registry)
+ try:
+ thumbnails.append(
+ _decode_media_thumbnail(
+ stream, image_format, max_width, max_height
+ )
+ )
+ finally:
+ stream.close()
+ except ImportError:
+ raise
+ except Exception as e:
+ logger.warning("Failed to decode media thumbnail: %s", e)
+ thumbnails.append(None)
+
+ return pa.array(thumbnails, type=pa.binary())
+
+ return media_thumbnail
+
+
+def _coerce_vector(value: Any) -> list[float] | None:
+ if value is None or not isinstance(value, list):
+ return None
+
+ vector = []
+ for item in value:
+ if isinstance(item, bool) or not isinstance(item, (int, float)):
+ return None
+ item = float(item)
+ if not math.isfinite(item):
+ return None
+ vector.append(item)
+ return vector
+
+
+def _make_vector_from_json():
+ def vector_from_json(values):
+ try:
+ import pyarrow as pa
+ except ImportError as e:
+ raise ImportError("pyarrow is required to return vector_from_json
results") from e
+
+ vectors = []
+ for raw_value in values.to_pylist():
+ if raw_value is None:
+ vectors.append(None)
+ continue
+
+ try:
+ parsed = json.loads(raw_value)
+ vectors.append(_coerce_vector(parsed))
+ except Exception:
+ vectors.append(None)
+
+ return pa.array(vectors, type=pa.list_(pa.float32()))
+
+ return vector_from_json
+
+
+def _make_vector_to_json():
+ def vector_to_json(values):
+ try:
+ import pyarrow as pa
+ except ImportError as e:
+ raise ImportError("pyarrow is required to return vector_to_json
results") from e
+
+ encoded = []
+ for raw_value in values.to_pylist():
+ vector = _coerce_vector(raw_value)
+ encoded.append(_json_dumps(vector) if vector is not None else None)
+
+ return pa.array(encoded, type=pa.string())
+
+ return vector_to_json
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index f6e2a6e..8cddde0 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -18,7 +18,7 @@
use std::collections::HashMap;
use std::sync::Arc;
-use arrow::datatypes::DataType as ArrowDataType;
+use arrow::datatypes::{DataType as ArrowDataType, Field as ArrowField};
use arrow::pyarrow::{FromPyArrow, ToPyArrow};
use datafusion::catalog::CatalogProvider;
use datafusion::logical_expr::{Signature, TypeSignature, Volatility};
@@ -129,12 +129,63 @@ pub struct PySQLContext {
}
impl PySQLContext {
- fn register_video_builtins(&self, py: Python<'_>) -> PyResult<()> {
+ fn vector_float32_type() -> ArrowDataType {
+ ArrowDataType::List(Arc::new(ArrowField::new(
+ "item",
+ ArrowDataType::Float32,
+ true,
+ )))
+ }
+
+ fn register_multimodal_builtins(&self, py: Python<'_>) -> PyResult<()> {
let functions = py.import("pypaimon_rust.functions")?;
let blob_reader_registry = Py::new(
py,
PyBlobReaderRegistry::new(self.inner.blob_reader_registry()),
)?;
+
+ let media_info_blob_reader_registry =
blob_reader_registry.clone_ref(py);
+ let media_info_func = functions
+ .getattr("_make_media_info")?
+ .call1((media_info_blob_reader_registry,))?
+ .unbind();
+ let media_info_udf = build_python_scalar_udf(
+ "media_info".to_string(),
+ media_info_func,
+ ArrowDataType::Utf8,
+ Signature::exact(vec![ArrowDataType::Binary],
Volatility::Volatile),
+ );
+ self.inner.ctx().register_udf(media_info_udf);
+
+ let thumbnail_blob_reader_registry =
blob_reader_registry.clone_ref(py);
+ let thumbnail_func = functions
+ .getattr("_make_media_thumbnail")?
+ .call1(("PNG", thumbnail_blob_reader_registry))?
+ .unbind();
+ let thumbnail_signature = Signature::one_of(
+ vec![
+ TypeSignature::Exact(vec![ArrowDataType::Binary]),
+ TypeSignature::Exact(vec![
+ ArrowDataType::Binary,
+ ArrowDataType::Int32,
+ ArrowDataType::Int32,
+ ]),
+ TypeSignature::Exact(vec![
+ ArrowDataType::Binary,
+ ArrowDataType::Int64,
+ ArrowDataType::Int64,
+ ]),
+ ],
+ Volatility::Volatile,
+ );
+ let thumbnail_udf = build_python_scalar_udf(
+ "media_thumbnail".to_string(),
+ thumbnail_func,
+ ArrowDataType::Binary,
+ thumbnail_signature,
+ );
+ self.inner.ctx().register_udf(thumbnail_udf);
+
let snapshot_blob_reader_registry = blob_reader_registry.clone_ref(py);
let snapshot_func = functions
.getattr("_make_video_snapshot")?
@@ -174,16 +225,44 @@ impl PySQLContext {
frame_signature,
);
self.inner.ctx().register_udf(frame_udf);
+
+ let vector_from_json_func = functions
+ .getattr("_make_vector_from_json")?
+ .call0()?
+ .unbind();
+ let vector_from_json_signature = Signature::one_of(
+ vec![
+ TypeSignature::Exact(vec![ArrowDataType::Utf8]),
+ TypeSignature::Exact(vec![ArrowDataType::LargeUtf8]),
+ ],
+ Volatility::Immutable,
+ );
+ let vector_from_json_udf = build_python_scalar_udf(
+ "vector_from_json".to_string(),
+ vector_from_json_func,
+ Self::vector_float32_type(),
+ vector_from_json_signature,
+ );
+ self.inner.ctx().register_udf(vector_from_json_udf);
+
+ let vector_to_json_func =
functions.getattr("_make_vector_to_json")?.call0()?.unbind();
+ let vector_to_json_udf = build_python_scalar_udf(
+ "vector_to_json".to_string(),
+ vector_to_json_func,
+ ArrowDataType::Utf8,
+ Signature::new(TypeSignature::Any(1), Volatility::Immutable),
+ );
+ self.inner.ctx().register_udf(vector_to_json_udf);
Ok(())
}
- fn warn_video_builtin_registration_failure(py: Python<'_>, err: PyErr) {
+ fn warn_multimodal_builtin_registration_failure(py: Python<'_>, err:
PyErr) {
if let Ok(warnings) = py.import("warnings") {
let category = py.get_type::<PyRuntimeWarning>();
let _ = warnings.call_method1(
"warn",
(
- format!("video built-ins could not be registered: {err}"),
+ format!("multimodal built-ins could not be registered:
{err}"),
category,
),
);
@@ -198,8 +277,8 @@ impl PySQLContext {
let ctx = Self {
inner: SQLContext::new(),
};
- if let Err(err) = ctx.register_video_builtins(py) {
- Self::warn_video_builtin_registration_failure(py, err);
+ if let Err(err) = ctx.register_multimodal_builtins(py) {
+ Self::warn_multimodal_builtin_registration_failure(py, err);
}
Ok(ctx)
}
diff --git a/bindings/python/tests/test_datafusion.py
b/bindings/python/tests/test_datafusion.py
index 5430b2a..e68698c 100644
--- a/bindings/python/tests/test_datafusion.py
+++ b/bindings/python/tests/test_datafusion.py
@@ -16,6 +16,7 @@
# under the License.
import io
+import json
import os
import struct
import sys
@@ -65,11 +66,14 @@ def write_sample_video(
container.mux(packet)
-def sample_image_bytes() -> bytes:
+def sample_image_bytes(
+ size: tuple[int, int] = (32, 32),
+ color: tuple[int, int, int] = (40, 120, 220),
+) -> bytes:
image_module = pytest.importorskip("PIL.Image")
output = io.BytesIO()
- image = image_module.new("RGB", (32, 32), color=(40, 120, 220))
+ image = image_module.new("RGB", size, color=color)
image.save(output, format="PNG")
return output.getvalue()
@@ -86,16 +90,24 @@ def
test_video_snapshot_builtin_registered_on_context_init():
"""
SELECT
video_snapshot(CAST(NULL AS BYTEA)) AS cover_png,
- video_frame(CAST(NULL AS BYTEA), 0) AS frame_png
+ video_frame(CAST(NULL AS BYTEA), 0) AS frame_png,
+ media_info(CAST(NULL AS BYTEA)) AS media_info_json,
+ media_thumbnail(CAST(NULL AS BYTEA)) AS thumbnail_png,
+ vector_from_json(CAST(NULL AS STRING)) AS vector_value,
+ vector_to_json(vector_from_json('[1.0, 2.5]')) AS vector_json
"""
)
table = pa.Table.from_batches(batches)
assert table["cover_png"].to_pylist() == [None]
assert table["frame_png"].to_pylist() == [None]
+ assert table["media_info_json"].to_pylist() == [None]
+ assert table["thumbnail_png"].to_pylist() == [None]
+ assert table["vector_value"].to_pylist() == [None]
+ assert json.loads(table["vector_json"].to_pylist()[0]) == [1.0, 2.5]
-def test_sql_context_survives_video_builtins_registration_failure(monkeypatch):
+def
test_sql_context_survives_multimodal_builtins_registration_failure(monkeypatch):
monkeypatch.setitem(
sys.modules,
"pypaimon_rust.functions",
@@ -104,7 +116,7 @@ def
test_sql_context_survives_video_builtins_registration_failure(monkeypatch):
with pytest.warns(
RuntimeWarning,
- match="video built-ins could not be registered",
+ match="multimodal built-ins could not be registered",
):
ctx = SQLContext()
@@ -336,6 +348,122 @@ def test_video_frame_accepts_frame_index():
ctx.sql("DROP TEMPORARY TABLE paimon.default.videos")
+def test_media_info_returns_json_for_image_and_video():
+ with tempfile.TemporaryDirectory() as warehouse:
+ video_path = Path(warehouse) / "sample.mp4"
+ write_sample_video(video_path)
+
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.register_batch(
+ "paimon.default.media",
+ pa.record_batch(
+ [
+ [1, 2],
+ pa.array(
+ [
+ sample_image_bytes(size=(48, 24)),
+ video_path.read_bytes(),
+ ],
+ type=pa.binary(),
+ ),
+ ],
+ names=["id", "content"],
+ ),
+ )
+
+ batches = ctx.sql(
+ """
+ SELECT id, media_info(content) AS info_json
+ FROM paimon.default.media
+ ORDER BY id
+ """
+ )
+ rows = pa.Table.from_batches(batches).to_pylist()
+ image_info = json.loads(rows[0]["info_json"])
+ video_info = json.loads(rows[1]["info_json"])
+
+ assert image_info["media_type"] == "image"
+ assert image_info["format"] == "png"
+ assert image_info["width"] == 48
+ assert image_info["height"] == 24
+
+ assert video_info["media_type"] == "video"
+ assert video_info["width"] == 32
+ assert video_info["height"] == 32
+ assert video_info["has_audio"] is False
+
+ ctx.sql("DROP TEMPORARY TABLE paimon.default.media")
+
+
+def test_media_thumbnail_handles_image_and_video():
+ image_module = pytest.importorskip("PIL.Image")
+
+ with tempfile.TemporaryDirectory() as warehouse:
+ video_path = Path(warehouse) / "sample.mp4"
+ write_sample_video(video_path)
+
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.register_batch(
+ "paimon.default.media",
+ pa.record_batch(
+ [
+ [1, 2],
+ pa.array(
+ [
+ sample_image_bytes(size=(64, 32)),
+ video_path.read_bytes(),
+ ],
+ type=pa.binary(),
+ ),
+ ],
+ names=["id", "content"],
+ ),
+ )
+
+ batches = ctx.sql(
+ """
+ SELECT
+ id,
+ media_thumbnail(content, 16, 16) AS thumbnail_png,
+ media_thumbnail(content, -1, 16) AS invalid_thumbnail_png
+ FROM paimon.default.media
+ ORDER BY id
+ """
+ )
+ rows = pa.Table.from_batches(batches).to_pylist()
+
+ for row in rows:
+ assert row["thumbnail_png"].startswith(PNG_SIGNATURE)
+ thumbnail = image_module.open(io.BytesIO(row["thumbnail_png"]))
+ assert thumbnail.width <= 16
+ assert thumbnail.height <= 16
+ assert row["invalid_thumbnail_png"] is None
+
+ ctx.sql("DROP TEMPORARY TABLE paimon.default.media")
+
+
+def test_vector_json_bridge_functions():
+ ctx = SQLContext()
+
+ batches = ctx.sql(
+ """
+ SELECT
+ vector_from_json('[1.0, 2.5, -3]') AS vector_value,
+ vector_from_json('not json') AS invalid_json,
+ vector_from_json('[true]') AS invalid_value,
+ vector_to_json(vector_from_json('[1, 2.5]')) AS vector_json
+ """
+ )
+ row = pa.Table.from_batches(batches).to_pylist()[0]
+
+ assert row["vector_value"] == [1.0, 2.5, -3.0]
+ assert row["invalid_json"] is None
+ assert row["invalid_value"] is None
+ assert json.loads(row["vector_json"]) == [1.0, 2.5]
+
+
def test_query_simple_table_via_catalog_provider():
catalog = PaimonCatalog({"warehouse": WAREHOUSE})
ctx = SessionContext()
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 6ae3b62..e983274 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -722,6 +722,57 @@ When the following conditions are met, `COUNT(*)`
retrieves exact row counts dir
- No LIMIT clause
- Filter predicates only involve partition columns (Exact level)
+## Python Multimodal Helper Functions
+
+When you use `pypaimon_rust.datafusion.SQLContext`, the Python binding
registers a small set of scalar helper functions for BLOB-backed media and
vector workflows. These helpers are Python-binding built-ins; they are not
registered by the Rust `paimon_datafusion::SQLContext`.
+
+Media helpers require the optional Python media dependencies:
+
+```shell
+pip install "pypaimon-rust[video]"
+```
+
+| Function | Return Type | Description |
+|---|---|---|
+| `media_info(blob)` | STRING | JSON metadata for image, video, or audio input
|
+| `media_thumbnail(blob)` | BINARY | PNG thumbnail, using a default 320x320
bounding box |
+| `media_thumbnail(blob, max_width, max_height)` | BINARY | PNG thumbnail
constrained to the given dimensions |
+| `video_snapshot(blob)` | BINARY | PNG frame near timestamp 0ms |
+| `video_snapshot(blob, timestamp_ms)` | BINARY | PNG frame near the given
timestamp |
+| `video_frame(blob, frame_index)` | BINARY | PNG frame by zero-based decoded
frame index |
+| `vector_from_json(json)` | `List<Float32>` | Converts a JSON float array
string into an Arrow float vector |
+| `vector_to_json(vector)` | STRING | Converts an Arrow float vector back to a
JSON array string |
+
+Invalid, NULL, unsupported, or undecodable media inputs return SQL `NULL`.
Media functions read either inline bytes or BLOB descriptor bytes when the
`SQLContext` has a registered Paimon catalog that can resolve the descriptor.
+
+Example:
+
+```sql
+SELECT
+ id,
+ media_info(content) AS info_json,
+ media_thumbnail(content, 160, 90) AS preview_png,
+ video_frame(content, 10) AS frame_png
+FROM paimon.my_db.assets;
+```
+
+Use `vector_from_json` to bridge JSON-encoded embeddings into lateral vector
search queries:
+
+```sql
+WITH queries AS (
+ SELECT id, vector_from_json(embedding_json) AS embedding
+ FROM paimon.my_db.query_embeddings
+)
+SELECT q.id AS query_id, r.id AS result_id
+FROM queries q
+CROSS JOIN LATERAL vector_search(
+ 'paimon.my_db.items',
+ 'embedding',
+ q.embedding,
+ 10
+) AS r;
+```
+
## Vector Search
Paimon supports approximate nearest neighbor (ANN) vector search via the
Lumina vector index. The `vector_search` table-valued function is registered as
a UDTF on the DataFusion session context.
