This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new a2815f50805 branch-4.1: [improvement](fe) Reduce cloud version sync
config (#66296) (#67929)
a2815f50805 is described below
commit a2815f50805a638d0c435a6e20cec83ab0ca59a4
Author: meiyi <[email protected]>
AuthorDate: Tue Sep 15 12:01:01 2026 +0800
branch-4.1: [improvement](fe) Reduce cloud version sync config (#66296)
(#67929)
pick https://github.com/apache/doris/pull/66296
---
.../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 | 2 +-
.../doris/cloud/catalog/CloudPartitionTest.java | 28 ++++++++-
.../apache/doris/cloud/rpc/VersionHelperTest.java | 70 ++++++++++++++++++++++
8 files changed, 152 insertions(+), 19 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 55a531c054b..031d2584590 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
@@ -3890,7 +3890,7 @@ public class Config extends ConfigBase {
@ConfField(description = {"存算分离模式下同步 table 和 partition version 的间隔. 所有
frontend 都会检查",
"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 = {"存算分离模式下是否启用同步 table 和 partition
version 的功能",
"Whether to enable the function of syncing table and partition
version in cloud mode"})
@@ -3905,7 +3905,10 @@ public class Config extends ConfigBase {
@ConfField(mutable = true, description = {"Get version task 包含的 table 或
partition 数目的 batch size",
"Maximal table or partition batch size of get version task."})
- 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 = {"schema change job 失败是否重试",
"Whether to enable retry when a schema change job fails, default
is 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 5aa46f8a57b..af7b21e2594 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
@@ -3545,6 +3545,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())
@@ -3558,7 +3562,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 d518721fabe..cfb9fabe907 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
@@ -228,6 +228,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<>();
}
@@ -243,7 +249,7 @@ public class CloudPartition extends Partition {
}
List<Long> versions = getSnapshotVisibleVersion(
- dbIds, tableIds, partitionIds, versionUpdateTimesMs,
waitForPendingTxns);
+ dbIds, tableIds, partitionIds, versionUpdateTimesMs,
waitForPendingTxns, maxAttempts);
// Cache visible version, see hasData() for details.
int size = versions.size();
@@ -290,8 +296,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);
}
@@ -345,7 +353,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, boolean waitForPendingTxns)
+ List<Long> versionUpdateTimesMs, boolean waitForPendingTxns, int
maxAttempts)
throws RpcException {
assert dbIds.size() == partitionIds.size() :
"partition ids size: " + partitionIds.size() + " should equals
to db ids size: " + dbIds.size();
@@ -367,7 +375,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 1353ec7ea8b..3a10a7dc162 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
@@ -637,7 +637,7 @@ public class OlapTableTest {
new MockUp<VersionHelper>() {
@Mock
- public Cloud.GetVersionResponse
getVersionFromMeta(Cloud.GetVersionRequest req) {
+ public Cloud.GetVersionResponse
getVersionFromMeta(Cloud.GetVersionRequest req, int maxAttempts) {
Cloud.GetVersionResponse.Builder builder =
Cloud.GetVersionResponse.newBuilder();
builder.setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
.setCode(Cloud.MetaServiceCode.OK).build());
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 de01b9a02f5..02d22ddcba8 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 mockit.Mock;
@@ -28,6 +29,8 @@ import mockit.MockUp;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
import java.util.ArrayList;
import java.util.Arrays;
@@ -72,6 +75,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,7 +124,7 @@ public class CloudPartitionTest {
new MockUp<VersionHelper>(VersionHelper.class) {
@Mock
- public Cloud.GetVersionResponse
getVersionFromMeta(Cloud.GetVersionRequest req) {
+ public Cloud.GetVersionResponse
getVersionFromMeta(Cloud.GetVersionRequest req, int maxAttempts) {
Cloud.GetVersionResponse.Builder builder =
Cloud.GetVersionResponse.newBuilder();
builder.setVersion(singleVersions.get(callCount[0]));
builder.addAllVersions(batchVersions.get(callCount[0]));
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]
