From afc0db22e9f58252592df72e2bfdcd0ccfff46dd Mon Sep 17 00:00:00 2001 From: deardeng Date: Tue, 30 Jun 2026 12:45:57 +0800 Subject: [PATCH] [fix](cloud) use snapshot read for table version to avoid txn conflict (#64647) PR #60467 added a non-snapshot txn->get(table_version_key) in commit_txn, commit_partition, commit_index and drop_partition to return table_version+1 to FE as a version-cache hint. A non-snapshot read adds table_version_key to the transaction's read-conflict range, while update_table_version() bumps the same key via atomic_add (otherwise conflict-free). As a result, concurrent commits/imports on the same table now conflict on the version key and fail with KV_TXN_CONFLICT, causing retries and throughput regression. Change these reads to snapshot reads. The returned value is only a hint for FE's version cache (reconciled by CloudTableAndPartitionVersionChecker) and the real increment is still done by atomic_add, so correctness is preserved while the read-conflict on the version key is removed. (cherry picked from commit 64df172ca8582442770ae9edb5a5db09b80e723f) --- .../meta-service/meta_service_partition.cpp | 18 +++++++++++++++--- cloud/src/meta-service/meta_service_txn.cpp | 18 +++++++++++++++--- 2 files changed, 30 insertions(+), 6 deletions(-) diff --git a/cloud/src/meta-service/meta_service_partition.cpp b/cloud/src/meta-service/meta_service_partition.cpp index c28abeadb0f34f..f4ed11c26c0268 100644 --- a/cloud/src/meta-service/meta_service_partition.cpp +++ b/cloud/src/meta-service/meta_service_partition.cpp @@ -335,7 +335,11 @@ void MetaServiceImpl::commit_index(::google::protobuf::RpcController* controller int64_t table_id = request->table_id(); std::string ver_key = table_version_key({instance_id, request->db_id(), table_id}); std::string ver_val; - err = txn->get(ver_key, &ver_val); + // snapshot read: the returned table version is only a hint for FE's version + // cache; the real increment is done by update_table_version() via atomic_add. + // A non-snapshot read would add ver_key to the read-conflict set and make + // concurrent commits on the same table conflict (KV_TXN_CONFLICT). + err = txn->get(ver_key, &ver_val, true); int64_t table_version = 0; if (err == TxnErrorCode::TXN_OK) { if (!txn->decode_atomic_int(ver_val, &table_version)) { @@ -845,7 +849,11 @@ void MetaServiceImpl::commit_partition_internal(const PartitionRequest* request, std::string ver_key = table_version_key({instance_id, request->db_id(), request->table_id()}); std::string ver_val; - err = txn->get(ver_key, &ver_val); + // snapshot read: the returned table version is only a hint for FE's version + // cache; the real increment is done by update_table_version() via atomic_add. + // A non-snapshot read would add ver_key to the read-conflict set and make + // concurrent commits on the same table conflict (KV_TXN_CONFLICT). + err = txn->get(ver_key, &ver_val, true); int64_t table_version = 0; if (err == TxnErrorCode::TXN_OK) { if (!txn->decode_atomic_int(ver_val, &table_version)) { @@ -1034,7 +1042,11 @@ void MetaServiceImpl::drop_partition(::google::protobuf::RpcController* controll std::string ver_key = table_version_key({instance_id, request->db_id(), request->table_id()}); std::string ver_val; - err = txn->get(ver_key, &ver_val); + // snapshot read: the returned table version is only a hint for FE's version + // cache; the real increment is done by update_table_version() via atomic_add. + // A non-snapshot read would add ver_key to the read-conflict set and make + // concurrent commits on the same table conflict (KV_TXN_CONFLICT). + err = txn->get(ver_key, &ver_val, true); int64_t table_version = 0; if (err == TxnErrorCode::TXN_OK) { if (!txn->decode_atomic_int(ver_val, &table_version)) { diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index 4694d0f14dbc7e..adcb18550ad3fb 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -1791,7 +1791,11 @@ void MetaServiceImpl::commit_txn_immediately( int64_t table_id = i.first; std::string ver_key = table_version_key({instance_id, db_id, table_id}); std::string ver_val; - err = txn->get(ver_key, &ver_val); + // snapshot read: the returned table version is only a hint for FE's version + // cache; the real increment is done by update_table_version() via atomic_add. + // A non-snapshot read would add ver_key to the read-conflict set and make + // concurrent commits on the same table conflict (KV_TXN_CONFLICT). + err = txn->get(ver_key, &ver_val, true); int64_t table_version = 0; if (err == TxnErrorCode::TXN_OK) { if (!txn->decode_atomic_int(ver_val, &table_version)) { @@ -2480,7 +2484,11 @@ void MetaServiceImpl::commit_txn_eventually( int64_t table_id = i.first; std::string ver_key = table_version_key({instance_id, db_id, table_id}); std::string ver_val; - err = txn->get(ver_key, &ver_val); + // snapshot read: the returned table version is only a hint for FE's version + // cache; the real increment is done by update_table_version() via atomic_add. + // A non-snapshot read would add ver_key to the read-conflict set and make + // concurrent commits on the same table conflict (KV_TXN_CONFLICT). + err = txn->get(ver_key, &ver_val, true); int64_t table_version = 0; if (err == TxnErrorCode::TXN_OK) { if (!txn->decode_atomic_int(ver_val, &table_version)) { @@ -2969,7 +2977,11 @@ void MetaServiceImpl::commit_txn_with_sub_txn(const CommitTxnRequest* request, int64_t table_id = i.first; std::string ver_key = table_version_key({instance_id, db_id, table_id}); std::string ver_val; - err = txn->get(ver_key, &ver_val); + // snapshot read: the returned table version is only a hint for FE's version + // cache; the real increment is done by update_table_version() via atomic_add. + // A non-snapshot read would add ver_key to the read-conflict set and make + // concurrent commits on the same table conflict (KV_TXN_CONFLICT). + err = txn->get(ver_key, &ver_val, true); int64_t table_version = 0; if (err == TxnErrorCode::TXN_OK) { if (!txn->decode_atomic_int(ver_val, &table_version)) {