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); + } +}