Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
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.");
}
// Blob files for the same field may span schema ids.
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
56 changes: 56 additions & 0 deletions test/inte/blob_table_inte_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -872,6 +872,62 @@ TEST_P(BlobTableInteTest, TestBasic) {
expected_row_tracking_array));
}

TEST_P(BlobTableInteTest, TestBlobFilesAcrossSchemaIds) {
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");

// Simulate the post-compaction layout: one normal file bridges blob files across schema ids.
std::vector<std::string> write_cols0 = {"f0", "f2"};
auto src_array0 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[0], fields_[2]}), R"([
[1, "a"],
[2, "b"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs0, WriteArray(table_path, {}, write_cols0, {src_array0}));
ASSERT_OK(Commit(table_path, commit_msgs0));

std::vector<std::string> blob_write_cols = {"f1"};
auto src_array1 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
["c"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
WriteArray(table_path, {}, blob_write_cols, {src_array1}));
SetFirstRowId(0, commit_msgs1);
ASSERT_OK(Commit(table_path, commit_msgs1));

auto f3 = arrow::field("f3", arrow::int64());
ASSERT_OK(WriteNextSchema(table_path,
{DataField(0, fields_[0]), DataField(1, fields_[1]),
DataField(2, fields_[2]), DataField(3, f3)},
/*highest_field_id=*/3, options));

auto src_array2 = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_[1]}), R"([
["d"]
])")
.ValueOrDie());
ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
WriteArray(table_path, {}, blob_write_cols, {src_array2}));
SetFirstRowId(1, commit_msgs2);
ASSERT_OK(Commit(table_path, commit_msgs2));

auto expected_array = std::dynamic_pointer_cast<arrow::StructArray>(
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_), R"([
[1, "c", "a"],
[2, "d", "b"]
])")
.ValueOrDie());
ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields_)->field_names(), expected_array));
}

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