924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3859221901
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -552,59 +628,307 @@ private void initPrunedPartitions() throws UserException
{
throw new UserException(ExceptionUtils.getRootCauseMessage(e), e);
}
partitionInit = true;
+ ensureHmsRuntimeGeneration();
}
@Override
public void startSplit(int numBackends) {
+ ensureHmsRuntimeGeneration();
if (prunedPartitions.isEmpty()) {
splitAssignment.finishSchedule();
+ releaseFsViewOnce();
return;
}
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
+ acquireFsView();
ExecutorService scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
+ Executor producerExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
long startTime = System.currentTimeMillis();
- CompletableFuture.runAsync(() -> {
- for (HivePartition partition : prunedPartitions) {
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
- }
- try {
- splittersOnFlight.acquire();
- } catch (InterruptedException e) {
- batchException.set(new UserException(e.getMessage(), e));
- break;
+ BatchFsViewOwner createdOwner = new BatchFsViewOwner(splitAssignment,
fsViewLease);
Review Comment:
Fixed in f1b97225035. The final BatchFsViewOwner is now registered with
SplitAssignment before producer submission, and close is idempotent while
requesting stop for every tracked terminal task. Lease release still waits for
terminal task accounting. Added an assignment.stop() test with an
already-started blocked task proving immediate interruption and deferred lease
close. Verified the related 90-test suite and ./build.sh --fe.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java:
##########
@@ -191,11 +191,16 @@ public void finishSchedule() {
}
public void stop() {
- if (isStop()) {
- return;
+ List<Closeable> resources;
+ synchronized (closeableResources) {
+ if (isStop()) {
+ return;
+ }
+ isStopped.set(true);
Review Comment:
Fixed in 1bcea055cf8. SplitAssignment.stop() is now strictly no-throw:
planning errors still surface from init()/getAssignedSplits(), while stop
always closes registered resources and lets FileQueryScanNode remove every
split source and coordinator cleanup continue. Added error-then-stop resource
cleanup and FileQueryScanNode source-removal regressions. Verified with 33
focused FE tests and ./build.sh --fe.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -521,23 +591,29 @@ private void getPartitionsSplits(List<HivePartition>
partitions, List<Split> spl
@Override
public List<Split> getSplits(int numBackends) throws UserException {
- if (incrementalRead && !incrementalRelation.fallbackFullTableScan()) {
- return getIncrementalSplits();
- }
- initPrunedPartitions();
- List<Split> splits = Collections.synchronizedList(new ArrayList<>());
+ ensureHmsRuntimeGeneration();
+ acquireFsView();
Review Comment:
Fixed in 2968a1b8d99. Non-fallback incremental planning now returns before
acquiring the shared fs-view, while generation checks remain before and after
split construction. HudiScanNodeTest covers a non-empty native/COW result and
an empty JNI/MOR result and verifies that neither path creates an fs-view lease.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/DorisHiveCatalog.java:
##########
@@ -0,0 +1,77 @@
+// 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.iceberg;
+
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.io.FileIO;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** Owns and closes the shared FileIO created by one Iceberg HiveCatalog
generation. */
+public class DorisHiveCatalog extends HiveCatalog {
+ private final AtomicBoolean closed = new AtomicBoolean();
+ private FileIO ownedFileIO;
+
+ @Override
+ public void initialize(String name, Map<String, String> properties) {
+ super.initialize(name, properties);
+ ownedFileIO = extractFileIO();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+ IOException closeFailure = null;
+ try {
+ super.close();
Review Comment:
Fixed in 1bcea055cf8. DorisHiveCatalog.close() now always executes both
inherited catalog cleanup and its owned FileIO cleanup. It preserves and
rethrows the first failure and attaches a later cleanup failure as suppressed,
so an unchecked metrics-reporter failure cannot skip FileIO release. Added a
throwing-reporter regression and verified repeated close remains idempotent.
Verified with 33 focused FE tests and ./build.sh --fe.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -133,7 +143,32 @@ IcebergSnapshotEntryKey.class,
IcebergSnapshotCacheValue.class, defaultEntryCach
public Table getIcebergTable(ExternalTable dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getIcebergTable();
+ IcebergTableCacheValue.Lease lease = statementLease(nameMapping);
+ if (lease != null) {
+ return lease.getIcebergTable();
Review Comment:
Fixed in 2968a1b8d99. updateTableProperties now uses
getWritableIcebergTable(dorisTable, this), matching the other mutation paths
and preserving expected metadata-ops/catalog-generation identity. The
validation tests now require the generation-bound writable acquisition on both
success and failure paths.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java:
##########
@@ -869,7 +871,7 @@ public void dropColumn(ExternalTable dorisTable, ColumnPath
columnPath, long upd
@Override
public void updateTableProperties(ExternalTable dorisTable, Map<String,
String> properties, long updateTime)
throws UserException {
- Table icebergTable = IcebergUtils.getIcebergTable(dorisTable);
+ Table icebergTable = IcebergUtils.getWritableIcebergTable(dorisTable,
this);
Review Comment:
Fixed. updateTableProperties now acquires a scoped WritableTableLease that
owns both the per-table FileIO cleanup and the promoted catalog-generation
lease. Success, commit failure, generation rejection, and wrapper-construction
failure all release the lease exactly once before refresh.
IcebergMetadataOpsValidationTest covers successful and failed commits; the
final class run passed all 60 tests.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -490,24 +645,51 @@ private SchemaCacheValue
loadSchemaCacheValue(IcebergSchemaCacheKey key, Table r
private void retireTableGeneration(NameMapping nameMapping,
@Nullable IcebergTableCacheValue previousValue,
IcebergTableCacheValue currentValue) {
- if (previousValue != null &&
previousValue.isSameOperationalGeneration(currentValue)) {
+ if (previousValue == null) {
return;
}
- MetaCacheEntry<IcebergSnapshotEntryKey, IcebergSnapshotCacheValue>
snapshots =
- snapshotEntry.getIfInitialized(nameMapping.getCtlId());
- if (snapshots != null) {
- // Projections of another metadata generation are unreachable.
Projections of the same
- // generation frozen on a previous handle keep that handle's
FileIO (vended credentials)
- // and location provider; scans bind to them, so they must be
rebuilt from the new handle.
- snapshots.invalidateIf((key, value) ->
key.getNameMapping().equals(nameMapping)
- && (!key.belongsTo(currentValue) ||
!sharesOperationalResources(currentValue, value)));
+ try {
+ if (previousValue.isSameOperationalGeneration(currentValue)) {
+ return;
+ }
+ MetaCacheEntry<IcebergSnapshotEntryKey, IcebergSnapshotCacheValue>
snapshots =
+ snapshotEntry.getIfInitialized(nameMapping.getCtlId());
+ if (snapshots != null) {
+ // Projections of another metadata generation are unreachable.
Projections of the same
+ // generation frozen on a previous handle keep that handle's
FileIO (vended credentials)
+ // and location provider; scans bind to them, so they must be
rebuilt from the new handle.
+ snapshots.invalidateIf((key, value) ->
key.getNameMapping().equals(nameMapping)
+ && (!key.belongsTo(currentValue) ||
!sharesOperationalResources(currentValue, value)));
+ }
+ Optional<String> currentUuid = currentValue.getTableUuid();
+ MetaCacheEntry<IcebergSchemaCacheKey, SchemaCacheValue> schemas =
+ schemaEntry.getIfInitialized(nameMapping.getCtlId());
+ if (schemas != null) {
+ schemas.invalidateIf(key ->
key.getNameMapping().equals(nameMapping)
+ && !key.getTableUuid().equals(currentUuid));
+ }
+ } finally {
+ // Caffeine REPLACED notifications intentionally do not run the
removal listener because
+ // the cache reservation transfers to the new generation. Resource
ownership does not:
+ // retire the old value here and let active statement/async leases
delay physical close.
+ previousValue.retire();
Review Comment:
Fixed. Projection reuse now requires the exact FileIO object, captured
authenticator, and both mapping flags. Distinct FileIO instances are never
treated as safe merely because their properties are equal. When consecutive
generations share the exact same FileIO, they share an atomic cleanup-owner
token before publication; the final generation or borrower closes it once, and
a failed ownership retain rejects the generation without a duplicate close.
Added focused replacement and retired-owner tests.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -282,7 +316,7 @@ public View getIcebergView(ExternalTable dorisTable) {
}
public IcebergSchemaCacheValue getIcebergSchemaCacheValue(NameMapping
nameMapping, long schemaId) {
- IcebergTableCacheValue tableValue =
tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ IcebergTableCacheValue tableValue = statementValue(nameMapping);
Review Comment:
Fixed. Table, snapshot, partition, and schema conversion now carry the same
captured authenticator and both schema-mapping options from the leased table
generation. Schema cache identity also includes table UUID, schema ID,
partition spec ID, and both mapping flags, so a retained generation cannot be
parsed with live catalog settings or a different partition spec. The final
focused run passed IcebergExternalMetaCacheTest and the related generation
tests.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -110,7 +114,12 @@ public IcebergExternalMetaCache(ExecutorService
refreshExecutor, ExternalMetaCac
this::loadTableCacheValue, defaultEntryCacheSpec(),
MetaCacheEntryInvalidation.forNameMapping(nameMapping ->
nameMapping))
.withSizeEstimator(this::prepareTableForCachePublication)
- .withReplacementListener(this::retireTableGeneration));
+ .withReplacementListener(this::retireTableGeneration)
+ .withRemovalListener(value -> value, (key, value) -> {
Review Comment:
Fixed in 9190fb0b8386. Both generation-fenced refresh workers now retain
local ownership until publication/admission returns `ADMITTED`; every
fence-loss, `DISABLED`, `NOT_CURRENT`, `REJECTED`, or exception path retires
the unpublished value in `finally`. Iceberg uses a dedicated
`IcebergTableCacheValue::retire` callback, so discarding a candidate releases
its FileIO/catalog references without running admitted-removal side effects
against the retained generation. Added deterministic mutation-fence and
weighted-rejection tests, including an Iceberg assertion that the retained
table, snapshot, and schema caches remain intact. The focused 139-test suite
and `./build.sh --fe` pass.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/DorisHiveCatalog.java:
##########
@@ -0,0 +1,80 @@
+// 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.iceberg;
+
+import com.google.common.base.Throwables;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.io.FileIO;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** Owns and closes the shared FileIO created by one Iceberg HiveCatalog
generation. */
+public class DorisHiveCatalog extends HiveCatalog {
+ private final AtomicBoolean closed = new AtomicBoolean();
+ private FileIO ownedFileIO;
+
+ @Override
+ public void initialize(String name, Map<String, String> properties) {
Review Comment:
Fixed in 2968a1b8d99. DorisHiveCatalog.initialize is failure-atomic: if
parent initialization fails after assigning FileIO, the override extracts and
closes that partial resource, suppresses any close failure on the original
exception, and rethrows it. DorisHiveCatalogTest uses a close-tracking FileIO
plus invalid client-pool configuration and verifies one close.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -204,8 +236,7 @@ private Table createQueryTable(
public IcebergSnapshotCacheValue getSnapshotCache(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- IcebergTableCacheValue tableValue =
- tableEntry.get(nameMapping.getCtlId()).get(nameMapping);
+ IcebergTableCacheValue tableValue = statementValue(nameMapping);
Review Comment:
Fixed in 2968a1b8d99. executeForGeneration now selects one captured (or
initialized fallback) ExecutionAuthenticator, passes that exact object into
loadSnapshotProjection and IcebergUtils.loadPartitionInfo, and binds it to the
snapshot value. The regression verifies that partition loading receives the
captured authenticator rather than re-reading the live catalog.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -110,7 +114,13 @@ public IcebergExternalMetaCache(ExecutorService
refreshExecutor, ExternalMetaCac
this::loadTableCacheValue, defaultEntryCacheSpec(),
MetaCacheEntryInvalidation.forNameMapping(nameMapping ->
nameMapping))
.withSizeEstimator(this::prepareTableForCachePublication)
- .withReplacementListener(this::retireTableGeneration));
+ .withReplacementListener(this::retireTableGeneration)
+ .withUnpublishedValueRetirer(IcebergTableCacheValue::retire)
+ .withRemovalListener(value -> value, (key, value) -> {
Review Comment:
Fixed in 2968a1b8d99. Resource-owning weighted entries can now opt out of
Caffeine soft values; the Iceberg table entry uses strong values until normal
bounded eviction, so removal always receives the exact cleanup token and
retires the table/catalog generation. The production
MetaCacheEntryDef-to-runtime path is covered, including explicit invalidation
delivering the exact value once.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/HudiFsViewCacheValue.java:
##########
@@ -0,0 +1,111 @@
+// 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;
+
+/**
+ * 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;
+ // The loader owns one transferable reference until getFsView hands this
exact generation to its first caller.
+ private int refCount = 1;
+ private boolean loaderReferenceAvailable = true;
+ private boolean evicted = false;
+ private boolean closed = false;
+
+ public HudiFsViewCacheValue(HoodieTableFileSystemView fsView) {
+ this.fsView = fsView;
+ }
+
+ public synchronized Lease tryAcquire() {
+ Lease lease;
+ if (loaderReferenceAvailable) {
+ loaderReferenceAvailable = false;
+ lease = new Lease(this, fsView);
+ } else if (evicted) {
+ return null;
+ } else {
+ refCount++;
+ lease = new Lease(this, fsView);
+ }
+ try {
+ // The cache uses expire-after-access without detached refresh.
Sync every foreground generation handoff
+ // so a continuously hot key still observes newly completed
commits.
+ fsView.sync();
Review Comment:
Fixed in 9190fb0b8386. The exact Hudi generation lease is now acquired under
the owner monitor, while `fsView.sync()` runs after leaving that monitor.
Eviction/reset can therefore retire the generation immediately during blocked
timeline I/O; the counted lease still delays physical close until sync/planning
releases it, and a sync exception closes the lease before propagating. Added a
deterministic blocked-sync vs eviction test, and `HudiFsViewCacheValueTest`
passes.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -521,23 +591,31 @@ private void getPartitionsSplits(List<HivePartition>
partitions, List<Split> spl
@Override
public List<Split> getSplits(int numBackends) throws UserException {
- if (incrementalRead && !incrementalRelation.fallbackFullTableScan()) {
- return getIncrementalSplits();
- }
- initPrunedPartitions();
- List<Split> splits = Collections.synchronizedList(new ArrayList<>());
+ ensureHmsRuntimeGeneration();
try {
+ if (incrementalRead &&
!incrementalRelation.fallbackFullTableScan()) {
+ List<Split> splits = getIncrementalSplits();
+ ensureHmsRuntimeGeneration();
+ return splits;
+ }
+ acquireFsView();
Review Comment:
Fixed. An empty full-scan pruning result now returns before acquiring or
synchronizing the Hudi filesystem view, while still checking that the catalog
generation did not move. This avoids unrelated remote timeline or storage I/O
for a scan already known to have no splits. Added
testEmptyFullScanDoesNotAcquireUnusedFsView; the final HudiScanNodeTest run
passed all 12 tests.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -2143,8 +2196,8 @@ public static List<Column>
getIcebergPartitionColumns(Optional<MvccSnapshot> sna
if (snapshotTable.isPresent()) {
// Schema ID alone cannot identify the partition spec;
metadata-only evolution may keep
// the same schema and snapshot IDs while changing spec(), so
derive both from T0.
- return buildTableSchemaCacheValue(dorisTable,
snapshotValue.getSnapshot().getSchemaId(),
- snapshotTable.get()).getPartitionColumns();
+ return getSchemaCacheValue(dorisTable,
snapshotValue.getSnapshot().getSchemaId(),
Review Comment:
Fixed in 19d58d3ebf7. The historical schema keeps the old column name while
the current Iceberg spec can reference the same source field after a rename, so
joining by the current name could omit the partition column. The projection now
joins PartitionField.sourceId to Doris Column.uniqueId, and the schema cache
key also carries the frozen table current schema ID so a warm entry cannot
cross a rename generation. IcebergUtilsTest now exercises historical `p` versus
current `p_renamed`; the focused Iceberg suite passes 119/119.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java:
##########
@@ -173,17 +221,108 @@ protected List<String>
listTableNamesFromRemote(SessionContext ctx, String dbNam
}
@Override
- public void onClose() {
+ public synchronized void onClose() {
+ ThreadPoolExecutor retiredExecutor = threadPoolWithPreAuth;
+ threadPoolWithPreAuth = null;
super.onClose();
- if (null != catalog) {
- try {
- if (catalog instanceof AutoCloseable) {
- ((AutoCloseable) catalog).close();
- }
- catalog = null;
- } catch (Exception e) {
- LOG.warn("Failed to close iceberg catalog: {}", getName(), e);
+ Catalog retiredCatalog = catalog;
+ catalog = null;
+ resourceTracker.retireCurrent(() -> {
+ closeCatalog(retiredCatalog);
+ if (retiredExecutor != null) {
+ ThreadPoolManager.shutdownExecutorService(retiredExecutor);
+ }
+ });
+ }
+
+ @Override
+ public synchronized void resetToUninitialized(boolean invalidCache) {
+ ExternalMetaCacheMgr cacheMgr =
Env.getCurrentEnv().getExtMetaCacheMgr();
+ resetCatalogRuntime(cacheMgr, invalidCache);
+ }
+
+ private void resetCatalogRuntime(ExternalMetaCacheMgr cacheMgr, boolean
invalidCache) {
+ cacheMgr.removeCatalogByEngine(getId(),
IcebergExternalMetaCache.ENGINE);
+ super.resetToUninitialized(invalidCache);
+ }
+
+ private void closeCatalog(Catalog retiredCatalog) {
+ if (retiredCatalog == null) {
+ return;
+ }
+ try {
+ if (retiredCatalog instanceof AutoCloseable) {
+ ((AutoCloseable) retiredCatalog).close();
Review Comment:
Fixed in 19d58d3ebf7. Iceberg HadoopCatalog.close() retires its lock manager
and reporter but does not close the catalog-wide FileIO shared by table
operations, so catalog refreshes could retain one FileIO generation each.
DorisHadoopCatalog now owns that exact FileIO, closes it idempotently after the
catalog closes, and also closes it when initialization fails after FileIO
creation. The Hadoop catalog property is wired to this implementation.
DorisHadoopCatalogTest covers normal repeated close and partial-initialization
failure.
--
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]
