[core] Add blobFieldName to BlobConsumer

Author: JingsongLi

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

@@ -26,5 +26,5 @@ public interface BlobConsumer {
      *
      * @return Whether to flush to output stream.
      */
-    boolean accept(BlobDescriptor blobDescriptor);
+    boolean accept(String blobFieldName, BlobDescriptor blobDescriptor);
 }

paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java

@@ -67,7 +67,12 @@ public void testBlobConsumer() throws Exception {
         FileStoreTable table = getTableDefault();
         List<BlobDescriptor> blobs = new ArrayList<>();
         try (BatchTableWrite write = table.newBatchWriteBuilder().newWrite()) {
-            write.withBlobConsumer(blobs::add);
+            write.withBlobConsumer(
+                    (blobFieldName, blobDescriptor) -> {
+                        assertThat(blobFieldName).isEqualTo("f2");
+                        blobs.add(blobDescriptor);
+                        return true;
+                    });
             write.write(dataDefault(0, 0));
             write.write(dataDefault(0, 0));
         }

paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java

@@ -79,7 +79,7 @@ public FormatReaderFactory createReaderFactory(
 
     @Override
     public FormatWriterFactory createWriterFactory(RowType type) {
-        return new BlobFormatWriterFactory();
+        return new BlobFormatWriterFactory(type);
     }
 
     @Override
@@ -99,9 +99,15 @@ public Optional<SimpleStatsExtractor> createStatsExtractor(
 
     private class BlobFormatWriterFactory implements FormatWriterFactory {
 
+        private final RowType type;
+
+        private BlobFormatWriterFactory(RowType type) {
+            this.type = type;
+        }
+
         @Override
         public FormatWriter create(PositionOutputStream out, String compression) {
-            return new BlobFormatWriter(out, writeConsumer);
+            return new BlobFormatWriter(out, writeConsumer, type);
         }
     }
 

paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.PositionOutputStream;
 import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.DeltaVarintCompressor;
 import org.apache.paimon.utils.LongArrayList;
 
@@ -48,15 +49,19 @@ public class BlobFormatWriter implements FileAwareFormatWriter {
 
     private final PositionOutputStream out;
     @Nullable private final BlobConsumer writeConsumer;
+    private final String blobFieldName;
     private final CRC32 crc32;
     private final byte[] tmpBuffer;
     private final LongArrayList lengths;
 
     private String pathString;
 
-    public BlobFormatWriter(PositionOutputStream out, @Nullable BlobConsumer writeConsumer) {
+    public BlobFormatWriter(
+            PositionOutputStream out, @Nullable BlobConsumer writeConsumer, RowType type) {
         this.out = out;
         this.writeConsumer = writeConsumer;
+        checkArgument(type.getFieldCount() == 1, "BlobFormatWriter only support one field.");
+        this.blobFieldName = type.getFieldNames().get(0);
         this.crc32 = new CRC32();
         this.tmpBuffer = new byte[4096];
         this.lengths = new LongArrayList(16);
@@ -102,7 +107,7 @@ public void addElement(InternalRow element) throws IOException {
 
         if (writeConsumer != null) {
             BlobDescriptor descriptor = new BlobDescriptor(pathString, blobPos, blobLength);
-            boolean flush = writeConsumer.accept(descriptor);
+            boolean flush = writeConsumer.accept(blobFieldName, descriptor);
             if (flush) {
                 out.flush();
             }