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.
+    //
+    // BlobBunch::Add only reaches the per-bunch schema/write-cols check when a second blob file is
+    // *strictly contiguous* with the previous one (first_row_id == expected_next_first_row_id_);
+    // the same-first-row-id and overlapping cases return early. Two 1-row f1 blobs at [0,0] and
+    // [1,1] are contiguous, but their ranges do not overlap, so on their own MergeOverlappingRanges
+    // would split them into different need_merge_files groups and each would form a singleton bunch
+    // (never hitting the check). We therefore write the non-blob columns f0/f2 as one file spanning
+    // both rows [0,1]; that file overlaps both f1 blobs and bridges them into a single group, where
+    // the two contiguous f1 blobs (schema 0 then schema 1) are added to the same bunch.
+    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. The non-blob columns f0/f2 are written as one file spanning
+    // both rows [0,1] -- this is the file that bridges the two f1 blobs into one group. The f1
+    // blob covers only row 0 -> [0,0], schema_id 0.
+    std::vector<std::string> write_cols_base = {"f0", "f2"};
+    auto base_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[2]}), R"([
+        [1, "b"],
+        [2, "d"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs_base, WriteArray(table_path, {}, write_cols_base, {base_array}));
+    SetFirstRowId(0, commit_msgs_base);
+
+    std::vector<std::string> write_cols_f1a = {"f1"};
+    auto f1a_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
+        ["a"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs_f1a, WriteArray(table_path, {}, write_cols_f1a, {f1a_array}));
+    SetFirstRowId(0, commit_msgs_f1a);
+
+    std::vector<std::shared_ptr<CommitMessage>> total_msgs;
+    total_msgs.insert(total_msgs.end(), commit_msgs_base.begin(), commit_msgs_base.end());
+    total_msgs.insert(total_msgs.end(), commit_msgs_f1a.begin(), commit_msgs_f1a.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: the f1 blob for row 1 -> [1,1], schema_id 1. It is strictly
+    // contiguous with the schema_id 0 f1 blob above ([0,0]), so both land in the same blob bunch.
+    std::vector<std::string> write_cols_f1b = {"f1"};
+    auto f1b_array = std::dynamic_pointer_cast<arrow::StructArray>(
+        arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
+        ["c"]
+    ])")
+            .ValueOrDie());
+    ASSERT_OK_AND_ASSIGN(auto commit_msgs_f1b, WriteArray(table_path, {}, write_cols_f1b, {f1b_array}));
+    SetFirstRowId(1, commit_msgs_f1b);
+    ASSERT_OK(Commit(table_path, commit_msgs_f1b));
+
+    // 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");