This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 5f51a85b6e9 [Fix](merge-on-write) cloud mow table should sync rowsets
in publish phase if compaction on other BE finished during this load (#37670)
5f51a85b6e9 is described below
commit 5f51a85b6e983f403ea2998959c8d5ff6a35d976
Author: bobhan1 <[email protected]>
AuthorDate: Sat Jul 13 21:53:54 2024 +0800
[Fix](merge-on-write) cloud mow table should sync rowsets in publish phase
if compaction on other BE finished during this load (#37670)
## Proposed changes
Due to https://github.com/apache/doris/pull/35838, when executing load
job, BE will not `sync_rowsets()` in publish phase if a compaction job
is finished on another BE on the same tablet between the commit phase
and the publish phase of the current load job. This PR let the meta
service return the tablet compaction stats along with the
getDeleteBitmapUpdateLockResponse to FE and FE will send them to BE to
let the BE know whether it should `sync_rowsets()` due to compaction on
other BEs.
---
.../cloud/cloud_engine_calc_delete_bitmap_task.cpp | 34 ++++++-
.../cloud/cloud_engine_calc_delete_bitmap_task.h | 7 ++
cloud/src/meta-service/meta_service.cpp | 26 +++++
.../transaction/CloudGlobalTransactionMgr.java | 96 ++++++++++++++++---
gensrc/proto/cloud.proto | 5 +
gensrc/thrift/AgentService.thrift | 3 +
...st_cloud_load_compaction_on_different_be.groovy | 106 +++++++++++++++++++++
7 files changed, 264 insertions(+), 13 deletions(-)
diff --git a/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
b/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
index 723bcdb48fa..22f6689ff23 100644
--- a/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
+++ b/be/src/cloud/cloud_engine_calc_delete_bitmap_task.cpp
@@ -69,9 +69,18 @@ Status CloudEngineCalcDeleteBitmapTask::execute() {
for (const auto& partition : _cal_delete_bitmap_req.partitions) {
int64_t version = partition.version;
- for (auto tablet_id : partition.tablet_ids) {
+ bool has_compaction_stats = partition.__isset.base_compaction_cnts &&
+
partition.__isset.cumulative_compaction_cnts &&
+ partition.__isset.cumulative_points;
+ for (size_t i = 0; i < partition.tablet_ids.size(); i++) {
+ auto tablet_id = partition.tablet_ids[i];
auto tablet_calc_delete_bitmap_ptr =
std::make_shared<CloudTabletCalcDeleteBitmapTask>(
_engine, this, tablet_id, transaction_id, version);
+ if (has_compaction_stats) {
+ tablet_calc_delete_bitmap_ptr->set_compaction_stats(
+ partition.base_compaction_cnts[i],
partition.cumulative_compaction_cnts[i],
+ partition.cumulative_points[i]);
+ }
auto submit_st = token->submit_func([=]() {
auto st = tablet_calc_delete_bitmap_ptr->handle();
if (!st.ok()) {
@@ -107,6 +116,14 @@
CloudTabletCalcDeleteBitmapTask::CloudTabletCalcDeleteBitmapTask(
fmt::format("CloudTabletCalcDeleteBitmapTask#_transaction_id={}",
_transaction_id));
}
+void CloudTabletCalcDeleteBitmapTask::set_compaction_stats(int64_t
ms_base_compaction_cnt,
+ int64_t
ms_cumulative_compaction_cnt,
+ int64_t
ms_cumulative_point) {
+ _ms_base_compaction_cnt = ms_base_compaction_cnt;
+ _ms_cumulative_compaction_cnt = ms_base_compaction_cnt;
+ _ms_cumulative_point = ms_cumulative_point;
+}
+
Status CloudTabletCalcDeleteBitmapTask::handle() const {
SCOPED_ATTACH_TASK(_mem_tracker);
int64_t t1 = MonotonicMicros();
@@ -122,7 +139,20 @@ Status CloudTabletCalcDeleteBitmapTask::handle() const {
}
int64_t max_version = tablet->max_version_unlocked();
int64_t t2 = MonotonicMicros();
- if (_version != max_version + 1) {
+
+ auto should_sync_rowsets_produced_by_compaction = [&]() {
+ if (_ms_base_compaction_cnt == -1) {
+ return true;
+ }
+
+ // some compaction jobs finished on other BEs during this load job
+ // we should sync rowsets and their delete bitmaps produced by
compaction jobs
+ std::shared_lock rlock(tablet->get_header_lock());
+ return _ms_base_compaction_cnt > tablet->base_compaction_cnt() ||
+ _ms_cumulative_compaction_cnt >
tablet->cumulative_compaction_cnt() ||
+ _ms_cumulative_point > tablet->cumulative_layer_point();
+ };
+ if (_version != max_version + 1 ||
should_sync_rowsets_produced_by_compaction()) {
auto sync_st = tablet->sync_rowsets();
if (sync_st.is<ErrorCode::INVALID_TABLET_STATE>()) [[unlikely]] {
_engine_calc_delete_bitmap_task->add_succ_tablet_id(_tablet_id);
diff --git a/be/src/cloud/cloud_engine_calc_delete_bitmap_task.h
b/be/src/cloud/cloud_engine_calc_delete_bitmap_task.h
index 6c399cd73e3..e3733d3e696 100644
--- a/be/src/cloud/cloud_engine_calc_delete_bitmap_task.h
+++ b/be/src/cloud/cloud_engine_calc_delete_bitmap_task.h
@@ -37,6 +37,9 @@ public:
int64_t transaction_id, int64_t version);
~CloudTabletCalcDeleteBitmapTask() = default;
+ void set_compaction_stats(int64_t ms_base_compaction_cnt, int64_t
ms_cumulative_compaction_cnt,
+ int64_t ms_cumulative_point);
+
Status handle() const;
private:
@@ -46,6 +49,10 @@ private:
int64_t _tablet_id;
int64_t _transaction_id;
int64_t _version;
+
+ int64_t _ms_base_compaction_cnt {-1};
+ int64_t _ms_cumulative_compaction_cnt {-1};
+ int64_t _ms_cumulative_point {-1};
std::shared_ptr<MemTrackerLimiter> _mem_tracker;
};
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index 7aac0816435..ecf054b68b6 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -1936,6 +1936,32 @@ void
MetaServiceImpl::get_delete_bitmap_update_lock(google::protobuf::RpcControl
}
}
+ bool require_tablet_stats =
+ request->has_require_compaction_stats() ?
request->require_compaction_stats() : false;
+ if (require_tablet_stats) {
+ // this request is from fe when it commits txn for MOW table, we send
the compaction stats
+ // along with the GetDeleteBitmapUpdateLockResponse which will be sent
to BE later to let
+ // BE eliminate unnecessary sync_rowsets() calls if possible
+ for (const auto& tablet_index : request->tablet_indexes()) {
+ TabletIndexPB idx(tablet_index);
+ TabletStatsPB tablet_stat;
+ internal_get_tablet_stats(code, msg, txn.get(), instance_id, idx,
tablet_stat, true);
+ if (code != MetaServiceCode::OK) {
+ response->clear_base_compaction_cnts();
+ response->clear_cumulative_compaction_cnts();
+ response->clear_cumulative_points();
+ LOG_WARNING(
+ "failed to get tablet stats when
get_delete_bitmap_update_lock, "
+ "lock_id={}, initiator={}, tablet_id={}",
+ request->lock_id(), request->initiator(),
tablet_index.tablet_id());
+ return;
+ }
+
response->add_base_compaction_cnts(tablet_stat.base_compaction_cnt());
+
response->add_cumulative_compaction_cnts(tablet_stat.cumulative_compaction_cnt());
+ response->add_cumulative_points(tablet_stat.cumulative_point());
+ }
+ }
+
lock_info.set_lock_id(request->lock_id());
lock_info.set_expiration(now + request->expiration());
bool found = false;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
index b433d8cfbe7..f3e106743d4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -56,6 +56,7 @@ import org.apache.doris.cloud.proto.Cloud.PrecommitTxnRequest;
import org.apache.doris.cloud.proto.Cloud.PrecommitTxnResponse;
import org.apache.doris.cloud.proto.Cloud.SubTxnInfo;
import org.apache.doris.cloud.proto.Cloud.TableStatsPB;
+import org.apache.doris.cloud.proto.Cloud.TabletIndexPB;
import org.apache.doris.cloud.proto.Cloud.TxnInfoPB;
import org.apache.doris.cloud.proto.Cloud.TxnStatusPB;
import org.apache.doris.cloud.proto.Cloud.UniqueIdPB;
@@ -582,16 +583,24 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
Map<Long, Map<Long, List<Long>>> backendToPartitionTablets =
Maps.newHashMap();
Map<Long, Partition> partitions = Maps.newHashMap();
Map<Long, Set<Long>> tableToPartitions = Maps.newHashMap();
- getPartitionInfo(tableList, tabletCommitInfos, tableToPartitions,
partitions, backendToPartitionTablets);
+ Map<Long, List<Long>> tableToTabletList = Maps.newHashMap();
+ Map<Long, TabletMeta> tabletToTabletMeta = Maps.newHashMap();
+ getPartitionInfo(tableList, tabletCommitInfos, tableToPartitions,
partitions, backendToPartitionTablets,
+ tableToTabletList, tabletToTabletMeta);
if (backendToPartitionTablets.isEmpty()) {
throw new UserException("The partition info is empty, table may be
dropped, txnid=" + transactionId);
}
- getDeleteBitmapUpdateLock(tableToPartitions, transactionId);
+ Map<Long, Long> baseCompactionCnts = Maps.newHashMap();
+ Map<Long, Long> cumulativeCompactionCnts = Maps.newHashMap();
+ Map<Long, Long> cumulativePoints = Maps.newHashMap();
+ getDeleteBitmapUpdateLock(tableToPartitions, transactionId,
tableToTabletList, tabletToTabletMeta,
+ baseCompactionCnts, cumulativeCompactionCnts,
cumulativePoints);
Map<Long, Long> partitionVersions = getPartitionVersions(partitions);
Map<Long, List<TCalcDeleteBitmapPartitionInfo>>
backendToPartitionInfos = getCalcDeleteBitmapInfo(
- backendToPartitionTablets, partitionVersions);
+ backendToPartitionTablets, partitionVersions,
baseCompactionCnts, cumulativeCompactionCnts,
+ cumulativePoints);
sendCalcDeleteBitmaptask(dbId, transactionId, backendToPartitionInfos);
}
@@ -599,7 +608,9 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
List<TabletCommitInfo> tabletCommitInfos,
Map<Long, Set<Long>> tableToParttions,
Map<Long, Partition> partitions,
- Map<Long, Map<Long, List<Long>>> backendToPartitionTablets) {
+ Map<Long, Map<Long, List<Long>>> backendToPartitionTablets,
+ Map<Long, List<Long>> tableToTabletList,
+ Map<Long, TabletMeta> tabletToTabletMeta) {
Map<Long, OlapTable> tableMap = Maps.newHashMap();
for (OlapTable olapTable : tableList) {
tableMap.put(olapTable.getId(), olapTable);
@@ -616,6 +627,13 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
continue;
}
+ tabletToTabletMeta.put(tabletIds.get(i), tabletMeta);
+
+ if (!tableToTabletList.containsKey(tableId)) {
+ tableToTabletList.put(tableId, Lists.newArrayList());
+ }
+ tableToTabletList.get(tableId).add(tabletIds.get(i));
+
long partitionId = tabletMeta.getPartitionId();
long backendId = tabletCommitInfos.get(i).getBackendId();
@@ -647,15 +665,32 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
}
private Map<Long, List<TCalcDeleteBitmapPartitionInfo>>
getCalcDeleteBitmapInfo(
- Map<Long, Map<Long, List<Long>>> backendToPartitionTablets,
Map<Long, Long> partitionVersions) {
+ Map<Long, Map<Long, List<Long>>> backendToPartitionTablets,
Map<Long, Long> partitionVersions,
+ Map<Long, Long> baseCompactionCnts, Map<Long, Long>
cumulativeCompactionCnts,
+ Map<Long, Long> cumulativePoints) {
Map<Long, List<TCalcDeleteBitmapPartitionInfo>>
backendToPartitionInfos = Maps.newHashMap();
for (Map.Entry<Long, Map<Long, List<Long>>> entry :
backendToPartitionTablets.entrySet()) {
List<TCalcDeleteBitmapPartitionInfo> partitionInfos =
Lists.newArrayList();
- for (Map.Entry<Long, List<Long>> partitionToTables :
entry.getValue().entrySet()) {
- Long partitionId = partitionToTables.getKey();
+ for (Map.Entry<Long, List<Long>> partitionToTablets :
entry.getValue().entrySet()) {
+ Long partitionId = partitionToTablets.getKey();
+ List<Long> tabletList = partitionToTablets.getValue();
TCalcDeleteBitmapPartitionInfo partitionInfo = new
TCalcDeleteBitmapPartitionInfo(partitionId,
partitionVersions.get(partitionId),
- partitionToTables.getValue());
+ tabletList);
+ if (!baseCompactionCnts.isEmpty() &&
!cumulativeCompactionCnts.isEmpty()
+ && !cumulativePoints.isEmpty()) {
+ List<Long> reqBaseCompactionCnts = Lists.newArrayList();
+ List<Long> reqCumulativeCompactionCnts =
Lists.newArrayList();
+ List<Long> reqCumulativePoints = Lists.newArrayList();
+ for (long tabletId : tabletList) {
+
reqBaseCompactionCnts.add(baseCompactionCnts.get(tabletId));
+
reqCumulativeCompactionCnts.add(cumulativeCompactionCnts.get(tabletId));
+
reqCumulativePoints.add(cumulativePoints.get(tabletId));
+ }
+ partitionInfo.setBaseCompactionCnts(reqBaseCompactionCnts);
+
partitionInfo.setCumulativeCompactionCnts(reqCumulativeCompactionCnts);
+ partitionInfo.setCumulativePoints(reqCumulativePoints);
+ }
partitionInfos.add(partitionInfo);
}
backendToPartitionInfos.put(entry.getKey(), partitionInfos);
@@ -663,8 +698,20 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
return backendToPartitionInfos;
}
- private void getDeleteBitmapUpdateLock(Map<Long, Set<Long>>
tableToParttions, long transactionId)
- throws UserException {
+ private void getDeleteBitmapUpdateLock(Map<Long, Set<Long>>
tableToParttions, long transactionId,
+ Map<Long, List<Long>> tableToTabletList, Map<Long, TabletMeta>
tabletToTabletMeta,
+ Map<Long, Long> baseCompactionCnts, Map<Long, Long>
cumulativeCompactionCnts,
+ Map<Long, Long> cumulativePoints) throws
UserException {
+ if
(DebugPointUtil.isEnable("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.sleep"))
{
+ DebugPoint debugPoint = DebugPointUtil.getDebugPoint(
+
"CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.sleep");
+ int t = debugPoint.param("sleep_time", 8);
+ try {
+ Thread.sleep(t * 1000);
+ } catch (InterruptedException e) {
+ LOG.info("error ", e);
+ }
+ }
StopWatch stopWatch = new StopWatch();
stopWatch.start();
for (Map.Entry<Long, Set<Long>> entry : tableToParttions.entrySet()) {
@@ -672,7 +719,19 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
builder.setTableId(entry.getKey())
.setLockId(transactionId)
.setInitiator(-1)
- .setExpiration(DELETE_BITMAP_LOCK_EXPIRATION_SECONDS);
+ .setExpiration(DELETE_BITMAP_LOCK_EXPIRATION_SECONDS)
+ .setRequireCompactionStats(true);
+ List<Long> tabletList = tableToTabletList.get(entry.getKey());
+ for (Long tabletId : tabletList) {
+ TabletMeta tabletMeta = tabletToTabletMeta.get(tabletId);
+ TabletIndexPB.Builder tabletIndexBuilder =
TabletIndexPB.newBuilder();
+ tabletIndexBuilder.setDbId(tabletMeta.getDbId());
+ tabletIndexBuilder.setTableId(tabletMeta.getTableId());
+ tabletIndexBuilder.setIndexId(tabletMeta.getIndexId());
+ tabletIndexBuilder.setPartitionId(tabletMeta.getPartitionId());
+ tabletIndexBuilder.setTabletId(tabletId);
+ builder.addTabletIndexes(tabletIndexBuilder);
+ }
final GetDeleteBitmapUpdateLockRequest request = builder.build();
GetDeleteBitmapUpdateLockResponse response = null;
@@ -716,6 +775,21 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
}
throw new UserException("Failed to get delete bitmap lock,
code: " + response.getStatus().getCode());
}
+
+ // record tablet's latest compaction stats from meta service and
send them to BEs
+ // to let BEs eliminate unnecessary sync_rowsets() calls if
possible
+ List<Long> respBaseCompactionCnts =
response.getBaseCompactionCntsList();
+ List<Long> respCumulativeCompactionCnts =
response.getCumulativeCompactionCntsList();
+ List<Long> respCumulativePoints =
response.getCumulativePointsList();
+ if (!respBaseCompactionCnts.isEmpty() &&
!respCumulativeCompactionCnts.isEmpty()
+ && !respCumulativePoints.isEmpty()) {
+ for (int i = 0; i < tabletList.size(); i++) {
+ long tabletId = tabletList.get(i);
+ baseCompactionCnts.put(tabletId,
respBaseCompactionCnts.get(i));
+ cumulativeCompactionCnts.put(tabletId,
respCumulativeCompactionCnts.get(i));
+ cumulativePoints.put(tabletId,
respCumulativePoints.get(i));
+ }
+ }
}
stopWatch.stop();
LOG.info("get delete bitmap lock successfully. txns: {}. time cost: {}
ms.",
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index 5179ea52a9e..d056c457a5b 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -1401,10 +1401,15 @@ message GetDeleteBitmapUpdateLockRequest {
optional int64 lock_id = 4;
optional int64 initiator = 5;
optional int64 expiration = 6;
+ optional bool require_compaction_stats = 7 [default = false];
+ repeated TabletIndexPB tablet_indexes = 8;
}
message GetDeleteBitmapUpdateLockResponse {
optional MetaServiceResponseStatus status = 1;
+ repeated int64 base_compaction_cnts = 2;
+ repeated int64 cumulative_compaction_cnts = 3;
+ repeated int64 cumulative_points = 4;
}
message GetRLTaskCommitAttachRequest {
diff --git a/gensrc/thrift/AgentService.thrift
b/gensrc/thrift/AgentService.thrift
index a03cb9df99b..76066f9d566 100644
--- a/gensrc/thrift/AgentService.thrift
+++ b/gensrc/thrift/AgentService.thrift
@@ -432,6 +432,9 @@ struct TCalcDeleteBitmapPartitionInfo {
1: required Types.TPartitionId partition_id
2: required Types.TVersion version
3: required list<Types.TTabletId> tablet_ids
+ 4: optional list<i64> base_compaction_cnts
+ 5: optional list<i64> cumulative_compaction_cnts
+ 6: optional list<i64> cumulative_points
}
struct TCalcDeleteBitmapRequest {
diff --git
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_load_compaction_on_different_be.groovy
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_load_compaction_on_different_be.groovy
new file mode 100644
index 00000000000..b09364b9877
--- /dev/null
+++
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_load_compaction_on_different_be.groovy
@@ -0,0 +1,106 @@
+// 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.
+
+suite("test_cloud_load_compaction_on_different_be","nonConcurrent") {
+ if (!isCloudMode()) {
+ return
+ }
+
+ def backends = sql_return_maparray('show backends')
+ def replicaNum = 0
+ def targetBackend = null
+ for (def be : backends) {
+ def alive = be.Alive.toBoolean()
+ def decommissioned = be.SystemDecommissioned.toBoolean()
+ if (alive && !decommissioned) {
+ replicaNum++
+ targetBackend = be
+ }
+ }
+
+ if (replicaNum < 2) {
+ return
+ }
+
+ GetDebugPoint().clearDebugPointsForAllFEs()
+
+ def tableName = "test_cloud_load_compaction_on_different_be"
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """ CREATE TABLE ${tableName}
+ (k int, v1 int, v2 int )
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH (k)
+ BUCKETS 1 PROPERTIES(
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write"="true",
+ "disable_auto_compaction" = "true");
+ """
+ (1..20).each{ id ->
+ sql """insert into ${tableName} select number, number, number from
numbers("number"="10");"""
+ }
+
+ def tabletStats = sql_return_maparray("show tablets from ${tableName};")
+ assertTrue(tabletStats.size() == 1)
+ def tabletId = tabletStats[0].TabletId
+ def tabletBackendId = tabletStats[0].BackendId
+ def tabletBackend
+ for (def be : backends) {
+ if (be.BackendId == tabletBackendId) {
+ tabletBackend = be
+ break;
+ }
+ }
+ logger.info("tablet ${tabletId} on backend ${tabletBackend.Host} with
backendId=${tabletBackend.BackendId}");
+
+ def idx = 0
+ for (idx = 0; idx < backends.size(); idx++) {
+ if (backends[idx].BackendId != tabletBackendId) {
+ break;
+ }
+ }
+ def anotherBackend = backends[idx]
+
+ GetDebugPoint().clearDebugPointsForAllFEs()
+ try {
+
GetDebugPoint().enableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.sleep",
[sleep_time: 10])
+
+ def t1 = Thread.start {
+ sql """insert into ${tableName} select number, number, number from
numbers("number"="10");"""
+ }
+
+ def t2 = Thread.start {
+ sleep(4);
+ logger.info("trigger compaction on another BE
${anotherBackend.Host} with backendId=${anotherBackend.BackendId}")
+ def (code, out, err) = be_run_full_compaction(anotherBackend.Host,
anotherBackend.HttpPort, tabletId)
+ logger.info("Run compaction: code=" + code + ", out=" + out + ",
err=" + err)
+ assertEquals(code, 0)
+ def compactJson = parseJson(out.trim())
+ assertEquals("success", compactJson.status.toLowerCase())
+ }
+
+ t1.join()
+ t2.join()
+
+ def res = sql "select k, count(*) from ${tableName} group by k having
count(*) > 1;"
+ assertTrue(res.size() == 0)
+ } catch (Exception e) {
+ logger.info(e.getMessage())
+ assertTrue(false)
+ } finally {
+
GetDebugPoint().disableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.sleep")
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
