fix(blob): allow blob files with different schema ids in a bunch

BlobBunch::Add required every file in a bunch to share the same schema id.
After schema evolution that adds/drops OTHER columns, a single blob field's
contiguous files can carry different schema ids (an old appended file plus a
compacted file), and MergeRangesAndSort groups them into one bunch by row id,
raising a spurious "All files in a blob bunch should have the same schema id."

Paimon Java (SpecialFieldBunch.add) guards this check with if (!isBlobFile(file))
so blob files are exempt -- a blob column's on-disk layout is schema-independent
and the bunch is read with the first file's schema regardless. The guard was
dropped when porting to C++. Restore it and add a regression unit test.

Author: duanyan.duan

src/paimon/core/operation/data_evolution_split_read.cpp

@@ -94,8 +94,18 @@ Status DataEvolutionSplitRead::BlobBunch::Add(const std::shared_ptr<DataFileMeta
             }
         }
         if (!files_.empty()) {
-            if (file->schema_id != files_[0]->schema_id) {
-                return Status::Invalid("All files in a blob bunch should have the same schema id.");
+            // Mirror Paimon Java (SpecialFieldBunch.add): the schema id must agree only for
+            // non-blob (vector-store) files. Blob files are allowed to span schema ids -- a
+            // blob column's on-disk layout is schema-independent, so after schema evolution
+            // (which adds/drops OTHER columns) the same blob field can have contiguous files
+            // written under different schema ids that MergeRangesAndSort groups into one
+            // bunch. The Java guard `if (!isBlobFile(file))` around this check was dropped
+            // when porting, which turned that legitimate case into a spurious error.
+            if (!BlobUtils::IsBlobFile(file->file_name)) {
+                if (file->schema_id != files_[0]->schema_id) {
+                    return Status::Invalid(
+                        "All files in a blob bunch should have the same schema id.");
+                }
             }
             if (file->write_cols != files_[0]->write_cols) {
                 return Status::Invalid(

src/paimon/core/operation/data_evolution_split_read_test.cpp

@@ -84,12 +84,12 @@ class DataEvolutionSplitReadTest : public ::testing::Test {
     std::shared_ptr<DataFileMeta> CreateBlobFile(
         const std::string& file_name, int64_t first_row_id, int64_t row_count,
         int64_t max_sequence_number,
-        const std::optional<std::vector<std::string>>& write_cols) const {
+        const std::optional<std::vector<std::string>>& write_cols, int64_t schema_id = 0) const {
         return DataFileMeta::ForAppend(file_name + ".blob", /*file_size=*/row_count,
                                        /*row_count=*/row_count,
                                        /*row_stats=*/SimpleStats::EmptyStats(),
                                        /*min_sequence_number=*/0, max_sequence_number,
-                                       /*schema_id=*/0, FileSource::Append(),
+                                       schema_id, FileSource::Append(),
                                        /*value_stats_cols=*/std::nullopt,
                                        /*external_path=*/std::nullopt, first_row_id, write_cols)
             .value();
@@ -247,6 +247,28 @@ TEST_F(DataEvolutionSplitReadTest, TestAddBlobFileWithDifferentWriteCols) {
                         "All files in a blob bunch should have the same write columns.");
 }
 
+TEST_F(DataEvolutionSplitReadTest, TestAddBlobFilesWithDifferentSchemaId) {
+    // A blob field written under schema 0 for the first rows and under schema 1 for the
+    // following rows (after schema evolution that adds/drops OTHER columns) must still be
+    // groupable into a single bunch: a blob column's on-disk layout is schema-independent.
+    // Same write cols, contiguous row ids, different schema id -> both added OK.
+    auto blob_entry =
+        CreateBlobFile("blob1", /*first_row_id=*/0, /*row_count=*/100,
+                       /*max_sequence_number=*/1,
+                       /*write_cols=*/std::optional<std::vector<std::string>>({"blob_col"}),
+                       /*schema_id=*/0);
+    auto blob_tail =
+        CreateBlobFile("blob2", /*first_row_id=*/100, /*row_count=*/200,
+                       /*max_sequence_number=*/1,
+                       /*write_cols=*/std::optional<std::vector<std::string>>({"blob_col"}),
+                       /*schema_id=*/1);
+    auto blob_bunch = std::make_shared<DataEvolutionSplitRead::BlobBunch>(
+        INT64_MAX, /*has_row_ids_selection=*/false);
+    ASSERT_OK(blob_bunch->Add(blob_entry));
+    ASSERT_OK(blob_bunch->Add(blob_tail));
+    ASSERT_EQ(blob_bunch->Files().size(), 2);
+}
+
 TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap) {
     auto blob_entry =
         CreateBlobFile("blob1", /*first_row_id=*/0, /*row_count=*/10,