From 343a3d7bb18882d7527bee63090caf9ef099f734 Mon Sep 17 00:00:00 2001 From: meiyi Date: Fri, 11 Sep 2026 17:03:10 +0800 Subject: [PATCH] [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. --- .../java/org/apache/doris/common/Config.java | 7 +- .../org/apache/doris/catalog/OlapTable.java | 6 +- .../doris/cloud/catalog/CloudPartition.java | 18 +++-- .../cloud/catalog/CloudSyncVersionDaemon.java | 14 +++- .../apache/doris/cloud/rpc/VersionHelper.java | 26 +++++-- .../apache/doris/catalog/OlapTableTest.java | 2 +- .../cloud/catalog/CloudPartitionTest.java | 28 +++++++- .../doris/cloud/rpc/VersionHelperTest.java | 70 +++++++++++++++++++ 8 files changed, 152 insertions(+), 19 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/VersionHelperTest.java 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 9010441127643c..3260b8721a987c 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 @@ -3894,7 +3894,7 @@ public void handle(Field field, String value) throws Exception { @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"}) @@ -3909,7 +3909,10 @@ public void handle(Field field, String value) throws Exception { @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 5aa46f8a57b10a..af7b21e259491c 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 @@ private static List getVisibleVersionInBatchFromMs(List tables) } public static List getVisibleVersionFromMeta(List dbIds, List tableIds) { + return getVisibleVersionFromMeta(dbIds, tableIds, Config.metaServiceRpcRetryTimes()); + } + + public static List getVisibleVersionFromMeta(List dbIds, List tableIds, int maxAttempts) { // get version rpc Cloud.GetVersionRequest request = Cloud.GetVersionRequest.newBuilder() .setRequestIp(FrontendOptions.getLocalHostAddressCached()) @@ -3558,7 +3562,7 @@ public static List getVisibleVersionFromMeta(List dbIds, List .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 d518721fabe88e..cfb9fabe907cd0 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 static List selectNonEmptyPartitionIds(List partiti // Return the visible version in order of the specified partition ids public static List getSnapshotVisibleVersionFromMs( List partitions, boolean waitForPendingTxns) throws RpcException { + return getSnapshotVisibleVersionFromMs( + partitions, waitForPendingTxns, Config.metaServiceRpcRetryTimes()); + } + + public static List getSnapshotVisibleVersionFromMs( + List partitions, boolean waitForPendingTxns, int maxAttempts) throws RpcException { if (partitions.isEmpty()) { return new ArrayList<>(); } @@ -243,7 +249,7 @@ public static List getSnapshotVisibleVersionFromMs( } List 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 static List getSnapshotVisibleVersion(List partitio 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 static List getSnapshotVisibleVersion(List partitio // // Return the visible version in order of the specified partition ids private static List getSnapshotVisibleVersion(List dbIds, List tableIds, List partitionIds, - List versionUpdateTimesMs, boolean waitForPendingTxns) + List 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 @@ private static List getSnapshotVisibleVersion(List dbIds, List 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 d65a6806eac9d7..6d485d959bc0e4 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.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 @@ protected void runAfterCatalogReady() { 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 tableVersionMap = syncTableVersions(); if (!tableVersionMap.isEmpty()) { @@ -121,7 +129,8 @@ private Future submitGetTableVersionTask(Map tableVersion List tableIds, List tables) { return GET_VERSION_THREAD_POOL.submit(() -> { try { - List versions = OlapTable.getVisibleVersionFromMeta(dbIds, tableIds); + List 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 @@ private void syncPartitionVersion(Map tableVersionMap) { private Future submitGetPartitionVersionTask(Set failedTables, List 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 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 703f8d2675cac6..d0003c504671d9 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 static Cloud.GetVersionResponse getVersionFromMeta(Cloud.GetVersionReques } 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 static Cloud.GetVersionResponse getVisibleVersion(Cloud.GetVersionRequest 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 1353ec7ea8b5ad..3a10a7dc1621bc 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 boolean isCloudMode() { new MockUp() { @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 de01b9a02f516a..02d22ddcba888e 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.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 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 void testIsCachedVersionExpired() { } + @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 mockedVersionHelper = Mockito.mockStatic(VersionHelper.class)) { + List 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 void testCachedVersion() throws RpcException { new MockUp(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 00000000000000..ffdb81a988ff5d --- /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 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 mockedProxy = Mockito.mockStatic(MetaServiceProxy.class)) { + mockedProxy.when(MetaServiceProxy::getInstance).thenReturn(proxy); + + Assert.assertSame(notFoundResponse, VersionHelper.getVisibleVersion(request, 3)); + } + + Mockito.verify(proxy).getVisibleVersionAsync(request); + } +}