[core][flink][spark] Support OSS Blob presigned URLs

docs/docs/multimodal-table/blob.mdx

@@ -502,6 +502,81 @@ CREATE TABLE descriptor_table (
 
 Paimon stores only serialized `BlobDescriptor` bytes in normal data files. Reading the blob follows the descriptor URI to access bytes, and writing requires descriptor input for those fields.
 
+## Presigned URLs for OSS Blobs
+
+Paimon can create a temporary HTTPS URL for an OSS-backed `BlobDescriptor`. The descriptor may
+refer to a byte range inside a larger `.blob` file. Paimon first materializes exactly that range as
+a separate OSS object, sets its content type from the requested extension, and then returns a
+presigned GET URL.
+
+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:
+
+```properties
+fs.oss.endpoint=https://oss-cn-hangzhou.aliyuncs.com
+```
+
+To obtain descriptors from a regular BLOB column, read the source table with
+`blob-as-descriptor=true`. The SQL functions have strict and error-tolerant forms:
+
+```sql
+-- Flink SQL
+SELECT sys.descriptor_to_presigned_url(
+    'default.image_table',
+    image,
+    'png',
+    INTERVAL '5' MINUTE)
+FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;
+
+SELECT sys.try_descriptor_to_presigned_url(
+    'default.image_table',
+    image,
+    'png',
+    INTERVAL '5' MINUTE)
+FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;
+```
+
+```sql
+-- Spark SQL
+ALTER TABLE image_table SET TBLPROPERTIES ('blob-as-descriptor' = 'true');
+
+SELECT sys.descriptor_to_presigned_url(
+    'default.image_table',
+    image,
+    'png',
+    INTERVAL '5' MINUTE)
+FROM image_table;
+```
+
+In both Flink and Spark, `source_table` must be a non-null string literal in
+`database.table` form. `catalog.database.table` is also accepted when the catalog matches the
+catalog that owns the function. Dynamic columns and `CASE` expressions are rejected during
+planning. The strict function propagates row errors; `try_descriptor_to_presigned_url` returns
+`NULL` for row-level errors. The validity interval must be positive whole seconds.
+
+The Java API uses the same table-root safety check:
+
+```java
+DataTable table = ...;
+Blob blob = ...;
+String url =
+        blob.toPresignedUrl(
+                table.fileIO(), table.location(), "png", Duration.ofMinutes(5));
+```
+
+The extension must contain only lower-case letters or digits after normalization and must map to a
+known MIME type. It must match the actual content. For direct model `image_url` inputs, use only
+image formats verified with the target model; PDF is not covered by this entry point.
+
+Materialized objects are keyed by the complete descriptor fingerprint and extension. Repeated
+short-term calls for the same descriptor reuse an object whose length, content type, and
+fingerprint still match, while generating a fresh URL. The first version does not resolve races
+with stale-cache cleanup or orphan-file cleaning.
+
+Treat the returned URL as a bearer credential: submit it immediately to the external service, do
+not persist it, and never write the URL or its signature/security-token query parameters to logs.
+
 ## Blob View
 
 Blob view is useful when a downstream table should reference BLOB values already stored in an upstream table, without copying the bytes or creating new `.blob` files. A blob view field stores only a small `BlobViewStruct` inline. When the field is read, Paimon resolves the referenced BLOB from the upstream table.

docs/docs/spark/sql-functions.md

@@ -94,6 +94,40 @@ SELECT sys.descriptor_to_string(content) FROM t WHERE id = '1';
 -- [BlobDescriptor{version=1', uri='/path/to/data-2c103f6f-3857-4062-abc3-2e260374a68e-1.blob', offset=4, length=1048576}]
 ```
 
+### descriptor_to_presigned_url
+
+`sys.descriptor_to_presigned_url($source_table, $descriptor, $extension, $validity)`
+
+Creates a temporary HTTPS GET URL for an OSS-backed blob descriptor. Use
+`sys.try_descriptor_to_presigned_url` to return `NULL` instead of failing on row-level errors.
+
+- `source_table` must be a non-null STRING literal in `database.table` form, or
+  `catalog.database.table` with the same catalog as the function.
+- `descriptor` is the serialized BINARY descriptor. Set `blob-as-descriptor=true` when reading a
+  normal BLOB column.
+- `extension` determines the materialized object's MIME type and must match the content.
+- `validity` is a positive day-time interval containing whole seconds.
+
+The catalog's `fs.oss.endpoint` must be a standard public HTTPS endpoint, for example
+`https://oss-cn-hangzhou.aliyuncs.com`.
+
+```sql
+ALTER TABLE image_table SET TBLPROPERTIES ('blob-as-descriptor' = 'true');
+
+SELECT sys.descriptor_to_presigned_url(
+    'default.image_table',
+    image,
+    'png',
+    INTERVAL '5' MINUTE)
+FROM image_table;
+```
+
+Repeated short-term calls for the same descriptor reuse the materialized object and issue a fresh
+URL. Treat the URL as a bearer credential: send it immediately to the consumer and never log or
+persist it. Direct model `image_url` use is supported only for image formats verified with that
+model; PDF is not covered. See [Blob Storage](../multimodal-table/blob#presigned-urls-for-oss-blobs)
+for Java and Flink examples, caching behavior, and current limitations.
+
 ## User-defined Function
 
 Paimon Spark supports three types of user-defined functions: lambda functions, file-based functions, and SQL functions.

paimon-common/src/main/java/org/apache/paimon/data/Blob.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.annotation.Public;
 import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.SeekableInputStream;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.utils.UriReader;
@@ -28,6 +29,7 @@
 import javax.annotation.Nullable;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.function.Supplier;
 
 /**
@@ -44,6 +46,11 @@ public interface Blob {
 
     SeekableInputStream newInputStream() throws IOException;
 
+    default String toPresignedUrl(
+            FileIO fileIO, Path tableRoot, String extension, Duration validity) throws IOException {
+        return fileIO.createBlobPresignedUrl(tableRoot, toDescriptor(), extension, validity);
+    }
+
     static Blob fromData(byte[] data) {
         return new BlobData(data);
     }

paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.annotation.Public;
 import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.fs.hadoop.HadoopFileIOLoader;
 import org.apache.paimon.fs.local.LocalFileIO;
 
@@ -262,6 +263,13 @@ default Optional<Path> unarchive(Path path, StorageType type) throws IOException
                 getClass().getName() + " does not support unarchive.");
     }
 
+    default String createBlobPresignedUrl(
+            Path tableRoot, BlobDescriptor descriptor, String extension, Duration validity)
+            throws IOException {
+        throw new UnsupportedOperationException(
+                getClass().getName() + " does not support creating blob presigned URLs.");
+    }
+
     /**
      * Override this method to empty, many FileIO implementation classes rely on static variables
      * and do not have the ability to close them.

paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java

@@ -19,9 +19,11 @@
 package org.apache.paimon.fs;
 
 import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.options.Options;
 
 import java.io.IOException;
+import java.time.Duration;
 
 /**
  * A {@link FileIO} for plugin jar. {@link FileIO} is serializable, so plugin FileIO should be
@@ -96,6 +98,17 @@ public boolean tryToWriteAtomic(Path path, String content) throws IOException {
         return wrap(() -> fileIO(path).tryToWriteAtomic(path, content));
     }
 
+    @Override
+    public String createBlobPresignedUrl(
+            Path tableRoot, BlobDescriptor descriptor, String extension, Duration validity)
+            throws IOException {
+        return wrap(
+                () ->
+                        fileIO(new Path(descriptor.uri()))
+                                .createBlobPresignedUrl(
+                                        tableRoot, descriptor, extension, validity));
+    }
+
     private FileIO fileIO(Path path) throws IOException {
         if (lazyFileIO == null) {
             synchronized (this) {

paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java

@@ -20,11 +20,13 @@
 
 import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.time.Duration;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
@@ -107,6 +109,17 @@ public boolean rename(Path src, Path dst) throws IOException {
         return wrap(() -> fileIO(src).rename(src, dst));
     }
 
+    @Override
+    public String createBlobPresignedUrl(
+            Path tableRoot, BlobDescriptor descriptor, String extension, Duration validity)
+            throws IOException {
+        return wrap(
+                () ->
+                        fileIO(new Path(descriptor.uri()))
+                                .createBlobPresignedUrl(
+                                        tableRoot, descriptor, extension, validity));
+    }
+
     @VisibleForTesting
     public FileIO fileIO(Path path) throws IOException {
         CacheKey cacheKey = new CacheKey(path.toUri().getScheme(), path.toUri().getAuthority());

paimon-common/src/main/java/org/apache/paimon/fs/cache/CachingFileIO.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.fs.cache;
 
 import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
@@ -32,6 +33,7 @@
 import javax.annotation.Nullable;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.EnumSet;
 import java.util.Map;
 import java.util.Set;
@@ -151,6 +153,13 @@ public boolean rename(Path src, Path dst) throws IOException {
         return delegate.rename(src, dst);
     }
 
+    @Override
+    public String createBlobPresignedUrl(
+            Path tableRoot, BlobDescriptor descriptor, String extension, Duration validity)
+            throws IOException {
+        return delegate.createBlobPresignedUrl(tableRoot, descriptor, extension, validity);
+    }
+
     @Override
     public boolean isObjectStore() {
         return delegate.isObjectStore();

paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
@@ -44,6 +45,7 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.time.Duration;
 import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -150,6 +152,16 @@ public boolean rename(Path src, Path dst) throws IOException {
         return fileIO().rename(src, dst);
     }
 
+    @Override
+    public String createBlobPresignedUrl(
+            Path tableRoot, BlobDescriptor descriptor, String extension, Duration validity)
+            throws IOException {
+        if (!path.equals(tableRoot)) {
+            throw new IOException("Table root does not match RESTTokenFileIO bound table root.");
+        }
+        return fileIO().createBlobPresignedUrl(tableRoot, descriptor, extension, validity);
+    }
+
     @Override
     public boolean isObjectStore() {
         try {

paimon-common/src/main/java/org/apache/paimon/utils/BlobDescriptorUtils.java

@@ -20,18 +20,50 @@
 
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
 
 import javax.annotation.Nullable;
 
+import java.io.IOException;
+import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 
 import static org.apache.paimon.CoreOptions.BLOB_DESCRIPTOR_PREFIX;
 
 /** Utils for {@link BlobDescriptor}. */
 public class BlobDescriptorUtils {
 
+    public static void validateTableRoot(Path tableRoot, BlobDescriptor descriptor)
+            throws IOException {
+        URI root = tableRoot.toUri().normalize();
+        URI blob;
+        try {
+            blob = new Path(descriptor.uri()).toUri().normalize();
+        } catch (RuntimeException e) {
+            throw new IOException("Invalid blob descriptor URI.", e);
+        }
+
+        if (!Objects.equals(normalizeScheme(root.getScheme()), normalizeScheme(blob.getScheme()))
+                || !Objects.equals(root.getAuthority(), blob.getAuthority())) {
+            throw new IOException(
+                    "Blob descriptor URI must use the same scheme and authority as table root.");
+        }
+
+        String rootPath = root.getPath();
+        String blobPath = blob.getPath();
+        String childPrefix = rootPath.endsWith("/") ? rootPath : rootPath + "/";
+        if (!blobPath.startsWith(childPrefix)) {
+            throw new IOException("Blob descriptor URI must be under table root.");
+        }
+    }
+
+    private static String normalizeScheme(@Nullable String scheme) {
+        return scheme == null ? null : scheme.toLowerCase();
+    }
+
     /**
      * Try to create a {@link CatalogContext} for input {@link BlobDescriptor}. This enables reading
      * descriptors from external storages which can be different from paimon's own.