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 one).
MergeRangesAndSort groups them into one bunch by row id, so the check raised a
spurious "All files in a blob bunch should have the same schema id."

A blob column's on-disk layout is schema-independent and the bunch is read with
the first file's schema regardless, so blob files may span schema ids. BlobBunch
only ever holds blob files (non-blob files are rejected up front), so the check is
removed. This matches Paimon Java, whose SpecialFieldBunch.add guards the same
check with if (!isBlobFile(file)).

Adds a unit test (TestAddBlobFilesWithDifferentSchemaId) and an integration test
(TestBlobBunchSpanningSchemaIds, which uses the existing WriteNextSchema helper to
simulate an ALTER TABLE between two appends of the same blob field).

Author: duanyyyyyyy

src/paimon/core/operation/data_evolution_split_read.cpp

@@ -94,9 +94,7 @@ 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.");
-            }
+            // Paimon Java does not check the schema id for blob files in a bunch.
             if (file->write_cols != files_[0]->write_cols) {
                 return Status::Invalid(
                     "All files in a blob bunch should have the same write columns.");

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,

test/inte/blob_table_inte_test.cpp

@@ -840,6 +840,85 @@ TEST_P(BlobTableInteTest, TestMultipleAppendsDifferentFirstRowIds) {
     }
 }
 
+TEST_P(BlobTableInteTest, TestBlobBunchSpanningSchemaIds) {
+    // Regression test: a blob field whose contiguous files were written under different schema
+    // ids (an ALTER TABLE happens between two appends) must still be groupable into a single
+    // blob bunch. Same shape as TestMultipleAppendsDifferentFirstRowIds, with a WriteNextSchema()
+    // inserted between the two commits so the second f1 blob carries schema_id 1 while the first
+    // keeps schema_id 0.
+    std::map<std::string, std::string> options = {{Options::MANIFEST_FORMAT, "orc"},
+                                                  {Options::FILE_FORMAT, GetParam()},
+                                                  {Options::FILE_SYSTEM, "local"},
+                                                  {Options::ROW_TRACKING_ENABLED, "true"},
+                                                  {Options::DATA_EVOLUTION_ENABLED, "true"}};
+    CreateTable(/*partition_keys=*/{}, options);
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+    auto schema = arrow::schema(fields_);
+
+    // First commit under schema 0, firstRowId = 0.
+    std::vector<std::string> write_cols1 = {"f0", "f1"};
+    auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
+        [1, "a"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs1, WriteArray(table_path, {}, write_cols1, {src_array1}));
+    SetFirstRowId(0, commit_msgs1);
+
+    std::vector<std::string> write_cols2 = {"f2"};
+    auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
+        ["b"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs2, WriteArray(table_path, {}, write_cols2, {src_array2}));
+    SetFirstRowId(0, commit_msgs2);
+
+    std::vector<std::shared_ptr<CommitMessage>> total_msgs;
+    total_msgs.insert(total_msgs.end(), commit_msgs1.begin(), commit_msgs1.end());
+    total_msgs.insert(total_msgs.end(), commit_msgs2.begin(), commit_msgs2.end());
+    ASSERT_OK(Commit(table_path, total_msgs));
+
+    // ALTER TABLE ADD COLUMN f3 -> schema 1. Files written afterwards carry schema_id 1.
+    ASSERT_OK(WriteNextSchema(
+        table_path,
+        {DataField(0, fields_[0]), DataField(1, fields_[1]), DataField(2, fields_[2]),
+         DataField(3, arrow::field("f3", arrow::utf8()))},
+        /*highest_field_id=*/3, options));
+
+    // Second commit under schema 1, firstRowId = 1. The f1 blob here has schema_id 1; it is
+    // contiguous with the schema_id 0 f1 blob above, so on read both land in one blob bunch.
+    std::vector<std::string> write_cols3 = {"f0", "f1"};
+    auto src_array3 = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[1]}), R"([
+        [2, "c"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs3, WriteArray(table_path, {}, write_cols3, {src_array3}));
+    SetFirstRowId(1, commit_msgs3);
+    ASSERT_OK(Commit(table_path, commit_msgs3));
+
+    std::vector<std::string> write_cols4 = {"f2"};
+    auto src_array4 = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[2]}), R"([
+        ["d"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs4, WriteArray(table_path, {}, write_cols4, {src_array4}));
+    SetFirstRowId(1, commit_msgs4);
+    ASSERT_OK(Commit(table_path, commit_msgs4));
+
+    // Without the fix the read fails with
+    // "All files in a blob bunch should have the same schema id."; with it both rows read.
+    auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
+        [1, "a", "b"],
+        [2, "c", "d"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK(ScanAndRead(table_path, schema->field_names(), expected_array));
+}
+
 TEST_P(BlobTableInteTest, TestMoreDataWithDataEvolution) {
     CreateTable();
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");