github-actions[bot] commented on code in PR #65915:
URL: https://github.com/apache/doris/pull/65915#discussion_r3766062115
##########
be/src/cloud/cloud_cumulative_compaction.cpp:
##########
@@ -491,34 +544,70 @@ 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);
+ 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;
+ }
+ if (rowset_meta->is_segments_overlapping()) {
+ break;
+ }
+
+ int64_t new_cumulative_point = compaction_policy->new_cumulative_point(
Review Comment:
[P1] Do not advance over raw time-series deltas
This loop passes untouched rowsets to `new_cumulative_point()`, whose
time-series implementation is a post-compaction rule. With the default level
threshold 1 it returns `end + 1` for every nonempty RUNNING-tablet rowset, so
raw singleton deltas `[2-2]`, `[3-3]` can produce an EMPTY_CUMULATIVE update
from 2 to 4 without ever being merged. Time-series point initialization
deliberately stops at the first singleton, and once these rowsets fall below
the new point the cumulative score excludes them while the time-series base
score is zero without a delete. Please use a pre-pick-safe predicate that
preserves raw singleton/level-0 inputs.
##########
cloud/src/meta-service/meta_service_job.cpp:
##########
@@ -1068,7 +1103,39 @@ void process_compaction_job(MetaServiceCode& code,
std::string& msg, std::string
}
}
- if (compaction_update_tablet_stats(compaction, stats, code, msg, now) ==
-1) {
+ // Older BEs only put the proposal snapshot in START. Use the recorded
snapshot when FINISH
+ // omits both counters so a stale proposal is ignored without rejecting
the rowset commit.
+ const bool has_finish_proposal_snapshot =
compaction.has_base_compaction_cnt();
+ const int64_t proposal_base_compaction_cnt =
+ has_finish_proposal_snapshot ? compaction.base_compaction_cnt()
+ :
recorded_compaction->base_compaction_cnt();
Review Comment:
[P1] Do not accept a legacy higher-first proposal from its START snapshot
This fallback cannot distinguish two legacy parallel jobs that both started
with the same counters. If `[2-4]` and `[5-7]` record `(0,0)` and the old BE
finishes `[5-7]` first without FINISH counters, its recorded snapshot still
matches current stats, so its old point proposal (for example 8) is accepted
and skips unresolved `[2-4]`. The new legacy test only exercises lower-first
order. Please make legacy FINISH proposals safe for higher-first completion as
well, or decline those unverifiable point advances.
##########
be/src/cloud/cloud_cumulative_compaction.cpp:
##########
@@ -244,12 +250,34 @@ Status CloudCumulativeCompaction::execute_compact() {
Status CloudCumulativeCompaction::modify_rowsets() {
// calculate new cumulative point
- int64_t input_cumulative_point = cloud_tablet()->cumulative_layer_point();
+ int64_t input_cumulative_point;
+ int64_t proposal_base_compaction_cnt;
+ int64_t proposal_cumulative_compaction_cnt;
+ TabletState input_tablet_state;
+ int64_t input_alter_version;
+ {
+ std::shared_lock rlock(_tablet->get_header_lock());
+ input_cumulative_point = cloud_tablet()->cumulative_layer_point();
+ proposal_base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
+ proposal_cumulative_compaction_cnt =
cloud_tablet()->cumulative_compaction_cnt();
+ input_tablet_state = _tablet->tablet_state();
+ input_alter_version = cloud_tablet()->alter_version();
+ }
auto compaction_policy =
cloud_tablet()->tablet_meta()->compaction_policy();
- int64_t new_cumulative_point =
- _engine.cumu_compaction_policy(compaction_policy)
- ->new_cumulative_point(cloud_tablet(), _output_rowset,
_last_delete_version,
- input_cumulative_point);
+ int64_t new_cumulative_point = input_cumulative_point;
+ if (!_enable_parallel_cumu_compaction &&
Review Comment:
[P1] Allow normal serial suffix compactions
`output.start > input_cumulative_point` is not specific to a NOTREADY
schema-change tablet. The size-based policy deliberately drops a large leading
rowset; for example, the existing
`pick_input_rowsets_large_head_not_repeated_when_output_below_promotion` test
leaves the point at 2 while selecting `[3-22]` on a normal RUNNING tablet. In
the default serial mode that valid compaction now reaches this branch and
aborts on `DORIS_CHECK_EQ(input_tablet_state, TABLET_NOTREADY)`. Please
distinguish the schema-change case explicitly and preserve the existing RUNNING
suffix behavior.
##########
cloud/src/meta-service/meta_service_job.cpp:
##########
@@ -832,6 +847,16 @@ int compaction_update_tablet_stats(const
TabletCompactionJobPB& compaction, Tabl
} else if (compaction.type() == TabletCompactionJobPB::BASE) {
// clang-format off
stats->set_base_compaction_cnt(stats->base_compaction_cnt() + 1);
+ if (compaction.input_versions_size() == 2 &&
+ stats->cumulative_point() > compaction.input_versions(0) &&
+ stats->cumulative_point() <= compaction.input_versions(1)) {
+ LOG_WARNING("cumulative point falls inside base compaction input
range")
+ .tag("job_id", compaction.id())
+ .tag("cumulative_point", stats->cumulative_point())
+ .tag("input_start_version", compaction.input_versions(0))
+ .tag("input_end_version", compaction.input_versions(1));
+ stats->set_cumulative_point(compaction.input_versions(1) + 1);
Review Comment:
[P1] Publish the repaired BASE point to the BE cache
When this moves a point inside BASE input `[2-7]` from 6 to 8, both
`CloudBaseCompaction::modify_rowsets` and the index-change BASE handler still
install output `[2-7]` and the new base counter without applying
`stats.cumulative_point()` or setting `last_sync_time_s = 0`. They also reset
approximate rowset stats, so the next prepare can skip synchronization and
operate with local point 6 inside the output while meta-service is at 8. Please
force a sync (or safely apply the returned point) whenever this repair advances
it.
--
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]
