This is an automated email from the ASF dual-hosted git repository.
hello-stephen 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 f959bb904df [fix](doris-catalog) avoid metadata refresh lock
contention (#68042)
f959bb904df is described below
commit f959bb904df2571f95edcc26838213141d3c8468
Author: re20052 <[email protected]>
AuthorDate: Thu Oct 22 15:05:14 2026 +0800
[fix](doris-catalog) avoid metadata refresh lock contention (#68042)
Problem Summary:
Concurrent queries against the same Remote Doris Catalog table with
`use_arrow_flight=false` can stall during FE planning.
Two factors compound the problem:
1. **Lock design:** `makeSureInitialized()` and metadata refresh share
the same table monitor, which the refresh holds throughout RPC,
deserialization, partition reconstruction, and cache updates. Arriving
requests block at initialization before reaching the reuse branch. After
a refresh completes, a thread leaving initialization may immediately
reacquire the monitor for another refresh while others remain blocked,
limiting result reuse.
2. **Large partition counts:** More partitions increase metadata
processing work and extend the monitor hold time, amplifying contention
and request buildup.
This PR uses current-and-successor `FutureTask<RemoteOlapTable>`
coordination:
- Refresh coordination holds the monitor only to create or select a
task; initialization remains synchronized.
- Callers run their selected refresh outside the monitor, without an
executor or background thread.
- A task is closed to new callers before its metadata RPC starts.
Requests arriving while it is unfinished wait for and share a successor
task instead of consuming the current task's result.
- The first late caller creates an open successor. Other late callers
join it while its predecessor is unfinished.
- After the predecessor completes, every successor caller may attempt to
run the task. `FutureTask.run()` ensures that only one caller executes
it.
- An interrupted waiter exits before starting the successor; another
current or future caller can still run the same successor.
- If a refresh fails, a later successor can retry. An interrupted waiter
restores its interrupt status without cancelling the shared task.
- No TTL cache is added.
The scope is one `RemoteDorisExternalTable` instance within one FE
process. The metadata RPC parameters, partition synchronization, and
cache update logic remain unchanged.
Co-authored-by: renyiyang.1 <[email protected]>
---
.../datasource/doris/RemoteDorisExternalTable.java | 136 +++++++----
.../doris/RemoteDorisExternalTableTest.java | 271 +++++++++++++++++++++
2 files changed, 353 insertions(+), 54 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java
index 9b057a0fbc7..d178072f0e6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java
@@ -38,15 +38,15 @@ import org.apache.logging.log4j.Logger;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
public class RemoteDorisExternalTable extends ExternalTable {
private static final Logger LOG =
LogManager.getLogger(RemoteDorisExternalTable.class);
private volatile List<Partition> partitions = Lists.newArrayList();
private volatile List<Partition> tempPartitions = Lists.newArrayList();
private volatile long tableId = -1;
- private volatile boolean isSyncOlapTable = false;
- private volatile RemoteOlapTable remoteOlapTable = null;
- private volatile Exception lastException = null;
+ private transient RefreshTask latestRefreshTask;
public RemoteDorisExternalTable(long id, String name, String remoteName,
RemoteDorisExternalCatalog catalog, ExternalDatabase db) {
@@ -62,62 +62,90 @@ public class RemoteDorisExternalTable extends ExternalTable
{
}
private RemoteOlapTable getDorisOlapTable() {
- if (!isSyncOlapTable) {
- synchronized (this) {
- if (!isSyncOlapTable) {
- try {
- isSyncOlapTable = true;
- remoteOlapTable = null;
- lastException = null; // clear previous exception
-
- List<Partition> cachedPartitions =
Lists.newArrayList(partitions);
- List<Partition> cachedTempPartitions =
Lists.newArrayList(tempPartitions);
- RemoteOlapTable olapTable =
((RemoteDorisExternalCatalog) catalog).getFeServiceClient()
- .getOlapTable(dbName, remoteName, tableId,
cachedPartitions, cachedTempPartitions);
- olapTable.setCatalog((RemoteDorisExternalCatalog)
catalog);
- olapTable.setDatabase((RemoteDorisExternalDatabase)
db);
-
- // Remove redundant nested synchronized block
- tableId = olapTable.getId();
- partitions =
Lists.newArrayList(olapTable.getPartitions());
- tempPartitions =
Lists.newArrayList(olapTable.getTempPartitions().getPartitions());
-
- olapTable.setId(id); // change id in case of possible
conflicts
- olapTable.invalidateBackendsIfNeed();
- remoteOlapTable = olapTable;
- } catch (Exception e) {
- // Save exception for waiting threads
- lastException = e;
- LOG.warn("Failed to get remote doris olap table:
{}.{}", dbName, remoteName, e);
- throw e; // Re-throw the exception
- } finally {
- isSyncOlapTable = false;
- this.notifyAll();
- }
- return remoteOlapTable;
- }
+ RefreshTask refreshTask;
+ synchronized (this) {
+ if (latestRefreshTask == null || latestRefreshTask.isStarted()) {
+ FutureTask<RemoteOlapTable> predecessor =
+ latestRefreshTask == null ? null :
latestRefreshTask.task;
+ latestRefreshTask = new RefreshTask(predecessor);
}
+ refreshTask = latestRefreshTask;
}
- synchronized (this) {
- while (isSyncOlapTable) {
- try {
- this.wait();
- } catch (InterruptedException e) {
- throw new AnalysisException("interrupted while getting
doris olap table", e);
- }
- }
+ refreshTask.run();
+ return getRefreshResult(refreshTask.task);
+ }
+
+ private class RefreshTask {
+ private final FutureTask<RemoteOlapTable> task;
+ private volatile FutureTask<RemoteOlapTable> predecessor;
+ private volatile boolean started;
+
+ private RefreshTask(FutureTask<RemoteOlapTable> predecessor) {
+ this.predecessor = predecessor;
+ this.task = new
FutureTask<>(RemoteDorisExternalTable.this::loadDorisOlapTable);
+ }
+
+ private boolean isStarted() {
+ return started;
+ }
- // If there is a saved exception, throw it with more details
- if (remoteOlapTable == null) {
- if (lastException != null) {
- throw new AnalysisException(
- "failed to get remote doris olap table: " +
Util.getRootCauseMessage(lastException),
- lastException);
- }
- throw new AnalysisException("failed to get remote doris olap
table");
+ private void run() {
+ FutureTask<RemoteOlapTable> previousTask = predecessor;
+ if (previousTask != null) {
+ waitForRefreshCompletion(previousTask);
+ predecessor = null;
}
- return remoteOlapTable;
+ started = true;
+ task.run();
+ }
+ }
+
+ private RemoteOlapTable loadDorisOlapTable() {
+ try {
+ List<Partition> cachedPartitions = Lists.newArrayList(partitions);
+ List<Partition> cachedTempPartitions =
Lists.newArrayList(tempPartitions);
+ RemoteOlapTable olapTable = ((RemoteDorisExternalCatalog)
catalog).getFeServiceClient()
+ .getOlapTable(dbName, remoteName, tableId,
cachedPartitions, cachedTempPartitions);
+ olapTable.setCatalog((RemoteDorisExternalCatalog) catalog);
+ olapTable.setDatabase((RemoteDorisExternalDatabase) db);
+
+ tableId = olapTable.getId();
+ partitions = Lists.newArrayList(olapTable.getPartitions());
+ tempPartitions =
Lists.newArrayList(olapTable.getTempPartitions().getPartitions());
+
+ olapTable.setId(id); // change id in case of possible conflicts
+ olapTable.invalidateBackendsIfNeed();
+ return olapTable;
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to get remote doris olap table: {}.{}", dbName,
remoteName, e);
+ throw e;
+ }
+ }
+
+ private RemoteOlapTable getRefreshResult(FutureTask<RemoteOlapTable>
refreshTask) {
+ try {
+ // The underlying Thrift RPC has its own timeout; this only waits
for the shared refresh result.
+ return refreshTask.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AnalysisException("interrupted while getting doris olap
table", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ throw new AnalysisException(
+ "failed to get remote doris olap table: " +
Util.getRootCauseMessage(cause),
+ cause);
+ }
+ }
+
+ private void waitForRefreshCompletion(FutureTask<RemoteOlapTable>
refreshTask) {
+ try {
+ refreshTask.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AnalysisException("interrupted while getting doris olap
table", e);
+ } catch (ExecutionException e) {
+ // A later generation can retry after an ordinary refresh failure.
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalTableTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalTableTest.java
new file mode 100644
index 00000000000..5382e0755d5
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalTableTest.java
@@ -0,0 +1,271 @@
+// 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.datasource.doris;
+
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.TempPartitions;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+import org.mockito.stubbing.OngoingStubbing;
+
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class RemoteDorisExternalTableTest {
+ private static final String DB_NAME = "test_db";
+ private static final String REMOTE_TABLE_NAME = "remote_test_table";
+
+ private RemoteDorisExternalTable table;
+ private FeServiceClient client;
+ private RemoteOlapTable versionOneTable;
+ private RemoteOlapTable versionTwoTable;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ RemoteDorisExternalCatalog catalog =
Mockito.mock(RemoteDorisExternalCatalog.class);
+ RemoteDorisExternalDatabase db =
Mockito.mock(RemoteDorisExternalDatabase.class);
+ client = Mockito.mock(FeServiceClient.class);
+ versionOneTable = createRemoteOlapTable(3L);
+ versionTwoTable = createRemoteOlapTable(5L);
+
+ Mockito.when(catalog.getId()).thenReturn(1L);
+ Mockito.doReturn(db).when(catalog).getDbOrAnalysisException(DB_NAME);
+ Mockito.when(catalog.getFeServiceClient()).thenReturn(client);
+ Mockito.when(db.getId()).thenReturn(2L);
+ Mockito.when(db.getFullName()).thenReturn(DB_NAME);
+ Mockito.when(db.getRemoteName()).thenReturn(DB_NAME);
+
+ table = new RemoteDorisExternalTable(
+ 4L, "test_table", REMOTE_TABLE_NAME, catalog, db);
+ }
+
+ @Test
+ public void testPostCommitCallersShareSuccessorRefresh() throws Exception {
+ CountDownLatch versionOneSnapshotTaken = new CountDownLatch(1);
+ CountDownLatch releaseVersionOneRpc = new CountDownLatch(1);
+ CountDownLatch versionTwoRpcStarted = new CountDownLatch(1);
+ CountDownLatch releaseVersionTwoRpc = new CountDownLatch(1);
+ AtomicInteger refreshCount = new AtomicInteger();
+ AtomicReference<RemoteOlapTable> remoteMetadata = new
AtomicReference<>(versionOneTable);
+ whenRefreshCalled().thenAnswer(invocation -> {
+ RemoteOlapTable snapshot = remoteMetadata.get();
+ if (refreshCount.incrementAndGet() == 1) {
+ versionOneSnapshotTaken.countDown();
+ await(releaseVersionOneRpc);
+ } else {
+ versionTwoRpcStarted.countDown();
+ await(releaseVersionTwoRpc);
+ }
+ return snapshot;
+ });
+
+ AtomicReference<OlapTable> ownerResult = new AtomicReference<>();
+ AtomicReference<OlapTable> firstWaiterResult = new AtomicReference<>();
+ AtomicReference<OlapTable> secondWaiterResult = new
AtomicReference<>();
+ AtomicReference<Throwable> ownerFailure = new AtomicReference<>();
+ AtomicReference<Throwable> firstWaiterFailure = new
AtomicReference<>();
+ AtomicReference<Throwable> secondWaiterFailure = new
AtomicReference<>();
+
+ Thread owner = startRefresh(ownerResult, ownerFailure);
+ Assertions.assertTrue(versionOneSnapshotTaken.await(5,
TimeUnit.SECONDS));
+
+ // The remote commit becomes visible after the first RPC captured V.
+ remoteMetadata.set(versionTwoTable);
+ Thread firstWaiter = startRefresh(firstWaiterResult,
firstWaiterFailure);
+ Thread secondWaiter = startRefresh(secondWaiterResult,
secondWaiterFailure);
+ waitUntilWaiting(firstWaiter);
+ waitUntilWaiting(secondWaiter);
+
+ releaseVersionOneRpc.countDown();
+ join(owner);
+ Assertions.assertTrue(versionTwoRpcStarted.await(5, TimeUnit.SECONDS));
+ releaseVersionTwoRpc.countDown();
+ join(firstWaiter);
+ join(secondWaiter);
+
+ Assertions.assertNull(ownerFailure.get());
+ Assertions.assertNull(firstWaiterFailure.get());
+ Assertions.assertNull(secondWaiterFailure.get());
+ Assertions.assertSame(versionOneTable, ownerResult.get());
+ Assertions.assertSame(versionTwoTable, firstWaiterResult.get());
+ Assertions.assertSame(versionTwoTable, secondWaiterResult.get());
+ verifyRefreshCount(2);
+ }
+
+ @Test
+ public void testFailedRefreshCanRetry() {
+ RuntimeException failure = new RuntimeException("refresh failed");
+ whenRefreshCalled().thenThrow(failure).thenReturn(versionOneTable);
+
+ AnalysisException exception =
+ Assertions.assertThrows(AnalysisException.class,
table::getOlapTable);
+ Assertions.assertSame(failure, exception.getCause());
+
+ Assertions.assertSame(versionOneTable, table.getOlapTable());
+ verifyRefreshCount(2);
+ }
+
+ @Test
+ public void testInterruptedWaiterDoesNotBlockLaterRefresh() throws
Exception {
+ CountDownLatch snapshotTaken = new CountDownLatch(1);
+ CountDownLatch releaseRpc = new CountDownLatch(1);
+ whenRefreshCalled()
+ .thenAnswer(invocation -> {
+ snapshotTaken.countDown();
+ await(releaseRpc);
+ return versionOneTable;
+ })
+ .thenReturn(versionTwoTable);
+
+ AtomicReference<OlapTable> ownerResult = new AtomicReference<>();
+ AtomicReference<OlapTable> waiterResult = new AtomicReference<>();
+ AtomicReference<Throwable> ownerFailure = new AtomicReference<>();
+ AtomicReference<Throwable> waiterFailure = new AtomicReference<>();
+
+ Thread owner = startRefresh(ownerResult, ownerFailure);
+ Assertions.assertTrue(snapshotTaken.await(5, TimeUnit.SECONDS));
+ Thread waiter = startRefresh(waiterResult, waiterFailure);
+ waitUntilWaiting(waiter);
+
+ waiter.interrupt();
+ join(waiter);
+ Assertions.assertInstanceOf(AnalysisException.class,
waiterFailure.get());
+
+ releaseRpc.countDown();
+ join(owner);
+ Assertions.assertNull(ownerFailure.get());
+ Assertions.assertSame(versionOneTable, ownerResult.get());
+
+ Assertions.assertSame(versionTwoTable, table.getOlapTable());
+ verifyRefreshCount(2);
+ }
+
+ @Test
+ public void testInterruptedWaiterDoesNotOrphanSharedSuccessor() throws
Exception {
+ CountDownLatch firstSnapshotTaken = new CountDownLatch(1);
+ CountDownLatch releaseFirstRpc = new CountDownLatch(1);
+ CountDownLatch successorRpcStarted = new CountDownLatch(1);
+ whenRefreshCalled()
+ .thenAnswer(invocation -> {
+ firstSnapshotTaken.countDown();
+ await(releaseFirstRpc);
+ return versionOneTable;
+ })
+ .thenAnswer(invocation -> {
+ successorRpcStarted.countDown();
+ return versionTwoTable;
+ });
+
+ AtomicReference<OlapTable> ownerResult = new AtomicReference<>();
+ AtomicReference<OlapTable> interruptedResult = new AtomicReference<>();
+ AtomicReference<OlapTable> survivingResult = new AtomicReference<>();
+ AtomicReference<Throwable> ownerFailure = new AtomicReference<>();
+ AtomicReference<Throwable> interruptedFailure = new
AtomicReference<>();
+ AtomicReference<Throwable> survivingFailure = new AtomicReference<>();
+
+ Thread owner = startRefresh(ownerResult, ownerFailure);
+ Assertions.assertTrue(firstSnapshotTaken.await(5, TimeUnit.SECONDS));
+ Thread interruptedWaiter = startRefresh(interruptedResult,
interruptedFailure);
+ Thread survivingWaiter = startRefresh(survivingResult,
survivingFailure);
+ waitUntilWaiting(interruptedWaiter);
+ waitUntilWaiting(survivingWaiter);
+
+ interruptedWaiter.interrupt();
+ join(interruptedWaiter);
+ Assertions.assertInstanceOf(AnalysisException.class,
interruptedFailure.get());
+
+ releaseFirstRpc.countDown();
+ join(owner);
+ Assertions.assertTrue(successorRpcStarted.await(5, TimeUnit.SECONDS));
+ join(survivingWaiter);
+
+ Assertions.assertNull(ownerFailure.get());
+ Assertions.assertNull(survivingFailure.get());
+ Assertions.assertSame(versionOneTable, ownerResult.get());
+ Assertions.assertSame(versionTwoTable, survivingResult.get());
+ verifyRefreshCount(2);
+ }
+
+ private RemoteOlapTable createRemoteOlapTable(long tableId) {
+ RemoteOlapTable olapTable = Mockito.mock(RemoteOlapTable.class);
+ TempPartitions tempPartitions = Mockito.mock(TempPartitions.class);
+ Mockito.when(olapTable.getId()).thenReturn(tableId);
+
Mockito.when(olapTable.getPartitions()).thenReturn(Collections.emptyList());
+ Mockito.when(olapTable.getTempPartitions()).thenReturn(tempPartitions);
+
Mockito.when(tempPartitions.getPartitions()).thenReturn(Collections.emptyList());
+ return olapTable;
+ }
+
+ private OngoingStubbing<RemoteOlapTable> whenRefreshCalled() {
+ return Mockito.when(client.getOlapTable(
+ ArgumentMatchers.eq(DB_NAME),
ArgumentMatchers.eq(REMOTE_TABLE_NAME),
+ ArgumentMatchers.anyLong(), ArgumentMatchers.anyList(),
ArgumentMatchers.anyList()));
+ }
+
+ private void verifyRefreshCount(int count) {
+ Mockito.verify(client, Mockito.times(count)).getOlapTable(
+ ArgumentMatchers.eq(DB_NAME),
ArgumentMatchers.eq(REMOTE_TABLE_NAME),
+ ArgumentMatchers.anyLong(), ArgumentMatchers.anyList(),
ArgumentMatchers.anyList());
+ }
+
+ private Thread startRefresh(AtomicReference<OlapTable> result,
AtomicReference<Throwable> failure) {
+ Thread thread = new Thread(() -> {
+ try {
+ result.set(table.getOlapTable());
+ } catch (Throwable t) {
+ failure.set(t);
+ }
+ });
+ thread.start();
+ return thread;
+ }
+
+ private static void await(CountDownLatch latch) throws
InterruptedException {
+ if (!latch.await(5, TimeUnit.SECONDS)) {
+ throw new AssertionError("timed out waiting for test latch");
+ }
+ }
+
+ private static void waitUntilWaiting(Thread thread) throws
InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (System.nanoTime() < deadline) {
+ Thread.State state = thread.getState();
+ if (state == Thread.State.WAITING || state ==
Thread.State.TIMED_WAITING) {
+ return;
+ }
+ if (!thread.isAlive()) {
+ throw new AssertionError("thread exited before waiting");
+ }
+ Thread.sleep(10);
+ }
+ throw new AssertionError("thread did not enter waiting state");
+ }
+
+ private static void join(Thread thread) throws InterruptedException {
+ thread.join(TimeUnit.SECONDS.toMillis(5));
+ Assertions.assertFalse(thread.isAlive(), "test thread did not finish");
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
