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
11 changes: 7 additions & 4 deletions be/src/olap/compaction.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -297,11 +297,14 @@ int64_t Compaction::get_avg_segment_rows() {
const auto& meta = _tablet->tablet_meta();
if (meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {
int64_t compaction_goal_size_mbytes = meta->time_series_compaction_goal_size_mbytes();
return (compaction_goal_size_mbytes * 1024 * 1024 * 2) /
(_input_rowsets_data_size / (_input_row_num + 1) + 1);
// The output segment rows should be less than total input rows
return std::min((compaction_goal_size_mbytes * 1024 * 1024 * 2) /
(_input_rowsets_data_size / (_input_row_num + 1) + 1),
_input_row_num + 1);
}
return config::vertical_compaction_max_segment_size /
(_input_rowsets_data_size / (_input_row_num + 1) + 1);
return std::min(config::vertical_compaction_max_segment_size /
(_input_rowsets_data_size / (_input_row_num + 1) + 1),
_input_row_num + 1);
}

CompactionMixin::CompactionMixin(StorageEngine& engine, TabletSharedPtr tablet,
Expand Down
34 changes: 31 additions & 3 deletions be/test/olap/base_compaction_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -39,11 +39,12 @@ namespace doris {
class TestBaseCompaction : public testing::Test {};

static RowsetSharedPtr create_rowset(Version version, int num_segments, bool overlapping,
int data_size) {
int data_size, int rows_num) {
auto rs_meta = std::make_shared<RowsetMeta>();
rs_meta->set_rowset_type(BETA_ROWSET); // important
rs_meta->_rowset_meta_pb.set_start_version(version.first);
rs_meta->_rowset_meta_pb.set_end_version(version.second);
rs_meta->_rowset_meta_pb.set_num_rows(rows_num);
rs_meta->set_num_segments(num_segments);
rs_meta->set_segments_overlap(overlapping ? OVERLAPPING : NONOVERLAPPING);
rs_meta->set_total_disk_size(data_size);
Expand All @@ -66,10 +67,10 @@ TEST_F(TestBaseCompaction, filter_input_rowset) {
BaseCompaction compaction(engine, tablet);
//std::vector<RowsetSharedPtr> rowsets;

RowsetSharedPtr init_rs = create_rowset({0, 1}, 1, false, 0);
RowsetSharedPtr init_rs = create_rowset({0, 1}, 1, false, 0, 100);
tablet->_rs_version_map.emplace(init_rs->version(), init_rs);
for (int i = 2; i < 30; ++i) {
RowsetSharedPtr rs = create_rowset({i, i}, 1, false, 1024);
RowsetSharedPtr rs = create_rowset({i, i}, 1, false, 1024, 100);
tablet->_rs_version_map.emplace(rs->version(), rs);
}
Status st = compaction.pick_rowsets_to_compact();
Expand All @@ -81,4 +82,31 @@ TEST_F(TestBaseCompaction, filter_input_rowset) {
EXPECT_EQ(compaction._input_rowsets.back()->end_version(), 21);
}

TEST_F(TestBaseCompaction, zero_input_rows) {
StorageEngine engine({});
TabletMetaSharedPtr tablet_meta;
tablet_meta.reset(new TabletMeta(2, 3, 15674, 15675, 6, 7, TTabletSchema(), 6, {{7, 8}},
UniqueId(9, 10), TTabletType::TABLET_TYPE_DISK,
TCompressionType::LZ4F));
TabletSharedPtr tablet(new Tablet(engine, tablet_meta, nullptr, CUMULATIVE_SIZE_BASED_POLICY));
tablet->_cumulative_point = 25;
BaseCompaction compaction(engine, tablet);
//std::vector<RowsetSharedPtr> rowsets;

RowsetSharedPtr init_rs = create_rowset({0, 1}, 1, false, 0, 0);
tablet->_rs_version_map.emplace(init_rs->version(), init_rs);
for (int i = 2; i < 30; ++i) {
RowsetSharedPtr rs = create_rowset({i, i}, 1, false, 1024, 0);
tablet->_rs_version_map.emplace(rs->version(), rs);
}
Status st = compaction.pick_rowsets_to_compact();
EXPECT_TRUE(st.ok());
EXPECT_EQ(compaction._input_rowsets.front()->start_version(), 0);
EXPECT_EQ(compaction._input_rowsets.front()->end_version(), 1);

EXPECT_EQ(compaction._input_rowsets.back()->start_version(), 21);
EXPECT_EQ(compaction._input_rowsets.back()->end_version(), 21);
EXPECT_EQ(compaction.get_avg_segment_rows(), 1);
}

} // namespace doris
Loading