Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
duanyyyyyyy wants to merge 1 commit into
base: main
Choose a base branch
from
+109 −7
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions src/paimon/core/operation/data_evolution_split_read.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.");
Expand Down
30 changes: 26 additions & 4 deletions src/paimon/core/operation/data_evolution_split_read_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,13 +83,13 @@ 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 {
int64_t max_sequence_number, 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(),
/*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();
Expand Down Expand Up @@ -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);
}

@lxy-9602 lxy-9602 Jul 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add an integration test in blob_table_inte_test.cpp that uses WriteNextSchema to simulate an ALTER TABLE scenario?

TEST_F(DataEvolutionSplitReadTest, TestRowIdSelectionWithOverlap) {
auto blob_entry =
CreateBlobFile("blob1", /*first_row_id=*/0, /*row_count=*/10,
Expand Down
82 changes: 82 additions & 0 deletions test/inte/blob_table_inte_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<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));
}

@lxy-9602 lxy-9602 Jul 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I tested this integration test locally against the unmodified/pre-fix code, and it still passes. So I don’t think this test actually hits the regression scenario.

The test creates two f1 blob files with adjacent row ranges: [0, 0] under schema 0 and [1, 1] under schema 1. However, RangeHelper::MergeOverlappingRanges only merges overlapping ranges, not adjacent ranges. As a result, these two files are split into different need_merge_files groups, and each group creates its own BlobBunch with only one blob file. The old schema-id check is therefore never exercised.

To cover the real scenario, the integration test needs to ensure that two blob files with different schema ids are added to the same BlobBunch, for example all with first row id 0.


TEST_P(BlobTableInteTest, TestMoreDataWithDataEvolution) {
CreateTable();
std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
Expand Down
Loading