diff --git a/cloud/src/meta-service/meta_service_job.cpp b/cloud/src/meta-service/meta_service_job.cpp index dad84f6a36ab0d..35aedd81852c28 100644 --- a/cloud/src/meta-service/meta_service_job.cpp +++ b/cloud/src/meta-service/meta_service_job.cpp @@ -97,6 +97,55 @@ bool check_compaction_input_verions(const TabletCompactionJobPB& compaction, return false; } +// Normalize compaction type for conflict detection. +// EMPTY_CUMULATIVE only updates cumulative_point and cumulative_compaction_cnt without producing +// any output rowset. It MUST be considered the same family as CUMULATIVE so that an EMPTY_CUMULATIVE +// cannot be accepted while a real CUMULATIVE is still running on the same tablet (which would +// otherwise advance cumulative_point past the input range of the in-flight cumu and let base +// compaction race with it). +static inline TabletCompactionJobPB::CompactionType normalize_compaction_type_for_conflict( + TabletCompactionJobPB::CompactionType t) { + return t == TabletCompactionJobPB::EMPTY_CUMULATIVE ? TabletCompactionJobPB::CUMULATIVE : t; +} + +// Two compaction jobs are considered to be in the same conflict family iff their normalized +// compaction types are equal, OR either side is FULL (full compaction conflicts with anything), +// OR they are different in (BASE, CUMULATIVE) - in this case the conflict still depends on whether +// their input version ranges overlap, callers should additionally consult version range checks. +static inline bool is_same_conflict_family(TabletCompactionJobPB::CompactionType a, + TabletCompactionJobPB::CompactionType b) { + return normalize_compaction_type_for_conflict(a) == normalize_compaction_type_for_conflict(b); +} + +// Whether a compaction type belongs to the "rowset compaction family", i.e. any compaction +// kind that operates on rowsets bounded by `cumulative_point`. Only members of this family +// are subject to the cross-type version-range conflict check below. Note that STOP_TOKEN is +// already filtered out by the caller (see the early-return on STOP_TOKEN in start_compaction_job), +// so it is intentionally excluded here. +static inline bool is_rowset_compaction_family(TabletCompactionJobPB::CompactionType t) { + return t == TabletCompactionJobPB::BASE || t == TabletCompactionJobPB::CUMULATIVE || + t == TabletCompactionJobPB::EMPTY_CUMULATIVE || t == TabletCompactionJobPB::FULL; +} + +// Whether two compaction jobs MAY conflict on the rowset range (regardless of the actual +// input version range). The caller still needs to compare `input_versions` to make the +// final decision. +// +// Conflict matrix (Plan D extends this from same-type-only to cross-type within the family): +// FULL vs anything-in-family : true (full compaction touches the whole rowset range) +// BASE vs BASE / CUMULATIVE : true (their rowset ranges may overlap around cumu_point) +// CUMU vs CUMU / BASE : true (symmetric of the above) +// EMPTY_CU vs anything-in-family : true (it advances cumu_point and would race with the others) +// +// Any compaction type outside this family (or that we don't yet model) is conservatively NOT +// considered conflicting here - if a new type is added later, the author MUST revisit this +// function and decide its conflict semantics explicitly, instead of silently inheriting the +// "everything conflicts" behaviour. +static inline bool may_conflict_by_type(TabletCompactionJobPB::CompactionType a, + TabletCompactionJobPB::CompactionType b) { + return is_rowset_compaction_family(a) && is_rowset_compaction_family(b); +} + void start_compaction_job(MetaServiceCode& code, std::string& msg, std::stringstream& ss, std::unique_ptr& txn, const StartTabletJobRequest* request, StartTabletJobResponse* response, std::string& instance_id, @@ -250,10 +299,16 @@ void start_compaction_job(MetaServiceCode& code, std::string& msg, std::stringst compaction.input_versions().empty()) || (compaction.has_check_input_versions_range() && !compaction.check_input_versions_range())) { - // Unknown input version range, doesn't support parallel compaction of same type + // Unknown input version range, doesn't support parallel compaction of same family. + // EMPTY_CUMULATIVE is normalized to CUMULATIVE here so it conflicts with an in-flight + // CUMULATIVE on the same tablet (otherwise EMPTY_CUMULATIVE could advance + // cumulative_point past the in-flight cumu's input range and let base compaction race + // with it). for (auto& c : compactions) { - if (c.type() != compaction.type() && c.type() != TabletCompactionJobPB::FULL) + if (!is_same_conflict_family(c.type(), compaction.type()) && + c.type() != TabletCompactionJobPB::FULL) { continue; + } if (c.id() == compaction.id()) return; // Same job, return OK to keep idempotency msg = fmt::format("compaction has already started, tablet_id={} job={}", tablet_id, proto_to_json(c)); @@ -270,8 +325,14 @@ void start_compaction_job(MetaServiceCode& code, std::string& msg, std::stringst a.input_versions(1) < b.input_versions(0); }; for (auto& c : compactions) { - if (c.type() != compaction.type() && c.type() != TabletCompactionJobPB::FULL) - continue; + // Plan D: BASE and CUMULATIVE on the same tablet may also conflict when their + // input version ranges overlap. Previously we only checked same-type conflicts, + // which left a window where BASE could be accepted with versions that overlap an + // in-flight CUMULATIVE (and vice versa) after cumulative_point was unsafely + // advanced. Now we treat any pair within (BASE, CUMULATIVE, EMPTY_CUMULATIVE, + // FULL) as potentially conflicting and rely on the input-version-range check + // below to make the final decision. + if (!may_conflict_by_type(c.type(), compaction.type())) continue; if (c.input_versions_size() > 0 && version_not_conflict(c, compaction)) continue; if (c.id() == compaction.id()) return; // Same job, return OK to keep idempotency msg = fmt::format("compaction has already started, tablet_id={} job={}", tablet_id, @@ -279,14 +340,21 @@ void start_compaction_job(MetaServiceCode& code, std::string& msg, std::stringst code = MetaServiceCode::JOB_TABLET_BUSY; // Unknown version range of started compaction, BE should not retry other version range if (c.input_versions_size() == 0) return; - // Notify version ranges in started compaction to BE, so BE can retry other version range + // Notify version ranges of all in-flight compactions that may conflict with the + // incoming one, so BE can retry on a non-overlapping range. The notification + // predicate is intentionally kept consistent with the conflict predicate above + // (`may_conflict_by_type`); previously only same-family ranges were surfaced, + // which left BE blind to cross-family (BASE vs CUMULATIVE) conflicts. + // + // An in-flight EMPTY_CUMULATIVE (or any other family member without a concrete + // [v_lo, v_hi]) carries no usable range; surfacing fabricated zeros would + // mislead BE retry. Skip such entries defensively here - the real conflict is + // already enforced by the version-range check above. for (auto& c : compactions) { - if (c.type() == compaction.type() || c.type() == TabletCompactionJobPB::FULL) { - // If there are multiple started compaction of same type, they all must has input version range - DCHECK_EQ(c.input_versions_size(), 2) << proto_to_json(c); - response->add_version_in_compaction(c.input_versions(0)); - response->add_version_in_compaction(c.input_versions(1)); - } + if (!may_conflict_by_type(c.type(), compaction.type())) continue; + if (c.input_versions_size() != 2) continue; + response->add_version_in_compaction(c.input_versions(0)); + response->add_version_in_compaction(c.input_versions(1)); } return; } diff --git a/cloud/test/meta_service_job_test.cpp b/cloud/test/meta_service_job_test.cpp index 1926f6c600a54c..089418745b3c89 100644 --- a/cloud/test/meta_service_job_test.cpp +++ b/cloud/test/meta_service_job_test.cpp @@ -4610,6 +4610,221 @@ TEST(MetaServiceJobTest, ParallelCumuCompactionTest) { ASSERT_EQ(res.status().code(), MetaServiceCode::OK); } +// Plan A regression test: EMPTY_CUMULATIVE must be considered the same conflict family as +// CUMULATIVE so that an EMPTY_CUMULATIVE submitted while a real CUMULATIVE is still active on the +// same tablet is rejected with JOB_TABLET_BUSY. Otherwise EMPTY_CUMULATIVE could advance +// cumulative_point past the in-flight cumu's input range and let base compaction race with it. +TEST(MetaServiceJobTest, EmptyCumulativeBlockedByCumulativeTest) { + auto meta_service = get_meta_service(); + + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + SyncPoint::get_instance()->clear_all_call_backs(); + }; + sp->set_call_back("get_instance_id", [&](auto&& args) { + auto* ret = try_any_cast_ret(args); + ret->first = instance_id; + ret->second = true; + }); + sp->enable_processing(); + + constexpr int64_t table_id = 1; + constexpr int64_t index_id = 2; + constexpr int64_t partition_id = 3; + constexpr int64_t tablet_id = 4; + ASSERT_NO_FATAL_FAILURE( + create_tablet(meta_service.get(), table_id, index_id, partition_id, tablet_id, false)); + + // Helper to start an EMPTY_CUMULATIVE job. EMPTY_CUMULATIVE has no input_versions and no + // expiration (only cumulative_point/cumulative_compaction_cnt are bumped), which lets it + // bypass `STALE_TABLET_CACHE` when both sides carry the same cumulative_compaction_cnt. + auto start_empty_cumu = [&](const std::string& job_id, const std::string& initiator, + int base_cnt, int cumu_cnt, StartTabletJobResponse& res) { + brpc::Controller cntl; + StartTabletJobRequest req; + req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id); + auto* compaction = req.mutable_job()->add_compaction(); + compaction->set_id(job_id); + compaction->set_initiator(initiator); + compaction->set_base_compaction_cnt(base_cnt); + compaction->set_cumulative_compaction_cnt(cumu_cnt); + compaction->set_type(TabletCompactionJobPB::EMPTY_CUMULATIVE); + long now = ::time(nullptr); + compaction->set_lease(now + 3); + meta_service->start_tablet_job(&cntl, &req, &res, nullptr); + }; + + // Step 1: An in-flight CUMULATIVE job [42326-42474] is registered first (mimics the + // scenario from the production log). + StartTabletJobResponse res; + start_compaction_job(meta_service.get(), tablet_id, "cumu1", "BE1", 0, 0, + TabletCompactionJobPB::CUMULATIVE, res, {42326, 42474}); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK); + + // Step 2: An EMPTY_CUMULATIVE arrives carrying the same cumulative_compaction_cnt as + // cumu1. Before the fix this was wrongly accepted because MS only compared raw enum types. + // After the fix, EMPTY_CUMULATIVE must be normalized to CUMULATIVE for conflict detection + // and rejected as JOB_TABLET_BUSY. + res.Clear(); + start_empty_cumu("empty1", "BE1", 0, 0, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); + // EMPTY_CUMULATIVE has no input_versions, so BE must NOT receive any version range hint + // (the BE retry on `version_in_compaction` is meaningless for EMPTY_CUMULATIVE). + EXPECT_EQ(res.version_in_compaction_size(), 0); + + // Step 3: Idempotency check - the same job_id submitted twice should still return OK. + res.Clear(); + start_compaction_job(meta_service.get(), tablet_id, "cumu1", "BE1", 0, 0, + TabletCompactionJobPB::CUMULATIVE, res, {42326, 42474}); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK); + + // Step 4: A BASE compaction arrives via the same code path used by EMPTY_CUMULATIVE - + // i.e. without `input_versions`. Because is_same_conflict_family(BASE, CUMULATIVE) is + // false, BASE should still be accepted on this branch (the cross-family conflict is + // enforced only on the version-range branch validated by Plan D test below). + res.Clear(); + start_compaction_job(meta_service.get(), tablet_id, "base1", "BE1", 0, 0, + TabletCompactionJobPB::BASE, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK); + + // Step 5: A second EMPTY_CUMULATIVE should also be rejected by the now-active CUMULATIVE. + // (Even though job_pb already contains an EMPTY_CUMULATIVE-equivalent, the same-family + // check primarily catches the CUMULATIVE side here.) + res.Clear(); + start_empty_cumu("empty2", "BE2", 0, 0, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); +} + +// Plan D regression test: when a CUMULATIVE compaction is already running with `input_versions` +// and `check_input_versions_range = true`, a BASE compaction whose version range overlaps with +// the in-flight CUMULATIVE must be rejected with JOB_TABLET_BUSY. Non-overlapping BASE jobs are +// still allowed, which is the typical safe case (BASE handles [0, cumu_point - 1] while +// CUMULATIVE handles versions above cumu_point). +TEST(MetaServiceJobTest, BaseCumulativeCrossTypeConflictTest) { + auto meta_service = get_meta_service(); + + auto sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + SyncPoint::get_instance()->clear_all_call_backs(); + }; + sp->set_call_back("get_instance_id", [&](auto&& args) { + auto* ret = try_any_cast_ret(args); + ret->first = instance_id; + ret->second = true; + }); + sp->enable_processing(); + + constexpr int64_t table_id = 1; + constexpr int64_t index_id = 2; + constexpr int64_t partition_id = 3; + constexpr int64_t tablet_id = 4; + ASSERT_NO_FATAL_FAILURE( + create_tablet(meta_service.get(), table_id, index_id, partition_id, tablet_id, false)); + + // Local helper: start a BASE compaction request that carries `input_versions` (matching + // production BE behaviour: cloud_base_compaction.cpp always calls add_input_versions). + // Note: BASE does NOT call set_check_input_versions_range, so it's left as default false + // BUT input_versions is non-empty - this routes the request into the "has input_versions" + // branch on MS, which is the branch Plan D guards. + auto start_base = [&](const std::string& job_id, const std::string& initiator, int base_cnt, + int cumu_cnt, std::pair versions, + StartTabletJobResponse& res) { + brpc::Controller cntl; + StartTabletJobRequest req; + req.mutable_job()->mutable_idx()->set_tablet_id(tablet_id); + auto* compaction = req.mutable_job()->add_compaction(); + compaction->set_id(job_id); + compaction->set_initiator(initiator); + compaction->set_base_compaction_cnt(base_cnt); + compaction->set_cumulative_compaction_cnt(cumu_cnt); + compaction->set_type(TabletCompactionJobPB::BASE); + long now = ::time(nullptr); + compaction->set_expiration(now + 12); + compaction->set_lease(now + 3); + compaction->add_input_versions(versions.first); + compaction->add_input_versions(versions.second); + // Intentionally NOT calling set_check_input_versions_range - BASE relies on the + // default false to mimic real BE behaviour. + meta_service->start_tablet_job(&cntl, &req, &res, nullptr); + }; + + // Step 1: A CUMULATIVE compaction with versions [10, 20] is started with parallel-cumu + // mode enabled (check_input_versions_range = true). This routes into the + // version-range-aware branch on MS. + StartTabletJobResponse res; + start_compaction_job(meta_service.get(), tablet_id, "cumu1", "BE1", 0, 0, + TabletCompactionJobPB::CUMULATIVE, res, {10, 20}); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK); + + // Step 2: BASE [5, 15] overlaps with CUMULATIVE [10, 20]. Plan D requires this to be + // rejected. Before the fix it would succeed (because the old `c.type() != compaction.type()` + // check skipped the active CUMULATIVE for a BASE submission). + res.Clear(); + start_base("base_overlap_left", "BE1", 0, 0, {5, 15}, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); + + // Step 3: BASE [15, 25] also overlaps. Should be rejected. + res.Clear(); + start_base("base_overlap_right", "BE1", 0, 0, {15, 25}, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); + + // Step 4: BASE [12, 18] is fully contained inside CUMULATIVE's range. Should be rejected. + res.Clear(); + start_base("base_overlap_inside", "BE1", 0, 0, {12, 18}, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); + + // Step 5: BASE [5, 25] fully covers the CUMULATIVE range. Should be rejected. + res.Clear(); + start_base("base_overlap_cover", "BE1", 0, 0, {5, 25}, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); + + // Step 6: BASE [0, 9] is BELOW the CUMULATIVE range. This is the typical safe case + // (base handles [0, cumu_point - 1]) and must still be accepted after Plan D. + res.Clear(); + start_base("base_safe_below", "BE1", 0, 0, {0, 9}, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg(); + + // Step 7: A second BASE [21, 30] is ABOVE the CUMULATIVE range AND non-overlapping with the + // already-accepted base_safe_below [0, 9]. This is also a safe non-overlap case - although + // unusual in production (BASE rarely operates above cumu_point), MS should accept it. + res.Clear(); + start_base("base_safe_above", "BE2", 0, 0, {21, 30}, res); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg(); + + // Step 8: A new CUMULATIVE [22, 28] overlaps with the just-accepted base_safe_above and + // must be rejected. Verifies the conflict is symmetric - CUMULATIVE submissions also + // see BASE jobs as conflicting. + res.Clear(); + start_compaction_job(meta_service.get(), tablet_id, "cumu_overlap_base", "BE1", 0, 0, + TabletCompactionJobPB::CUMULATIVE, res, {22, 28}); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); + // The version_in_compaction notification predicate is kept consistent with the conflict + // predicate (`may_conflict_by_type`): every in-flight job in the rowset compaction family + // (BASE / CUMULATIVE) is surfaced so BE can pick a non-overlapping range to retry. + // Active jobs at this point: cumu1[10,20], base_safe_below[0,9], base_safe_above[21,30]. + // All three carry concrete input_versions so all three must be reported. + ASSERT_EQ(res.version_in_compaction_size(), 6); + EXPECT_EQ(res.version_in_compaction(0), 10); + EXPECT_EQ(res.version_in_compaction(1), 20); + EXPECT_EQ(res.version_in_compaction(2), 0); + EXPECT_EQ(res.version_in_compaction(3), 9); + EXPECT_EQ(res.version_in_compaction(4), 21); + EXPECT_EQ(res.version_in_compaction(5), 30); + + // Step 9: A new CUMULATIVE [30, 35] does not overlap with cumu1 [10, 20] but DOES overlap + // with base_safe_above [21, 30] (sharing version 30). Must be rejected. + res.Clear(); + start_compaction_job(meta_service.get(), tablet_id, "cumu_overlap_base2", "BE1", 0, 0, + TabletCompactionJobPB::CUMULATIVE, res, {30, 35}); + ASSERT_EQ(res.status().code(), MetaServiceCode::JOB_TABLET_BUSY) << res.status().msg(); + + // Step 10: A new CUMULATIVE [31, 40] is fully above all active jobs and must be accepted. + res.Clear(); + start_compaction_job(meta_service.get(), tablet_id, "cumu_safe_above", "BE1", 0, 0, + TabletCompactionJobPB::CUMULATIVE, res, {31, 40}); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.status().msg(); +} + TEST(MetaServiceJobTest, SchemaChangeJobPersistTest) { auto meta_service = get_meta_service();