This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 836b1d484e [python] Preserve schema fields in Python multimodal inputs
(#10058)
836b1d484e is described below
commit 836b1d484e612a504cee69b5d361cade9b32d8d5
Author: chaoyang <[email protected]>
AuthorDate: Tue Sep 22 10:44:33 2026 +0800
[python] Preserve schema fields in Python multimodal inputs (#10058)
---
paimon-python/pypaimon/multimodal/table.py | 12 +++++
.../pypaimon/tests/multimodal_table_test.py | 57 ++++++++++++++++++++++
2 files changed, 69 insertions(+)
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
index 4931da1dd1..81f302c499 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -582,6 +582,18 @@ def _blob_columns(table):
def _to_arrow_table(data, target_schema=None):
if target_schema is not None:
data = _serialize_blob_values(data, target_schema)
+ if isinstance(data, list):
+ # Untyped from_pylist only discovers columns present in the first
row.
+ data = {
+ field.name: [row.get(field.name) for row in data]
+ for field in target_schema
+ }
+ if isinstance(data, dict):
+ data = dict(data)
+ for field in target_schema:
+ if field.name in data and pa.types.is_map(field.type):
+ # Inferred STRUCT/LIST arrays cannot be cast back to MAP.
+ data[field.name] = pa.array(data[field.name],
type=field.type)
if isinstance(data, pa.Table):
table = data
elif isinstance(data, pa.RecordBatch):
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index 71160f8a8d..acca5be9d9 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -972,6 +972,63 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(["id", "name"], result.column_names)
self.assertEqual([1, 2], result["id"].to_pylist())
+ def test_add_preserves_fields_missing_from_first_row(self):
+ table = self.conn.create_table(
+ "sparse_rows", schema=_schema({
+ "id": pa.int32(), "caption": pa.string(), "quality":
pa.float64(),
+ "image": pa.large_binary(), "missing": pa.string(),
+ }), options=_PARQUET_OPTIONS)
+ table.add([
+ {"id": "1"},
+ {"id": "2", "caption": "keep-me", "quality": 0.9,
+ "image": b"payload", "extra": "ignored"},
+ ])
+ scalar, blobs = table.scan().read_blobs("image")
+ rows = sorted(zip(scalar.to_pylist(), blobs["image"]), key=lambda row:
row[0]["id"])
+ self.assertEqual([
+ ({"id": 1, "caption": None, "quality": None, "missing": None},
None),
+ ({"id": 2, "caption": "keep-me", "quality": 0.9, "missing": None},
b"payload"),
+ ], rows)
+
+ def test_add_python_map_values(self):
+ schema = _schema({
+ "id": pa.int32(), "assets": pa.map_(pa.string(),
pa.large_binary()),
+ "scores": pa.map_(pa.string(), pa.int64()),
+ })
+ rows = [
+ {"id": 1, "assets": {"image": pmm.Blob.from_data(b"body"),
"missing": None},
+ "scores": {"quality": 9}},
+ {"id": 2, "assets": [("empty", b""), ("image", b"second")],
"scores": []},
+ {"id": 3, "assets": [], "scores": None},
+ {"id": 4, "assets": None, "scores": None},
+ ]
+ for columnar in (False, True):
+ with self.subTest(columnar=columnar):
+ table = self.conn.create_table(
+ "map_input_%s" % columnar, schema=schema,
options=_PARQUET_OPTIONS)
+ data = {name: [row[name] for row in rows] for name in
schema.names} if columnar else rows
+ table.add(data)
+ scalar, blobs = table.scan().read_blobs("assets")
+ self.assertEqual({
+ 1: [("image", b"body"), ("missing", None)],
+ 2: [("empty", b""), ("image", b"second")], 3: [], 4: None,
+ }, dict(zip(scalar["id"].to_pylist(), blobs["assets"])))
+ self.assertEqual({1: [("quality", 9)], 2: [], 3: None, 4:
None},
+ dict(zip(scalar["id"].to_pylist(),
scalar["scores"].to_pylist())))
+
+ def test_python_input_preserves_safe_casts_and_empty_rows(self):
+ from pypaimon.multimodal.table import _to_arrow_table
+
+ schema = _schema({"id": pa.int32(), "caption": pa.string()})
+ for data in ([{"id": 1.5}], {"id": [1.5]}):
+ with self.assertRaises(pa.ArrowInvalid):
+ _to_arrow_table(data, schema)
+ self.assertEqual([], _to_arrow_table([], schema).to_pylist())
+ self.assertEqual([{"id": None, "caption": None}] * 2,
+ _to_arrow_table([{}, {}], schema).to_pylist())
+ self.assertEqual([{"id": 2, "caption": None}],
+ _to_arrow_table({"id": ["2"]}, schema).to_pylist())
+
def test_add_scan_where_select_limit(self):
users = self.conn.create_table(
"users",
