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 a52df860f8 [python] Fix mounted BLOB reads bypassing table FileIO
(#10621)
a52df860f8 is described below
commit a52df860f87dd74a220b8aa18d8b7db460147155
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Oct 17 22:09:53 2026 +0800
[python] Fix mounted BLOB reads bypassing table FileIO (#10621)
---
.../pypaimon/filesystem/oss_blob_presigner.py | 24 ++-----
paimon-python/pypaimon/filesystem/uri_utils.py | 46 ++++++++++++
paimon-python/pypaimon/multimodal/blob_store.py | 16 +++++
.../pypaimon/tests/multimodal_table_test.py | 83 +++++++++++++++++++++-
4 files changed, 149 insertions(+), 20 deletions(-)
diff --git a/paimon-python/pypaimon/filesystem/oss_blob_presigner.py
b/paimon-python/pypaimon/filesystem/oss_blob_presigner.py
index 3f02e3d6b6..a5231b1258 100644
--- a/paimon-python/pypaimon/filesystem/oss_blob_presigner.py
+++ b/paimon-python/pypaimon/filesystem/oss_blob_presigner.py
@@ -20,10 +20,11 @@
import hashlib
import logging
import posixpath
-import re
from datetime import timedelta
from urllib.parse import unquote, urlparse, urlsplit
+from pypaimon.filesystem.uri_utils import normalize_uri_path,
table_relative_path
+
_LOGGER = logging.getLogger(__name__)
@@ -104,24 +105,9 @@ def _validity_seconds(validity, has_security_token=False)
-> int:
def _validate_table_root(table_root, descriptor):
root = urlparse(str(table_root))
blob = urlparse(descriptor.uri)
- root_path = _normalize_path(root.path)
- blob_path = _normalize_path(blob.path)
- if ((root.scheme or '').lower() != (blob.scheme or '').lower()
- or root.netloc != blob.netloc):
- raise ValueError(
- "Blob descriptor URI must use the same scheme and authority "
- "as table root.")
- child_prefix = root_path if root_path.endswith('/') else root_path + '/'
- if not blob_path.startswith(child_prefix):
- raise ValueError("Blob descriptor URI must be under table root.")
- return blob._replace(path=blob_path)
-
-
-def _normalize_path(path):
- normalized = posixpath.normpath(re.sub('/+', '/', path or '/'))
- if path.startswith('/') and not normalized.startswith('/'):
- normalized = '/' + normalized
- return normalized
+ relative_path = table_relative_path(table_root, descriptor.uri)
+ return blob._replace(
+ path=posixpath.join(normalize_uri_path(root.path), relative_path))
def _head_object_if_exists(bucket, key):
diff --git a/paimon-python/pypaimon/filesystem/uri_utils.py
b/paimon-python/pypaimon/filesystem/uri_utils.py
new file mode 100644
index 0000000000..03ba7c64ec
--- /dev/null
+++ b/paimon-python/pypaimon/filesystem/uri_utils.py
@@ -0,0 +1,46 @@
+# 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.
+
+"""URI path containment shared by managed BLOB readers and presigning."""
+
+import posixpath
+import re
+from urllib.parse import urlparse
+
+
+def normalize_uri_path(path):
+ normalized = posixpath.normpath(re.sub('/+', '/', path or '/'))
+ if path.startswith('/') and not normalized.startswith('/'):
+ normalized = '/' + normalized
+ return normalized
+
+
+def table_relative_path(table_root, uri):
+ """Return a normalized child path, rejecting URIs outside the table
root."""
+ root = urlparse(str(table_root))
+ child = urlparse(str(uri))
+ if ((root.scheme or '').lower() != (child.scheme or '').lower()
+ or root.netloc != child.netloc):
+ raise ValueError(
+ "Blob descriptor URI must use the same scheme and authority "
+ "as table root.")
+
+ root_path = normalize_uri_path(root.path)
+ child_path = normalize_uri_path(child.path)
+ child_prefix = root_path if root_path.endswith('/') else root_path + '/'
+ if not child_path.startswith(child_prefix):
+ raise ValueError("Blob descriptor URI must be under table root.")
+ return child_path[len(child_prefix):]
diff --git a/paimon-python/pypaimon/multimodal/blob_store.py
b/paimon-python/pypaimon/multimodal/blob_store.py
index 061402ba87..2d92b3ee92 100644
--- a/paimon-python/pypaimon/multimodal/blob_store.py
+++ b/paimon-python/pypaimon/multimodal/blob_store.py
@@ -20,9 +20,11 @@ import re
from dataclasses import dataclass
from datetime import timedelta
from typing import BinaryIO, Dict, Iterable, List, Mapping, Optional, Sequence
+from urllib.parse import urlparse
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.filesystem.uri_utils import table_relative_path
from pypaimon.schema.data_types import is_blob_type
from pypaimon.table.row.blob import Blob, BlobData, BlobDescriptor
from pypaimon.table.row.generic_row import GenericRow
@@ -70,6 +72,20 @@ class BlobObject:
def open(self) -> BinaryIO:
descriptor = _descriptor_for_range(self.descriptor, self.range_header)
+ if self.table_root is not None:
+ parsed = urlparse(descriptor.uri)
+ if not (parsed.params or parsed.query or parsed.fragment):
+ try:
+ relative_path = table_relative_path(
+ self.table_root, descriptor.uri)
+ except ValueError:
+ pass
+ else:
+ # Use the table root's exact spelling for FUSE prefix
mapping.
+ managed_uri = self.table_root.rstrip("/") + "/" +
relative_path
+ return Blob.from_file(
+ self.file_io, managed_uri, descriptor.offset,
+ descriptor.length).new_input_stream()
uri_reader = self.file_io.uri_reader_factory.create(descriptor.uri)
return Blob.from_descriptor(uri_reader, descriptor).new_input_stream()
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index 629fffbca0..7154d27de7 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -1142,11 +1142,92 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(len(data), result.size)
self.assertEqual("videos/2", more[0].key)
obj = store.get_object("videos/1")
- self.assertEqual(data, obj.read())
+ with patch.object(obj.file_io, "new_input_stream",
side_effect=AssertionError(
+ "external object must use its URI reader")):
+ self.assertEqual(data, obj.read())
self.assertEqual("video/mp4", obj.columns["media_type"])
self.assertEqual(external_path,
store.head_object("videos/1").descriptor.uri)
self.assertEqual(data, store.get_object("videos/2").read())
+ def test_blob_store_managed_object_reads_through_fuse_file_io(self):
+ from pypaimon.common.file_io import FileIO
+ from pypaimon.filesystem.local_file_io import FuseLocalFileIO
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ table = self.conn.create_table(
+ "mounted_objects",
+ schema=_schema({"key": pa.string(), "payload": pa.large_binary()}),
+ options=_PARQUET_OPTIONS,
+ )
+ store = table.blobs(column="payload")
+ store.put_object("example.hdf5", body=b"hdf5-payload")
+ obj = store.get_object("example.hdf5")
+
+ remote_root = "oss://test-bucket/table"
+ remote_uri = remote_root + obj.descriptor.uri[len(obj.table_root):]
+ mounted = replace(
+ obj,
+ descriptor=BlobDescriptor(
+ remote_uri, obj.descriptor.offset, obj.descriptor.length),
+ file_io=FuseLocalFileIO(remote_root, obj.table_root),
+ table_root=remote_root,
+ )
+
+ with patch.object(FileIO, "get", side_effect=AssertionError(
+ "managed object must use the table FileIO")):
+ with mounted.open() as stream:
+ self.assertEqual(b"hdf5-payload", stream.read())
+ self.assertEqual(b"df5", replace(
+ mounted, range_header="bytes=1-3").read())
+
+ suffix = remote_uri[len(remote_root):]
+ for variant in (
+ "OSS://test-bucket/table" + suffix,
+ remote_root + "//./" + suffix.lstrip("/"),
+ remote_root + "/sub/../" + suffix.lstrip("/")):
+ with self.subTest(uri=variant), patch.object(
+ mounted.file_io.uri_reader_factory, "create",
+ side_effect=AssertionError("managed object must not use a
URI reader")):
+ normalized = replace(
+ mounted,
+ descriptor=BlobDescriptor(
+ variant, obj.descriptor.offset, obj.descriptor.length),
+ )
+ self.assertEqual(b"hdf5-payload", normalized.read())
+
+ def test_blob_store_external_uris_do_not_use_fuse_file_io(self):
+ from pypaimon.filesystem.local_file_io import FuseLocalFileIO
+ from pypaimon.multimodal.blob_store import BlobObject
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ class ExternalReader:
+
+ def new_input_stream(self, uri):
+ return io.BytesIO(b"external")
+
+ root = "oss://test-bucket/table"
+ file_io = FuseLocalFileIO(root, self.temp_dir)
+ for uri in (
+ root + "/../other/payload.bin",
+ "oss://other-bucket/table/payload.bin",
+ "oss://test-bucket/table-sibling/payload.bin",
+ "s3://test-bucket/table/payload.bin",
+ root + "/payload.bin?token=external"):
+ with self.subTest(uri=uri), patch.object(
+ file_io, "new_input_stream",
+ side_effect=AssertionError("external object must not use
table FileIO")), \
+ patch.object(file_io.uri_reader_factory, "create",
+ return_value=ExternalReader()) as
create_reader:
+ obj = BlobObject(
+ key="external",
+ descriptor=BlobDescriptor(uri, 0, len(b"external")),
+ columns={},
+ file_io=file_io,
+ table_root=root,
+ )
+ self.assertEqual(b"external", obj.read())
+ create_reader.assert_called_once_with(uri)
+
def test_blob_store_put_object_uri_streams_into_managed_blob(self):
table = self.conn.create_table(
"managed_only_objects",
