From f47b53fe9474f53753b01331ab005e5f42fd0300 Mon Sep 17 00:00:00 2001 From: duanyyyyyyy Date: Wed, 8 Jul 2026 19:13:42 +0800 Subject: [PATCH] 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). --- .../operation/data_evolution_split_read.cpp | 4 +- .../data_evolution_split_read_test.cpp | 30 ++++++- test/inte/blob_table_inte_test.cpp | 82 +++++++++++++++++++ 3 files changed, 109 insertions(+), 7 deletions(-) diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index fab21ac01..129c427ea 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -94,9 +94,7 @@ Status DataEvolutionSplitRead::BlobBunch::Add(const std::shared_ptrschema_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."); diff --git a/src/paimon/core/operation/data_evolution_split_read_test.cpp b/src/paimon/core/operation/data_evolution_split_read_test.cpp index aa6f9c9e1..0dddd900b 100644 --- a/src/paimon/core/operation/data_evolution_split_read_test.cpp +++ b/src/paimon/core/operation/data_evolution_split_read_test.cpp @@ -83,13 +83,13 @@ class DataEvolutionSplitReadTest : public ::testing::Test { std::shared_ptr CreateBlobFile( const std::string& file_name, int64_t first_row_id, int64_t row_count, - int64_t max_sequence_number, - const std::optional>& write_cols) const { + int64_t max_sequence_number, const std::optional>& 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(), + /*min_sequence_number=*/0, max_sequence_number, 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>({"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>({"blob_col"}), + /*schema_id=*/1); + auto blob_bunch = std::make_shared( + 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, diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index 0338dcf84..4ee38fe53 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -840,6 +840,88 @@ 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 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 write_cols_base = {"f0", "f2"}; + auto base_array = std::dynamic_pointer_cast( + 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 write_cols_f1a = {"f1"}; + auto f1a_array = std::dynamic_pointer_cast( + 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> 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 write_cols_f1b = {"f1"}; + auto f1b_array = std::dynamic_pointer_cast( + 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::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");