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 eb31fd8139 [python] Populate IOConfig on Daft blob File columns (#8388)
eb31fd8139 is described below
commit eb31fd8139dc76f66848176b913585658e5a7acc
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Jun 30 18:58:32 2026 +0800
[python] Populate IOConfig on Daft blob File columns (#8388)
Reading a blob column via the Daft integration (read_paimon) on a
REST/DLF + OSS catalog: the column
comes back as Daft's native File, but any native File op fails:
`df["blob_data"].open().read()
DaftError::External Failed to load Credentials for store: opendal(oss)`
This PR populates IOConfig on Daft blob File columns.
---
paimon-python/pypaimon/daft/daft_blob.py | 18 ++-
paimon-python/pypaimon/daft/daft_datasource.py | 44 +++++-
paimon-python/pypaimon/daft/daft_io_config.py | 72 +++++++++
paimon-python/pypaimon/daft/daft_paimon.py | 9 +-
.../pypaimon/tests/daft/daft_blob_test.py | 169 +++++++++++++++++++++
.../pypaimon/tests/daft/daft_catalog_rest_test.py | 3 +-
6 files changed, 306 insertions(+), 9 deletions(-)
diff --git a/paimon-python/pypaimon/daft/daft_blob.py
b/paimon-python/pypaimon/daft/daft_blob.py
index 07691883f9..d249352e03 100644
--- a/paimon-python/pypaimon/daft/daft_blob.py
+++ b/paimon-python/pypaimon/daft/daft_blob.py
@@ -59,8 +59,12 @@ def _deserialize_one(data: bytes) -> tuple[str, int, int]:
return uri, offset, length
-def blob_column_to_file_array(column: pa.Array) -> pa.Array:
- """Convert a large_binary column of serialized BlobDescriptors to a
FileReference-compatible struct."""
+def blob_column_to_file_array(column: pa.Array, io_config_bytes: bytes | None
= None) -> pa.Array:
+ """Convert a large_binary column of serialized BlobDescriptors to a
File-compatible struct.
+
+ ``io_config_bytes`` (serialized IOConfig) is embedded into each File so
native File ops carry
+ credentials; when None the io_config is left null and ops fall back to
Daft's global IOConfig.
+ """
urls: list[str | None] = []
offsets: list[int | None] = []
lengths: list[int | None] = []
@@ -78,10 +82,18 @@ def blob_column_to_file_array(column: pa.Array) -> pa.Array:
lengths.append(length)
n = len(urls)
+ if io_config_bytes is None:
+ io_configs: pa.Array = pa.nulls(n, type=pa.large_binary())
+ else:
+ # Only populate io_config for valid rows; keep null rows null.
+ io_configs = pa.array(
+ [io_config_bytes if u is not None else None for u in urls],
+ type=pa.large_binary(),
+ )
return pa.StructArray.from_arrays(
[
pa.array(urls, type=pa.large_utf8()),
- pa.nulls(n, type=pa.large_binary()),
+ io_configs,
pa.array(offsets, type=pa.int64()),
pa.array(lengths, type=pa.int64()),
],
diff --git a/paimon-python/pypaimon/daft/daft_datasource.py
b/paimon-python/pypaimon/daft/daft_datasource.py
index 5308ff0178..68acae19dc 100644
--- a/paimon-python/pypaimon/daft/daft_datasource.py
+++ b/paimon-python/pypaimon/daft/daft_datasource.py
@@ -180,6 +180,7 @@ class _PaimonPKSplitTask(DataSourceTask):
predicate: Predicate | None = None,
output_columns: list[str] | None = None,
blob_column_names: set[str] | None = None,
+ explicit_io_config_bytes: bytes | None = None,
) -> None:
self._table_catalog_options = table_catalog_options
self._table_identifier = table_identifier
@@ -192,6 +193,7 @@ class _PaimonPKSplitTask(DataSourceTask):
self._predicate = predicate
self._output_columns = output_columns
self._blob_column_names = blob_column_names or set()
+ self._explicit_io_config_bytes = explicit_io_config_bytes
@property
def schema(self) -> Schema:
@@ -212,19 +214,48 @@ class _PaimonPKSplitTask(DataSourceTask):
if self._predicate is not None:
read_builder = read_builder.with_filter(self._predicate)
+ blob_io_config_bytes = self._blob_io_config_bytes(table) if
self._blob_column_names else None
reader = read_builder.new_read().to_arrow_batch_reader([self._split])
for batch in iter(reader.read_next_batch, None):
if self._output_columns is not None:
batch = batch.select(self._output_columns)
if self._blob_column_names:
- batch = _convert_blob_columns(batch, self._blob_column_names)
+ batch = _convert_blob_columns(batch, self._blob_column_names,
blob_io_config_bytes)
rb = RecordBatch.from_arrow_record_batches([batch], batch.schema)
if self._blob_column_names:
rb = _cast_blob_columns_to_file(rb, self._blob_column_names)
yield rb
-
-def _convert_blob_columns(batch: pa.RecordBatch, blob_column_names: set[str])
-> pa.RecordBatch:
+ def _blob_io_config_bytes(self, table: FileStoreTable) -> bytes | None:
+ """Serialized IOConfig embedded into blob File columns, in priority
order: refreshed
+ REST-DLF token / catalog creds (refreshed at read time, so long reads
don't freeze a
+ short STS token), then the explicit read_paimon io_config, then the
OSS env alias."""
+ from pypaimon.daft.daft_io_config import (
+ _convert_paimon_catalog_options_to_file_io_config,
+ _with_oss_alias,
+ serialize_io_config,
+ )
+ from pypaimon.daft.daft_paimon import _enrich_options_with_rest_token
+
+ enriched =
_enrich_options_with_rest_token(self._table_catalog_options, table)
+ io_config = _convert_paimon_catalog_options_to_file_io_config(enriched)
+ if io_config is not None:
+ return serialize_io_config(io_config)
+ if self._explicit_io_config_bytes is not None:
+ # oss:// blobs need the s3 alias even from an explicit io_config
(opendal File.open is broken).
+ if urlparse(str(getattr(table, "table_path", "") or "")).scheme ==
"oss":
+ from daft.io import IOConfig
+ return
serialize_io_config(_with_oss_alias(IOConfig._from_serialized(self._explicit_io_config_bytes)))
+ return self._explicit_io_config_bytes
+ io_config =
_convert_paimon_catalog_options_to_file_io_config(enriched,
require_credentials=False)
+ return serialize_io_config(io_config) if io_config is not None else
None
+
+
+def _convert_blob_columns(
+ batch: pa.RecordBatch,
+ blob_column_names: set[str],
+ io_config_bytes: bytes | None = None,
+) -> pa.RecordBatch:
"""Replace serialized BlobDescriptor columns with the File physical struct
layout."""
from pypaimon.daft.daft_blob import FILE_PHYSICAL_TYPE,
blob_column_to_file_array
@@ -233,7 +264,7 @@ def _convert_blob_columns(batch: pa.RecordBatch,
blob_column_names: set[str]) ->
for i, field in enumerate(batch.schema):
col = batch.column(i)
if field.name in blob_column_names and
(pa.types.is_large_binary(field.type) or pa.types.is_binary(field.type)):
- arrays.append(blob_column_to_file_array(col))
+ arrays.append(blob_column_to_file_array(col, io_config_bytes))
fields.append(pa.field(field.name, FILE_PHYSICAL_TYPE,
nullable=field.nullable))
else:
arrays.append(col)
@@ -272,8 +303,10 @@ class PaimonDataSource(DataSource):
table: FileStoreTable,
storage_config: StorageConfig,
catalog_options: dict[str, str],
+ explicit_io_config_bytes: bytes | None = None,
) -> None:
self._storage_config = storage_config
+ self._explicit_io_config_bytes = explicit_io_config_bytes
self._catalog_options = dict(catalog_options or {})
self._table_catalog_options = {
**_extract_catalog_options(table),
@@ -291,6 +324,7 @@ class PaimonDataSource(DataSource):
def __getstate__(self) -> dict[str, Any]:
return {
"_multithreaded_io": self._storage_config.multithreaded_io,
+ "_explicit_io_config_bytes": self._explicit_io_config_bytes,
"_catalog_options": self._catalog_options,
"_table_catalog_options": self._table_catalog_options,
"_table_identifier": self._table_identifier,
@@ -302,6 +336,7 @@ class PaimonDataSource(DataSource):
}
def __setstate__(self, state: dict[str, Any]) -> None:
+ self._explicit_io_config_bytes = state.get("_explicit_io_config_bytes")
self._catalog_options = state["_catalog_options"]
self._table_catalog_options = state["_table_catalog_options"]
self._table_identifier = state["_table_identifier"]
@@ -485,6 +520,7 @@ class PaimonDataSource(DataSource):
read_pushdowns.reader_predicate,
read_pushdowns.task_columns,
self._blob_column_names,
+ self._explicit_io_config_bytes,
)
def explain_scan(self, pushdowns: Pushdowns, verbose: bool = False) ->
PaimonScanExplain:
diff --git a/paimon-python/pypaimon/daft/daft_io_config.py
b/paimon-python/pypaimon/daft/daft_io_config.py
index ef3da7e98d..96063a3850 100644
--- a/paimon-python/pypaimon/daft/daft_io_config.py
+++ b/paimon-python/pypaimon/daft/daft_io_config.py
@@ -23,6 +23,19 @@ from urllib.parse import urlparse
from daft.io import IOConfig, S3Config
+def serialize_io_config(io_config: IOConfig) -> bytes:
+ """Serialize an IOConfig to the bytes Daft's File struct embeds, via its
__reduce__."""
+ reducer = io_config.__reduce__()
+ if (reducer[0] is not IOConfig._from_serialized
+ or not isinstance(reducer[1], tuple) or not reducer[1]
+ or not isinstance(reducer[1][0], bytes)):
+ raise TypeError(
+ f"Unexpected IOConfig.__reduce__ shape ({reducer!r}); "
+ "Daft may have changed its IOConfig serialization format."
+ )
+ return reducer[1][0]
+
+
def _convert_paimon_catalog_options_to_io_config(catalog_options: dict[str,
str]) -> IOConfig | None:
"""Convert pypaimon catalog options to Daft IOConfig.
@@ -87,3 +100,62 @@ def
_convert_paimon_catalog_options_to_io_config(catalog_options: dict[str, str]
)
return io_config if any_props_set else None
+
+
+def _convert_paimon_catalog_options_to_file_io_config(
+ catalog_options: dict[str, str], require_credentials: bool = True
+) -> IOConfig | None:
+ """IOConfig for blob File ops. OSS routes through Daft's S3 client (oss://
aliased to s3,
+ virtual-hosted) because File.open() over OpenDAL/OSS is broken on some
Daft builds.
+ require_credentials=False keeps a config without a key pair so the env
chain is used."""
+ warehouse = catalog_options.get("warehouse", "")
+ if (urlparse(warehouse).scheme if warehouse else "") != "oss":
+ if catalog_options.get("fs.s3.accessKeyId") and
catalog_options.get("fs.s3.accessKeySecret"):
+ return
_convert_paimon_catalog_options_to_io_config(catalog_options)
+ # No complete pair: None when required (explicit wins), else
endpoint/region-only for env.
+ if require_credentials:
+ return None
+ endpoint = catalog_options.get("fs.s3.endpoint")
+ region = catalog_options.get("fs.s3.region")
+ if not endpoint and not region:
+ return None
+ return IOConfig(s3=S3Config(endpoint_url=endpoint, region_name=region))
+
+ endpoint = catalog_options.get("fs.oss.endpoint")
+ if endpoint and not endpoint.startswith(("http://";, "https://";)):
+ endpoint = f"https://{endpoint}";
+ key_id = catalog_options.get("fs.oss.accessKeyId")
+ access_key = catalog_options.get("fs.oss.accessKeySecret")
+ token = catalog_options.get("fs.oss.securityToken")
+ if not (key_id and access_key):
+ # No complete pair: None when required (explicit wins), else alias
only for the env chain.
+ if require_credentials:
+ return None
+ key_id = access_key = token = None
+ return IOConfig(
+ s3=S3Config(
+ endpoint_url=endpoint,
+ region_name=catalog_options.get("fs.oss.region"),
+ key_id=key_id,
+ access_key=access_key,
+ session_token=token,
+ force_virtual_addressing=True,
+ ),
+ protocol_aliases={"oss": "s3"},
+ )
+
+
+def _with_oss_alias(io_config: IOConfig) -> IOConfig:
+ """Add the oss->s3 alias + virtual-hosted to a caller's io_config,
deriving creds from an
+ OpenDAL OSS backend if the S3 config has none, so File.open() works on
oss:// blobs."""
+ s3 = io_config.s3
+ if s3 is None or s3.key_id is None:
+ oss = (io_config.opendal_backends or {}).get("oss") or {}
+ if oss:
+ s3 = S3Config(
+ endpoint_url=oss.get("endpoint"),
region_name=oss.get("region"),
+ key_id=oss.get("access_key_id"),
access_key=oss.get("access_key_secret"),
+ session_token=oss.get("security_token"),
+ )
+ s3 = (s3 or S3Config()).replace(force_virtual_addressing=True)
+ return io_config.replace(s3=s3,
protocol_aliases={**dict(io_config.protocol_aliases or {}), "oss": "s3"})
diff --git a/paimon-python/pypaimon/daft/daft_paimon.py
b/paimon-python/pypaimon/daft/daft_paimon.py
index 29825fbc11..c419d4ab8c 100644
--- a/paimon-python/pypaimon/daft/daft_paimon.py
+++ b/paimon-python/pypaimon/daft/daft_paimon.py
@@ -90,11 +90,15 @@ def _source_for_table(
from pypaimon.daft.daft_datasource import PaimonDataSource
from pypaimon.daft.daft_io_config import (
_convert_paimon_catalog_options_to_io_config,
+ serialize_io_config,
)
if catalog_options is None:
catalog_options = {}
+ # Keep the caller's io_config as a blob File fallback when nothing else is
derivable.
+ explicit_io_config_bytes = serialize_io_config(io_config) if io_config is
not None else None
+
io_config = io_config or _convert_paimon_catalog_options_to_io_config(
_enrich_options_with_rest_token(catalog_options, table)
)
@@ -104,7 +108,10 @@ def _source_for_table(
storage_config = StorageConfig(multithreaded_io, io_config)
return PaimonDataSource(
- table, storage_config=storage_config, catalog_options=catalog_options
+ table,
+ storage_config=storage_config,
+ catalog_options=catalog_options,
+ explicit_io_config_bytes=explicit_io_config_bytes,
)
diff --git a/paimon-python/pypaimon/tests/daft/daft_blob_test.py
b/paimon-python/pypaimon/tests/daft/daft_blob_test.py
new file mode 100644
index 0000000000..594278ca45
--- /dev/null
+++ b/paimon-python/pypaimon/tests/daft/daft_blob_test.py
@@ -0,0 +1,169 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import unittest
+
+import pyarrow as pa
+import pytest
+
+pypaimon = pytest.importorskip("pypaimon")
+daft = pytest.importorskip("daft")
+
+from daft.datatype import DataType
+from daft.io import IOConfig, S3Config
+
+from pypaimon.daft.daft_blob import blob_column_to_file_array
+from pypaimon.daft.daft_compat import (
+ file_range_position_field,
+ file_range_size_field,
+ has_file_range_reads,
+)
+from pypaimon.daft.daft_io_config import (
+ _convert_paimon_catalog_options_to_file_io_config,
+ serialize_io_config,
+)
+from pypaimon.daft.daft_datasource import _PaimonPKSplitTask
+from pypaimon.table.row.blob import BlobDescriptor
+
+
+def _descriptor_column(specs):
+ """large_binary column of serialized BlobDescriptors (None for gaps)."""
+ out = [None if s is None else BlobDescriptor(*s).serialize() for s in
specs]
+ return pa.array(out, type=pa.large_binary())
+
+
+class BlobColumnToFileArrayTest(unittest.TestCase):
+ # Pure config/arrow tests run on any installed Daft; only File-cast needs
file ranges.
+
+ def test_blob_column_to_file_array(self):
+ # io_config is null without creds; embedded only for valid (non-null)
rows with creds.
+ col = _descriptor_column([("oss://b/k1", 0, 5), None, ("oss://b/k2",
8, 9)])
+ bare = blob_column_to_file_array(col)
+ self.assertEqual(bare.field("url").to_pylist(), ["oss://b/k1", None,
"oss://b/k2"])
+ self.assertEqual(bare.field(file_range_position_field()).to_pylist(),
[0, None, 8])
+ self.assertEqual(bare.field(file_range_size_field()).to_pylist(), [5,
None, 9])
+ self.assertEqual(bare.field("io_config").null_count, 3)
+
+ blob = serialize_io_config(IOConfig(s3=S3Config(key_id="AK",
access_key="SK")))
+ self.assertEqual(blob_column_to_file_array(col,
blob).field("io_config").to_pylist(),
+ [blob, None, blob])
+
+ def test_serialize_io_config_roundtrips(self):
+ s3 = IOConfig(s3=S3Config(key_id="AK", access_key="SK",
session_token="TOK"))
+
self.assertEqual(IOConfig._from_serialized(serialize_io_config(s3)).s3.session_token,
"TOK")
+ # OSS uses Daft's OpenDAL backend, which serializes differently from
S3Config.
+ oss = {"access_key_id": "AK", "endpoint":
"https://oss-test.example.com";, "bucket": "b"}
+ cfg = IOConfig(opendal_backends={"oss": oss})
+
self.assertEqual(IOConfig._from_serialized(serialize_io_config(cfg)).opendal_backends["oss"],
oss)
+
+ def test_file_io_config_routes_oss_through_s3(self):
+ # OSS -> Daft S3 client (File.open() over OpenDAL/OSS is broken on
some Daft builds).
+ cfg = _convert_paimon_catalog_options_to_file_io_config({
+ "warehouse": "oss://b", "fs.oss.endpoint": "oss-test.example.com",
+ "fs.oss.region": "test-region", "fs.oss.accessKeyId": "AK",
+ "fs.oss.accessKeySecret": "SK", "fs.oss.securityToken": "TOK",
+ })
+ self.assertEqual(cfg.s3.key_id, "AK")
+ self.assertEqual(cfg.s3.endpoint_url, "https://oss-test.example.com";)
+ self.assertTrue(cfg.s3.force_virtual_addressing)
+ self.assertEqual(dict(cfg.protocol_aliases)["oss"], "s3")
+ # No credentials: None when required; oss->s3 alias otherwise
(env/instance creds).
+
self.assertIsNone(_convert_paimon_catalog_options_to_file_io_config({"warehouse":
"oss://b"}))
+ env = _convert_paimon_catalog_options_to_file_io_config({"warehouse":
"oss://b"}, require_credentials=False)
+ self.assertEqual(dict(env.protocol_aliases)["oss"], "s3")
+
+ def test_explicit_io_config_used_when_no_derivable_credentials(self):
+ # Explicit io_config must reach blob File columns when no complete
creds are derivable.
+ explicit = serialize_io_config(IOConfig(s3=S3Config(key_id="USERKEY",
access_key="SK")))
+
+ def blob_key(catalog_options):
+ task = _PaimonPKSplitTask(catalog_options, None, None, {}, None,
None,
+ blob_column_names={"x"},
explicit_io_config_bytes=explicit)
+ return
IOConfig._from_serialized(task._blob_io_config_bytes(None)).s3.key_id
+
+ self.assertEqual(blob_key({}), "USERKEY")
+ for opts in (
+ {"warehouse": "oss://b", "fs.oss.endpoint":
"oss-test.example.com"},
+ {"warehouse": "oss://b", "fs.oss.accessKeyId": "PARTIAL"}, # no
secret
+ {"warehouse": "s3://b", "fs.s3.endpoint":
"https://s3.example.com"},
+ {"warehouse": "s3a://b", "fs.s3.accessKeyId": "PARTIAL"}, # no
secret
+ ):
+ self.assertEqual(blob_key(opts), "USERKEY")
+
+ def test_with_oss_alias_augments_explicit_config(self):
+ # s3 config -> add oss->s3 alias + virtual-hosted, keep creds.
+ from pypaimon.daft.daft_io_config import _with_oss_alias
+ s3 = _with_oss_alias(IOConfig(s3=S3Config(key_id="AK",
access_key="SK")))
+ self.assertEqual(dict(s3.protocol_aliases)["oss"], "s3")
+ self.assertTrue(s3.s3.force_virtual_addressing)
+ self.assertEqual(s3.s3.key_id, "AK")
+ # opendal OSS config -> converted to S3Config + alias.
+ od = _with_oss_alias(IOConfig(opendal_backends={"oss": {
+ "access_key_id": "AK", "access_key_secret": "SK", "endpoint":
"https://oss-test.example.com"}}))
+ self.assertEqual(dict(od.protocol_aliases)["oss"], "s3")
+ self.assertEqual(od.s3.key_id, "AK")
+ self.assertEqual(od.s3.endpoint_url, "https://oss-test.example.com";)
+
+ def test_explicit_io_config_gets_oss_alias_for_oss_blobs(self):
+ # An oss:// table's explicit-fallback io_config gets the s3 alias so
File.open() works.
+ class _OssTable:
+ table_path = "oss://b/db.db/t"
+
+ explicit = serialize_io_config(IOConfig(s3=S3Config(key_id="USERKEY",
access_key="SK")))
+ task = _PaimonPKSplitTask({}, None, None, {}, None, None,
+ blob_column_names={"x"},
explicit_io_config_bytes=explicit)
+ cfg =
IOConfig._from_serialized(task._blob_io_config_bytes(_OssTable()))
+ self.assertEqual(cfg.s3.key_id, "USERKEY")
+ self.assertEqual(dict(cfg.protocol_aliases)["oss"], "s3")
+
+ def test_oss_partial_credentials_cleared_for_env(self):
+ # OSS half key pair (no secret): keep only the oss->s3 alias, drop the
partial key.
+ opts = {"warehouse": "oss://b", "fs.oss.endpoint":
"oss-test.example.com",
+ "fs.oss.accessKeyId": "PARTIAL"}
+
self.assertIsNone(_convert_paimon_catalog_options_to_file_io_config(opts))
+ cfg = _convert_paimon_catalog_options_to_file_io_config(opts,
require_credentials=False)
+ self.assertIsNone(cfg.s3.key_id)
+ self.assertEqual(dict(cfg.protocol_aliases)["oss"], "s3")
+
+ def test_non_oss_endpoint_kept_for_env_without_credentials(self):
+ # Custom-endpoint S3 (MinIO/Ceph), no key pair: None when required;
env keeps endpoint, no creds.
+ opts = {"warehouse": "s3a://b", "fs.s3.endpoint":
"https://minio.example.com"}
+
self.assertIsNone(_convert_paimon_catalog_options_to_file_io_config(opts))
+ env = _convert_paimon_catalog_options_to_file_io_config(opts,
require_credentials=False)
+ self.assertEqual(env.s3.endpoint_url, "https://minio.example.com";)
+ self.assertIsNone(env.s3.key_id)
+ # A half key is dropped; the endpoint is still kept.
+ partial = _convert_paimon_catalog_options_to_file_io_config(
+ {"warehouse": "s3://b", "fs.s3.endpoint":
"https://minio.example.com";,
+ "fs.s3.accessKeyId": "PARTIAL"}, require_credentials=False)
+ self.assertEqual(partial.s3.endpoint_url, "https://minio.example.com";)
+ self.assertIsNone(partial.s3.key_id)
+
+ @pytest.mark.skipif(not has_file_range_reads(), reason="daft >= 0.7.11
required for File range metadata")
+ def test_cast_to_file_reconstructs_io_config(self):
+ # The crux: embedded bytes must survive the cast to DataType.file().
+ blob = serialize_io_config(IOConfig(s3=S3Config(key_id="AK",
region_name="test-region")))
+ arr = blob_column_to_file_array(_descriptor_column([("s3://b/k", 0,
4)]), blob)
+ df = daft.from_arrow(pa.table({"f": arr}))
+ df = df.with_column("f", df["f"].cast(DataType.file()))
+ restored =
IOConfig._from_serialized(df.to_arrow().column("f")[0].as_py()["io_config"])
+ self.assertEqual(restored.s3.key_id, "AK")
+ self.assertEqual(restored.s3.region_name, "test-region")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/daft/daft_catalog_rest_test.py
b/paimon-python/pypaimon/tests/daft/daft_catalog_rest_test.py
index 2936fded1c..afcecfbd1c 100644
--- a/paimon-python/pypaimon/tests/daft/daft_catalog_rest_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_catalog_rest_test.py
@@ -243,12 +243,13 @@ class DaftRestReadTest(RESTBaseTest):
captured = {}
original_init = PaimonDataSource.__init__
- def spy_init(_self, table, storage_config, catalog_options):
+ def spy_init(_self, table, storage_config, catalog_options, **kwargs):
captured["catalog_options"] = dict(catalog_options)
return original_init(
_self, table,
storage_config=storage_config,
catalog_options=catalog_options,
+ **kwargs,
)
with patch.object(PaimonDataSource, "__init__", spy_init):
