Skip to content

Commit 100ca71

Browse files
wybmergify[bot]
wyb
authored and
mergify[bot]
committed
[BugFix] Fix BE crash in primary key auto-increment partial update apply (#76119)
Signed-off-by: wyb <wybb86@gmail.com> (cherry picked from commit ce03ba6) # Conflicts: # be/src/storage/tablet_updates.cpp # be/test/storage/rowset/rowset_test.cpp # be/test/storage/tablet_updates_test.cpp
1 parent 41b7e69 commit 100ca71

4 files changed

Lines changed: 306 additions & 8 deletions

File tree

‎be/src/storage/rowset/segment_rewriter.cpp‎

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -131,14 +131,25 @@ Status SegmentRewriter::rewrite_auto_increment(const std::string& src_path, cons
131131
seg_options.chunk_size = num_rows;
132132
seg_options.temporary_data = true;
133133

134-
auto res = rowset->segments()[segment_id]->new_iterator(src_schema, seg_options);
135-
auto& itr = res.value();
134+
ASSIGN_OR_RETURN(auto itr, rowset->segments()[segment_id]->new_iterator(src_schema, seg_options));
136135

137136
if (itr) {
138137
auto st = itr->get_next(read_chunk);
139-
DCHECK_EQ(read_chunk->num_rows(), num_rows);
138+
itr->close();
139+
// Do NOT swallow the read error: a transient read failure (page crc / decompress / io) here would
140+
// otherwise leave read_chunk short, and the downstream append_chunk would silently emit a segment whose
141+
// key columns have fewer rows than the value columns (segment num_rows=0 while value columns=N).
142+
TEST_SYNC_POINT_CALLBACK("SegmentRewriter::rewrite_auto_increment:get_next", &st);
143+
RETURN_IF_ERROR(st);
144+
TEST_SYNC_POINT_CALLBACK("SegmentRewriter::rewrite_auto_increment:read_chunk", read_chunk);
145+
if (UNLIKELY(read_chunk->num_rows() != num_rows)) {
146+
auto msg = "rewrite_auto_increment: read " + std::to_string(read_chunk->num_rows()) +
147+
" rows from partial segment " + src_path + " but expected " + std::to_string(num_rows) +
148+
" rows (partial-update/auto-increment segment read inconsistency)";
149+
LOG(ERROR) << msg;
150+
return Status::InternalError(msg);
151+
}
140152
}
141-
itr->close();
142153

143154
WritableFileOptions wopts{.sync_on_close = true, .mode = FileSystem::CREATE_OR_OPEN_WITH_TRUNCATE};
144155
ASSIGN_OR_RETURN(auto wfile, fs->new_writable_file(wopts, dest_path));

‎be/src/storage/tablet_updates.cpp‎

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,23 @@
2121
#include <filesystem>
2222
#include <memory>
2323

24+
<<<<<<< HEAD
25+
=======
26+
#include "base/debug/trace.h"
27+
#include "base/failpoint/fail_point.h"
28+
#include "base/random/random.h"
29+
#include "base/testutil/sync_point.h"
30+
#include "base/time/time.h"
31+
#include "base/utility/defer_op.h"
32+
#include "base/utility/pretty_printer.h"
33+
#include "base/utility/scoped_cleanup.h"
34+
#include "column/chunk_factory.h"
35+
#include "column/chunk_schema_helper.h"
36+
#include "common/config_compaction_fwd.h"
37+
#include "common/config_exec_fwd.h"
38+
#include "common/config_primary_key_fwd.h"
39+
#include "common/config_storage_fwd.h"
40+
>>>>>>> ce03ba620f ([BugFix] Fix BE crash in primary key auto-increment partial update apply (#76119))
2441
#include "common/status.h"
2542
#include "common/tracer.h"
2643
#include "exec/schema_scanner/schema_be_tablets_scanner.h"
@@ -5389,8 +5406,20 @@ Status TabletUpdates::get_column_values(const std::vector<uint32_t>& column_ids,
53895406
LOG(WARNING) << "Fail to open " << seg_path << ": " << segment.status();
53905407
return segment.status();
53915408
}
5392-
if ((*segment)->num_rows() == 0) {
5393-
continue;
5409+
uint32_t segment_num_rows = (*segment)->num_rows();
5410+
// Test hook: allow forcing a 0-row segment to exercise the integrity guard below.
5411+
TEST_SYNC_POINT_CALLBACK("TabletUpdates::get_column_values:segment_num_rows", &segment_num_rows);
5412+
if (segment_num_rows == 0) {
5413+
// A rssid appears here only with non-empty rowids resolved from the primary index, so a 0-row
5414+
// segment means the index/rowset-meta disagree with the on-disk segment data. Silently skipping
5415+
// would return short read columns and later crash in append_selective; fail loudly instead.
5416+
std::string msg = strings::Substitute(
5417+
"segment $0 (rssid $1) has 0 rows but $2 rowids are requested in tablet $3; primary "
5418+
"index/rowset-meta vs segment-data inconsistency (possibly a corrupt partial-update / "
5419+
"auto-increment segment rewrite output)",
5420+
seg_path, rssid, rowids.size(), _tablet.tablet_id());
5421+
LOG(ERROR) << msg;
5422+
return Status::Corruption(msg);
53945423
}
53955424
GetDeltaColumnContext ctx;
53965425
RETURN_IF_ERROR(ctx.prepareGetDeltaColumnContext((*segment), _tablet.data_dir()->get_meta(),
@@ -5452,8 +5481,18 @@ Status TabletUpdates::get_column_values(const std::vector<uint32_t>& column_ids,
54525481
LOG(WARNING) << "Fail to open " << seg_path << ": " << segment.status();
54535482
return segment.status();
54545483
}
5455-
if ((*segment)->num_rows() == 0) {
5456-
return Status::OK();
5484+
uint32_t segment_num_rows = (*segment)->num_rows();
5485+
// Test hook: allow forcing a 0-row segment to exercise the integrity guard below.
5486+
TEST_SYNC_POINT_CALLBACK("TabletUpdates::get_column_values:auto_increment_segment_num_rows", &segment_num_rows);
5487+
if (segment_num_rows == 0) {
5488+
// Same integrity guard as the main read loop: a 0-row segment with resolved rowids means the
5489+
// primary index / rowset meta disagree with the on-disk segment data.
5490+
std::string msg = strings::Substitute(
5491+
"auto-increment: segment $0 (rssid $1) has 0 rows but $2 rowids are requested in tablet $3 "
5492+
"(primary index/rowset-meta vs segment-data inconsistency)",
5493+
seg_path, rssid, rowids.size(), _tablet.tablet_id());
5494+
LOG(ERROR) << msg;
5495+
return Status::Corruption(msg);
54575496
}
54585497
GetDeltaColumnContext ctx;
54595498
RETURN_IF_ERROR(ctx.prepareGetDeltaColumnContext((*segment), _tablet.data_dir()->get_meta(),

‎be/test/storage/rowset/rowset_test.cpp‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,12 @@
3939
#include <string>
4040
#include <vector>
4141

42+
<<<<<<< HEAD
43+
=======
44+
#include "base/testutil/assert.h"
45+
#include "base/testutil/sync_point.h"
46+
#include "column/chunk_factory.h"
47+
>>>>>>> ce03ba620f ([BugFix] Fix BE crash in primary key auto-increment partial update apply (#76119))
4248
#include "column/datum_tuple.h"
4349
#include "fs/fs_util.h"
4450
#include "gen_cpp/data.pb.h"
@@ -1023,6 +1029,139 @@ TEST_F(RowsetTest, SegmentRewriterAutoIncrementTest) {
10231029
ASSERT_EQ(segment->num_rows(), num_rows);
10241030
}
10251031

1032+
// Regression test for the PK auto-increment partial-update 0-row-segment crash (write side / root cause).
1033+
// rewrite_auto_increment must NOT swallow a failed read of the partial segment: the read status used to be
1034+
// dropped (only a DCHECK guarded the row count, compiled out in release), leaving read_chunk short and letting
1035+
// append_chunk emit a corrupt segment (segment num_rows=0 while value columns=N). The read error must surface.
1036+
TEST_F(RowsetTest, SegmentRewriterAutoIncrementReadErrorTest) {
1037+
std::shared_ptr<TabletSchema> partial_tablet_schema = TabletSchemaHelper::create_tablet_schema(
1038+
{create_int_key_pb(1), create_int_key_pb(2), create_int_key_pb(3)});
1039+
1040+
RowsetWriterContext writer_context;
1041+
create_rowset_writer_context(12345, partial_tablet_schema, &writer_context);
1042+
writer_context.writer_type = kHorizontal;
1043+
std::unique_ptr<RowsetWriter> rowset_writer;
1044+
ASSERT_TRUE(RowsetFactory::create_rowset_writer(writer_context, &rowset_writer).ok());
1045+
1046+
int32_t chunk_size = 3000;
1047+
size_t num_rows = 3000;
1048+
{
1049+
std::vector<uint32_t> column_indexes{0, 1, 2};
1050+
auto schema = ChunkHelper::convert_schema(partial_tablet_schema, column_indexes);
1051+
auto chunk = ChunkFactory::new_chunk(schema, chunk_size);
1052+
for (auto i = 0; i < num_rows / chunk_size + 1; ++i) {
1053+
chunk->reset();
1054+
auto cols = chunk->columns();
1055+
for (auto j = 0; j < chunk_size && i * chunk_size + j < num_rows; ++j) {
1056+
cols[0]->as_mutable_ptr()->append_datum(Datum(static_cast<int32_t>(i * chunk_size + j)));
1057+
cols[1]->as_mutable_ptr()->append_datum(Datum(static_cast<int32_t>(i * chunk_size + j + 1)));
1058+
cols[2]->as_mutable_ptr()->append_datum(Datum(static_cast<int32_t>(i * chunk_size + j + 2)));
1059+
}
1060+
std::unique_ptr<SegmentPB> seg_info = std::make_unique<SegmentPB>();
1061+
ASSERT_OK(rowset_writer->flush_chunk(*chunk, seg_info.get()));
1062+
}
1063+
}
1064+
RowsetSharedPtr rowset = rowset_writer->build().value();
1065+
rowset->load();
1066+
std::string file_name = Rowset::segment_file_path(rowset->rowset_path(), rowset->rowset_id(), 0);
1067+
1068+
std::shared_ptr<TabletSchema> tablet_schema = TabletSchemaHelper::create_tablet_schema(
1069+
{create_int_key_pb(1), create_int_key_pb(2), create_int_value_pb(3), create_int_value_pb(4)});
1070+
std::vector<uint32_t> read_column_ids{2, 3};
1071+
MutableColumns write_columns(read_column_ids.size());
1072+
for (auto i = 0; i < read_column_ids.size(); ++i) {
1073+
auto tablet_column = tablet_schema->column(read_column_ids[i]);
1074+
auto column = ChunkFactory::column_from_field_type(tablet_column.type(), tablet_column.is_nullable());
1075+
write_columns[i] = column->clone_empty();
1076+
for (auto j = 0; j < num_rows; ++j) {
1077+
write_columns[i]->append_datum(Datum(static_cast<int32_t>(j + read_column_ids[i])));
1078+
}
1079+
}
1080+
AutoIncrementPartialUpdateState auto_increment_partial_update_state;
1081+
auto_increment_partial_update_state.init(rowset.get(), partial_tablet_schema, 2, 0);
1082+
auto_increment_partial_update_state.write_column.reset(std::move(write_columns[0]));
1083+
write_columns.erase(write_columns.begin());
1084+
auto dst_file_name = Rowset::segment_temp_file_path(rowset->rowset_path(), rowset->rowset_id(), 0);
1085+
std::vector<uint32_t> column_ids{3};
1086+
1087+
// Inject a failed read at the partial-segment get_next; rewrite_auto_increment must return the error.
1088+
SyncPoint::GetInstance()->EnableProcessing();
1089+
SyncPoint::GetInstance()->SetCallBack("SegmentRewriter::rewrite_auto_increment:get_next",
1090+
[](void* arg) { *(Status*)arg = Status::Corruption("injected read error"); });
1091+
auto st = SegmentRewriter::rewrite_auto_increment(file_name, dst_file_name, tablet_schema,
1092+
auto_increment_partial_update_state, column_ids, &write_columns);
1093+
SyncPoint::GetInstance()->ClearCallBack("SegmentRewriter::rewrite_auto_increment:get_next");
1094+
SyncPoint::GetInstance()->DisableProcessing();
1095+
ASSERT_FALSE(st.ok()) << "rewrite_auto_increment must not swallow a partial-segment read error";
1096+
ASSERT_TRUE(st.is_corruption()) << st.to_string();
1097+
}
1098+
1099+
// Covers the row-count guard in rewrite_auto_increment: if get_next succeeds but read_chunk comes back with
1100+
// fewer rows than the segment (a short read), the rebuilt row would be malformed, so rewrite must reject it.
1101+
TEST_F(RowsetTest, SegmentRewriterAutoIncrementRowCountMismatchTest) {
1102+
std::shared_ptr<TabletSchema> partial_tablet_schema = TabletSchemaHelper::create_tablet_schema(
1103+
{create_int_key_pb(1), create_int_key_pb(2), create_int_key_pb(3)});
1104+
1105+
RowsetWriterContext writer_context;
1106+
create_rowset_writer_context(12345, partial_tablet_schema, &writer_context);
1107+
writer_context.writer_type = kHorizontal;
1108+
std::unique_ptr<RowsetWriter> rowset_writer;
1109+
ASSERT_TRUE(RowsetFactory::create_rowset_writer(writer_context, &rowset_writer).ok());
1110+
1111+
int32_t chunk_size = 3000;
1112+
size_t num_rows = 3000;
1113+
{
1114+
std::vector<uint32_t> column_indexes{0, 1, 2};
1115+
auto schema = ChunkHelper::convert_schema(partial_tablet_schema, column_indexes);
1116+
auto chunk = ChunkFactory::new_chunk(schema, chunk_size);
1117+
for (auto i = 0; i < num_rows / chunk_size + 1; ++i) {
1118+
chunk->reset();
1119+
auto cols = chunk->columns();
1120+
for (auto j = 0; j < chunk_size && i * chunk_size + j < num_rows; ++j) {
1121+
cols[0]->as_mutable_ptr()->append_datum(Datum(static_cast<int32_t>(i * chunk_size + j)));
1122+
cols[1]->as_mutable_ptr()->append_datum(Datum(static_cast<int32_t>(i * chunk_size + j + 1)));
1123+
cols[2]->as_mutable_ptr()->append_datum(Datum(static_cast<int32_t>(i * chunk_size + j + 2)));
1124+
}
1125+
std::unique_ptr<SegmentPB> seg_info = std::make_unique<SegmentPB>();
1126+
ASSERT_OK(rowset_writer->flush_chunk(*chunk, seg_info.get()));
1127+
}
1128+
}
1129+
RowsetSharedPtr rowset = rowset_writer->build().value();
1130+
rowset->load();
1131+
std::string file_name = Rowset::segment_file_path(rowset->rowset_path(), rowset->rowset_id(), 0);
1132+
1133+
std::shared_ptr<TabletSchema> tablet_schema = TabletSchemaHelper::create_tablet_schema(
1134+
{create_int_key_pb(1), create_int_key_pb(2), create_int_value_pb(3), create_int_value_pb(4)});
1135+
std::vector<uint32_t> read_column_ids{2, 3};
1136+
MutableColumns write_columns(read_column_ids.size());
1137+
for (auto i = 0; i < read_column_ids.size(); ++i) {
1138+
auto tablet_column = tablet_schema->column(read_column_ids[i]);
1139+
auto column = ChunkFactory::column_from_field_type(tablet_column.type(), tablet_column.is_nullable());
1140+
write_columns[i] = column->clone_empty();
1141+
for (auto j = 0; j < num_rows; ++j) {
1142+
write_columns[i]->append_datum(Datum(static_cast<int32_t>(j + read_column_ids[i])));
1143+
}
1144+
}
1145+
AutoIncrementPartialUpdateState auto_increment_partial_update_state;
1146+
auto_increment_partial_update_state.init(rowset.get(), partial_tablet_schema, 2, 0);
1147+
auto_increment_partial_update_state.write_column.reset(std::move(write_columns[0]));
1148+
write_columns.erase(write_columns.begin());
1149+
auto dst_file_name = Rowset::segment_temp_file_path(rowset->rowset_path(), rowset->rowset_id(), 0);
1150+
std::vector<uint32_t> column_ids{3};
1151+
1152+
// get_next succeeds, but the read comes back short (empty read_chunk); rewrite must reject the row-count mismatch.
1153+
SyncPoint::GetInstance()->EnableProcessing();
1154+
SyncPoint::GetInstance()->SetCallBack("SegmentRewriter::rewrite_auto_increment:read_chunk",
1155+
[](void* arg) { ((Chunk*)arg)->reset(); });
1156+
auto st = SegmentRewriter::rewrite_auto_increment(file_name, dst_file_name, tablet_schema,
1157+
auto_increment_partial_update_state, column_ids, &write_columns);
1158+
SyncPoint::GetInstance()->ClearCallBack("SegmentRewriter::rewrite_auto_increment:read_chunk");
1159+
SyncPoint::GetInstance()->DisableProcessing();
1160+
ASSERT_FALSE(st.ok()) << "rewrite_auto_increment must reject a short partial-segment read";
1161+
ASSERT_TRUE(st.is_internal_error()) << st.to_string();
1162+
EXPECT_NE(st.to_string().find("segment read inconsistency"), std::string::npos) << st.to_string();
1163+
}
1164+
10261165
TEST_F(RowsetTest, SegmentDeleteWriteTest) {
10271166
auto tablet = create_tablet(12345, 1111);
10281167
int64_t num_rows = 1024;

0 commit comments

Comments
 (0)