github-actions[bot] commented on code in PR #65915:
URL: https://github.com/apache/doris/pull/65915#discussion_r3780750768
##########
be/src/cloud/cloud_cumulative_compaction.cpp:
##########
@@ -491,34 +544,73 @@ Status CloudCumulativeCompaction::garbage_collection() {
return st;
}
-Status CloudCumulativeCompaction::pick_rowsets_to_compact() {
- _input_rowsets.clear();
+Status CloudCumulativeCompaction::advance_cumulative_point_before_pick(
+ int64_t min_conflict_version) {
+ if (!_enable_parallel_cumu_compaction) {
+ return Status::OK();
+ }
- std::vector<RowsetSharedPtr> candidate_rowsets;
+ auto compaction_policy =
+
_engine.cumu_compaction_policy(cloud_tablet()->tablet_meta()->compaction_policy());
+ int64_t input_cumulative_point;
+ int64_t output_cumulative_point;
+ std::vector<RowsetSharedPtr> candidates;
{
std::shared_lock rlock(_tablet->get_header_lock());
+ input_cumulative_point = cloud_tablet()->cumulative_layer_point();
+ output_cumulative_point = input_cumulative_point;
_base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
_cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
- int64_t candidate_version = std::max(
- std::max(cloud_tablet()->cumulative_layer_point(),
_max_conflict_version + 1),
- cloud_tablet()->alter_version() + 1);
- // Get all rowsets whose version >= `candidate_version` as candidate
rowsets
cloud_tablet()->traverse_rowsets_unlocked(
- [&candidate_rowsets, candidate_version](const RowsetSharedPtr&
rs) {
- if (rs->start_version() >= candidate_version) {
- candidate_rowsets.push_back(rs);
+ [&candidates, input_cumulative_point,
+ min_conflict_version](const RowsetSharedPtr& rs) {
+ if (rs->start_version() >= input_cumulative_point &&
+ rs->end_version() < min_conflict_version) {
+ candidates.push_back(rs);
}
});
}
- if (candidate_rowsets.empty()) {
- return Status::Error<CUMULATIVE_NO_SUITABLE_VERSION>(
- "no suitable versions: candidate rowsets empty");
+ std::sort(candidates.begin(), candidates.end(), Rowset::comparator);
+ const bool is_time_series_policy = compaction_policy->name() ==
CUMULATIVE_TIME_SERIES_POLICY;
+ Version no_delete_version {-1, -1};
+ for (const auto& rowset : candidates) {
+ if (rowset->start_version() != output_cumulative_point) {
+ break;
+ }
+
+ auto rowset_meta = rowset->rowset_meta();
+ if (rowset_meta->has_delete_predicate()) {
+ output_cumulative_point = rowset->end_version() + 1;
+ continue;
+ }
+ // A time-series singleton is a raw delta. Its post-compaction point
rule is not safe here.
+ if (rowset_meta->is_segments_overlapping() ||
+ (is_time_series_policy && rowset_meta->is_singleton_delta())) {
Review Comment:
[P1] Recover completed time-series singleton outputs
`is_singleton_delta()` only checks `start_version == end_version`, so this
also stops at a non-overlapping output produced by compacting one overlapping
singleton. With the default threshold 1, let higher `[5-5]` finish before lower
`[2-4]`: the higher job correctly leaves the point at 2, then the lower job
moves it to 5. The next pre-pick pass stops at the committed `[5-5]`, while the
ordinary picker rejects that sole non-overlapping rowset; without a later write
or an unrelated explicit FULL repair, ordinary cumulative scheduling cannot
move the point to 6. This is distinct from preserving a raw singleton. Please
retain reliable completed-output provenance (or an equivalent completion
frontier) so recovery can cross this output without treating raw `[v-v]` deltas
as compacted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
