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 b727e235dc [java][python] Support OSS blob presigned URLs (#10040)
b727e235dc is described below
commit b727e235dc342c8f5143cc4277c8c1bcff2def99
Author: XiaoHongbo <[email protected]>
AuthorDate: Tue Sep 22 11:27:58 2026 +0800
[java][python] Support OSS blob presigned URLs (#10040)
---
docs/docs/multimodal-table/blob-references.mdx | 22 +-
docs/docs/pypaimon/blob-store.md | 14 +
.../org/apache/paimon/rest/RESTTokenFileIO.java | 90 +++-
.../apache/paimon/rest/RESTTokenFileIOTest.java | 129 +++++
.../org/apache/paimon/oss/OSSBlobPresigner.java | 25 +-
.../java/org/apache/paimon/oss/OSSFileIOTest.java | 12 +-
.../pypaimon/catalog/rest/rest_token_file_io.py | 71 ++-
paimon-python/pypaimon/common/file_io.py | 11 +
.../pypaimon/filesystem/caching_file_io.py | 3 +
.../pypaimon/filesystem/oss_blob_presigner.py | 250 ++++++++++
paimon-python/pypaimon/filesystem/oss_file_io.py | 58 ++-
.../pypaimon/filesystem/resolving_file_io.py | 4 +
paimon-python/pypaimon/multimodal/blob_store.py | 14 +
paimon-python/pypaimon/table/row/blob.py | 5 +
.../pypaimon/tests/multimodal_table_test.py | 21 +
.../pypaimon/tests/oss_blob_presigner_test.py | 530 +++++++++++++++++++++
.../pypaimon/tests/rest/rest_token_file_io_test.py | 121 +++++
17 files changed, 1318 insertions(+), 62 deletions(-)
diff --git a/docs/docs/multimodal-table/blob-references.mdx
b/docs/docs/multimodal-table/blob-references.mdx
index 0c81bf5a0e..898b2fb23e 100644
--- a/docs/docs/multimodal-table/blob-references.mdx
+++ b/docs/docs/multimodal-table/blob-references.mdx
@@ -186,9 +186,9 @@ scheme and authority. For a descriptor forwarded from
another table, pass the
original owning table as `source_table`; using the downstream descriptor table
or
an arbitrary external object path fails this check.
-Configure the catalog with the standard public HTTPS OSS endpoint. Internal
endpoints and
-endpoints without the `https` scheme are rejected because external consumers
must be able to fetch
-the returned URL:
+Use an HTTPS OSS endpoint. Java and PyPaimon preserve the configured endpoint,
including
+Internal endpoints, in the returned URL. Both the application and the URL
consumer must be
+able to reach that endpoint. For consumers that require public access,
configure a public endpoint:
```properties
fs.oss.endpoint=https://oss-cn-hangzhou.aliyuncs.com
@@ -246,6 +246,22 @@ String url =
table.fileIO(), table.location(), Duration.ofMinutes(5));
```
+PyPaimon exposes the same operation for descriptor-backed `Blob` values. Pass
the owning table's
+FileIO and storage path so PyPaimon can validate that the descriptor belongs
to the table:
+
+```python
+from datetime import timedelta
+
+url = blob.to_presigned_url(
+ table.file_io,
+ table.table_path,
+ timedelta(minutes=5),
+)
+```
+
+Install the OSS dependencies with `pip install 'pypaimon[oss]'`. For DLF REST
catalogs, PyPaimon
+refreshes the STS credentials before creating the materialized object and URL.
+
The materialized object's bytes are unchanged. Consumers must inspect the
content instead of
relying on a URL suffix or a format-specific content type. For direct model
`image_url` inputs, use
only formats verified with the target model; PDF is not covered by this entry
point.
diff --git a/docs/docs/pypaimon/blob-store.md b/docs/docs/pypaimon/blob-store.md
index faeb717d2f..e5d8cebab4 100644
--- a/docs/docs/pypaimon/blob-store.md
+++ b/docs/docs/pypaimon/blob-store.md
@@ -154,6 +154,20 @@ for obj_info in objects:
print(obj_info.key, obj_info.size, obj_info.columns)
```
+### Create a Presigned URL
+
+For an OSS-backed object, generate a temporary URL without downloading its
bytes:
+
+```python
+from datetime import timedelta
+
+video = table.blobs(column="video", key_column="id").get_object("video-001")
+video_url = video.to_presigned_url(timedelta(minutes=30))
+```
+
+The URL preserves the configured OSS endpoint and any byte range selected by
+`get_object`. The descriptor must belong to the table.
+
### Delete Objects
`delete_object` deletes one object key. `delete_objects` deletes a batch of
keys
diff --git
a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
index d3618a3c14..6e469db3ee 100644
--- a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java
@@ -191,7 +191,35 @@ public class RESTTokenFileIO implements FileIO {
if (!path.equals(tableRoot)) {
throw new IOException("Table root does not match RESTTokenFileIO
bound table root.");
}
- return fileIO().createBlobPresignedUrl(tableRoot, descriptor,
validity);
+ if (validity == null
+ || validity.isZero()
+ || validity.isNegative()
+ || validity.getNano() != 0) {
+ throw new IOException("Blob presigned URL validity must be
positive whole seconds.");
+ }
+ final long validityMillis;
+ try {
+ validityMillis = validity.toMillis();
+ } catch (ArithmeticException e) {
+ throw new IOException("Blob presigned URL validity is too large.",
e);
+ }
+
+ FileIOWithToken first = fileIOWithToken(validityMillis);
+ String url = first.fileIO.createBlobPresignedUrl(tableRoot,
descriptor, validity);
+ if (hasRemainingLifetime(first.token, validityMillis)) {
+ return url;
+ }
+
+ // The first call materialized the range. Refresh and sign the cached
object again so the
+ // returned URL has the requested lifetime.
+ FileIOWithToken refreshed = fileIOWithToken(validityMillis);
+ url = refreshed.fileIO.createBlobPresignedUrl(tableRoot, descriptor,
validity);
+ if (!hasRemainingLifetime(refreshed.token, validityMillis)) {
+ throw new IOException(
+ "Requested presigned URL validity exceeds the remaining "
+ + "REST credential lifetime after refresh.");
+ }
+ return url;
}
@Override
@@ -204,21 +232,31 @@ public class RESTTokenFileIO implements FileIO {
}
public FileIO fileIO() throws IOException {
- tryToRefreshToken();
+ return fileIOWithToken(0).fileIO;
+ }
+
+ private FileIOWithToken fileIOWithToken(long minimumValidityMillis) throws
IOException {
+ tryToRefreshToken(minimumValidityMillis);
+ RESTToken currentToken = token;
+ if (!hasRemainingLifetime(currentToken, minimumValidityMillis)) {
+ throw new IOException(
+ "Requested presigned URL validity exceeds the remaining "
+ + "REST credential lifetime after refresh.");
+ }
- FileIO fileIO = FILE_IO_CACHE.getIfPresent(token);
+ FileIO fileIO = FILE_IO_CACHE.getIfPresent(currentToken);
if (fileIO != null) {
- return fileIO;
+ return new FileIOWithToken(fileIO, currentToken);
}
synchronized (FILE_IO_CACHE) {
- fileIO = FILE_IO_CACHE.getIfPresent(token);
+ fileIO = FILE_IO_CACHE.getIfPresent(currentToken);
if (fileIO != null) {
- return fileIO;
+ return new FileIOWithToken(fileIO, currentToken);
}
Options options = catalogContext.options();
- options = new Options(RESTUtil.merge(options.toMap(),
token.token()));
+ options = new Options(RESTUtil.merge(options.toMap(),
currentToken.token()));
options.set(FILE_IO_ALLOW_CACHE, false);
CatalogContext context =
CatalogContext.create(
@@ -227,25 +265,49 @@ public class RESTTokenFileIO implements FileIO {
catalogContext.preferIO(),
catalogContext.fallbackIO());
fileIO = FileIO.get(path, context);
- FILE_IO_CACHE.put(token, fileIO);
- return fileIO;
+ FILE_IO_CACHE.put(currentToken, fileIO);
+ return new FileIOWithToken(fileIO, currentToken);
}
}
+ private boolean hasRemainingLifetime(RESTToken signingToken, long
minimumValidityMillis) {
+ return signingToken == null
+ || signingToken.expireAtMillis() - currentTimeMillis() >=
minimumValidityMillis;
+ }
+
+ long currentTimeMillis() {
+ return System.currentTimeMillis();
+ }
+
private void tryToRefreshToken() {
- if (shouldRefresh()) {
+ tryToRefreshToken(0);
+ }
+
+ private void tryToRefreshToken(long minimumValidityMillis) {
+ if (shouldRefresh(minimumValidityMillis)) {
synchronized (this) {
- if (shouldRefresh()) {
+ if (shouldRefresh(minimumValidityMillis)) {
refreshToken();
}
}
}
}
- private boolean shouldRefresh() {
+ private boolean shouldRefresh(long minimumValidityMillis) {
return token == null
- || token.expireAtMillis() - System.currentTimeMillis()
- < TOKEN_EXPIRATION_SAFE_TIME_MILLIS;
+ || token.expireAtMillis() - currentTimeMillis()
+ < Math.max(TOKEN_EXPIRATION_SAFE_TIME_MILLIS,
minimumValidityMillis);
+ }
+
+ private static class FileIOWithToken {
+
+ private final FileIO fileIO;
+ private final RESTToken token;
+
+ private FileIOWithToken(FileIO fileIO, RESTToken token) {
+ this.fileIO = fileIO;
+ this.token = token;
+ }
}
private void refreshToken() {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java
b/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java
index 8aefb6b4a6..e80f097fcf 100644
---
a/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java
@@ -35,12 +35,14 @@ import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -97,6 +99,133 @@ class RESTTokenFileIOTest {
.hasMessageContaining("bound table root");
}
+ @Test
+ void testPresigningRefreshesForRequestedLifetime() throws IOException {
+ checkPresignedLifetime(
+ Duration.ofHours(3), Duration.ofHours(2), Duration.ofHours(4),
true, false);
+ }
+
+ @Test
+ void testPresigningReusesSufficientLifetime() throws IOException {
+ checkPresignedLifetime(
+ Duration.ofMinutes(30), Duration.ofHours(2),
Duration.ofHours(4), false, false);
+ }
+
+ @Test
+ void testPresigningAcceptsWhenOneSecondExceedsValidity() throws
IOException {
+ checkPresignedLifetime(
+ Duration.ofMinutes(90),
+ Duration.ofMinutes(90).plusSeconds(1),
+ Duration.ofHours(3),
+ false,
+ false);
+ }
+
+ @Test
+ void testPresigningRejectsInsufficientRefreshedLifetime() throws
IOException {
+ checkPresignedLifetime(
+ Duration.ofHours(3), Duration.ofHours(2), Duration.ofHours(2),
true, true);
+ }
+
+ private void checkPresignedLifetime(
+ Duration validity,
+ Duration initialLifetime,
+ Duration refreshedLifetime,
+ boolean refresh,
+ boolean rejected)
+ throws IOException {
+ Path root = new Path("oss://bucket/table");
+ BlobDescriptor descriptor = new
BlobDescriptor("oss://bucket/table/data.blob", 0, 1);
+ FileIO delegate = mock(FileIO.class);
+ when(delegate.exists(any())).thenReturn(true);
+ when(delegate.createBlobPresignedUrl(root, descriptor, validity))
+ .thenReturn("https://signed";);
+ FileIOLoader loader = mock(FileIOLoader.class);
+ when(loader.load(any())).thenReturn(delegate);
+ when(loader.getScheme()).thenReturn("oss");
+ RESTApi api = mock(RESTApi.class);
+ Identifier identifier = Identifier.create("db", "table");
+ long now = System.currentTimeMillis();
+ when(api.loadTableToken(identifier))
+ .thenReturn(
+ new GetTableTokenResponse(
+ Collections.singletonMap(
+ "test.token",
UUID.randomUUID().toString()),
+ now + initialLifetime.toMillis()),
+ new GetTableTokenResponse(
+ Collections.singletonMap(
+ "test.token",
UUID.randomUUID().toString()),
+ now + refreshedLifetime.toMillis()));
+ RESTTokenFileIO fileIO =
+ new RESTTokenFileIO(
+ CatalogContext.create(new Options(), loader, null),
api, identifier, root) {
+ @Override
+ long currentTimeMillis() {
+ return now;
+ }
+ };
+ fileIO.validToken();
+ if (rejected) {
+ assertThatThrownBy(() -> fileIO.createBlobPresignedUrl(root,
descriptor, validity))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("credential lifetime after refresh");
+ verify(delegate, never()).createBlobPresignedUrl(any(), any(),
any());
+ } else {
+ assertThat(fileIO.createBlobPresignedUrl(root, descriptor,
validity))
+ .isEqualTo("https://signed";);
+ verify(delegate).createBlobPresignedUrl(root, descriptor,
validity);
+ }
+ verify(api, times(refresh ? 2 : 1)).loadTableToken(identifier);
+ }
+
+ @Test
+ void testPresigningRefreshesAndResignsAfterMaterialization() throws
IOException {
+ Path root = new Path("oss://bucket/table");
+ BlobDescriptor descriptor = new
BlobDescriptor("oss://bucket/table/data.blob", 0, 1);
+ Duration validity = Duration.ofMinutes(30);
+ AtomicLong now = new AtomicLong(1700000000000L);
+ FileIO delegate = mock(FileIO.class);
+ when(delegate.exists(any())).thenReturn(true);
+ when(delegate.createBlobPresignedUrl(root, descriptor, validity))
+ .thenAnswer(
+ ignored -> {
+ if (now.get() == 1700000000000L) {
+ now.addAndGet(Duration.ofHours(2).toMillis());
+ return "https://first";;
+ }
+ return "https://refreshed";;
+ });
+ FileIOLoader loader = mock(FileIOLoader.class);
+ when(loader.load(any())).thenReturn(delegate);
+ when(loader.getScheme()).thenReturn("oss");
+ RESTApi api = mock(RESTApi.class);
+ Identifier identifier = Identifier.create("db", "table");
+ when(api.loadTableToken(identifier))
+ .thenReturn(
+ new GetTableTokenResponse(
+ Collections.singletonMap(
+ "test.token",
UUID.randomUUID().toString()),
+ now.get() + Duration.ofHours(2).toMillis()),
+ new GetTableTokenResponse(
+ Collections.singletonMap(
+ "test.token",
UUID.randomUUID().toString()),
+ now.get() + Duration.ofHours(4).toMillis()));
+ RESTTokenFileIO fileIO =
+ new RESTTokenFileIO(
+ CatalogContext.create(new Options(), loader, null),
api, identifier, root) {
+ @Override
+ long currentTimeMillis() {
+ return now.get();
+ }
+ };
+
+ fileIO.validToken();
+ assertThat(fileIO.createBlobPresignedUrl(root, descriptor, validity))
+ .isEqualTo("https://refreshed";);
+ verify(delegate, times(2)).createBlobPresignedUrl(root, descriptor,
validity);
+ verify(api, times(2)).loadTableToken(identifier);
+ }
+
@Test
void testFileIOCreationFailureSurfacesAsCheckedIOException() throws
IOException {
Path tableRoot = new Path("resttoken-broken://bucket/table");
diff --git
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSBlobPresigner.java
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSBlobPresigner.java
index 09aefa263b..55f9089e32 100644
---
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSBlobPresigner.java
+++
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSBlobPresigner.java
@@ -45,14 +45,12 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
-import java.util.Locale;
/** Materializes blob ranges and creates OSS presigned URLs. */
public final class OSSBlobPresigner {
private static final String BLOB_FINGERPRINT_METADATA =
"paimon-blob-descriptor-sha256";
private static final String BLOB_CONTENT_TYPE = "application/octet-stream";
- private static final String OSS_INTERNAL_ENDPOINT_SUFFIX =
"-internal.aliyuncs.com";
private static final long BLOB_COPY_MIN_PART_SIZE = 100L * 1024 * 1024;
private static final long MAX_MULTIPART_UPLOAD_PARTS = 10_000;
private static final char[] HEX_CHARS = "0123456789abcdef".toCharArray();
@@ -96,7 +94,6 @@ public final class OSSBlobPresigner {
targetKey,
Date.from(Instant.now().plus(validity)),
HttpMethod.GET);
- url = usePublicEndpoint(url);
validatePresignedUrl(client, url, bucket, targetKey);
return url.toString();
} catch (IOException e) {
@@ -222,7 +219,7 @@ public final class OSSBlobPresigner {
private static void validatePresignedUrl(
OSSClient client, URL url, String bucket, String targetKey) throws
Exception {
URI endpoint = client.getEndpoint();
- String expectedHost = publicEndpointHost(bucket + "." +
endpoint.getHost());
+ String expectedHost = bucket + "." + endpoint.getHost();
if (!"https".equalsIgnoreCase(endpoint.getScheme())
|| !"https".equalsIgnoreCase(url.getProtocol())
|| !expectedHost.equalsIgnoreCase(url.getHost())
@@ -230,24 +227,4 @@ public final class OSSBlobPresigner {
throw new IOException("OSS client generated a presigned URL for an
invalid target.");
}
}
-
- private static URL usePublicEndpoint(URL url) throws IOException {
- String publicHost = publicEndpointHost(url.getHost());
- if (publicHost.equals(url.getHost())) {
- return url;
- }
- String file = url.getFile();
- if (url.getRef() != null) {
- file += "#" + url.getRef();
- }
- return new URL(url.getProtocol(), publicHost, url.getPort(), file);
- }
-
- private static String publicEndpointHost(String host) {
- if
(!host.toLowerCase(Locale.ROOT).endsWith(OSS_INTERNAL_ENDPOINT_SUFFIX)) {
- return host;
- }
- return host.substring(0, host.length() -
OSS_INTERNAL_ENDPOINT_SUFFIX.length())
- + ".aliyuncs.com";
- }
}
diff --git
a/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOTest.java
b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOTest.java
index 0e68472e8a..e801d25f81 100644
---
a/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOTest.java
+++
b/paimon-filesystems/paimon-oss-impl/src/test/java/org/apache/paimon/oss/OSSFileIOTest.java
@@ -152,7 +152,7 @@ public class OSSFileIOTest {
}
@Test
- public void
testCreateBlobPresignedUrlUsesPublicEndpointWithoutChangingPath() throws
Exception {
+ public void testCreateBlobPresignedUrlPreservesInternalEndpoint() throws
Exception {
OSSClient client = mock(OSSClient.class);
BlobDescriptor descriptor =
new
BlobDescriptor("oss://bucket/table/-internal.aliyuncs.com/source.blob", 0, 1);
@@ -176,7 +176,7 @@ public class OSSFileIOTest {
descriptor,
Duration.ofMinutes(5)));
-
assertThat(url.getHost()).isEqualTo("bucket.oss-cn-hangzhou.aliyuncs.com");
+
assertThat(url.getHost()).isEqualTo("bucket.oss-cn-hangzhou-internal.aliyuncs.com");
assertThat(url.getPath())
.isEqualTo(
"/table/-internal.aliyuncs.com/_bloburl_"
@@ -316,7 +316,13 @@ public class OSSFileIOTest {
}
@Test
- public void testCreateBlobPresignedUrlRejectsInvalidTarget() {
+ public void testCreateBlobPresignedUrlRejectsInvalidTarget() throws
Exception {
+ assertInvalidPresignedUrl(
+ "https://oss-cn-hangzhou-internal.aliyuncs.com";,
+ "https://bucket.oss-cn-hangzhou.aliyuncs.com/table/_bloburl_";
+ + sha256Hex(
+ new
BlobDescriptor("oss://bucket/table/source.blob", 0, 1)
+ .serialize()));
assertInvalidPresignedUrl(
"https://oss-cn-hangzhou.aliyuncs.com";,
"http://bucket.oss-cn-hangzhou.aliyuncs.com/table/_bloburl_hash";);
diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
index 6a87d004eb..2931712851 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py
@@ -18,6 +18,7 @@
import logging
import threading
import time
+from datetime import timedelta
from typing import Optional, Union
from cachetools import TTLCache
@@ -86,21 +87,33 @@ class RESTTokenFileIO(FileIO):
self._init_file_io_cache()
def file_io(self) -> FileIO:
+ return self._file_io_with_token()[0]
+
+ def _file_io_with_token(self, minimum_validity_millis: int = 0):
with self._file_io_cache_lock:
- self.try_to_refresh_token()
+ if minimum_validity_millis:
+ self.try_to_refresh_token(minimum_validity_millis)
+ if not self._has_remaining_lifetime(
+ self.token, minimum_validity_millis):
+ raise ValueError(
+ "Requested presigned URL validity exceeds the
remaining "
+ "REST credential lifetime after refresh.")
+ else:
+ self.try_to_refresh_token()
if self.token is None:
- return FileIO.get(self.path, self.catalog_options or
Options({}))
+ return (FileIO.get(
+ self.path, self.catalog_options or Options({})), None)
cache_key = self.token
cache = self._file_io_cache
file_io = cache.get(cache_key)
if file_io is not None:
- return file_io
+ return file_io, cache_key
merged_properties = RESTUtil.merge(
self.catalog_options.to_map() if self.catalog_options else {},
- self.token.token
+ cache_key.token
)
if self.catalog_options:
dlf_oss_endpoint =
self.catalog_options.get(CatalogOptions.DLF_OSS_ENDPOINT)
@@ -110,7 +123,7 @@ class RESTTokenFileIO(FileIO):
file_io = FileIO.get(self.path, merged_options)
cache[cache_key] = file_io
- return file_io
+ return file_io, cache_key
def _merge_token_with_catalog_options(self, token: dict) -> dict:
"""Merge token with catalog options, DLF OSS endpoint should override
the standard OSS endpoint."""
@@ -154,6 +167,31 @@ class RESTTokenFileIO(FileIO):
def try_to_write_atomic(self, path: str, content: str) -> bool:
return self.file_io().try_to_write_atomic(path, content)
+ def create_blob_presigned_url(self, table_root, descriptor, validity) ->
str:
+ if table_root != self.path:
+ raise ValueError("Table root does not match RESTTokenFileIO bound
table root.")
+ if not isinstance(validity, timedelta):
+ raise TypeError("Blob presigned URL validity must be
datetime.timedelta.")
+ if validity <= timedelta(0) or validity.microseconds != 0:
+ raise ValueError("Blob presigned URL validity must be positive
whole seconds.")
+ validity_millis = int(validity.total_seconds() * 1000)
+ file_io, signing_token = self._file_io_with_token(validity_millis)
+ url = file_io.create_blob_presigned_url(
+ table_root, descriptor, validity)
+ if self._has_remaining_lifetime(signing_token, validity_millis):
+ return url
+
+ # The first call materialized the range. Refresh and sign the cached
+ # object again so the returned URL has the requested lifetime.
+ file_io, signing_token = self._file_io_with_token(validity_millis)
+ url = file_io.create_blob_presigned_url(
+ table_root, descriptor, validity)
+ if not self._has_remaining_lifetime(signing_token, validity_millis):
+ raise ValueError(
+ "Requested presigned URL validity exceeds the remaining "
+ "REST credential lifetime after refresh.")
+ return url
+
def write_parquet(self, path: str, data, compression: str = 'zstd',
zstd_level: int = 1, **kwargs):
return self.file_io().write_parquet(path, data, compression,
zstd_level, **kwargs)
@@ -207,14 +245,14 @@ class RESTTokenFileIO(FileIO):
return auth_provider.token or 'anonymous'
return 'anonymous'
- def try_to_refresh_token(self):
+ def try_to_refresh_token(self, minimum_validity_millis: int = 0):
identifier_str = self._build_cache_key()
- if self.token is not None and not self._is_token_expired(self.token):
+ if self.token is not None and not self._is_token_expired(self.token,
minimum_validity_millis):
return
cached_token = self._get_cached_token(identifier_str)
- if cached_token and not self._is_token_expired(cached_token):
+ if cached_token and not self._is_token_expired(cached_token,
minimum_validity_millis):
self.token = cached_token
return
@@ -222,12 +260,12 @@ class RESTTokenFileIO(FileIO):
with global_lock:
cached_token = self._get_cached_token(identifier_str)
- if cached_token and not self._is_token_expired(cached_token):
+ if cached_token and not self._is_token_expired(cached_token,
minimum_validity_millis):
self.token = cached_token
return
token_to_check = cached_token if cached_token else self.token
- if token_to_check is None or
self._is_token_expired(token_to_check):
+ if token_to_check is None or
self._is_token_expired(token_to_check, minimum_validity_millis):
self.refresh_token()
self._set_cached_token(identifier_str, self.token)
@@ -239,11 +277,20 @@ class RESTTokenFileIO(FileIO):
with self._TOKEN_LOCKS_LOCK:
self._TOKEN_CACHE[identifier_str] = token
- def _is_token_expired(self, token: Optional[RESTToken]) -> bool:
+ def _is_token_expired(
+ self, token: Optional[RESTToken], minimum_validity_millis: int =
0) -> bool:
if token is None:
return True
current_time = int(time.time() * 1000)
- return (token.expire_at_millis - current_time) <
RESTApi.TOKEN_EXPIRATION_SAFE_TIME_MILLIS
+ return (token.expire_at_millis - current_time) < max(
+ RESTApi.TOKEN_EXPIRATION_SAFE_TIME_MILLIS, minimum_validity_millis)
+
+ @staticmethod
+ def _has_remaining_lifetime(
+ token: Optional[RESTToken], minimum_validity_millis: int) -> bool:
+ return (token is None or
+ token.expire_at_millis - int(time.time() * 1000)
+ >= minimum_validity_millis)
def _get_global_token_lock(self, identifier_str: str) -> threading.Lock:
with self._TOKEN_LOCKS_LOCK:
diff --git a/paimon-python/pypaimon/common/file_io.py
b/paimon-python/pypaimon/common/file_io.py
index 23b2e51248..56c239de11 100644
--- a/paimon-python/pypaimon/common/file_io.py
+++ b/paimon-python/pypaimon/common/file_io.py
@@ -525,6 +525,17 @@ class FileIO(ABC):
target_file = f"{target_directory.rstrip('/')}/{file_name}" if
target_directory else file_name
self.copy_file(source_file, target_file, overwrite)
+ def create_blob_presigned_url(self, table_root, descriptor, validity) ->
str:
+ """Create a temporary URL for a descriptor-backed BLOB.
+
+ File systems which support presigning override this method.
``table_root``
+ identifies the table which owns ``descriptor`` and ``validity`` is a
+ :class:`datetime.timedelta` containing positive whole seconds.
+ """
+ raise NotImplementedError(
+ "{} does not support creating blob presigned URLs.".format(
+ type(self).__name__))
+
def read_overwritten_file_utf8(self, path: str) -> Optional[str]:
retry_number = 0
exception = None
diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py
b/paimon-python/pypaimon/filesystem/caching_file_io.py
index 8b10166461..3fad543d8d 100644
--- a/paimon-python/pypaimon/filesystem/caching_file_io.py
+++ b/paimon-python/pypaimon/filesystem/caching_file_io.py
@@ -436,6 +436,9 @@ class CachingFileIO(FileIO):
def try_to_write_atomic(self, *args, **kwargs):
return self._delegate.try_to_write_atomic(*args, **kwargs)
+ def create_blob_presigned_url(self, *args, **kwargs):
+ return self._delegate.create_blob_presigned_url(*args, **kwargs)
+
def write_parquet(self, *args, **kwargs):
return self._delegate.write_parquet(*args, **kwargs)
diff --git a/paimon-python/pypaimon/filesystem/oss_blob_presigner.py
b/paimon-python/pypaimon/filesystem/oss_blob_presigner.py
new file mode 100644
index 0000000000..3f02e3d6b6
--- /dev/null
+++ b/paimon-python/pypaimon/filesystem/oss_blob_presigner.py
@@ -0,0 +1,250 @@
+# 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.
+
+"""Materialize OSS BLOB ranges and create temporary GET URLs."""
+
+import hashlib
+import logging
+import posixpath
+import re
+from datetime import timedelta
+from urllib.parse import unquote, urlparse, urlsplit
+
+
+_LOGGER = logging.getLogger(__name__)
+
+_BLOB_FINGERPRINT_METADATA = "paimon-blob-descriptor-sha256"
+_BLOB_FINGERPRINT_HEADER = "x-oss-meta-" + _BLOB_FINGERPRINT_METADATA
+_BLOB_CONTENT_TYPE = "application/octet-stream"
+_BLOB_COPY_MIN_PART_SIZE = 100 * 1024 * 1024
+_MAX_MULTIPART_UPLOAD_PARTS = 10_000
+
+
+def create_presigned_url(
+ bucket, table_root, descriptor, validity, sse_headers=None,
+ has_security_token=False) -> str:
+ """Create a presigned URL with the same object layout as Java Paimon."""
+ endpoint = urlsplit(bucket.endpoint)
+ if endpoint.scheme.lower() != "https" or not endpoint.hostname:
+ raise ValueError("OSS BLOB presigning requires an HTTPS endpoint.")
+ validity_seconds = _validity_seconds(validity, has_security_token)
+ source = _validate_table_root(table_root, descriptor)
+ # Keep the compatibility URI rules in OssFileIO, which also owns bucket
+ # extraction for oss://AK:SK@endpoint/bucket/object-key URIs.
+ from pypaimon.filesystem.oss_file_io import OssFileIO
+ source_key = OssFileIO._extract_oss_object_key(source.geturl())
+ if not source_key:
+ raise ValueError("Blob descriptor URI must contain an OSS object key.")
+ sse_headers = dict(sse_headers or {})
+ fingerprint = hashlib.sha256(descriptor.serialize()).hexdigest()
+ parent_end = source_key.rfind('/') + 1
+ target_key = source_key[:parent_end] + "_bloburl_" + fingerprint
+
+ try:
+ target = _head_object_if_exists(bucket, target_key)
+ if not _matches(
+ target, descriptor.length, fingerprint, sse_headers):
+ source_metadata = bucket.head_object(source_key)
+ _validate_range(descriptor, source_metadata.content_length)
+ _materialize(
+ bucket,
+ source_key,
+ target_key,
+ descriptor,
+ fingerprint,
+ sse_headers,
+ )
+ target = bucket.head_object(target_key)
+ if not _matches(
+ target, descriptor.length, fingerprint, sse_headers):
+ raise OSError(
+ "Materialized blob object metadata does not match "
+ "descriptor.")
+
+ url = bucket.sign_url(
+ 'GET', target_key, validity_seconds, slash_safe=True)
+ _validate_presigned_url(bucket, url, target_key)
+ return url
+ except (OSError, ValueError):
+ raise
+ except Exception as error:
+ raise OSError("Failed to create blob presigned URL.") from error
+
+
+def _validity_seconds(validity, has_security_token=False) -> int:
+ if not isinstance(validity, timedelta):
+ raise TypeError(
+ "Blob presigned URL validity must be datetime.timedelta.")
+ if validity <= timedelta(0) or validity.microseconds != 0:
+ raise ValueError(
+ "Blob presigned URL validity must be positive whole seconds.")
+ seconds = validity.days * 24 * 60 * 60 + validity.seconds
+ maximum = 43200 if has_security_token else 604800
+ if seconds > maximum:
+ raise ValueError(
+ "OSS V4 presigned URL validity must not exceed {} seconds."
+ .format(maximum))
+ return seconds
+
+
+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
+
+
+def _head_object_if_exists(bucket, key):
+ try:
+ return bucket.head_object(key)
+ except Exception as error:
+ if getattr(error, 'code', None) in ('NoSuchKey', 'NoSuchObject'):
+ return None
+ raise
+
+
+def _matches(metadata, length, fingerprint, sse_headers) -> bool:
+ if metadata is None:
+ return False
+ headers = getattr(metadata, 'headers', {}) or {}
+ metadata_fingerprint = next(
+ (value for key, value in headers.items()
+ if key.lower() == _BLOB_FINGERPRINT_HEADER),
+ None,
+ )
+ return (
+ metadata.content_length == length
+ and metadata.content_type == _BLOB_CONTENT_TYPE
+ and metadata_fingerprint == fingerprint
+ and _matches_sse_headers(headers, sse_headers)
+ )
+
+
+def _matches_sse_headers(actual_headers, expected_headers) -> bool:
+ if not expected_headers:
+ return True
+ actual = {
+ str(key).lower(): value
+ for key, value in actual_headers.items()
+ }
+ case_insensitive_values = {
+ 'x-oss-server-side-encryption',
+ 'x-oss-server-side-data-encryption',
+ }
+ for key, expected in expected_headers.items():
+ normalized_key = str(key).lower()
+ found = actual.get(normalized_key)
+ if found is None:
+ return False
+ if normalized_key in case_insensitive_values:
+ if str(found).upper() != str(expected).upper():
+ return False
+ elif found != expected:
+ return False
+ return True
+
+
+def _validate_range(descriptor, source_length):
+ if (descriptor.offset < 0
+ or descriptor.length < 0
+ or descriptor.offset > source_length
+ or descriptor.length > source_length - descriptor.offset):
+ raise ValueError("Blob descriptor range is outside the source object.")
+
+
+def _materialize(
+ bucket,
+ source_key,
+ target_key,
+ descriptor,
+ fingerprint,
+ sse_headers):
+ import oss2
+
+ headers = dict(sse_headers)
+ headers.update({
+ 'Content-Type': _BLOB_CONTENT_TYPE,
+ _BLOB_FINGERPRINT_HEADER: fingerprint,
+ })
+ if descriptor.length == 0:
+ bucket.put_object(target_key, b'', headers=headers)
+ return
+
+ upload_id = None
+ try:
+ initiated = bucket.init_multipart_upload(target_key, headers=headers)
+ upload_id = initiated.upload_id
+ part_size = max(
+ _BLOB_COPY_MIN_PART_SIZE,
+ descriptor.length // _MAX_MULTIPART_UPLOAD_PARTS + 1,
+ )
+ parts = []
+ copied = 0
+ part_number = 1
+ while copied < descriptor.length:
+ size = min(part_size, descriptor.length - copied)
+ result = bucket.upload_part_copy(
+ bucket.bucket_name,
+ source_key,
+ (descriptor.offset + copied,
+ descriptor.offset + copied + size - 1),
+ target_key,
+ upload_id,
+ part_number,
+ )
+ parts.append(oss2.models.PartInfo(part_number, result.etag))
+ copied += size
+ part_number += 1
+ bucket.complete_multipart_upload(target_key, upload_id, parts)
+ upload_id = None
+ except Exception:
+ if upload_id is not None:
+ try:
+ bucket.abort_multipart_upload(target_key, upload_id)
+ except Exception:
+ _LOGGER.warning(
+ "Failed to abort OSS BLOB multipart upload %s for %s",
+ upload_id, target_key, exc_info=True)
+ raise
+
+
+def _validate_presigned_url(bucket, url, target_key):
+ endpoint = urlsplit(bucket.endpoint)
+ actual = urlsplit(url)
+ expected_host = bucket.bucket_name + '.' + (endpoint.hostname or '')
+ if (endpoint.scheme.lower() != 'https'
+ or actual.scheme.lower() != 'https'
+ or (actual.hostname or '').lower() != expected_host.lower()
+ or unquote(actual.path) != '/' + target_key):
+ raise OSError(
+ "OSS client generated a presigned URL for an invalid target.")
diff --git a/paimon-python/pypaimon/filesystem/oss_file_io.py
b/paimon-python/pypaimon/filesystem/oss_file_io.py
index 23b5859738..1dd0b40dc5 100644
--- a/paimon-python/pypaimon/filesystem/oss_file_io.py
+++ b/paimon-python/pypaimon/filesystem/oss_file_io.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""OSS conditional metadata creation, reusing the existing Arrow/Jindo
FileIO."""
+"""OSS-specific operations, reusing the existing Arrow/Jindo FileIO."""
import re
from urllib.parse import urlparse
@@ -38,9 +38,7 @@ class OssFileIO(PyArrowFileIO):
if uri.scheme:
if uri.scheme != 'oss' or self._extract_oss_bucket(path) !=
self._oss_bucket:
raise ValueError("Atomic write must target the configured OSS
bucket")
- key = re.sub(r'/+', '/', uri.path).lstrip('/')
- if '@' in uri.netloc:
- key = key.partition('/')[2]
+ key = self._extract_oss_object_key(path)
path = 'oss://{}/{}'.format(self._oss_bucket, key)
else:
key = path
@@ -77,6 +75,33 @@ class OssFileIO(PyArrowFileIO):
finally:
session.session.close()
+ def create_blob_presigned_url(self, table_root, descriptor, validity) ->
str:
+ try:
+ import oss2
+ except ImportError as error:
+ raise ImportError(
+ "OSS BLOB presigning requires oss2. Install pypaimon[oss] "
+ "or pypaimon[jindo]."
+ ) from error
+
+ from pypaimon.filesystem.oss_blob_presigner import create_presigned_url
+
+ session = oss2.Session()
+ try:
+ bucket = self._create_oss_bucket(
+ session, self._extract_oss_bucket(descriptor.uri))
+ return create_presigned_url(
+ bucket,
+ table_root,
+ descriptor,
+ validity,
+ sse_headers=self._sse_headers(),
+ has_security_token=bool(
+ self.properties.get(OssOptions.OSS_SECURITY_TOKEN)),
+ )
+ finally:
+ session.session.close()
+
def _supports_atomic_write(self, bucket):
"""Cache the publication mode for this instance; failed queries remain
retryable."""
import oss2
@@ -97,7 +122,22 @@ class OssFileIO(PyArrowFileIO):
self._oss_bucket, versioning)
return self._atomic_write_supported
- def _create_oss_bucket(self, session):
+ @staticmethod
+ def _extract_oss_object_key(location) -> str:
+ """Return an OSS object key for standard and credential URI forms."""
+ uri = urlparse(location)
+ if uri.scheme and uri.scheme.lower() != 'oss':
+ raise ValueError("Not an OSS URI: {}".format(location))
+ key = re.sub(r'/+', '/', uri.path).lstrip('/')
+ netloc = uri.netloc or ''
+ if ((getattr(uri, 'username', None)
+ or getattr(uri, 'password', None))
+ or '@' in netloc):
+ # Legacy URI: oss://AK:SK@endpoint/bucket/object-key
+ key = key.partition('/')[2]
+ return key
+
+ def _create_oss_bucket(self, session, bucket_name=None):
"""Build the metadata client with one V4 credential path for both AK
and STS."""
import oss2
from oss2.credentials import StaticCredentialsProvider
@@ -122,7 +162,13 @@ class OssFileIO(PyArrowFileIO):
raise ValueError("Set fs.oss.region for OSS V4 signing when the
endpoint is not regional")
provider = StaticCredentialsProvider(access_key, secret_key, token)
auth = oss2.ProviderAuthV4(provider)
- return oss2.Bucket(auth, endpoint, self._oss_bucket, session=session,
region=region)
+ return oss2.Bucket(
+ auth,
+ endpoint,
+ bucket_name or self._oss_bucket,
+ session=session,
+ region=region,
+ )
def _sse_headers(self):
"""Match Java OSSFileIO's SSE resolution, including the native option
fallback."""
diff --git a/paimon-python/pypaimon/filesystem/resolving_file_io.py
b/paimon-python/pypaimon/filesystem/resolving_file_io.py
index e8f1c79d87..e4ca45394e 100644
--- a/paimon-python/pypaimon/filesystem/resolving_file_io.py
+++ b/paimon-python/pypaimon/filesystem/resolving_file_io.py
@@ -100,6 +100,10 @@ class ResolvingFileIO(FileIO):
def try_to_write_atomic(self, path: str, content: str) -> bool:
return self._get_fileio(path).try_to_write_atomic(path, content)
+ def create_blob_presigned_url(self, table_root, descriptor, validity) ->
str:
+ return self._get_fileio(descriptor.uri).create_blob_presigned_url(
+ table_root, descriptor, validity)
+
def get_file_size(self, path: str) -> int:
return self._get_fileio(path).get_file_size(path)
diff --git a/paimon-python/pypaimon/multimodal/blob_store.py
b/paimon-python/pypaimon/multimodal/blob_store.py
index bed1ff22a0..53831750b5 100644
--- a/paimon-python/pypaimon/multimodal/blob_store.py
+++ b/paimon-python/pypaimon/multimodal/blob_store.py
@@ -18,6 +18,7 @@
import io
import re
from dataclasses import dataclass
+from datetime import timedelta
from typing import BinaryIO, Dict, Iterable, List, Mapping, Optional, Sequence
from pypaimon.common.options.core_options import CoreOptions
@@ -58,6 +59,7 @@ class BlobObject:
columns: Dict[str, object]
file_io: object
range_header: Optional[str] = None
+ table_root: Optional[str] = None
@property
def size(self) -> int:
@@ -77,6 +79,17 @@ class BlobObject:
with self.open() as stream:
return stream.read()
+ def to_presigned_url(self, validity: timedelta) -> str:
+ """Create a temporary URL for this object, honoring its selected byte
range.
+
+ Requires a table-bound object and a FileIO that supports presigning.
+ """
+ if self.table_root is None:
+ raise ValueError("BlobObject must be bound to a table root for
presigning.")
+ descriptor = _descriptor_for_range(self.descriptor, self.range_header)
+ return self.file_io.create_blob_presigned_url(
+ self.table_root, descriptor, validity)
+
class BlobStore:
"""S3-like object facade over a Paimon multimodal table BLOB column."""
@@ -172,6 +185,7 @@ class BlobStore:
columns=info.columns,
file_io=self._raw_table.file_io,
range_header=range,
+ table_root=self._raw_table.table_path,
)
def head_object(
diff --git a/paimon-python/pypaimon/table/row/blob.py
b/paimon-python/pypaimon/table/row/blob.py
index fb700721de..f2d666b64d 100644
--- a/paimon-python/pypaimon/table/row/blob.py
+++ b/paimon-python/pypaimon/table/row/blob.py
@@ -550,6 +550,11 @@ class Blob(ABC):
def new_input_stream(self) -> BinaryIO:
pass
+ def to_presigned_url(self, file_io, table_root: str, validity) -> str:
+ """Create a temporary URL for this descriptor-backed BLOB."""
+ return file_io.create_blob_presigned_url(
+ table_root, self.to_descriptor(), validity)
+
@staticmethod
def from_data(data: bytes) -> 'Blob':
return BlobData(data)
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index acca5be9d9..3d4de045ee 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -21,6 +21,8 @@ import os
import shutil
import tempfile
import unittest
+from dataclasses import replace
+from datetime import timedelta
from unittest.mock import patch
import pyarrow as pa
@@ -518,6 +520,25 @@ class MultimodalTableTest(unittest.TestCase):
cat = store.get_object("images/cat.jpg")
self.assertEqual(b"cat-image-v1", cat.read())
+ self.assertEqual(table.raw_table.table_path, cat.table_root)
+ with patch.object(cat.file_io, 'create_blob_presigned_url',
+ return_value="https://signed-url";) as presign:
+ validity = timedelta(minutes=30)
+ self.assertEqual("https://signed-url";,
cat.to_presigned_url(validity))
+ presign.assert_called_once_with(cat.table_root, cat.descriptor,
validity)
+ presign.reset_mock()
+ ranged = store.get_object("images/cat.jpg", range="bytes=1-4")
+ self.assertEqual("https://signed-url";,
ranged.to_presigned_url(validity))
+ root, descriptor, expiry = presign.call_args[0]
+ self.assertEqual(cat.table_root, root)
+ self.assertEqual(cat.descriptor.uri, descriptor.uri)
+ self.assertEqual(cat.descriptor.offset + 1, descriptor.offset)
+ self.assertEqual(4, descriptor.length)
+ self.assertEqual(validity, expiry)
+ presign.reset_mock()
+ with self.assertRaisesRegex(ValueError, "bound to a table root"):
+ replace(cat, table_root=None).to_presigned_url(validity)
+ presign.assert_not_called()
self.assertEqual(b"at-i", store.get_object(
"images/cat.jpg", range="bytes=1-4").read())
clipped = store.get_object("images/cat.jpg", range="bytes=10-999")
diff --git a/paimon-python/pypaimon/tests/oss_blob_presigner_test.py
b/paimon-python/pypaimon/tests/oss_blob_presigner_test.py
new file mode 100644
index 0000000000..b0886a8122
--- /dev/null
+++ b/paimon-python/pypaimon/tests/oss_blob_presigner_test.py
@@ -0,0 +1,530 @@
+# 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 hashlib
+from datetime import timedelta
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+
+from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO
+from pypaimon.common.file_io import FileIO
+from pypaimon.common.options import Options
+from pypaimon.filesystem.caching_file_io import CachingFileIO
+from pypaimon.filesystem.oss_blob_presigner import create_presigned_url
+from pypaimon.filesystem.oss_file_io import OssFileIO
+from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
+from pypaimon.table.row.blob import Blob, BlobDescriptor
+
+oss2 = pytest.importorskip("oss2")
+
+
+_FINGERPRINT = (
+ "cdba7930b06372d607da30924e956ec0"
+ "c1e5f0a27eac8155b2d6a2973022f1a8"
+)
+_TARGET_KEY = "table/bucket-0/_bloburl_" + _FINGERPRINT
+
+
+def _descriptor(offset=10, length=20):
+ return BlobDescriptor(
+ "oss://bucket/table/bucket-0/source.blob", offset, length)
+
+
+def _metadata(length=20, fingerprint=_FINGERPRINT, sse_headers=None):
+ headers = {
+ "x-oss-meta-paimon-blob-descriptor-sha256": fingerprint,
+ }
+ headers.update(sse_headers or {})
+ return SimpleNamespace(
+ content_length=length,
+ content_type="application/octet-stream",
+ headers=headers,
+ )
+
+
+def _bucket(endpoint="https://oss-cn-hangzhou.aliyuncs.com";):
+ bucket = mock.MagicMock()
+ bucket.bucket_name = "bucket"
+ bucket.endpoint = endpoint
+ bucket.sign_url.return_value = (
+ endpoint.replace("oss-", "bucket.oss-")
+ + "/"
+ + _TARGET_KEY
+ + "?x-oss-signature=test"
+ )
+ return bucket
+
+
+def _missing():
+ return oss2.exceptions.ServerError(
+ 404, {}, b"", {"Code": "NoSuchKey", "Message": "missing"})
+
+
+def test_blob_api_delegates_to_file_io():
+ descriptor = _descriptor()
+ blob = Blob.from_descriptor(mock.MagicMock(), descriptor)
+ file_io = mock.MagicMock()
+ validity = timedelta(minutes=5)
+ file_io.create_blob_presigned_url.return_value = "https://example";
+
+ assert blob.to_presigned_url(
+ file_io, "oss://bucket/table", validity) == "https://example";
+ file_io.create_blob_presigned_url.assert_called_once_with(
+ "oss://bucket/table", descriptor, validity)
+
+
+def test_inline_blob_cannot_create_presigned_url():
+ file_io = mock.MagicMock()
+ with pytest.raises(RuntimeError, match="can not convert to descriptor"):
+ Blob.from_data(b"inline").to_presigned_url(
+ file_io, "oss://bucket/table", timedelta(minutes=5))
+ file_io.create_blob_presigned_url.assert_not_called()
+
+
[email protected]("token", [None, "sts-token"])
+def test_oss_file_io_uses_descriptor_bucket_and_closes_session(token):
+ descriptor = _descriptor()
+ validity = timedelta(minutes=5)
+ bucket = _bucket()
+ session = mock.MagicMock()
+ file_io = OssFileIO.__new__(OssFileIO)
+ file_io._create_oss_bucket = mock.MagicMock(return_value=bucket)
+ file_io.properties = Options({
+ 'fs.oss.server-side-encryption': 'AES256',
+ 'fs.oss.securityToken': token,
+ })
+
+ with mock.patch('oss2.Session', return_value=session), mock.patch(
+ 'pypaimon.filesystem.oss_blob_presigner.create_presigned_url',
+ return_value='https://example') as create:
+ assert file_io.create_blob_presigned_url(
+ 'oss://bucket/table', descriptor, validity) == 'https://example'
+
+ file_io._create_oss_bucket.assert_called_once_with(session, 'bucket')
+ create.assert_called_once_with(
+ bucket,
+ 'oss://bucket/table',
+ descriptor,
+ validity,
+ sse_headers={'x-oss-server-side-encryption': 'AES256'},
+ has_security_token=bool(token),
+ )
+ session.session.close.assert_called_once_with()
+
+
+def test_cache_hit_only_creates_fresh_url():
+ bucket = _bucket()
+ bucket.head_object.return_value = _metadata()
+
+ url = create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(minutes=5),
+ )
+
+ assert url.startswith("https://bucket.oss-cn-hangzhou.aliyuncs.com/";)
+ bucket.head_object.assert_called_once_with(_TARGET_KEY)
+ bucket.sign_url.assert_called_once_with(
+ 'GET', _TARGET_KEY, 300, slash_safe=True)
+ bucket.init_multipart_upload.assert_not_called()
+
+
+def test_materializes_descriptor_range_and_reuses_java_object_key():
+ bucket = _bucket()
+ bucket.head_object.side_effect = [
+ _missing(),
+ SimpleNamespace(content_length=100),
+ _metadata(),
+ ]
+ bucket.init_multipart_upload.return_value = SimpleNamespace(
+ upload_id="upload-id")
+ bucket.upload_part_copy.return_value = SimpleNamespace(etag="etag")
+
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(seconds=30),
+ )
+
+ assert bucket.head_object.call_args_list == [
+ mock.call(_TARGET_KEY),
+ mock.call("table/bucket-0/source.blob"),
+ mock.call(_TARGET_KEY),
+ ]
+ bucket.init_multipart_upload.assert_called_once_with(
+ _TARGET_KEY,
+ headers={
+ "Content-Type": "application/octet-stream",
+ "x-oss-meta-paimon-blob-descriptor-sha256": _FINGERPRINT,
+ },
+ )
+ bucket.upload_part_copy.assert_called_once_with(
+ "bucket",
+ "table/bucket-0/source.blob",
+ (10, 29),
+ _TARGET_KEY,
+ "upload-id",
+ 1,
+ )
+ parts = bucket.complete_multipart_upload.call_args[0][2]
+ assert [(part.part_number, part.etag) for part in parts] == [(1, "etag")]
+
+
[email protected]("sse_headers", [
+ {"x-oss-server-side-encryption": "AES256"},
+ {
+ "x-oss-server-side-encryption": "KMS",
+ "x-oss-server-side-encryption-key-id": "cmk-123",
+ },
+])
+def test_materialization_preserves_server_side_encryption(sse_headers):
+ bucket = _bucket()
+ bucket.head_object.side_effect = [
+ _missing(),
+ SimpleNamespace(content_length=100),
+ _metadata(sse_headers=sse_headers),
+ ]
+ bucket.init_multipart_upload.return_value = SimpleNamespace(
+ upload_id="upload-id")
+ bucket.upload_part_copy.return_value = SimpleNamespace(etag="etag")
+
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(seconds=30),
+ sse_headers=sse_headers,
+ )
+
+ expected_headers = dict(sse_headers)
+ expected_headers.update({
+ "Content-Type": "application/octet-stream",
+ "x-oss-meta-paimon-blob-descriptor-sha256": _FINGERPRINT,
+ })
+ bucket.init_multipart_upload.assert_called_once_with(
+ _TARGET_KEY, headers=expected_headers)
+
+
[email protected]("actual_sse", [
+ {},
+ {"x-oss-server-side-encryption": "AES256"},
+], ids=["unencrypted", "different-method"])
+def test_cache_rejects_object_with_incompatible_server_side_encryption(
+ actual_sse):
+ expected_sse = {
+ "x-oss-server-side-encryption": "KMS",
+ "x-oss-server-side-encryption-key-id": "cmk-123",
+ }
+ bucket = _bucket()
+ bucket.head_object.side_effect = [
+ _metadata(sse_headers=actual_sse),
+ SimpleNamespace(content_length=100),
+ _metadata(sse_headers=expected_sse),
+ ]
+ bucket.init_multipart_upload.return_value = SimpleNamespace(
+ upload_id="upload-id")
+ bucket.upload_part_copy.return_value = SimpleNamespace(etag="etag")
+
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(seconds=30),
+ sse_headers=expected_sse,
+ )
+
+ bucket.init_multipart_upload.assert_called_once()
+
+
+def test_cache_reuses_object_with_matching_server_side_encryption():
+ expected_sse = {
+ "x-oss-server-side-encryption": "kms",
+ "x-oss-server-side-encryption-key-id": "cmk-123",
+ }
+ bucket = _bucket()
+ bucket.head_object.return_value = _metadata(sse_headers={
+ "X-Oss-Server-Side-Encryption": "KMS",
+ "X-Oss-Server-Side-Encryption-Key-Id": "cmk-123",
+ })
+
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(seconds=30),
+ sse_headers=expected_sse,
+ )
+
+ bucket.init_multipart_upload.assert_not_called()
+
+
+def test_legacy_oss_uri_strips_bucket_from_source_and_target_keys():
+ descriptor = BlobDescriptor(
+ "oss://AK:SK@endpoint/bucket/table/bucket-0/source.blob", 10, 20)
+ fingerprint = hashlib.sha256(descriptor.serialize()).hexdigest()
+ target_key = "table/bucket-0/_bloburl_" + fingerprint
+ bucket = _bucket()
+ bucket.sign_url.return_value = (
+ "https://bucket.oss-cn-hangzhou.aliyuncs.com/";
+ + target_key
+ + "?x-oss-signature=test"
+ )
+ bucket.head_object.side_effect = [
+ _missing(),
+ SimpleNamespace(content_length=100),
+ _metadata(fingerprint=fingerprint),
+ ]
+ bucket.init_multipart_upload.return_value = SimpleNamespace(
+ upload_id="upload-id")
+ bucket.upload_part_copy.return_value = SimpleNamespace(etag="etag")
+
+ create_presigned_url(
+ bucket,
+ "oss://AK:SK@endpoint/bucket/table",
+ descriptor,
+ timedelta(seconds=30),
+ )
+
+ assert bucket.head_object.call_args_list == [
+ mock.call(target_key),
+ mock.call("table/bucket-0/source.blob"),
+ mock.call(target_key),
+ ]
+ bucket.upload_part_copy.assert_called_once_with(
+ "bucket",
+ "table/bucket-0/source.blob",
+ (10, 29),
+ target_key,
+ "upload-id",
+ 1,
+ )
+
+
+def test_zero_length_descriptor_uses_put_object():
+ descriptor = _descriptor(offset=100, length=0)
+ fingerprint = hashlib.sha256(descriptor.serialize()).hexdigest()
+ target_key = "table/bucket-0/_bloburl_" + fingerprint
+ bucket = _bucket()
+ bucket.sign_url.return_value = (
+ "https://bucket.oss-cn-hangzhou.aliyuncs.com/";
+ + target_key
+ + "?x-oss-signature=test"
+ )
+ sse_headers = {"x-oss-server-side-encryption": "AES256"}
+ bucket.head_object.side_effect = [
+ _missing(),
+ SimpleNamespace(content_length=100),
+ _metadata(
+ length=0,
+ fingerprint=fingerprint,
+ sse_headers=sse_headers,
+ ),
+ ]
+
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ descriptor,
+ timedelta(minutes=5),
+ sse_headers=sse_headers,
+ )
+
+ expected_headers = dict(sse_headers)
+ expected_headers.update({
+ "Content-Type": "application/octet-stream",
+ "x-oss-meta-paimon-blob-descriptor-sha256": fingerprint,
+ })
+ bucket.put_object.assert_called_once_with(
+ target_key,
+ b'',
+ headers=expected_headers,
+ )
+ bucket.init_multipart_upload.assert_not_called()
+
+
[email protected]("descriptor", [
+ _descriptor(offset=-1, length=20),
+ _descriptor(offset=10, length=-1),
+ _descriptor(offset=90, length=20),
+])
+def test_rejects_invalid_source_range(descriptor):
+ bucket = _bucket()
+ bucket.head_object.side_effect = [
+ _missing(), SimpleNamespace(content_length=100)]
+ with pytest.raises(ValueError, match="range is outside"):
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ descriptor,
+ timedelta(minutes=5),
+ )
+ bucket.init_multipart_upload.assert_not_called()
+
+
[email protected]("table_root", [
+ "s3://bucket/table",
+ "oss://other/table",
+ "oss://bucket/other",
+ "oss://bucket/table-sibling",
+])
+def test_rejects_descriptor_outside_table_root(table_root):
+ with pytest.raises(
+ ValueError, match="scheme and authority|under table root"):
+ create_presigned_url(
+ _bucket(), table_root, _descriptor(), timedelta(minutes=5))
+
+
[email protected]("validity", [
+ timedelta(0),
+ timedelta(seconds=-1),
+ timedelta(microseconds=1),
+])
+def test_validity_must_be_positive_whole_seconds(validity):
+ with pytest.raises(ValueError, match="positive whole seconds"):
+ create_presigned_url(
+ _bucket(), "oss://bucket/table", _descriptor(), validity)
+
+
+def test_internal_endpoint_is_preserved():
+ bucket = _bucket("https://oss-cn-hangzhou-internal.aliyuncs.com";)
+ bucket.head_object.return_value = _metadata()
+
+ url = create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(minutes=5),
+ )
+
+ assert
url.startswith("https://bucket.oss-cn-hangzhou-internal.aliyuncs.com/";)
+ assert url == bucket.sign_url.return_value
+
+
[email protected]("endpoint, generated_endpoint", [
+ ("https://oss-cn-hangzhou-internal.aliyuncs.com";,
+ "https://oss-cn-hangzhou.aliyuncs.com";),
+ ("https://oss-cn-hangzhou.aliyuncs.com";,
+ "https://oss-cn-hangzhou-internal.aliyuncs.com";),
+])
+def test_rejects_url_with_different_endpoint(endpoint, generated_endpoint):
+ bucket = _bucket(endpoint)
+ bucket.head_object.return_value = _metadata()
+ bucket.sign_url.return_value =
_bucket(generated_endpoint).sign_url.return_value
+ with pytest.raises(OSError, match="invalid target"):
+ create_presigned_url(
+ bucket, "oss://bucket/table", _descriptor(), timedelta(minutes=5))
+
+
+def test_rejects_non_https_endpoint_before_any_remote_call():
+ bucket = _bucket("http://oss-cn-hangzhou.aliyuncs.com";)
+ bucket.head_object.return_value = _metadata()
+ with pytest.raises(ValueError, match="HTTPS endpoint"):
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(minutes=5),
+ )
+
+ assert bucket.mock_calls == []
+
+
+def test_failed_materialization_aborts_upload_and_wraps_error():
+ bucket = _bucket()
+ bucket.head_object.side_effect = [
+ _missing(), SimpleNamespace(content_length=100)]
+ bucket.init_multipart_upload.return_value = SimpleNamespace(
+ upload_id="upload-id")
+ bucket.upload_part_copy.side_effect = RuntimeError("copy failed")
+
+ with pytest.raises(OSError, match="Failed to create") as caught:
+ create_presigned_url(
+ bucket,
+ "oss://bucket/table",
+ _descriptor(),
+ timedelta(minutes=5),
+ )
+
+ assert isinstance(caught.value.__cause__, RuntimeError)
+ bucket.abort_multipart_upload.assert_called_once_with(
+ _TARGET_KEY, "upload-id")
+
+
+def test_file_io_wrappers_delegate_presigning():
+ descriptor = _descriptor()
+ validity = timedelta(minutes=5)
+ delegate = mock.MagicMock(spec=FileIO)
+ delegate.create_blob_presigned_url.return_value = "https://example";
+
+ caching = CachingFileIO(delegate, None)
+ assert caching.create_blob_presigned_url(
+ "oss://bucket/table", descriptor, validity) == "https://example";
+
+ resolving = ResolvingFileIO(Options({}))
+ with mock.patch.object(
+ resolving, '_get_fileio', return_value=delegate) as get_file_io:
+ assert resolving.create_blob_presigned_url(
+ "oss://bucket/table", descriptor, validity) == "https://example";
+ get_file_io.assert_called_once_with(descriptor.uri)
+
+ rest = RESTTokenFileIO.__new__(RESTTokenFileIO)
+ rest.path = "oss://bucket/table"
+ with mock.patch.object(
+ rest, '_file_io_with_token', return_value=(delegate, None)):
+ assert rest.create_blob_presigned_url(
+ "oss://bucket/table", descriptor, validity) == "https://example";
+
+
[email protected]("has_security_token, maximum", [
+ (False, 604800), (True, 43200),
+])
+def test_validity_upper_bound_before_any_remote_call(has_security_token,
maximum):
+ bucket = _bucket()
+ with pytest.raises(ValueError, match="must not exceed"):
+ create_presigned_url(
+ bucket, "oss://bucket/table", _descriptor(),
+ timedelta(seconds=maximum + 1),
+ has_security_token=has_security_token)
+ assert bucket.mock_calls == []
+ bucket.head_object.return_value = _metadata()
+ create_presigned_url(
+ bucket, "oss://bucket/table", _descriptor(),
+ timedelta(seconds=maximum), has_security_token=has_security_token)
+ bucket.sign_url.assert_called_once_with(
+ 'GET', _TARGET_KEY, maximum, slash_safe=True)
+
+
+def test_abort_failure_is_logged_without_replacing_copy_error(caplog):
+ bucket = _bucket()
+ bucket.head_object.side_effect = [
+ _missing(), SimpleNamespace(content_length=100)]
+ bucket.init_multipart_upload.return_value =
SimpleNamespace(upload_id="upload-id")
+ copy_error = RuntimeError("copy failed")
+ abort_error = RuntimeError("abort failed")
+ bucket.upload_part_copy.side_effect = copy_error
+ bucket.abort_multipart_upload.side_effect = abort_error
+ with pytest.raises(OSError) as caught:
+ create_presigned_url(
+ bucket, "oss://bucket/table", _descriptor(), timedelta(minutes=5))
+ assert caught.value.__cause__ is copy_error
+ record = next(r for r in caplog.records if "Failed to abort" in r.message)
+ assert record.exc_info[1] is abort_error
+ assert "upload-id" in record.message
diff --git a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
index cf80fc95d1..4e94b21d72 100644
--- a/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_token_file_io_test.py
@@ -20,6 +20,7 @@ import pickle
import tempfile
import time
import unittest
+from datetime import timedelta
from unittest.mock import patch, MagicMock
from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO
@@ -29,6 +30,7 @@ from pypaimon.common.identifier import Identifier
from pypaimon.common.options import Options
from pypaimon.common.options.config import CatalogOptions, OssOptions
from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.table.row.blob import BlobDescriptor
class RESTTokenFileIOTest(unittest.TestCase):
@@ -47,6 +49,125 @@ class RESTTokenFileIOTest(unittest.TestCase):
if os.path.exists(self.temp_dir):
shutil.rmtree(self.temp_dir)
+ def test_blob_presigned_url_bound_table_root(self):
+ root = "oss://bucket/table-a"
+ file_io = RESTTokenFileIO(self.identifier, root, self.catalog_options)
+ descriptor = BlobDescriptor(root + "/data/video.blob", 10, 20)
+ validity = timedelta(minutes=30)
+ delegate = MagicMock()
+ delegate.create_blob_presigned_url.return_value = "https://signed-url";
+ with patch.object(
+ file_io, '_file_io_with_token',
+ return_value=(delegate, None)):
+ self.assertEqual(
+ file_io.create_blob_presigned_url(root, descriptor, validity),
+ "https://signed-url";)
+ delegate.create_blob_presigned_url.assert_called_once_with(
+ root, descriptor, validity)
+
+ def test_blob_presigned_url_rejects_other_table_before_resolving_io(self):
+ file_io = RESTTokenFileIO(
+ self.identifier, "oss://bucket/table-a", self.catalog_options)
+ other_root = "oss://bucket/table-b"
+ descriptor = BlobDescriptor(other_root + "/data/video.blob", 10, 20)
+ with patch.object(file_io, '_file_io_with_token') as resolve:
+ with self.assertRaisesRegex(ValueError, "bound table root"):
+ file_io.create_blob_presigned_url(
+ other_root, descriptor, timedelta(minutes=30))
+ resolve.assert_not_called()
+
+ def test_presigned_url_refreshes_credentials_for_requested_lifetime(self):
+ self._check_presigned_url_lifetime(3, 4, True)
+
+ def test_presigned_url_reuses_credentials_for_short_validity(self):
+ self._check_presigned_url_lifetime(0.5, 4, False)
+
+ def test_presigned_url_accepts_when_one_second_exceeds_validity(self):
+ self._check_presigned_url_lifetime(
+ 1.5, 3, False, current_token_extra_seconds=1)
+
+ def test_presigned_url_rejects_insufficient_refreshed_lifetime(self):
+ self._check_presigned_url_lifetime(3, 2, True, rejected=True)
+
+ def _check_presigned_url_lifetime(
+ self, requested_hours, refreshed_hours, should_refresh,
rejected=False,
+ current_token_extra_seconds=None):
+ root = "oss://bucket/table"
+ file_io = RESTTokenFileIO(self.identifier, root, self.catalog_options)
+ now = 1700000000
+ token_properties = {OssOptions.OSS_SECURITY_TOKEN.key(): "test-token"}
+ if current_token_extra_seconds is None:
+ old_expiry = now + 2 * 3600
+ else:
+ old_expiry = (
+ now + requested_hours * 3600 + current_token_extra_seconds)
+ old_token = RESTToken(token_properties, int(old_expiry * 1000))
+ new_token = RESTToken(token_properties, (now + refreshed_hours * 3600)
* 1000)
+ file_io.token = old_token
+ descriptor = BlobDescriptor(root + "/video.blob", 0, 10)
+ validity = timedelta(hours=requested_hours)
+
+ def refresh():
+ file_io.token = new_token
+
+ with patch('pypaimon.catalog.rest.rest_token_file_io.time.time',
return_value=now), \
+ patch.object(file_io, '_build_cache_key',
return_value='presigning-test'), \
+ patch.object(file_io, '_get_cached_token',
return_value=old_token), \
+ patch.object(file_io, '_set_cached_token') as cache_token, \
+ patch.object(file_io, 'refresh_token', side_effect=refresh) as
refresh_token, \
+ patch('pypaimon.catalog.rest.rest_token_file_io.FileIO.get')
as get_io:
+ presign = get_io.return_value.create_blob_presigned_url
+ presign.side_effect = ["https://first-url";,
"https://refreshed-url";]
+ if rejected:
+ with self.assertRaisesRegex(ValueError, "credential lifetime
after refresh"):
+ file_io.create_blob_presigned_url(root, descriptor,
validity)
+ get_io.assert_not_called()
+ else:
+ self.assertEqual("https://first-url";,
file_io.create_blob_presigned_url(
+ root, descriptor, validity))
+ self.assertEqual(1, presign.call_count)
+ presign.assert_called_with(root, descriptor, validity)
+ self.assertEqual(int(should_refresh), refresh_token.call_count)
+ if should_refresh:
+
cache_token.assert_called_once_with(file_io._build_cache_key(), new_token)
+ else:
+ cache_token.assert_not_called()
+
+ def test_presigned_url_refreshes_and_resigns_after_materialization(self):
+ root = "oss://bucket/table"
+ file_io = RESTTokenFileIO(self.identifier, root, self.catalog_options)
+ now = [1700000000]
+ token_properties = {OssOptions.OSS_SECURITY_TOKEN.key(): "test-token"}
+ old_token = RESTToken(token_properties, (now[0] + 2 * 3600) * 1000)
+ new_token = RESTToken(token_properties, (now[0] + 4 * 3600) * 1000)
+ file_io.token = old_token
+ descriptor = BlobDescriptor(root + "/video.blob", 0, 10)
+ validity = timedelta(minutes=30)
+
+ def refresh():
+ file_io.token = new_token
+
+ def presign(*_):
+ if now[0] == 1700000000:
+ now[0] += 2 * 3600
+ return "https://first-url";
+ return "https://refreshed-url";
+
+ with patch('pypaimon.catalog.rest.rest_token_file_io.time.time',
+ side_effect=lambda: now[0]), \
+ patch.object(file_io, '_build_cache_key',
return_value='presigning-test'), \
+ patch.object(file_io, '_get_cached_token',
return_value=old_token), \
+ patch.object(file_io, '_set_cached_token'), \
+ patch.object(file_io, 'refresh_token', side_effect=refresh) as
refresh_token, \
+ patch('pypaimon.catalog.rest.rest_token_file_io.FileIO.get')
as get_io:
+ get_io.return_value.create_blob_presigned_url.side_effect = presign
+ self.assertEqual(
+ "https://refreshed-url";,
+ file_io.create_blob_presigned_url(root, descriptor, validity))
+ self.assertEqual(
+ 2, get_io.return_value.create_blob_presigned_url.call_count)
+ refresh_token.assert_called_once_with()
+
def test_new_output_stream_path_conversion_and_parent_creation(self):
"""Test new_output_stream correctly handles URI paths and creates
parent directories."""
with patch.object(RESTTokenFileIO, 'try_to_refresh_token'):
