924060929 commented on code in PR #66914: URL: https://github.com/apache/doris/pull/66914#discussion_r3838741640
########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogResourceTracker.java: ########## @@ -0,0 +1,196 @@ +// 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.connector.iceberg; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +/** Keeps catalog generations alive while cached tables loaded through them still have borrowers. */ +final class IcebergCatalogResourceTracker { + + private final List<Generation> generations = new ArrayList<>(); + private Generation current = new Generation(); + private int loadsInProgress; + private boolean closed; + + IcebergCatalogResourceTracker() { + generations.add(current); + } + + synchronized LoadGuard beginLoad() { + if (closed) { + throw new IllegalStateException("Iceberg catalog resources are already closed"); + } + loadsInProgress++; + current.retain(); + return new LoadGuard(this, generations.size() - 1, current); + } + + /** Atomically rotates the resource generation with publication of the corresponding REST delegate. */ + synchronized void rotate(Runnable retiredCleanup, Runnable publishReplacement) { + if (closed) { + throw new IllegalStateException("Iceberg catalog resources are already closed"); + } + Generation retired = current; + current = new Generation(); + current.retain(loadsInProgress); + generations.add(current); + publishReplacement.run(); + retired.retire(retiredCleanup); + } + + synchronized void close(Runnable currentCleanup) { + if (closed) { + return; + } + closed = true; + current.retire(currentCleanup); + } + + private synchronized ResourceLease promote(int firstGeneration) { + List<Generation> retained = new ArrayList<>(); + retained.add(generations.get(firstGeneration)); + for (int i = firstGeneration + 1; i < generations.size(); i++) { + // rotate() pre-retains one reference in every new generation for each load already in progress. + retained.add(generations.get(i)); + } + completeLoad(); + return new ResourceLease(retained); + } + + private synchronized void abortLoad(int firstGeneration) { + for (int i = firstGeneration + 1; i < generations.size(); i++) { + generations.get(i).release(); + } + completeLoad(); + } + + private void completeLoad() { + loadsInProgress--; + if (loadsInProgress < 0) { + throw new IllegalStateException("Iceberg catalog load guard completed too many times"); + } + } + + static final class LoadGuard implements AutoCloseable { + private final IcebergCatalogResourceTracker tracker; + private final int firstGeneration; + private final Generation initialGeneration; + private final AtomicBoolean transferred = new AtomicBoolean(); + + private LoadGuard(IcebergCatalogResourceTracker tracker, int firstGeneration, Generation initialGeneration) { + this.tracker = tracker; + this.firstGeneration = firstGeneration; + this.initialGeneration = initialGeneration; + } + + ResourceLease promote() { + if (!transferred.compareAndSet(false, true)) { + throw new IllegalStateException("Iceberg catalog load guard was already completed"); + } + return tracker.promote(firstGeneration); + } + + @Override + public void close() { + if (transferred.compareAndSet(false, true)) { + try { + initialGeneration.release(); + } finally { + tracker.abortLoad(firstGeneration); + } + } + } + } + + static final class ResourceLease implements AutoCloseable { + private final List<Generation> generations; + private final AtomicBoolean closed = new AtomicBoolean(); + + private ResourceLease(List<Generation> generations) { + this.generations = generations; + } + + @Override + public void close() { + if (closed.compareAndSet(false, true)) { + RuntimeException failure = null; + for (Generation generation : generations) { + try { + generation.release(); + } catch (RuntimeException e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + } + if (failure != null) { + throw failure; + } + } + } + } + + private static final class Generation { + private int references; + private boolean retired; + private boolean cleaned; + private Runnable cleanup; + + private synchronized void retain() { + if (cleaned) { + throw new IllegalStateException("Iceberg catalog generation was already cleaned"); + } + references++; + } + + private synchronized void retain(int count) { + if (cleaned) { + throw new IllegalStateException("Iceberg catalog generation was already cleaned"); + } + references += count; + } + + private synchronized void retire(Runnable cleanup) { + if (retired) { + return; + } + retired = true; + this.cleanup = cleanup; + maybeCleanup(); + } + + private synchronized void release() { + if (references <= 0) { + throw new IllegalStateException("Iceberg catalog generation released too many times"); + } + references--; + maybeCleanup(); + } + + private void maybeCleanup() { + if (retired && references == 0 && !cleaned) { + cleaned = true; + cleanup.run(); Review Comment: 已修复。retired generation 的 cleanup 在 finally 中清空，完成后不再保留旧 REST delegate/object graph；重复 rotation 测试已覆盖。 ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/ReauthenticatingRestSessionCatalog.java: ########## @@ -124,9 +134,22 @@ private synchronized void reauthenticate(RESTSessionCatalog attemptedOn, Runtime + "then retrying the request once.", name(), cause); RESTSessionCatalog replacement = delegateBuilder.get(); RESTSessionCatalog wedged = delegate; - delegate = replacement; + if (resourceTracker == null) { + delegate = replacement; + closeReplacedDelegate(wedged); + return; + } + try { + invalidateTables.run(); + } catch (RuntimeException e) { + LOG.warn("Failed to retire Iceberg table cache before replacing REST client of catalog {}", name(), e); + } + resourceTracker.rotate(() -> closeReplacedDelegate(wedged), () -> delegate = replacement); Review Comment: 已修复。replacement build 与 tracker acceptance 分离；close/rotation 竞态中未发布的 replacement 会被直接关闭，同时保留原始 401 结果。 ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergStatementScope.java: ########## @@ -83,13 +83,97 @@ static Table sharedTable(ConnectorSession session, String dbName, String tableNa () -> snapshotReadTable(loader.get())); } + /** + * Statement-scoped variant for a table borrowed from {@link IcebergTableCache}. The memoized holder is + * {@link AutoCloseable}, so the engine's statement-scope teardown releases the borrower only after scan + * pumps have quiesced. Cache eviction and statement completion may happen in either order; the underlying + * FileIO is closed only after both owners release it. + */ + static Table sharedBorrowedTable(ConnectorSession session, String dbName, String tableName, + Supplier<IcebergTableCache.TableLease> loader, Supplier<Table> unscopedLoader) { + if (session == null || session.getStatementScope() == ConnectorStatementScope.NONE) { + // NONE has no statement-end callback, so it cannot safely own a lease. Preserve its original direct + // load-every-time behavior; creating a lease here would drop its only close handle and leak forever. + return snapshotReadTable(unscopedLoader.get()); + } + ScopedBorrow borrowed = ConnectorStatementScopes.resolveInStatement( + session, TABLE_NAMESPACE, dbName, tableName, () -> new ScopedBorrow(loader.get())); + return borrowed.table; + } + + /** Statement-owned direct table for credential-dependent catalogs where cross-query caching is disabled. */ + static Table sharedTrackedTable(ConnectorSession session, String dbName, String tableName, + IcebergCatalogResourceTracker resourceTracker, Supplier<Table> loader) { + if (session == null || session.getStatementScope() == ConnectorStatementScope.NONE) { + return snapshotReadTable(loader.get()); + } + TrackedTable tracked = ConnectorStatementScopes.resolveInStatement( + session, TABLE_NAMESPACE, dbName, tableName, + () -> new TrackedTable(resourceTracker.load(loader), true)); + return tracked.table(); + } + + private static final class ScopedBorrow implements AutoCloseable { + private final IcebergTableCache.TableLease lease; + private final Table table; + + private ScopedBorrow(IcebergTableCache.TableLease lease) { + this.lease = lease; + this.table = snapshotReadTable(lease.table()); + } + + @Override + public void close() { + lease.close(); + } + } + /** Loads the mutable table used only by write planning and transaction creation. */ static Table sharedWritableTable( ConnectorSession session, String dbName, String tableName, Supplier<Table> loader) { return ConnectorStatementScopes.resolveInStatement( session, WRITABLE_TABLE_NAMESPACE, dbName, tableName, loader); } + /** Mutable table paired with the exact catalog generation that produced it. */ + static TrackedTable sharedTrackedWritableTable(ConnectorSession session, String dbName, String tableName, + IcebergCatalogResourceTracker resourceTracker, Supplier<Table> loader) { + if (session == null || session.getStatementScope() == ConnectorStatementScope.NONE) { + return new TrackedTable(resourceTracker.load(loader), false); + } + return ConnectorStatementScopes.resolveInStatement( + session, WRITABLE_TABLE_NAMESPACE, dbName, tableName, + () -> new TrackedTable(resourceTracker.load(loader), true)); + } + + static final class TrackedTable implements AutoCloseable { + private final IcebergCatalogResourceTracker.TrackedResource<Table> tracked; + private final boolean statementOwned; + + private TrackedTable(IcebergCatalogResourceTracker.TrackedResource<Table> tracked, + boolean statementOwned) { + this.tracked = tracked; + this.statementOwned = statementOwned; + } + + Table table() { + return tracked.resource(); + } + + IcebergCatalogResourceTracker.ResourceLease retainLease() { + return tracked.retainLease(); + } + + boolean isStatementOwned() { + return statementOwned; + } + + @Override + public void close() { + tracked.close(); Review Comment: 已修复。direct TrackedTable 同时持有 catalog generation 和 table-owned FileIO cleanup，最后一个 statement/transaction owner 释放后才执行 flavor-aware close。 ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorTransaction.java: ########## @@ -1335,7 +1369,11 @@ public void rollback() { @Override public void close() { - // No resources to release: the SDK transaction holds no connections of its own. + IcebergStatementScope.TrackedTableLease lease = tableLease; Review Comment: 已修复。beginWrite publication 与 close 由 beginLock/terminal state 线性化；close 会阻止晚到 begin，竞争失败的 lease 在本地释放。 -- 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]
