[core] Fix: blob meta should contains a filter to match normal data file meta (#6412)

Author: leaves12138

paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionSplitGenerator.java

@@ -97,6 +97,8 @@ public static List<List<DataFileMeta>> split(List<DataFileMeta> files) {
                                             f2.maxSequenceNumber(), f1.maxSequenceNumber());
                                 }));
 
+        files = filterBlob(files);
+
         // Split files by firstRowId
         long lastRowId = -1;
         long checkRowIdStart = 0;
@@ -128,4 +130,26 @@ public static List<List<DataFileMeta>> split(List<DataFileMeta> files) {
 
         return splitByRowId;
     }
+
+    private static List<DataFileMeta> filterBlob(List<DataFileMeta> files) {
+        List<DataFileMeta> result = new ArrayList<>();
+        long rowIdStart = -1;
+        long rowIdEnd = -1;
+        for (DataFileMeta file : files) {
+            if (file.firstRowId() == null) {
+                result.add(file);
+                continue;
+            }
+            if (!isBlobFile(file.fileName())) {
+                rowIdStart = file.firstRowId();
+                rowIdEnd = file.firstRowId() + file.rowCount();
+                result.add(file);
+            } else {
+                if (file.firstRowId() >= rowIdStart && file.firstRowId() < rowIdEnd) {
+                    result.add(file);
+                }
+            }
+        }
+        return result;
+    }
 }

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobTestBase.scala

@@ -65,10 +65,14 @@ class BlobTestBase extends PaimonSparkTestBase {
 
       sql(
         "CREATE TABLE t (id INT, data STRING, picture BINARY) TBLPROPERTIES ('row-tracking.enabled'='true', 'data-evolution.enabled'='true', 'blob-field'='picture', 'blob-as-descriptor'='true')")
-      sql("INSERT INTO t VALUES (1, 'paimon', X'" + bytesToHex(blobDescriptor.serialize()) + "')")
-
+      sql(
+        "INSERT INTO t VALUES (1, 'paimon', X'" + bytesToHex(blobDescriptor.serialize()) + "'),"
+          + "(5, 'paimon', X'" + bytesToHex(blobDescriptor.serialize()) + "'),"
+          + "(2, 'paimon', X'" + bytesToHex(blobDescriptor.serialize()) + "'),"
+          + "(3, 'paimon', X'" + bytesToHex(blobDescriptor.serialize()) + "'),"
+          + "(4, 'paimon', X'" + bytesToHex(blobDescriptor.serialize()) + "')")
       val newDescriptorBytes =
-        sql("SELECT picture FROM t").collect()(0).get(0).asInstanceOf[Array[Byte]]
+        sql("SELECT picture FROM t WHERE id = 1").collect()(0).get(0).asInstanceOf[Array[Byte]]
       val newBlobDescriptor = BlobDescriptor.deserialize(newDescriptorBytes)
       val options = new Options()
       options.set("warehouse", tempDBDir.toString)
@@ -79,7 +83,7 @@ class BlobTestBase extends PaimonSparkTestBase {
 
       sql("ALTER TABLE t SET TBLPROPERTIES ('blob-as-descriptor'='false')")
       checkAnswer(
-        sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t"),
+        sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t WHERE id = 1"),
         Seq(Row(1, "paimon", blobData, 0, 1))
       )
     }