github-actions[bot] commented on code in PR #68042:
URL: https://github.com/apache/doris/pull/68042#discussion_r4024408493
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java:
##########
@@ -62,62 +62,58 @@ protected synchronized void makeSureInitialized() {
}
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());
Review Comment:
[P1] Preserve the post-commit metadata boundary
`isDone()` lets a caller that starts after a remote transaction becomes
visible join a refresh whose server-side snapshot predates that commit.
`FrontendServiceImpl.getOlapTableMeta()` snapshots table/partition metadata
under the remote table read lock; after that lock is released, V+1 can become
visible while this task is still transporting or rebuilding a large response. A
later query then selects this unfinished task and plans with version V, so
committed rows can be invisible to a subsequent query. The removed monitor
scope forced that later caller to wait and issue a new post-commit RPC, and the
existing Remote Doris regression suite explicitly expects a select after insert
to fetch the new partition version. Please preserve a generation/invalidation
boundary that prevents post-snapshot callers from reusing earlier metadata.
This coordination protocol also has no deterministic upstream test: existing
command tests mock `getOlapTable()`, and the cited manual concurrency run was
not performed on this head. Please add latch/barrier-based FE coverage for one
RPC/result within a shareable cohort, a fresh successor after completion or
invalidation (including the V-to-V+1 schedule above), retry after exceptional
completion, and interruption of one waiter without cancelling the shared task.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalTable.java:
##########
@@ -62,62 +62,58 @@ protected synchronized void makeSureInitialized() {
}
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;
- }
+ FutureTask<RemoteOlapTable> refreshTask;
+ boolean shouldRun;
+ synchronized (this) {
+ if (currentRefreshTask == null || currentRefreshTask.isDone()) {
+ currentRefreshTask = new
FutureTask<>(this::loadDorisOlapTable);
+ shouldRun = true;
+ } else {
+ shouldRun = false;
}
+ refreshTask = currentRefreshTask;
}
- synchronized (this) {
- while (isSyncOlapTable) {
- try {
- this.wait();
- } catch (InterruptedException e) {
- throw new AnalysisException("interrupted while getting
doris olap table", e);
- }
- }
+ if (shouldRun) {
+ refreshTask.run();
+ }
+ return getRefreshResult(refreshTask);
+ }
- // 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");
- }
- return remoteOlapTable;
+ 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());
Review Comment:
[P1] Preserve fatal `Error` propagation
`FutureTask.run()` captures every `Throwable`, so an `OutOfMemoryError`
while deserializing/copying the large partition set (or another JVM `Error`)
reaches this branch as the `ExecutionException` cause. Wrapping it
unconditionally in Nereids `AnalysisException` turns a fatal VM/invariant
failure into an ordinary query-analysis error; on the direct query path that
materially changes control flow, and the old `catch (Exception)` implementation
did not catch the creator's `Error`. Please rethrow `Error` causes unchanged
before wrapping expected refresh exceptions, and add a focused test for this
boundary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
