[core] Introduce blob v2

docs/content/append-table/blob.md

@@ -94,7 +94,19 @@ For details about the blob file format structure, see [File Format - BLOB]({{< r
       <td>No</td>
       <td style="word-wrap: break-word;">false</td>
       <td>Boolean</td>
-      <td>When set to true, the blob field input is treated as a serialized BlobDescriptor. Paimon reads from the descriptor's URI and streams the data into Paimon's blob files in small chunks, avoiding loading the entire blob into memory. This is useful for writing very large blobs that cannot fit in memory. When reading, if set to true, returns the BlobDescriptor bytes; if false, returns actual blob bytes.</td>
+      <td>Controls read output format for blob fields. When set to true, queries return serialized BlobDescriptor bytes; when false, queries return actual blob bytes. This option is dynamic and can be changed with <code>ALTER TABLE ... SET</code>.</td>
+    </tr>
+    <tr>
+      <td><h5>blob-descriptor-field</h5></td>
+      <td>No</td>
+      <td style="word-wrap: break-word;">(none)</td>
+      <td>String</td>
+      <td>
+        Comma-separated BLOB field names stored as serialized <code>BlobDescriptor</code> bytes inline in normal data files.
+        By default, all blob fields store blob bytes in separate <code>.blob</code> files.
+        If configured, one table can mix:
+        some BLOB fields in <code>.blob</code> files and some as descriptor references.
+      </td>
     </tr>
     <tr>
       <td><h5>blob.target-file-size</h5></td>
@@ -217,31 +229,18 @@ SELECT id, name FROM image_table;
 SELECT * FROM image_table WHERE id = 1;
 ```
 
-### Blob Descriptor Mode
+### Blob Read Output Mode (`blob-as-descriptor`)
 
-When you want to store references from external blob data (stored in object storage) without loading the entire blob into memory, you can use the `blob-as-descriptor` option:
+`blob-as-descriptor` only controls how blob values are returned when reading.
 
 ```sql
--- Create table in descriptor mode
-CREATE TABLE blob_table (
-    id INT,
-    name STRING,
-    image BYTES
-) WITH (
-    'row-tracking.enabled' = 'true',
-    'data-evolution.enabled' = 'true',
-    'blob-field' = 'image',
-    'blob-as-descriptor' = 'true'
-);
-
--- Insert with serialized BlobDescriptor bytes
--- The BlobDescriptor contains: version (1 byte) + uri_length (4 bytes) + uri_bytes + offset (8 bytes) + length (8 bytes)
--- Paimon will read from the descriptor's URI and stream data into Paimon's blob files in small chunks, avoiding loading the entire blob into memory
-INSERT INTO blob_table VALUES (1, 'photo', X'<serialized_blob_descriptor_hex>');
+-- Return descriptor bytes
+ALTER TABLE blob_table SET ('blob-as-descriptor' = 'true');
+SELECT image FROM blob_table;
 
--- Toggle this setting to control read output format:
+-- Return actual blob bytes
 ALTER TABLE blob_table SET ('blob-as-descriptor' = 'false');
-SELECT * FROM blob_table;  -- Returns actual blob bytes from Paimon storage
+SELECT image FROM blob_table;
 ```
 
 ## Java API Usage
@@ -389,7 +388,7 @@ public class BlobTableExample {
 }
 ```
 
-### Inserting Blob Data
+### Construct blob from different sources
 
 ```java
 // From byte array (data already in memory)
@@ -442,17 +441,13 @@ long offset = descriptor.offset();  // Starting position in the file
 long length = descriptor.length();  // Length of the blob data
 ```
 
-### Blob Descriptor Mode
-
-The `blob-as-descriptor` option enables **memory-efficient writing** for very large blobs. When enabled, you provide a `BlobDescriptor` pointing to external data, and Paimon streams the data from the external source into Paimon's `.blob` files without loading the entire blob into memory.
+### Descriptor-Aware Write Behavior
 
-**How it works:**
-1. **Writing**: You provide a serialized `BlobDescriptor` (containing URI, offset, length) as the blob field value
-2. **Paimon copies the data**: Paimon reads from the descriptor's URI in small chunks (e.g., 1024 bytes at a time) and writes to Paimon's `.blob` files
-3. **Data is stored in Paimon**: The blob data IS copied to Paimon storage, but in a streaming fashion
+Paimon write path is descriptor-aware automatically:
 
-**Key benefit:**
-- **Memory efficiency**: For very large blobs (e.g., gigabyte-sized videos), you don't need to load the entire file into memory. Paimon streams the data incrementally.
+1. For blob fields stored in `.blob` files, input can be either blob bytes or a `BlobDescriptor`.
+2. For fields configured in `blob-descriptor-field`, Paimon stores descriptor bytes inline in data files (no `.blob` files for those fields), and input must be a descriptor.
+3. This behavior does not depend on `blob-as-descriptor`.
 
 ```java
 import org.apache.paimon.catalog.Catalog;
@@ -484,21 +479,21 @@ public class BlobDescriptorExample {
         Catalog catalog = CatalogFactory.createCatalog(catalogContext);
         catalog.createDatabase("my_db", true);
 
-        // Create table with blob-as-descriptor enabled
+        // Create table: store "video" as descriptor bytes inline
         Schema schema = Schema.newBuilder()
                 .column("id", DataTypes.INT())
                 .column("name", DataTypes.STRING())
                 .column("video", DataTypes.BLOB())
                 .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
                 .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
-                .option(CoreOptions.BLOB_AS_DESCRIPTOR.key(), "true")  // This is not necessary in java api
+                .option(CoreOptions.BLOB_DESCRIPTOR_FIELD.key(), "video")
                 .build();
 
         Identifier tableId = Identifier.create("my_db", "video_table");
         catalog.createTable(tableId, schema, true);
         Table table = catalog.getTable(tableId);
 
-        // Write large blob using descriptor (memory-efficient)
+        // Write blob using descriptor reference
         writeLargeBlobWithDescriptor(table);
 
         // Read blob data
@@ -514,7 +509,7 @@ public class BlobDescriptorExample {
             // For a very large file (e.g., 2GB video), instead of loading into memory:
             //   byte[] hugeVideo = Files.readAllBytes(...);  // This would cause OutOfMemoryError!
             //
-            // Use BlobDescriptor to let Paimon stream the data:
+            // Create a descriptor reference to external blob
             String externalUri = "s3://my-bucket/videos/large_video.mp4";
             long fileSize = 2L * 1024 * 1024 * 1024;  // 2GB
 
@@ -524,8 +519,6 @@ public class BlobDescriptorExample {
             UriReader uriReader = UriReader.fromFile(fileIO);
             Blob blob = Blob.fromDescriptor(uriReader, descriptor);
 
-            // Write the serialized descriptor as blob data
-            // Paimon will read from the URI and copy data to .blob files in chunks
             GenericRow row = GenericRow.of(
                     1,
                     BinaryString.fromString("large_video"),
@@ -535,7 +528,7 @@ public class BlobDescriptorExample {
             commit.commit(write.prepareCommit());
         }
 
-        System.out.println("Successfully wrote large blob using descriptor mode");
+        System.out.println("Successfully wrote large blob using descriptor reference");
     }
 
     private static void readBlobData(Table table) throws Exception {
@@ -548,20 +541,19 @@ public class BlobDescriptorExample {
             String name = row.getString(1).toString();
             Blob blob = row.getBlob(2);
 
-            // The blob data is now stored in Paimon's .blob files
-            // blob.toDescriptor() returns a descriptor pointing to Paimon's internal storage
+            // Field is configured in blob-descriptor-field, so descriptor is stored inline
             BlobDescriptor descriptor = blob.toDescriptor();
             System.out.println("Row " + id + ": " + name);
-            System.out.println("  Paimon blob URI: " + descriptor.uri());
+            System.out.println("  Blob URI: " + descriptor.uri());
             System.out.println("  Length: " + descriptor.length());
         });
     }
 }
 ```
 
-**Reading blob data with different modes:**
+**Reading blob data with different output modes:**
 
-The `blob-as-descriptor` option also affects how data is returned when reading:
+The `blob-as-descriptor` option affects only read output:
 
 ```sql
 -- When blob-as-descriptor = true: Returns BlobDescriptor bytes (reference to Paimon blob file)
@@ -573,33 +565,29 @@ ALTER TABLE video_table SET ('blob-as-descriptor' = 'false');
 SELECT * FROM video_table;  -- Returns actual blob bytes from Paimon storage
 ```
 
+### Blob storage mode: DESCRIPTOR ONLY
+
+If you want downstream tables to **reuse** upstream blob files (no copying and no new <code>.blob</code> files), configure the target blob field(s):
+
+```sql
+'blob-descriptor-field' = 'image'
+```
+
+For these configured fields, Paimon stores only serialized <code>BlobDescriptor</code> bytes in normal data files. Reading the blob follows the descriptor URI to access bytes, and writing requires descriptor input for those fields.
+
 ## Limitations
 
-1. **Single Blob Field**: Currently, only one blob field per table is supported.
-2. **Append Table Only**: Blob type is designed for append-only tables. Primary key tables are not supported.
-3. **No Predicate Pushdown**: Blob columns cannot be used in filter predicates.
-4. **No Statistics**: Statistics collection is not supported for blob columns.
-5. **Required Options**: `row-tracking.enabled` and `data-evolution.enabled` must be set to `true`.
+1. **Append Table Only**: Blob type is designed for append-only tables. Primary key tables are not supported.
+2. **No Predicate Pushdown**: Blob columns cannot be used in filter predicates.
+3. **No Statistics**: Statistics collection is not supported for blob columns.
+4. **Required Options**: `row-tracking.enabled` and `data-evolution.enabled` must be set to `true`.
 
 ## Best Practices
 
 1. **Use Column Projection**: Always select only the columns you need. Avoid `SELECT *` if you don't need blob data.
 
 2. **Set Appropriate Target File Size**: Configure `blob.target-file-size` based on your blob sizes. Larger values mean fewer files but larger individual files.
 
-3. **Consider Descriptor Mode**: For very large blobs that cannot fit in memory, use `blob-as-descriptor` mode to stream data from external sources into Paimon without loading the entire blob into memory.
+3. **Use Descriptor Fields When Reusing External Blob Files**: Configure `blob-descriptor-field` for fields that should keep descriptor references instead of writing new `.blob` files.
 
 4. **Use Partitioning**: Partition your blob tables by date or other dimensions to improve query performance and data management.
-
-```sql
-CREATE TABLE partitioned_blob_table (
-    id INT,
-    name STRING,
-    image BYTES,
-    dt STRING
-) PARTITIONED BY (dt) WITH (
-    'row-tracking.enabled' = 'true',
-    'data-evolution.enabled' = 'true',
-    'blob-field' = 'image'
-);
-```

docs/layouts/shortcodes/generated/core_configuration.html

@@ -68,6 +68,12 @@
             <td>Boolean</td>
             <td>Whether to consider blob file size as a factor when performing scan splitting.</td>
         </tr>
+        <tr>
+            <td><h5>blob-descriptor-field</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>Comma-separated BLOB field names to store as serialized BlobDescriptor bytes inline in data files.</td>
+        </tr>
         <tr>
             <td><h5>blob.target-file-size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2146,6 +2146,16 @@ public InlineElement getDescription() {
                             "Specifies column names that should be stored as blob type. "
                                     + "This is used when you want to treat a BYTES column as a BLOB.");
 
+    @Immutable
+    public static final ConfigOption<String> BLOB_DESCRIPTOR_FIELD =
+            key("blob-descriptor-field")
+                    .stringType()
+                    .noDefaultValue()
+                    .withFallbackKeys("blob.stored-descriptor-fields")
+                    .withDescription(
+                            "Comma-separated BLOB field names to store as serialized BlobDescriptor "
+                                    + "bytes inline in data files.");
+
     public static final ConfigOption<Boolean> BLOB_AS_DESCRIPTOR =
             key("blob-as-descriptor")
                     .booleanType()
@@ -2710,6 +2720,22 @@ public boolean blobSplitByFileSize() {
                 .orElse(!options.get(BLOB_AS_DESCRIPTOR));
     }
 
+    /**
+     * Resolve blob fields that should be stored as serialized descriptor bytes in data files.
+     *
+     * <p>If this option is not set, all blob fields are stored in '.blob' files by default.
+     */
+    public Set<String> blobDescriptorField() {
+        return options.getOptional(BLOB_DESCRIPTOR_FIELD)
+                .map(
+                        s ->
+                                Arrays.stream(s.split(","))
+                                        .map(String::trim)
+                                        .filter(str -> !str.isEmpty())
+                                        .collect(Collectors.toSet()))
+                .orElse(Collections.emptySet());
+    }
+
     public long compactionFileSize(boolean hasPrimaryKey) {
         // file size to join the compaction, we don't process on middle file size to avoid
         // compact a same file twice (the compression is not calculate so accurately. the output

paimon-api/src/main/java/org/apache/paimon/types/BlobType.java

@@ -23,6 +23,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 
 /**
  * Data type of binary large object.
@@ -67,13 +68,25 @@ public <R> R accept(DataTypeVisitor<R> visitor) {
     }
 
     public static Pair<RowType, RowType> splitBlob(RowType rowType) {
+        return splitBlob(rowType, java.util.Collections.emptySet());
+    }
+
+    /**
+     * Split row fields into normal fields and blob-file fields.
+     *
+     * <p>Blob fields contained in {@code blobDescriptorField} are treated as normal fields (stored
+     * inline as serialized descriptor bytes), while other blob fields are treated as blob-file
+     * fields.
+     */
+    public static Pair<RowType, RowType> splitBlob(
+            RowType rowType, Set<String> blobDescriptorFields) {
         List<DataField> fields = rowType.getFields();
         List<DataField> normalFields = new ArrayList<>();
         List<DataField> blobFields = new ArrayList<>();
 
         for (DataField field : fields) {
             DataTypeRoot type = field.type().getTypeRoot();
-            if (type == DataTypeRoot.BLOB) {
+            if (type == DataTypeRoot.BLOB && !blobDescriptorFields.contains(field.name())) {
                 blobFields.add(field);
             } else {
                 normalFields.add(field);

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

@@ -45,8 +45,8 @@
 public class BlobDescriptor implements Serializable {
 
     private static final long serialVersionUID = 1L;
-
-    private static final byte CURRENT_VERSION = 1;
+    private static final long MAGIC = 0x424C4F4244455343L; // "BLOBDESC"
+    private static final byte CURRENT_VERSION = 2;
 
     private final byte version;
     private final String uri;
@@ -113,11 +113,12 @@ public byte[] serialize() {
         byte[] uriBytes = uri.getBytes(UTF_8);
         int uriLength = uriBytes.length;
 
-        int totalSize = 1 + 4 + uriLength + 8 + 8;
+        int totalSize = 1 + 8 + 4 + uriLength + 8 + 8;
         ByteBuffer buffer = ByteBuffer.allocate(totalSize);
         buffer.order(ByteOrder.LITTLE_ENDIAN);
 
         buffer.put(version);
+        buffer.putLong(MAGIC);
         buffer.putInt(uriLength);
         buffer.put(uriBytes);
 
@@ -132,14 +133,26 @@ public static BlobDescriptor deserialize(byte[] bytes) {
         buffer.order(ByteOrder.LITTLE_ENDIAN);
 
         byte version = buffer.get();
-        if (version != CURRENT_VERSION) {
+        if (version > CURRENT_VERSION) {
             throw new UnsupportedOperationException(
-                    "Expecting BlobDescriptor version to be "
+                    "Expecting BlobDescriptor version to be less than or equal to "
                             + CURRENT_VERSION
                             + ", but found "
                             + version
                             + ".");
         }
+
+        if (version > 1) {
+            long magic = buffer.getLong();
+            if (MAGIC != magic) {
+                throw new IllegalArgumentException(
+                        "Invalid BlobDescriptor: missing magic header. Expected magic: "
+                                + MAGIC
+                                + ", but found: "
+                                + magic);
+            }
+        }
+
         int uriLength = buffer.getInt();
         byte[] uriBytes = new byte[uriLength];
         buffer.get(uriBytes);
@@ -149,4 +162,18 @@ public static BlobDescriptor deserialize(byte[] bytes) {
         long length = buffer.getLong();
         return new BlobDescriptor(version, uri, offset, length);
     }
+
+    public static boolean isBlobDescriptor(byte[] bytes) {
+        if (bytes.length < 9) {
+            return false;
+        }
+        ByteBuffer buffer = ByteBuffer.wrap(bytes);
+        buffer.order(ByteOrder.LITTLE_ENDIAN);
+
+        byte version = buffer.get();
+        if (version > CURRENT_VERSION) {
+            return false;
+        }
+        return MAGIC == buffer.getLong();
+    }
 }