fix

Author: JingsongLi

paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/data/Spark4InternalRowWithBlob.scala

@@ -25,11 +25,11 @@ import org.apache.paimon.utils.InternalRowUtils.copyInternalRow
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.unsafe.types.VariantVal
 
-class Spark4InternalRowWithBlob(rowType: RowType, blobFieldIndex: Int, blobAsDescriptor: Boolean)
+class Spark4InternalRowWithBlob(rowType: RowType, blobFields: Set[Int], blobAsDescriptor: Boolean)
   extends Spark4InternalRow(rowType) {
 
   override def getBinary(ordinal: Int): Array[Byte] = {
-    if (ordinal == blobFieldIndex) {
+    if (blobFields.contains(ordinal)) {
       if (blobAsDescriptor) {
         row.getBlob(ordinal).toDescriptor.serialize()
       } else {

paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala

@@ -57,9 +57,9 @@ class Spark4Shim extends SparkShim {
 
   override def createSparkInternalRowWithBlob(
       rowType: RowType,
-      blobFieldIndex: Int,
+      blobFields: Set[Int],
       blobAsDescriptor: Boolean): SparkInternalRow = {
-    new Spark4InternalRowWithBlob(rowType, blobFieldIndex, blobAsDescriptor)
+    new Spark4InternalRowWithBlob(rowType, blobFields, blobAsDescriptor)
   }
 
   override def createSparkArrayData(elementType: DataType): SparkArrayData = {