github-actions[bot] commented on code in PR #67639:
URL: https://github.com/apache/doris/pull/67639#discussion_r4348127562
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java:
##########
@@ -1807,7 +1831,8 @@ private void commitTransactionWithSubTxns(long dbId,
List<Table> tableList, long
}
}
- executeCommitTxnRequest(builder, tableList, transactionId, false,
null, null, new ArrayList<>(tabletIds));
+ executeCommitTxnRequest(builder, tableList, transactionId, false,
null, tabletCommitInfos,
Review Comment:
[P1] Do not publish subtransactions with the parent notification
This newly forwards subtransaction tablet lists into a protocol carrying
only the parent id and final partition version. The first statement uses the
parent id but can own an earlier version. Enable FE
`enable_notify_be_after_load_txn_commit` and BE
`enable_cloud_make_rs_visible_on_be`, disable MS
`advance_txn_lazy_commit_during_reads`, and use a non-TSO duplicate-key table
with empty-rowset metadata skipping. If statement 1 writes T1 at v2 and
statement 2 another bucket at v3, T1's v2 rowset is queued locally as v3. A
background sync fetches true v2 without filling v3 because that MS option
suppresses `partition_max_version`. The next T1 v4 notification applies pending
v3 and v4, exposing the first rowset twice at v2 and v3. Keep subtransactions
out of this shortcut or send each subtransaction id with its actual version;
owner-only notifications can remain independent.
##########
be/src/cloud/cloud_index_change_compaction.cpp:
##########
@@ -48,6 +50,12 @@ Status CloudIndexChangeCompaction::prepare_compact() {
}
RETURN_IF_ERROR(cloud_tablet()->sync_rowsets());
+ if (config::enable_compaction_rw_separation &&
+ static_cast<CloudClusterInfo*>(ExecEnv::GetInstance()->cluster_info())
+ ->should_skip_compaction(cloud_tablet())) {
Review Comment:
[P1] Preserve BUILD INDEX execution across compute groups
`BUILD INDEX` is routed to the submitting session's compute group:
`SchemaChangeHandler` saves it and `IndexChangeJob` restores it before choosing
BEs. That group need not own the tablet's last load. Loading a tablet in active
group A and building its index from B now fails here;
`EngineCloudIndexChangeTask` returns this preparation error and FE eventually
cancels the job. With tablets last loaded by different active groups, no single
submitting group can finish the index build. Route index tasks to each owner or
distinguish explicit index maintenance from automatic compaction before
applying this gate.
##########
regression-test/suites/cloud_p0/compaction_rw_separation/test_compaction_rw_separation_lazy_commit_owner.groovy:
##########
@@ -0,0 +1,138 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import groovy.json.JsonSlurper
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite('test_compaction_rw_separation_lazy_commit_owner', 'docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'cloud_tablet_rebalancer_interval_second=1',
+ ]
+ options.beConfigs += [
+ 'enable_compaction_rw_separation=true',
+ 'enable_cloud_make_rs_visible_on_be=false',
+ 'enable_cloud_txn_lazy_commit=true',
+ 'compaction_cluster_takeover_timeout_ms=600000',
+ 'cluster_status_cache_refresh_interval_sec=5',
+ 'tablet_sync_interval_s=3600',
+ 'schedule_sync_tablets_interval_s=3600',
+ 'file_cache_enter_disk_resource_limit_mode_percent=99',
+ 'disable_auto_compaction=true',
+ ]
+ options.msConfigs += [
+ 'enable_multi_version_status=true',
+ 'enable_cloud_txn_lazy_commit=true',
+ 'cloud_txn_lazy_commit_fuzzy_possibility=100',
+ ]
+ options.cloudMode = true
+
+ def getTabletStatus = { ip, port, tabletId ->
+ def response = new URL(
+
"http://${ip}:${port}/api/compaction/show?tablet_id=${tabletId}";).text
+ return new JsonSlurper().parseText(response)
+ }
+
+ def getBeIpAndPort = { clusterName ->
+ def backends = sql "SHOW BACKENDS"
+ def clusterBes = backends.findAll {
+ it[19].contains("\"compute_group_name\" : \"${clusterName}\"")
+ }
+ assertFalse(clusterBes.isEmpty(), "No BE found for cluster:
${clusterName}")
+ def firstBe = clusterBes[0]
+ return [ip: firstBe[1], httpPort: firstBe[4]]
+ }
+
+ def runAndWaitFullCompaction = { be, tabletId, boolean expectSuccess ->
+ def before = getTabletStatus(be.ip, be.httpPort, tabletId)
+ def beforeSuccess = before['last full success time']
+ def beforeFailure = before['last full failure time']
+ def (code, out, err) = curl('POST',
+
"http://${be.ip}:${be.httpPort}/api/compaction/run?tablet_id=${tabletId}&compact_type=full";)
+ assertEquals(0, code, "Failed to trigger full compaction: ${out},
${err}")
+ def submitResult = new JsonSlurper().parseText(out.trim())
+ assertEquals('success', submitResult.status.toLowerCase(),
Review Comment:
[P2] Accept synchronous rejection in the negative compaction cases
Every `expectSuccess=false` call reaches the new owner check in
`CloudFullCompaction::prepare_compact`, which returns `BE_NO_SUITABLE_VERSION`
synchronously from `_submit_full_compaction_task`. The HTTP action propagates
that error and serializes it through `Status::to_json()`, so the response
status is the error code rather than `success`. This unconditional assertion
therefore fails on the intended previous-owner rejection. The same assertion
pattern appears in the load-owner, schema-change-owner and versioned-read
suites, making all four fail their negative cases. Branch on the expected
outcome and validate the ownership rejection response for these calls.
##########
cloud/src/meta-service/txn_lazy_committer.cpp:
##########
@@ -547,6 +588,134 @@ void make_committed_txn_visible(const std::string&
instance_id, int64_t db_id, i
(txn_info.status() == TxnStatusPB::TXN_STATUS_VISIBLE));
if (txn_info.status() == TxnStatusPB::TXN_STATUS_COMMITTED) {
+ std::vector<int64_t> candidate_tablet_ids(
+ txn_info.last_active_candidate_tablet_ids().begin(),
+ txn_info.last_active_candidate_tablet_ids().end());
+ std::ranges::sort(candidate_tablet_ids);
+ txn_info.clear_last_active_candidate_tablet_ids();
+ if (!candidate_tablet_ids.empty()) {
+ std::unordered_map<int64_t, TabletIndexPB> tablet_indexes;
+ if (txn_info.versioned_read()) {
+ CloneChainReader meta_reader(instance_id, txn_kv.get(),
resource_mgr);
+ err = meta_reader.get_tablet_indexes(txn.get(),
candidate_tablet_ids,
+ &tablet_indexes);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format(
+ "failed to get versioned tablet indexes for lazy
owner result, "
+ "txn_id={}, err={}",
+ txn_id, err);
+ return;
+ }
+ } else {
+ // A non-versioned instance has no versioned tablet-index
documents. Read the
+ // single-version exact keys in bounded batches while
retaining the same final
+ // transaction and its conflict ranges.
+ for (size_t begin = 0; begin < candidate_tablet_ids.size();
+ begin += config::max_tablet_index_num_per_batch) {
+ const size_t end = std::min(begin +
config::max_tablet_index_num_per_batch,
+ candidate_tablet_ids.size());
+ std::vector<std::string> index_keys;
+ index_keys.reserve(end - begin);
+ for (size_t i = begin; i < end; ++i) {
+ index_keys.push_back(
+ meta_tablet_idx_key({instance_id,
candidate_tablet_ids[i]}));
+ }
+ std::vector<std::optional<std::string>> index_values;
+ err = txn->batch_get(&index_values, index_keys);
+ if (err != TxnErrorCode::TXN_OK) {
+ code = cast_as<ErrCategory::READ>(err);
+ msg = fmt::format(
+ "failed to read tablet indexes for lazy owner
result, txn_id={}, "
+ "err={}",
+ txn_id, err);
+ return;
+ }
+ DCHECK_EQ(index_values.size(), end - begin);
+ for (size_t i = begin; i < end; ++i) {
+ if (!index_values[i - begin].has_value()) {
+ continue;
+ }
+ TabletIndexPB index;
+ if (!index.ParseFromString(*index_values[i - begin])) {
+ code = MetaServiceCode::PROTOBUF_PARSE_ERR;
+ msg = fmt::format(
+ "malformed tablet index for lazy owner
result, txn_id={}, "
+ "tablet_id={}",
+ txn_id, candidate_tablet_ids[i]);
+ return;
+ }
+ tablet_indexes.emplace(candidate_tablet_ids[i],
std::move(index));
+ }
+ }
+ }
+ // The owner result is best effort and must never keep a committed
txn from becoming
+ // visible, e.g. when a tablet was dropped after the partition
conversion.
+ if (tablet_indexes.size() != candidate_tablet_ids.size()) {
+ LOG(WARNING) << "omit lazy owner result for tablets without
index, txn_id="
+ << txn_id << " expected=" <<
candidate_tablet_ids.size()
+ << " actual=" << tablet_indexes.size();
+ std::erase_if(candidate_tablet_ids, [&](int64_t tablet_id) {
+ return !tablet_indexes.contains(tablet_id);
+ });
+ }
+ txn_info.clear_last_active_tablet_ids();
+ txn_info.clear_last_active_epochs();
+ bool owner_result_at_capacity = false;
+ for (size_t begin = 0; begin < candidate_tablet_ids.size();
+ begin += config::max_tablet_index_num_per_batch) {
+ size_t end = std::min(begin +
config::max_tablet_index_num_per_batch,
+ candidate_tablet_ids.size());
+ std::vector<std::string> stats_keys;
+ stats_keys.reserve(end - begin);
+ for (size_t i = begin; i < end; ++i) {
+ int64_t tablet_id = candidate_tablet_ids[i];
+ const auto& idx = tablet_indexes.at(tablet_id);
+ stats_keys.push_back(
+ stats_tablet_key({instance_id, idx.table_id(),
idx.index_id(),
+ idx.partition_id(), tablet_id}));
+ }
+ std::vector<std::optional<std::string>> stats_values;
+ err = txn->batch_get(&stats_values, stats_keys);
Review Comment:
[P1] Bound owner lookups by the final transaction budget
These batches retain all non-snapshot read conflicts in one transaction. A
load touching 28,500 tablets with three-byte IDs fits the 90KB candidate bound
(85,500 bytes plus a small TxnInfo envelope). With a 36-byte instance id, the
71-byte index keys and 93-byte stats keys require 28,500 *
((2*71+33)+(2*93+33)) = 11,229,000 bytes of read conflicts alone, exceeding
[FDB's 10,000,000-byte transaction
limit](https://apple.github.io/foundationdb/known-limitations.html#large-transactions).
Existing partition/batch conversion can publish this load, but finalization
then fails and every retry reads the same candidates, leaving the transaction
COMMITTED. The serialized-value test does not cover this footprint. Omit excess
notification work or move/snapshot these best-effort reads outside mandatory
finalization.
##########
cloud/src/meta-service/meta_service_job.cpp:
##########
@@ -1982,24 +2162,62 @@ void process_schema_change_job(MetaServiceCode& code,
std::string& msg, std::str
.tag("msg", msg);
return;
}
+ merge_tablet_stats(load_stats, detached_stats);
+ copy_last_active_cluster_info(*stats, load_stats);
}
schema_change_update_tablet_stats(schema_change, stats, num_remove_rows,
size_remove_rowsets,
num_remove_rowsets, num_remove_segments,
index_size_remove_rowsets,
segment_size_remove_rowsets);
+
auto stats_key = stats_tablet_key(
{instance_id, new_table_id, new_index_id, new_partition_id,
new_tablet_id});
+ if (!has_valid_last_active_cluster(*stats)) {
+ TabletStatsPB old_tablet_stats;
+ if (is_versioned_read) {
+ internal_get_load_tablet_stats(code, msg, reader, txn.get(),
instance_id,
+ old_tablet_idx, old_tablet_stats,
false);
+ } else {
+ internal_get_tablet_stats(code, msg, txn.get(), instance_id,
old_tablet_idx,
+ old_tablet_stats, false);
+ }
+ if (code != MetaServiceCode::OK) {
+ LOG(WARNING) << "failed to get old tablet owner for schema change,
tablet_id="
+ << tablet_id << " code=" << code << " msg=" << msg;
+ return;
+ }
+ copy_last_active_cluster_info(old_tablet_stats, *stats);
+ }
+
+ copy_last_active_cluster_info(*stats, load_stats);
+
auto stats_val = stats->SerializeAsString();
txn->put(stats_key, stats_val);
if (is_versioned_write) {
+ TabletStatsPB compact_stats(*stats);
+ compact_stats.clear_last_active_cluster_id();
+ compact_stats.clear_last_active_time_ms();
+ compact_stats.clear_last_active_epoch();
+ compact_stats.clear_last_active_cluster_status();
+ compact_stats.clear_last_active_cluster_status_mtime_ms();
+ std::string compact_stats_val = compact_stats.SerializeAsString();
std::string compact_stats_key =
versioned::tablet_compact_stats_key({instance_id,
new_tablet_id});
- versioned_put(txn.get(), compact_stats_key, stats_val);
+ versioned_put(txn.get(), compact_stats_key, compact_stats_val);
+
+ std::string load_stats_key =
versioned::tablet_load_stats_key({instance_id, new_tablet_id});
+ if (!versioned::document_put(txn.get(), load_stats_key,
std::move(load_stats))) {
Review Comment:
[P1] Drain pending lazy publication before writing new load stats
On a versioned non-MoW table, a dual-write lazy load can reserve
versionstamp S1 while historical schema conversion runs. Schema-change finish
does not drain pending lazy publication and now writes this entire load-stats
document at a newer S2. Later `convert_tmp_rowsets` reads that document, adds
the load's counters, and writes them using its reserved S1. Latest-version
reads still select S2, permanently omitting that committed load's rows, bytes,
rowsets and segments; subsequent loads preserve the deficit. This sequential
ordering produces no transaction conflict, and the aggregate overlay restores
only owner fields. Drain pending lazy commits or avoid inserting a numeric
load-stats version ahead of their reserved stamps.
--
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]
