[flink] Support creating table with blob in flink sql. (#6351)

leaves12138 · web-flow · commit daf482e2866c · 2025-09-30T11:45:45.000+08:00
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -50,6 +50,12 @@
             <td>Boolean</td>
             <td>Whether to create underlying storage when reading and writing the table.</td>
         </tr>
+        <tr>
+            <td><h5>blob.field</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>Specify the blob field.</td>
+        </tr>
         <tr>
             <td><h5>bucket</h5></td>
             <td style="word-wrap: break-word;">-1</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1989,6 +1989,12 @@ public InlineElement getDescription() {
                     .defaultValue(false)
                     .withDescription("Format table file path only contain partition value.");
 
+    public static final ConfigOption<String> BLOB_FIELD =
+            key("blob.field")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Specify the blob field.");
+
     private final Options options;
 
     public CoreOptions(Map<String, String> options) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java
@@ -18,22 +18,17 @@
 
 package org.apache.paimon.io;
 
-import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fileindex.FileIndexOptions;
 import org.apache.paimon.format.FileFormat;
-import org.apache.paimon.format.SimpleStatsCollector;
-import org.apache.paimon.format.avro.AvroFileFormat;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.manifest.FileSource;
-import org.apache.paimon.statistics.NoneSimpleColStatsCollector;
 import org.apache.paimon.statistics.SimpleColStatsCollector;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.LongCounter;
 
 import javax.annotation.Nullable;
 
-import java.util.Arrays;
 import java.util.List;
 
 /** {@link RollingFileWriterImpl} for data files containing {@link InternalRow}. */
@@ -58,7 +53,7 @@ public RowDataRollingFileWriter(
                 () ->
                         new RowDataFileWriter(
                                 fileIO,
-                                createFileWriterContext(
+                                RollingFileWriter.createFileWriterContext(
                                         fileFormat, writeSchema, statsCollectors, fileCompression),
                                 pathFactory.newPath(),
                                 writeSchema,
@@ -72,34 +67,4 @@ public RowDataRollingFileWriter(
                                 writeCols),
                 targetFileSize);
     }
-
-    @VisibleForTesting
-    static FileWriterContext createFileWriterContext(
-            FileFormat fileFormat,
-            RowType rowType,
-            SimpleColStatsCollector.Factory[] statsCollectors,
-            String fileCompression) {
-        return new FileWriterContext(
-                fileFormat.createWriterFactory(rowType),
-                createStatsProducer(fileFormat, rowType, statsCollectors),
-                fileCompression);
-    }
-
-    private static SimpleStatsProducer createStatsProducer(
-            FileFormat fileFormat,
-            RowType rowType,
-            SimpleColStatsCollector.Factory[] statsCollectors) {
-        boolean isDisabled =
-                Arrays.stream(SimpleColStatsCollector.create(statsCollectors))
-                        .allMatch(p -> p instanceof NoneSimpleColStatsCollector);
-        if (isDisabled) {
-            return SimpleStatsProducer.disabledProducer();
-        }
-        if (fileFormat instanceof AvroFileFormat) {
-            SimpleStatsCollector collector = new SimpleStatsCollector(rowType, statsCollectors);
-            return SimpleStatsProducer.fromCollector(collector);
-        }
-        return SimpleStatsProducer.fromExtractor(
-                fileFormat.createStatsExtractor(rowType, statsCollectors).orElse(null));
-    }
 }
diff --git a/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/io/RollingFileWriterTest.java
@@ -71,7 +71,7 @@ public void initialize(String identifier, boolean statsDenseStore) {
                         () ->
                                 new RowDataFileWriter(
                                         LocalFileIO.create(),
-                                        RowDataRollingFileWriter.createFileWriterContext(
+                                        RollingFileWriter.createFileWriterContext(
                                                 fileFormat,
                                                 SCHEMA,
                                                 SimpleColStatsCollector.createFullStatsFactories(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -151,6 +151,7 @@
 import static org.apache.paimon.flink.FlinkCatalogOptions.DISABLE_CREATE_TABLE_IN_DEFAULT_DB;
 import static org.apache.paimon.flink.FlinkCatalogOptions.LOG_SYSTEM_AUTO_REGISTER;
 import static org.apache.paimon.flink.FlinkCatalogOptions.REGISTER_TIMEOUT;
+import static org.apache.paimon.flink.LogicalTypeConversion.toBlobType;
 import static org.apache.paimon.flink.LogicalTypeConversion.toDataType;
 import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
 import static org.apache.paimon.flink.log.LogStoreRegister.registerLogSystem;
@@ -1050,6 +1051,16 @@ public static Schema fromCatalogTable(CatalogBaseTable catalogTable) {
         RowType rowType = (RowType) schema.toPhysicalRowDataType().getLogicalType();
 
         Map<String, String> options = new HashMap<>(catalogTable.getOptions());
+        String blobName = options.get(CoreOptions.BLOB_FIELD.key());
+        if (blobName != null) {
+            checkArgument(
+                    options.containsKey(CoreOptions.DATA_EVOLUTION_ENABLED.key()),
+                    "When setting '"
+                            + CoreOptions.BLOB_FIELD.key()
+                            + "', you must also set '"
+                            + CoreOptions.DATA_EVOLUTION_ENABLED.key()
+                            + "'");
+        }
         // Serialize virtual columns and watermark to the options
         // This is what Flink SQL needs, the storage itself does not need them
         options.putAll(columnOptions(schema));
@@ -1069,7 +1080,9 @@ public static Schema fromCatalogTable(CatalogBaseTable catalogTable) {
                         field ->
                                 schemaBuilder.column(
                                         field.getName(),
-                                        toDataType(field.getType()),
+                                        field.getName().equals(blobName)
+                                                ? toBlobType(field.getType())
+                                                : toDataType(field.getType()),
                                         columnComments.get(field.getName())));
 
         return schemaBuilder.build();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowData.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowData.java
@@ -41,7 +41,7 @@
 /** Convert to Flink row data. */
 public class FlinkRowData implements RowData {
 
-    private InternalRow row;
+    protected InternalRow row;
 
     public FlinkRowData(InternalRow row) {
         this.row = row;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowDataWithBlob.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkRowDataWithBlob.java
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.data.InternalRow;
+
+/** Convert to Flink row data with blob. */
+public class FlinkRowDataWithBlob extends FlinkRowData {
+
+    private final int blobField;
+
+    public FlinkRowDataWithBlob(InternalRow row, int blobField) {
+        super(row);
+        this.blobField = blobField;
+    }
+
+    @Override
+    public byte[] getBinary(int pos) {
+        return pos == blobField ? row.getBlob(pos).toData() : row.getBinary(pos);
+    }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeConversion.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeConversion.java
@@ -18,13 +18,18 @@
 
 package org.apache.paimon.flink;
 
+import org.apache.paimon.types.BlobType;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.RowType;
 
+import org.apache.flink.table.types.logical.BinaryType;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VarBinaryType;
 
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /** Conversion between {@link LogicalType} and {@link DataType}. */
 public class LogicalTypeConversion {
 
@@ -37,6 +42,13 @@ public static LogicalType toLogicalType(DataType dataType) {
         return dataType.accept(DataTypeToLogicalType.INSTANCE);
     }
 
+    public static BlobType toBlobType(LogicalType logicalType) {
+        checkArgument(
+                logicalType instanceof BinaryType || logicalType instanceof VarBinaryType,
+                "Expected BinaryType or VarBinaryType, but got: " + logicalType);
+        return new BlobType();
+    }
+
     public static RowType toDataType(org.apache.flink.table.types.logical.RowType logicalType) {
         return (RowType) toDataType(logicalType, new AtomicInteger(-1));
     }
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java
@@ -23,6 +23,7 @@
 import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
 import org.apache.paimon.flink.utils.TableScanUtils;
 import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.RowType;
 
 import org.apache.flink.api.connector.source.SourceReader;
 import org.apache.flink.api.connector.source.SourceReaderContext;
@@ -50,14 +51,16 @@ public FileStoreSourceReader(
             FileStoreSourceReaderMetrics metrics,
             IOManager ioManager,
             @Nullable Long limit,
-            @Nullable NestedProjectedRowData rowData) {
+            @Nullable NestedProjectedRowData rowData,
+            @Nullable RowType readType) {
         // limiter is created in SourceReader, it can be shared in all split readers
         super(
                 () ->
                         new FileStoreSourceSplitReader(
                                 tableRead.withIOManager(ioManager),
                                 RecordLimiter.create(limit),
-                                metrics),
+                                metrics,
+                                readType),
                 (element, output, state) ->
                         FlinkRecordsWithSplitIds.emitRecord(
                                 readerContext, element, output, state, metrics, rowData),
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceSplitReader.java
@@ -20,12 +20,15 @@
 
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.flink.FlinkRowData;
+import org.apache.paimon.flink.FlinkRowDataWithBlob;
 import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.reader.RecordReader.RecordIterator;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Pool;
 
 import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
@@ -77,12 +80,13 @@ public class FileStoreSourceSplitReader
     public FileStoreSourceSplitReader(
             TableRead tableRead,
             @Nullable RecordLimiter limiter,
-            FileStoreSourceReaderMetrics metrics) {
+            FileStoreSourceReaderMetrics metrics,
+            @Nullable RowType readType) {
         this.tableRead = tableRead;
         this.limiter = limiter;
         this.splits = new LinkedList<>();
         this.pool = new Pool<>(1);
-        this.pool.add(new FileStoreRecordIterator());
+        this.pool.add(new FileStoreRecordIterator(readType));
         this.paused = false;
         this.metrics = metrics;
         this.wakeup = new AtomicBoolean(false);
@@ -260,6 +264,20 @@ private class FileStoreRecordIterator implements BulkFormat.RecordIterator<RowDa
 
         private final MutableRecordAndPosition<RowData> recordAndPosition =
                 new MutableRecordAndPosition<>();
+        @Nullable private final Integer blobField;
+
+        private FileStoreRecordIterator(@Nullable RowType rowType) {
+            this.blobField = rowType == null ? null : blobFieldIndex(rowType);
+        }
+
+        private Integer blobFieldIndex(RowType rowType) {
+            for (int i = 0; i < rowType.getFieldCount(); i++) {
+                if (rowType.getTypeAt(i).getTypeRoot() == DataTypeRoot.BLOB) {
+                    return i;
+                }
+            }
+            return null;
+        }
 
         public FileStoreRecordIterator replace(RecordIterator<InternalRow> iterator) {
             this.iterator = iterator;
@@ -283,7 +301,10 @@ public RecordAndPosition<RowData> next() {
                 return null;
             }
 
-            recordAndPosition.setNext(new FlinkRowData(row));
+            recordAndPosition.setNext(
+                    blobField == null
+                            ? new FlinkRowData(row)
+                            : new FlinkRowDataWithBlob(row, blobField));
             currentNumRead++;
             if (limiter != null) {
                 limiter.increment();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSource.java
@@ -67,13 +67,15 @@ public SourceReader<RowData, FileStoreSourceSplit> createReader(SourceReaderCont
                 new FileStoreSourceReaderMetrics(metricGroup);
         TableRead tableRead =
                 readBuilder.newRead().withMetricRegistry(new FlinkMetricRegistry(metricGroup));
+
         return new FileStoreSourceReader(
                 context,
                 tableRead,
                 sourceReaderMetrics,
                 ioManager,
                 limit,
-                NestedProjectedRowData.copy(rowData));
+                NestedProjectedRowData.copy(rowData),
+                readBuilder.readType());
     }
 
     @Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileStoreSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileStoreSource.java
@@ -65,7 +65,13 @@ public SourceReader<RowData, FileStoreSourceSplit> createReader(SourceReaderCont
         FileStoreSourceReaderMetrics sourceReaderMetrics =
                 new FileStoreSourceReaderMetrics(context.metricGroup());
         return new AlignedSourceReader(
-                context, readBuilder.newRead(), sourceReaderMetrics, ioManager, limit, rowData);
+                context,
+                readBuilder.newRead(),
+                sourceReaderMetrics,
+                ioManager,
+                limit,
+                rowData,
+                readBuilder.readType());
     }
 
     @Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedSourceReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedSourceReader.java
@@ -25,6 +25,7 @@
 import org.apache.paimon.flink.source.FileStoreSourceSplitState;
 import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
 import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.ReflectionUtils;
 
 import org.apache.flink.api.connector.source.ExternallyInducedSourceReader;
@@ -60,8 +61,9 @@ public AlignedSourceReader(
             FileStoreSourceReaderMetrics metrics,
             IOManager ioManager,
             @Nullable Long limit,
-            @Nullable NestedProjectedRowData rowData) {
-        super(readerContext, tableRead, metrics, ioManager, limit, rowData);
+            @Nullable NestedProjectedRowData rowData,
+            @Nullable RowType readType) {
+        super(readerContext, tableRead, metrics, ioManager, limit, rowData, readType);
         this.nextCheckpointId = null;
         try {
             // In lower versions of Flink, the SplitFetcherManager does not provide the getQueue
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceSplitReaderTest.java
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedSourceReaderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedSourceReaderTest.java
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DataEvolutionPaimonWriter.scala
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
