This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 fbeda73811c [improvement](fe) Reduce cloud version sync config (#66296)
fbeda73811c is described below
commit fbeda73811c488cb4302d1fe7927505ba0e354a1
Author: meiyi <[email protected]>
AuthorDate: Fri Sep 11 17:03:10 2026 +0800
[improvement](fe) Reduce cloud version sync config (#66296)
1. Cloud table and partition version synchronization ran every 20
seconds, batched up to 2000 version reads, and reused the global Meta
Service retry limit of 200, which could create high concurrent FDB read
pressure and amplify failed requests. Increase the sync interval to 60
seconds, reduce the batch size to 200, and limit background get-version
tasks to 3 attempts without changing other callers.
2. Skip the daemon when both global/default cache TTLs are finite while
retaining proactive refresh whenever either cache never expires.
---
.../main/java/org/apache/doris/common/Config.java | 7 ++-
.../java/org/apache/doris/catalog/OlapTable.java | 6 +-
.../apache/doris/cloud/catalog/CloudPartition.java | 18 ++++--
.../cloud/catalog/CloudSyncVersionDaemon.java | 14 ++++-
.../org/apache/doris/cloud/rpc/VersionHelper.java | 26 +++++---
.../org/apache/doris/catalog/OlapTableTest.java | 18 +++---
.../doris/cloud/catalog/CloudPartitionTest.java | 43 ++++++++++---
.../apache/doris/cloud/rpc/VersionHelperTest.java | 70 ++++++++++++++++++++++
8 files changed, 170 insertions(+), 32 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 54a16d46685..3ddf8ba9ec9 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -3495,7 +3495,7 @@ public class Config extends ConfigBase {
@ConfField(description = "Cloud table and partition version syncer
interval. All frontends will perform the "
+ "checking.")
- public static int cloud_version_syncer_interval_second = 20;
+ public static int cloud_version_syncer_interval_second = 60;
@ConfField(mutable = true, description = "Whether to enable the function
of syncing table and partition version "
+ "in cloud mode.")
@@ -3508,7 +3508,10 @@ public class Config extends ConfigBase {
public static int cloud_sync_version_task_threads_num = 4;
@ConfField(mutable = true, description = "Maximum table or partition batch
size for get version tasks.")
- public static int cloud_get_version_task_batch_size = 2000;
+ public static int cloud_get_version_task_batch_size = 200;
+
+ @ConfField(mutable = true, description = "Maximum retry times for cloud
version syncer get version tasks.")
+ public static int cloud_version_syncer_get_version_retry_times = 3;
@ConfField(mutable = true, description = "Whether to enable retry when a
schema change job fails, default is true.")
public static boolean enable_schema_change_retry = true;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
index e36e9456c3e..17cac68dc68 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
@@ -3738,6 +3738,10 @@ public class OlapTable extends Table implements
MTMVRelatedTableIf, GsonPostProc
}
public static List<Long> getVisibleVersionFromMeta(List<Long> dbIds,
List<Long> tableIds) {
+ return getVisibleVersionFromMeta(dbIds, tableIds,
Config.metaServiceRpcRetryTimes());
+ }
+
+ public static List<Long> getVisibleVersionFromMeta(List<Long> dbIds,
List<Long> tableIds, int maxAttempts) {
// get version rpc
Cloud.GetVersionRequest request = Cloud.GetVersionRequest.newBuilder()
.setRequestIp(FrontendOptions.getLocalHostAddressCached())
@@ -3751,7 +3755,7 @@ public class OlapTable extends Table implements
MTMVRelatedTableIf, GsonPostProc
.build();
try {
- Cloud.GetVersionResponse resp =
VersionHelper.getVersionFromMeta(request);
+ Cloud.GetVersionResponse resp =
VersionHelper.getVersionFromMeta(request, maxAttempts);
if (resp.getStatus().getCode() != Cloud.MetaServiceCode.OK) {
throw new RpcException("get table visible version",
"unexpected status " + resp.getStatus());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
index eab872ecdb3..86c3b72409c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
@@ -239,6 +239,12 @@ public class CloudPartition extends Partition {
// Return the visible version in order of the specified partition ids
public static List<Long> getSnapshotVisibleVersionFromMs(
List<CloudPartition> partitions, boolean waitForPendingTxns)
throws RpcException {
+ return getSnapshotVisibleVersionFromMs(
+ partitions, waitForPendingTxns,
Config.metaServiceRpcRetryTimes());
+ }
+
+ public static List<Long> getSnapshotVisibleVersionFromMs(
+ List<CloudPartition> partitions, boolean waitForPendingTxns, int
maxAttempts) throws RpcException {
if (partitions.isEmpty()) {
return new ArrayList<>();
}
@@ -255,7 +261,7 @@ public class CloudPartition extends Partition {
}
List<Long> versions = getSnapshotVisibleVersion(
- dbIds, tableIds, partitionIds, versionUpdateTimesMs,
commitTsos, waitForPendingTxns);
+ dbIds, tableIds, partitionIds, versionUpdateTimesMs,
commitTsos, waitForPendingTxns, maxAttempts);
// Cache visible version, see hasData() for details.
int size = versions.size();
@@ -304,8 +310,10 @@ public class CloudPartition extends Partition {
return Collections.emptyList();
}
- long cloudPartitionVersionCacheTtlMs = ConnectContext.get() == null ? 0
- :
ConnectContext.get().getSessionVariable().cloudPartitionVersionCacheTtlMs;
+ ConnectContext ctx = ConnectContext.get();
+ long cloudPartitionVersionCacheTtlMs = ctx == null
+ ?
VariableMgr.getDefaultSessionVariable().cloudPartitionVersionCacheTtlMs
+ : ctx.getSessionVariable().cloudPartitionVersionCacheTtlMs;
if (cloudPartitionVersionCacheTtlMs <= 0) { // No cached versions will
be used
return getSnapshotVisibleVersionFromMs(partitions, false);
}
@@ -359,7 +367,7 @@ public class CloudPartition extends Partition {
//
// Return the visible version in order of the specified partition ids
private static List<Long> getSnapshotVisibleVersion(List<Long> dbIds,
List<Long> tableIds, List<Long> partitionIds,
- List<Long> versionUpdateTimesMs, List<Long> commitTsos, boolean
waitForPendingTxns)
+ List<Long> versionUpdateTimesMs, List<Long> commitTsos, boolean
waitForPendingTxns, int maxAttempts)
throws RpcException {
assert dbIds.size() == partitionIds.size() :
"partition ids size: " + partitionIds.size() + " should equals
to db ids size: " + dbIds.size();
@@ -381,7 +389,7 @@ public class CloudPartition extends Partition {
if (LOG.isDebugEnabled()) {
LOG.debug("getVisibleVersion use CloudPartition {}",
partitionIds.toString());
}
- Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(req);
+ Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(req,
maxAttempts);
if (resp.getStatus().getCode() != MetaServiceCode.OK) {
throw new RpcException("get visible version", "unexpected status "
+ resp.getStatus());
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
index d65a6806eac..6d485d959bc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
@@ -24,6 +24,7 @@ import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.Table;
import org.apache.doris.common.Config;
import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.qe.VariableMgr;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -58,6 +59,13 @@ public class CloudSyncVersionDaemon extends MasterDaemon {
if (!Config.cloud_enable_version_syncer) {
return;
}
+ // This daemon has no ConnectContext, so use the global/default TTLs
to decide whether
+ // the shared version caches need proactive refresh. Finite TTLs
refresh lazily on reads,
+ // while Long.MAX_VALUE never expires and requires this daemon to keep
the cache current.
+ if
(VariableMgr.getDefaultSessionVariable().cloudPartitionVersionCacheTtlMs !=
Long.MAX_VALUE
+ &&
VariableMgr.getDefaultSessionVariable().cloudTableVersionCacheTtlMs !=
Long.MAX_VALUE) {
+ return;
+ }
LOG.info("begin sync cloud table and partition version");
Map<OlapTable, Long> tableVersionMap = syncTableVersions();
if (!tableVersionMap.isEmpty()) {
@@ -121,7 +129,8 @@ public class CloudSyncVersionDaemon extends MasterDaemon {
List<Long> tableIds, List<OlapTable> tables) {
return GET_VERSION_THREAD_POOL.submit(() -> {
try {
- List<Long> versions =
OlapTable.getVisibleVersionFromMeta(dbIds, tableIds);
+ List<Long> versions = OlapTable.getVisibleVersionFromMeta(
+ dbIds, tableIds,
Config.cloud_version_syncer_get_version_retry_times);
for (int i = 0; i < tables.size(); i++) {
OlapTable table = tables.get(i);
long version = versions.get(i);
@@ -190,7 +199,8 @@ public class CloudSyncVersionDaemon extends MasterDaemon {
private Future<Void> submitGetPartitionVersionTask(Set<Long> failedTables,
List<CloudPartition> partitions) {
return GET_VERSION_THREAD_POOL.submit(() -> {
try {
- CloudPartition.getSnapshotVisibleVersionFromMs(partitions,
false);
+ CloudPartition.getSnapshotVisibleVersionFromMs(
+ partitions, false,
Config.cloud_version_syncer_get_version_retry_times);
} catch (Exception e) {
LOG.warn("get partition version error", e);
Set<Long> failedTableIds = partitions.stream().map(p ->
p.getTableId())
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/VersionHelper.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/VersionHelper.java
index 703f8d2675c..d0003c50467 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/VersionHelper.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/VersionHelper.java
@@ -38,10 +38,15 @@ public class VersionHelper {
// Call get_version() from meta service, and save the elapsed to summary
profile.
public static Cloud.GetVersionResponse
getVersionFromMeta(Cloud.GetVersionRequest req)
throws RpcException {
+ return getVersionFromMeta(req, Config.metaServiceRpcRetryTimes());
+ }
+
+ public static Cloud.GetVersionResponse
getVersionFromMeta(Cloud.GetVersionRequest req, int maxAttempts)
+ throws RpcException {
long startAt = System.nanoTime();
boolean isTableVersion = req.getIsTableVersion();
try {
- return getVisibleVersion(req);
+ return getVisibleVersion(req, maxAttempts);
} finally {
SummaryProfile profile = getSummaryProfile();
if (profile != null) {
@@ -56,8 +61,13 @@ public class VersionHelper {
}
public static Cloud.GetVersionResponse
getVisibleVersion(Cloud.GetVersionRequest request) throws RpcException {
+ return getVisibleVersion(request, Config.metaServiceRpcRetryTimes());
+ }
+
+ public static Cloud.GetVersionResponse
getVisibleVersion(Cloud.GetVersionRequest request, int maxAttempts)
+ throws RpcException {
int tryTimes = 0;
- while (tryTimes++ < Config.metaServiceRpcRetryTimes()) {
+ while (tryTimes++ < maxAttempts) {
Cloud.GetVersionResponse resp = getVisibleVersionInternal(request,
Config.default_get_version_from_ms_timeout_second * 1000);
if (resp != null) {
@@ -73,14 +83,16 @@ public class VersionHelper {
resp.getStatus(), tryTimes);
}
// sleep random millis, retry rpc failed
- if (tryTimes > Config.metaServiceRpcRetryTimes() / 2) {
- sleepSeveralMs(500, 1000);
- } else {
- sleepSeveralMs(20, 200);
+ if (tryTimes < maxAttempts) {
+ if (tryTimes > maxAttempts / 2) {
+ sleepSeveralMs(500, 1000);
+ } else {
+ sleepSeveralMs(20, 200);
+ }
}
}
- LOG.warn("get version from meta service failed after retry {} times",
tryTimes);
+ LOG.warn("get version from meta service failed after retry {} times",
maxAttempts);
throw new RpcException("get version from meta service", "failed after
retry n times");
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java
index 9d8e7e608d9..2671b51b379 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableTest.java
@@ -715,14 +715,16 @@ public class OlapTableTest {
mockedConfig.when(Config::isNotCloudMode).thenReturn(false);
mockedConfig.when(Config::isCloudMode).thenReturn(true);
- mockedVH.when(() ->
VersionHelper.getVersionFromMeta(Mockito.any())).thenAnswer(invocation -> {
- Cloud.GetVersionResponse.Builder builder =
Cloud.GetVersionResponse.newBuilder();
- builder.setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
- .setCode(Cloud.MetaServiceCode.OK).build());
- builder.addAllVersions(batchVersions.get(callCount[0]));
- callCount[0]++;
- return builder.build();
- });
+ mockedVH.when(() -> VersionHelper.getVersionFromMeta(
+ Mockito.any(Cloud.GetVersionRequest.class),
Mockito.anyInt()))
+ .thenAnswer(invocation -> {
+ Cloud.GetVersionResponse.Builder builder =
Cloud.GetVersionResponse.newBuilder();
+
builder.setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.OK).build());
+
builder.addAllVersions(batchVersions.get(callCount[0]));
+ callCount[0]++;
+ return builder.build();
+ });
ConnectContext ctx = new ConnectContext();
ctx.setSessionVariable(new SessionVariable());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
index 8ba448a46cd..9a33dcf70af 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
@@ -21,6 +21,7 @@ import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.rpc.VersionHelper;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.qe.VariableMgr;
import org.apache.doris.rpc.RpcException;
import org.junit.jupiter.api.Assertions;
@@ -28,6 +29,7 @@ import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import org.mockito.stubbing.Answer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -72,6 +74,29 @@ public class CloudPartitionTest {
}
+ @Test
+ public void
testSnapshotVisibleVersionUsesDefaultCacheTtlWithoutConnectContext() throws
RpcException {
+ ConnectContext.remove();
+ SessionVariable defaultSessionVariable =
VariableMgr.getDefaultSessionVariable();
+ long originalCacheTtlMs =
defaultSessionVariable.cloudPartitionVersionCacheTtlMs;
+ try {
+ defaultSessionVariable.cloudPartitionVersionCacheTtlMs =
Long.MAX_VALUE;
+ CloudPartition cachedPartition = createPartition(1, 2, 3);
+ cachedPartition.setCachedVisibleVersion(2, 10086L);
+
+ try (MockedStatic<VersionHelper> mockedVersionHelper =
Mockito.mockStatic(VersionHelper.class)) {
+ List<Long> versions = CloudPartition.getSnapshotVisibleVersion(
+ Arrays.asList(cachedPartition));
+
+ Assertions.assertEquals(Arrays.asList(2L), versions);
+ mockedVersionHelper.verifyNoInteractions();
+ }
+ } finally {
+ defaultSessionVariable.cloudPartitionVersionCacheTtlMs =
originalCacheTtlMs;
+ ConnectContext.remove();
+ }
+ }
+
@Test
public void testCachedVersion() throws RpcException {
// Create ConnectContext with SessionVariable
@@ -98,14 +123,18 @@ public class CloudPartitionTest {
// CHECKSTYLE ON
try (MockedStatic<VersionHelper> mockedVersionHelper =
Mockito.mockStatic(VersionHelper.class)) {
+ Answer<Cloud.GetVersionResponse> getVersionAnswer = invocation -> {
+ Cloud.GetVersionResponse.Builder builder =
Cloud.GetVersionResponse.newBuilder();
+ builder.setVersion(singleVersions.get(callCount[0]));
+ builder.addAllVersions(batchVersions.get(callCount[0]));
+ ++callCount[0];
+ return builder.build();
+ };
mockedVersionHelper.when(() ->
VersionHelper.getVersionFromMeta(Mockito.any(Cloud.GetVersionRequest.class)))
- .thenAnswer(invocation -> {
- Cloud.GetVersionResponse.Builder builder =
Cloud.GetVersionResponse.newBuilder();
- builder.setVersion(singleVersions.get(callCount[0]));
-
builder.addAllVersions(batchVersions.get(callCount[0]));
- ++callCount[0];
- return builder.build();
- });
+ .thenAnswer(getVersionAnswer);
+ mockedVersionHelper.when(() -> VersionHelper.getVersionFromMeta(
+ Mockito.any(Cloud.GetVersionRequest.class),
Mockito.anyInt()))
+ .thenAnswer(getVersionAnswer);
ctx.getSessionVariable().cloudPartitionVersionCacheTtlMs = -1; //
disable cache
{
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/VersionHelperTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/VersionHelperTest.java
new file mode 100644
index 00000000000..ffdb81a988f
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/VersionHelperTest.java
@@ -0,0 +1,70 @@
+// 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.
+
+package org.apache.doris.cloud.rpc;
+
+import org.apache.doris.cloud.proto.Cloud;
+import org.apache.doris.rpc.RpcException;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.concurrent.CompletableFuture;
+
+public class VersionHelperTest {
+ @Test
+ public void testGetVisibleVersionUsesSpecifiedMaxAttempts() throws
RpcException {
+ Cloud.GetVersionRequest request =
Cloud.GetVersionRequest.newBuilder().build();
+ Cloud.GetVersionResponse failedResponse =
Cloud.GetVersionResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.KV_TXN_GET_ERR))
+ .build();
+ MetaServiceProxy proxy = Mockito.mock(MetaServiceProxy.class);
+ Mockito.when(proxy.getVisibleVersionAsync(request))
+ .thenReturn(CompletableFuture.completedFuture(failedResponse));
+
+ try (MockedStatic<MetaServiceProxy> mockedProxy =
Mockito.mockStatic(MetaServiceProxy.class)) {
+ mockedProxy.when(MetaServiceProxy::getInstance).thenReturn(proxy);
+
+ Assert.assertThrows(RpcException.class, () ->
VersionHelper.getVisibleVersion(request, 3));
+ }
+
+ Mockito.verify(proxy,
Mockito.times(3)).getVisibleVersionAsync(request);
+ }
+
+ @Test
+ public void testGetVisibleVersionStopsOnVersionNotFound() throws
RpcException {
+ Cloud.GetVersionRequest request =
Cloud.GetVersionRequest.newBuilder().build();
+ Cloud.GetVersionResponse notFoundResponse =
Cloud.GetVersionResponse.newBuilder()
+ .setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(Cloud.MetaServiceCode.VERSION_NOT_FOUND))
+ .build();
+ MetaServiceProxy proxy = Mockito.mock(MetaServiceProxy.class);
+ Mockito.when(proxy.getVisibleVersionAsync(request))
+
.thenReturn(CompletableFuture.completedFuture(notFoundResponse));
+
+ try (MockedStatic<MetaServiceProxy> mockedProxy =
Mockito.mockStatic(MetaServiceProxy.class)) {
+ mockedProxy.when(MetaServiceProxy::getInstance).thenReturn(proxy);
+
+ Assert.assertSame(notFoundResponse,
VersionHelper.getVisibleVersion(request, 3));
+ }
+
+ Mockito.verify(proxy).getVisibleVersionAsync(request);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
