github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3820138546
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/HudiExternalMetaCache.java:
##########
@@ -103,7 +105,18 @@ public HoodieTableMetaClient
getHoodieTableMetaClient(NameMapping nameMapping) {
}
public HoodieTableFileSystemView getFsView(NameMapping nameMapping) {
- return
fsViewEntry.get(nameMapping.getCtlId()).get(HudiFsViewCacheKey.of(nameMapping));
+ return
fsViewEntry.get(nameMapping.getCtlId()).get(HudiFsViewCacheKey.of(nameMapping)).acquire();
+ }
+
+ public void releaseFsView(NameMapping nameMapping) {
+ MetaCacheEntry<HudiFsViewCacheKey, HudiFsViewCacheValue> entry =
+ fsViewEntry.getIfInitialized(nameMapping.getCtlId());
+ if (entry != null) {
+ HudiFsViewCacheValue value =
entry.getIfPresent(HudiFsViewCacheKey.of(nameMapping));
Review Comment:
[P1] Release the exact filesystem-view generation that was acquired
`getFsView()` discards the wrapper identity, while this path re-resolves the
current key. If W1 is invalidated or replaced before the scan finishes, this
either finds nothing (W1 stays evicted with refCount 1 forever) or decrements
W2, possibly below zero, while W1 leaks. Return/store a lease for the exact
`HudiFsViewCacheValue` and release that instance; do not use a cache lookup for
release.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -533,6 +547,8 @@ public List<Split> getSplits(int numBackends) throws
UserException {
});
} catch (Exception e) {
throw new UserException(ExceptionUtils.getRootCauseMessage(e), e);
+ } finally {
+ releaseFsViewOnce();
Review Comment:
[P1] Tie non-batch release to the whole task group
`initPrunedPartitions()` executes before this `try`, so pruning failure
bypasses release. Conversely, await interruption or submission rejection can
enter this `finally` while already accepted file-listing tasks still use
`fsView`, allowing eviction to close under them. Put pruning and structured
cancel/join of every accepted task under one owner and release only after the
task group is terminal.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/HudiFsViewCacheValue.java:
##########
@@ -0,0 +1,62 @@
+// 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.hudi;
+
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Reference-counted wrapper around a shared {@link HoodieTableFileSystemView}.
+ *
+ * <p>The underlying fs view is cached per table and shared by concurrent scan
nodes. Closing it while
+ * another thread is still planning splits is unsafe, so the cache only closes
the view after the entry has
+ * been evicted AND all acquired references have been released.
+ */
+public class HudiFsViewCacheValue {
+ private final HoodieTableFileSystemView fsView;
+ private final AtomicInteger refCount = new AtomicInteger(0);
+ private volatile boolean evicted = false;
+ private volatile boolean closed = false;
+
+ public HudiFsViewCacheValue(HoodieTableFileSystemView fsView) {
+ this.fsView = fsView;
+ }
+
+ public HoodieTableFileSystemView acquire() {
+ refCount.incrementAndGet();
Review Comment:
[P1] Linearize acquisition with eviction
`acquire()` is outside the synchronized close state. The removal listener
can run after cache lookup but before this increment, see count 0, close
`fsView`, and then `acquire()` returns the closed object. Make acquire/evict
one synchronized transition (for example, `tryAcquire` that fails after
eviction and makes the caller retry).
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -279,6 +282,36 @@ private List<org.apache.iceberg.DeleteFile>
loadDeleteFiles(org.apache.iceberg.M
return deleteFiles;
}
+ private boolean shouldCloseTableFileIO(CatalogIf catalog,
IcebergMetadataOps ops, Table table) {
+ if (!(catalog instanceof IcebergExternalCatalog)) {
+ return false;
+ }
+ IcebergExternalCatalog icebergCatalog = (IcebergExternalCatalog)
catalog;
+ String type = icebergCatalog.getIcebergCatalogType();
+ if (IcebergExternalCatalog.ICEBERG_GLUE.equals(type)
+ || IcebergExternalCatalog.ICEBERG_DLF.equals(type)) {
Review Comment:
[P1] Treat DLF FileIO as catalog-owned
On this branch `DLFCatalog.initialize()` creates one `FileIO` in
`HiveCompatibleCatalog.fileIO`, and every `newTableOps()` passes that same
object to `DLFTableOperations`. Evicting any table here closes the IO still
used by every other DLF table. Return false for DLF and close the shared IO
once at catalog shutdown; add a two-table ownership test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/MetaCacheEntry.java:
##########
@@ -107,7 +118,12 @@ public MetaCacheEntry(String name, @Nullable Function<K,
V> loader, CacheSpec ca
maxSize,
true,
null);
- this.loadingData =
cacheFactory.buildCache(this::loadFromDefaultLoader, refreshExecutor);
+ if (removalListener != null) {
+ this.loadingData = cacheFactory.buildCacheWithAsyncRemovalListener(
Review Comment:
[P1] Define cleanup for values that never remain in Caffeine
Resource listeners only observe admitted values. With caching disabled or
invalidation during a manual miss, this method can return a loaded value
without caching it, so no listener ever retires it; in the post-put generation
race it removes/schedules close and then returns the same value. Hudi/Iceberg
resource entries therefore leak or hand callers a value being closed. Add an
explicit ownership/lease path for bypassed and suppressed loads.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -597,6 +614,7 @@ public void startSplit(int numBackends) {
System.currentTimeMillis() -
startTime);
}
splitAssignment.finishSchedule();
+ releaseFsViewOnce();
Review Comment:
[P1] Release after all submitted batch work, not all original partitions
The producer breaks on `batchException`, stop, or interruption and can
submit fewer than `prunedPartitions.size()`. Only submitted workers increment
this counter, so the equality can never be reached and the lease remains
forever; executor rejection also bypasses it. Track producer completion plus
actual submitted/in-flight workers and run one terminal release for success,
stop, error, interruption, and rejection.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java:
##########
@@ -126,6 +127,26 @@ public Catalog getCatalog() {
return catalog;
}
+ /**
+ * Returns the catalog-level FileIO for session catalogs that expose one
internally (e.g. REST).
+ * Returns null when the catalog does not have a separate catalog-level
FileIO or it cannot be determined.
+ */
+ public FileIO getCatalogFileIO() {
+ if (catalog == null) {
+ return null;
+ }
+ try {
+ if (catalog instanceof org.apache.iceberg.rest.RESTSessionCatalog)
{
Review Comment:
[P2] Check the RESTCatalog runtime type
`CatalogUtil.buildIcebergCatalog` returns `RESTCatalog` in Iceberg 1.10.1;
it contains a `RESTSessionCatalog` delegate but is not one. This `instanceof`
is therefore always false, `getCatalogFileIO()` always returns null, and
`shouldCloseTableFileIO()` never enables the REST eviction cleanup this patch
claims. Use the SDK-supported tracker/lifecycle or handle the actual wrapper
and test shared versus vended IO.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -92,7 +93,8 @@ public IcebergExternalMetaCache(ExecutorService
refreshExecutor) {
super(ENGINE, refreshExecutor);
tableEntry = registerEntry(MetaCacheEntryDef.of(ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class,
this::loadTableCacheValue, defaultEntryCacheSpec(),
- MetaCacheEntryInvalidation.forNameMapping(nameMapping ->
nameMapping)));
+ MetaCacheEntryInvalidation.forNameMapping(nameMapping ->
nameMapping),
+ this::closeTableCacheValue));
Review Comment:
[P1] Do not close table IO while escaped table generations are active
This entry returns raw `Table`/snapshot values with no lease. Scans, sinks,
transactions, and actions retain those values beyond cache lookup, and the
frozen snapshot operations delegate `io()` to the old operations. A refresh or
invalidation therefore closes `FileIO` under an active statement. Use an
exact-generation lease or rely on Iceberg's `FileIOTracker`, which defers close
until `TableOperations` is unreferenced.
--
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]
