Copilot commented on code in PR #65915:
URL: https://github.com/apache/doris/pull/65915#discussion_r3629882438
##########
be/src/cloud/cloud_cumulative_compaction.cpp:
##########
@@ -47,6 +50,18 @@
CloudCumulativeCompaction::CloudCumulativeCompaction(CloudStorageEngine& engine,
CloudCumulativeCompaction::~CloudCumulativeCompaction() = default;
+int64_t CloudCumulativeCompaction::_refresh_conflict_versions() {
+ std::vector<std::shared_ptr<CloudCumulativeCompaction>> cumu_compactions;
+ _engine.get_cumu_compaction(_tablet->tablet_id(), cumu_compactions);
+ for (const auto& cumu : cumu_compactions) {
+ _min_conflict_version =
+ std::min(_min_conflict_version,
cumu->_input_rowsets.front()->start_version());
+ _max_conflict_version =
+ std::max(_max_conflict_version,
cumu->_input_rowsets.back()->end_version());
+ }
+ return _max_conflict_version;
+}
Review Comment:
`_refresh_conflict_versions()` accumulates into `_min_conflict_version` /
`_max_conflict_version` without resetting them, so the conflict window can
become permanently stale after in-flight compactions complete. This can cause
`pick_rowsets_to_compact()` to keep avoiding version ranges that are no longer
conflicting, potentially stalling cumulative compaction. Reset to defaults at
the start of this method (e.g., min=INT64_MAX, max=0) and recompute from
current in-flight compactions; if none exist, keep defaults.
##########
be/src/cloud/cloud_cumulative_compaction.cpp:
##########
@@ -47,6 +50,18 @@
CloudCumulativeCompaction::CloudCumulativeCompaction(CloudStorageEngine& engine,
CloudCumulativeCompaction::~CloudCumulativeCompaction() = default;
+int64_t CloudCumulativeCompaction::_refresh_conflict_versions() {
+ std::vector<std::shared_ptr<CloudCumulativeCompaction>> cumu_compactions;
+ _engine.get_cumu_compaction(_tablet->tablet_id(), cumu_compactions);
+ for (const auto& cumu : cumu_compactions) {
+ _min_conflict_version =
+ std::min(_min_conflict_version,
cumu->_input_rowsets.front()->start_version());
+ _max_conflict_version =
+ std::max(_max_conflict_version,
cumu->_input_rowsets.back()->end_version());
+ }
Review Comment:
This reads `cumu->_input_rowsets.front()/back()` without guarding against an
empty `_input_rowsets` (or concurrent mutation). If an in-flight compaction is
registered before its inputs are populated, this will crash. Consider storing
immutable `input_start_version/input_end_version` on the compaction object when
inputs are picked, and have `_refresh_conflict_versions()` use those; or at
minimum, skip/guard empty `_input_rowsets` with a check to avoid dereferencing.
##########
be/src/cloud/cloud_cumulative_compaction_policy.cpp:
##########
@@ -35,6 +35,33 @@
namespace doris {
+int64_t CloudCumulativeCompactionPolicy::calculate_cumulative_point(
+ CloudTablet* tablet, const std::vector<RowsetSharedPtr>& rowsets,
+ const RowsetSharedPtr& output_rowset, Version& last_delete_version,
+ int64_t input_cumulative_point) {
+ DORIS_CHECK(std::find(rowsets.begin(), rowsets.end(), output_rowset) !=
rowsets.end());
Review Comment:
This membership check relies on `shared_ptr` identity (`operator==`) instead
of semantic identity (rowset id / version range). If a caller passes an
`output_rowset` instance that represents the same rowset but is not the same
`shared_ptr` object as the one stored in `rowsets`, this will abort even though
the input is logically valid. Prefer matching by a stable identifier (e.g.,
rowset id) or by version range, and avoid fatal checks here (return
`input_cumulative_point` or DCHECK) since this is used on a production path.
##########
be/src/cloud/cloud_cumulative_compaction.cpp:
##########
@@ -244,12 +252,56 @@ Status CloudCumulativeCompaction::execute_compact() {
Status CloudCumulativeCompaction::modify_rowsets() {
// calculate new cumulative point
- int64_t input_cumulative_point = cloud_tablet()->cumulative_layer_point();
- 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 prepare_max_conflict_version = _max_conflict_version;
+ int64_t max_conflict_version = _refresh_conflict_versions();
+ int64_t input_cumulative_point;
+ bool output_already_passed;
+ std::vector<RowsetSharedPtr> rowsets;
+ {
+ std::shared_lock rlock(_tablet->get_header_lock());
+ input_cumulative_point = cloud_tablet()->cumulative_layer_point();
+ DORIS_CHECK(input_cumulative_point <= _output_rowset->start_version()
||
+ input_cumulative_point > _output_rowset->end_version());
+ output_already_passed = input_cumulative_point >
_output_rowset->end_version();
Review Comment:
This `DORIS_CHECK` can abort if the cumulative point advances into the
output rowset’s version range between compaction prepare and modify (a
plausible race under parallel compaction). Instead of fatal-checking, treat
this as an expected concurrency outcome: either set `output_already_passed` (or
return without advancing), or return a non-OK `Status` that causes the
compaction to retry/resync.
##########
be/src/cloud/cloud_cumulative_compaction_policy.cpp:
##########
@@ -35,6 +35,33 @@
namespace doris {
+int64_t CloudCumulativeCompactionPolicy::calculate_cumulative_point(
+ CloudTablet* tablet, const std::vector<RowsetSharedPtr>& rowsets,
+ const RowsetSharedPtr& output_rowset, Version& last_delete_version,
+ int64_t input_cumulative_point) {
+ DORIS_CHECK(std::find(rowsets.begin(), rowsets.end(), output_rowset) !=
rowsets.end());
+ int64_t cumulative_point = input_cumulative_point;
+ Version no_delete_version {-1, -1};
+ for (const auto& rowset : rowsets) {
+ DORIS_CHECK_EQ(rowset->start_version(), cumulative_point);
+ if (rowset->rowset_meta()->has_delete_predicate()) {
+ cumulative_point = rowset->end_version() + 1;
+ continue;
+ }
Review Comment:
`DORIS_CHECK_EQ(rowset->start_version(), cumulative_point)` will hard-abort
the process if there is a version gap or if the provided rowset list isn't
perfectly continuous (which can happen due to concurrent tablet changes between
traversal and evaluation, or due to legitimate version holes). Since the goal
is to conservatively advance the cumulative point, consider handling
non-continuity by returning the current `cumulative_point` (or downgrading to
`DCHECK`/logging) instead of crashing.
--
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]
