github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3831225091
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java:
##########
@@ -984,21 +984,84 @@ public PlSqlOperation getPlSqlOperation() {
// with "Split source X is released". These executors are finalized when
the next query starts
// on this connection, or when the connection is torn down. See #62259.
private final List<StmtExecutor> flightSqlDeferredExecutors = new
ArrayList<>();
+ private boolean flightSqlDeferredExecutorsSealed;
+ private int flightSqlResultPublishers;
- public void addFlightSqlDeferredExecutor(StmtExecutor executor) {
+ public boolean addFlightSqlDeferredExecutor(StmtExecutor executor) {
synchronized (flightSqlDeferredExecutors) {
+ if (flightSqlDeferredExecutorsSealed) {
+ return false;
+ }
flightSqlDeferredExecutors.add(executor);
+ return true;
+ }
+ }
+
+ /** Linearizes GetFlightInfo publication with the terminal session seal. */
+ public boolean canPublishFlightSqlResult() {
+ synchronized (flightSqlDeferredExecutors) {
+ return !flightSqlDeferredExecutorsSealed;
+ }
+ }
+
+ public boolean beginFlightSqlResultPublication() {
+ synchronized (flightSqlDeferredExecutors) {
+ if (flightSqlDeferredExecutorsSealed) {
+ return false;
+ }
+ flightSqlResultPublishers++;
+ return true;
+ }
+ }
+
+ public void endFlightSqlResultPublication() {
+ List<StmtExecutor> toClose = null;
+ synchronized (flightSqlDeferredExecutors) {
+ if (--flightSqlResultPublishers == 0 &&
flightSqlDeferredExecutorsSealed) {
Review Comment:
[P1] Do not return a FlightInfo after teardown wins publication
There is still a seal window after the producer's final
`canPublishFlightSqlResult()` check: Java evaluates the return expression
before running its `finally`. If teardown seals while this publisher is active,
it defers its drain; this decrement then reaches zero and finalizes the
registered executor, coordinator, and detached Hudi/Iceberg resources, but the
already-evaluated `FlightInfo` is still returned. The client therefore receives
a ticket whose DoGet query is already released. Make publication completion
report when teardown won (or atomically commit publication so teardown cannot
drain it), and add a barrier test that seals after the final check succeeds but
before this decrement.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -105,12 +113,46 @@ public IcebergExternalMetaCache(ExecutorService
refreshExecutor) {
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();
+ }
+ // Background/bootstrap callers without a StatementContext have no
deterministic release boundary.
+ // Load directly instead of borrowing a cache generation that could be
evicted while they use it.
+ return loadTable(nameMapping);
+ }
+
+ /** Returns the executor owned by the exact table generation retained by
this statement. */
+ ThreadPoolExecutor getIcebergTableExecutor(ExternalTable dorisTable) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ IcebergTableCacheValue.Lease lease = statementLease(nameMapping);
+ if (lease == null || lease.getPlanningExecutor() == null) {
+ return dorisTable.getCatalog().getThreadPoolWithPreAuth();
+ }
+ return lease.getPlanningExecutor();
+ }
+
+ /** Runs a bounded metadata operation while retaining the exact table
generation it uses. */
+ <T> T withIcebergTable(ExternalTable dorisTable, Function<Table, T>
action) {
+ NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
+ IcebergTableCacheValue.Lease statementLease =
statementLease(nameMapping);
+ if (statementLease != null) {
+ return action.apply(statementLease.getIcebergTable());
+ }
+ try (IcebergTableCacheValue.Lease operationLease =
borrow(nameMapping)) {
+ return action.apply(operationLease.getIcebergTable());
+ }
}
public IcebergSnapshotCacheValue getSnapshotCache(ExternalTable
dorisTable) {
NameMapping nameMapping = dorisTable.getOrBuildNameMapping();
- return
tableEntry.get(nameMapping.getCtlId()).get(nameMapping).getLatestSnapshotCacheValue();
+ IcebergTableCacheValue.Lease lease = statementLease(nameMapping);
+ if (lease != null) {
+ return lease.getLatestSnapshotCacheValue();
+ }
+ Table table = loadTable(nameMapping);
Review Comment:
[P1] Give background snapshot loads a bounded generation owner
`DictionaryManager` polls Iceberg source versions without a
`ConnectContext`, so every poll reaches this branch. It loads a raw table
outside both `TableLoadContext` and `IcebergTableCacheValue`, then resolves the
planning executor separately. A concurrent reset can therefore close G1 while
the projection still uses it or pair the G1 table with G2's executor, and
per-table FileIO created by recurring direct loads never reaches the cached
value's retirement callback. Run the projection under one exact native/HMS load
lease and close it after deriving non-runtime metadata (or return an explicit
owner), with a no-context polling/reset regression.
--
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]
