github-actions[bot] commented on code in PR #66914:
URL: https://github.com/apache/doris/pull/66914#discussion_r3836067551
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java:
##########
@@ -1005,21 +1005,96 @@ public void clear() {
// 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 boolean endFlightSqlResultPublication() {
+ List<StmtExecutor> toClose = null;
+ boolean published;
+ synchronized (flightSqlDeferredExecutors) {
+ published = !flightSqlDeferredExecutorsSealed;
+ if (--flightSqlResultPublishers == 0 &&
flightSqlDeferredExecutorsSealed) {
+ toClose = drainFlightSqlDeferredExecutors();
+ flightSqlDeferredExecutors.notifyAll();
+ }
}
+ finalizeFlightSqlDeferredExecutors(toClose);
+ return published;
}
public void closeFlightSqlDeferredExecutors() {
- List<StmtExecutor> toClose;
+ closeFlightSqlDeferredExecutors(false);
+ }
+
+ /** Prevents a session teardown race from accepting an executor after the
final drain. */
+ public void sealAndCloseFlightSqlDeferredExecutors() {
+ closeFlightSqlDeferredExecutors(true);
+ }
+
+ private void closeFlightSqlDeferredExecutors(boolean seal) {
+ List<StmtExecutor> toClose = null;
synchronized (flightSqlDeferredExecutors) {
- if (flightSqlDeferredExecutors.isEmpty()) {
- return;
+ if (seal) {
+ flightSqlDeferredExecutorsSealed = true;
+ // The result channel is destroyed immediately after this
method returns. Wait until every
+ // admitted publisher has either committed or observed the
seal, so a losing local-result
+ // publisher cannot insert Arrow buffers after the channel's
one-time invalidation.
+ boolean interrupted = false;
+ while (flightSqlResultPublishers != 0) {
Review Comment:
[P1] Do not wait for the active Flight publisher before canceling its query.
KILL CONNECTION calls FlightSqlConnectContext.kill(true), which enters
unregisterConnection and blocks in this loop; cancelQuery() is ordered only
after unregister returns. A GetFlightInfo request stuck in execution therefore
holds the publisher count while the killer cannot reach the cancellation that
would release it. Publish the seal, cancel the active executor, then wait/drain
and close the channel, with a latch test showing KILL terminates a blocked
publisher.
##########
fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/DorisFlightSqlProducerTest.java:
##########
@@ -207,4 +209,69 @@ public void
testGetFlightInfoFinalizesDeferredExecutorWhenSchemaFetchFails() thr
producer.close();
}
}
+
+ @Test
+ public void
testGetFlightInfoFailsWhenTeardownDrainsRegisteredQueryBeforePublication()
throws Exception {
+ assertTeardownPreventsFlightInfoPublication(true);
+ }
+
+ @Test
+ public void
testGetFlightInfoFailsWhenTeardownSealsBeforeQueryRegistration() throws
Exception {
+ assertTeardownPreventsFlightInfoPublication(false);
+ }
+
+ @Test
+ public void testPublicationCompletionAtomicallyObservesTerminalSeal() {
+ ConnectContext ctx = new ConnectContext();
+ StmtExecutor deferred = Mockito.mock(StmtExecutor.class);
+ Assert.assertTrue(ctx.beginFlightSqlResultPublication());
+ Assert.assertTrue(ctx.addFlightSqlDeferredExecutor(deferred));
+
+ ctx.sealAndCloseFlightSqlDeferredExecutors();
Review Comment:
[P1] Run teardown on a different thread here; this test currently deadlocks
itself. beginFlightSqlResultPublication() increments the only publisher count,
and sealAndCloseFlightSqlDeferredExecutors() waits for that count to reach
zero, but this same thread cannot call endFlightSqlResultPublication() until
the next line. The two helper-backed tests repeat the cycle at line 254 from
the producer's admitted request thread, so all three tests hang. Please
coordinate publisher and teardown with latches, let the publisher reach end,
and join teardown with a bound.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergProcedureOps.java:
##########
@@ -230,6 +249,15 @@ private List<ConnectorRewriteGroup>
planInAuthScope(IcebergTableHandle handle,
return
groups.stream().map(IcebergProcedureOps::toConnectorRewriteGroup).collect(Collectors.toList());
}
+ private <T> T withCatalogLease(Supplier<T> operation) {
+ if (resourceTracker == null) {
+ return operation.get();
+ }
+ try (IcebergCatalogResourceTracker.TrackedResource<T> tracked =
resourceTracker.load(operation)) {
Review Comment:
[P1] Close the procedure table's owned FileIO along with this catalog lease.
Both procedure paths load a raw Table inside operation and discard it, while
this wrapper retains and releases only the catalog generation around the
returned result. Glue/S3Tables therefore leave each per-table S3FileIO open,
and REST tables with a distinct response IO never reach cachedTableCleanup
either; repeated EXECUTE or rewrite planning accumulates those resources until
GC/catalog teardown. Please make the loaded table an operation-owned tracked
resource, run the flavor-aware cleanup in finally on success and failure, and
cover Glue/S3Tables plus REST IO ownership.
##########
fe/fe-core/src/test/java/org/apache/doris/qe/ConnectContextTest.java:
##########
@@ -811,4 +816,37 @@ public void
testCloseFlightSqlDeferredExecutorsFinalizesRemainingWhenOneFails()
Mockito.verify(failing, Mockito.times(1)).finalizeArrowFlightQuery();
Mockito.verify(healthy, Mockito.times(1)).finalizeArrowFlightQuery();
}
+
+ @Test
+ public void testSessionTeardownRejectsLateDeferredExecutorRegistration() {
+ ConnectContext ctx = new ConnectContext();
+ StmtExecutor late = Mockito.mock(StmtExecutor.class);
+
+ ctx.sealAndCloseFlightSqlDeferredExecutors();
+
+ Assert.assertFalse("an executor registered after the final teardown
drain must not become unreachable",
+ ctx.addFlightSqlDeferredExecutor(late));
+ ctx.closeFlightSqlDeferredExecutors();
+ Mockito.verifyNoInteractions(late);
+ }
+
+ @Test
+ public void testSessionSealWaitsForAdmittedResultPublisher() throws
Exception {
+ ConnectContext ctx = new ConnectContext();
+ Assert.assertTrue(ctx.beginFlightSqlResultPublication());
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> teardown =
executor.submit(ctx::sealAndCloseFlightSqlDeferredExecutors);
+ try {
+ teardown.get(100, TimeUnit.MILLISECONDS);
Review Comment:
[P2] Add a barrier proving the teardown task entered the seal before
treating this timeout as evidence. On a loaded CI executor the submitted task
may remain unscheduled for 100 ms; the test then catches TimeoutException, ends
publication, and the task starts afterward and returns immediately, so it
passes without exercising the wait at all. Coordinate method entry/observed
seal with a latch or test seam, then release the publisher and join both sides.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -2319,6 +2368,13 @@ private static int getFormatVersion(Table table) {
return formatVersion;
}
+ private <T> T executeAuthenticated(Callable<T> operation) throws Exception
{
+ if (resourceTracker == null) {
+ return context.executeAuthenticated(operation);
+ }
+ return resourceTracker.call(() ->
context.executeAuthenticated(operation));
Review Comment:
[P1] Retire table-owned FileIOs for the DDL operations guarded here.
resourceTracker.call() retains only the catalog generation and cannot see
Tables loaded or created inside an operation that returns null. Column/schema,
branch/tag, and partition DDL all load a raw Table in IcebergCatalogOps and
discard it after commit, and createTable discards the SDK's returned Table;
Glue/S3Tables per-table IO and distinct REST table IO therefore accumulate
until GC/catalog teardown. Give these operations a table owner with
flavor-aware cleanup in finally, including success and post-load/commit-failure
tests.
--
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]
