This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a72dcaa17b1 [fix](fe) Keep an ADBC catalog's driver open until the
calls still using it finish (#69874)
a72dcaa17b1 is described below
commit a72dcaa17b127265a5d713d811157d638d5f6fde
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sat Oct 31 08:44:19 2026 +0000
[fix](fe) Keep an ADBC catalog's driver open until the calls still using it
finish (#69874)
### What problem does this PR solve?
Issue Number: None
Related PR: #66331, #68353
Problem Summary:
**In short:** dropping or altering an ADBC catalog can crash the FE
process. Closing the catalog frees the native ADBC objects at once,
including the connections and results that other threads are still
reading, such as the column-statistics loads that a query on the catalog
has just started. This kills the FE in about one External regression run
in 75, whatever the PR under test changes. With this PR, closing a
client only refuses new calls, and the last call still in flight
releases the driver as it leaves. DROP and ALTER CATALOG still return
without waiting.
This commit was first pushed as the last commit of #68353, which is
still in review. It shares nothing with the rest of that PR, so it is
carved out here to land on its own.
#### Background
An ADBC catalog (#66331) reads its metadata through `AdbcClient`, which
owns one native `AdbcDatabase` and one Arrow allocator per catalog.
Every metadata call borrows a fresh connection through
`AdbcClient.withConnection` and closes it afterwards. FE talks to the C
driver through adbc-driver-jni 0.24. In that bridge, closing a parent
closes everything still open under it, on the closing thread:
`JniDatabase.close()` closes every connection still open on the
database, and `JniConnection.close()` closes every result (`getObjects`
and similar readers) still open on the connection. The bridge only
checks a handle before a call starts, and the C driver does not check at
all, so freeing an object that another thread is still using is a
use-after-free in native code, which takes the whole JVM down rather
than failing one call.
Column statistics reach the driver too. When a query plans against an
ADBC table, the STATS_FETCH pool loads column statistics in the
background. A load that finds nothing in the statistics table asks the
table itself: `PluginDrivenExternalTable.getColumnStatistic` →
`AdbcConnectorMetadata.getTableHandle` → `withConnection` →
`getObjects`. With `"meta.cache.adbc.metadata.enable" = "false"`, every
such load calls into the driver.
#### The problem, and what it cost
- `AdbcClient.close()` closed the database and then the allocator
immediately, whatever was still in flight. DROP CATALOG and ALTER
CATALOG ... SET PROPERTIES close the catalog's connector this way, on
their own thread (since #69665, on the `ExternalResourceReclaimer` pool;
before that, synchronously on the statement thread, with the same race).
- A statistics load that was reading a `getObjects` result, or was
inside the driver on its connection, then read memory that had just been
freed. Where it dies depends on where that thread is at that moment.
- In CI, the External regression pipeline lost its FE in 11 of the 834
runs between 2026-10-22 and 2026-10-30. Each time, 54 to 56 suites then
failed with `Communications link failure`, so the PR under test showed a
red run it did not cause. All 11 died at the same moment:
`test_adbc_sqlite_catalog_scan` drops its catalog with the metadata
cache turned off right after querying it. In the same millisecond,
fe.log shows the allocator closing with `Memory was leaked by query` and
a statistics load failing with `Failed to read the ADBC getObjects
result: ArrowArrayStream is already closed`. That is the same race,
caught by a thread that came back only after the close had finished.
| FE crash in `fe.out` | Runs |
| --- | --- |
| `SIGSEGV` in
`nanoarrow::ArrayStreamFactory<VectorArrayStream>::get_next_wrapper+0x4`
| 6 |
| `SIGSEGV` in `PrivateArrowSchemaDeepCopy+0x46` | 2 |
| glibc `double free or corruption (fasttop)` | 2 |
| no message | 1 |
| `SIGSEGV` in `sqlite3FindTable` (an earlier run, 2026-10-06) | 1 |
- Users can hit this too, not only CI: DROP or ALTER an ADBC catalog
while a statement or a statistics load is still reading it.
#### How this PR fixes it
- `AdbcClient` counts the calls inside `withConnection` (`enter` /
`leave`, under the client's monitor, which also opens the database on
first use).
- `close()` marks the client closed, so new calls are refused with
`AdbcClient has been closed`. It releases the database and the allocator
itself only when no call is in flight. Otherwise it returns at once, and
the last call to leave releases them.
- A deferred release that fails is logged with the driver's path, as the
catalog already logs a failed connector close. It is not thrown at that
last caller, whose own work succeeded and which did not close anything.
- `close()` never waits for the calls in flight, so DROP and ALTER
CATALOG are not held up by a slow remote source.
#### Results
Unit tests (0 skipped). The first case runs against the real JNI bridge
and the SQLite driver, the others against a recording database:
| `AdbcClientTest` case | `close()` releasing at once, as before this PR
| This PR |
| --- | --- | --- |
| a call still on its connection when the client is closed keeps using
it | fails: `Native connection handle is closed` | passes |
| a call still reading a `getObjects` result when the client is closed
reads it to the end | fails: `the database was released under an open
connection` | passes |
| the last call to leave a closed client releases the database exactly
once, after it leaves | fails: `the database was released under an open
connection` | passes |
| a release that fails as the last call leaves does not fail that call |
fails: `the database was released under an open connection` | passes |
| an idle client releases at once | passes | passes |
Freeing only the allocator at once, and leaving the database to the last
call, fails the `getObjects` case alone: `Attempting operation on
allocator when allocator is closed`.
A stress program (not committed) closes a client 200 times while four
threads loop over `getObjects` and `getTableSchema` through
`withConnection` against the SQLite driver:
| | `AdbcClient` of master | This PR |
| --- | --- | --- |
| Runs | 5 | 5 |
| JVM crashed | 5 of 5, 1.1 to 1.3 s after the JVM started: `SIGSEGV` in
`PrivateArrowSchemaDeepCopy+0x40`, under
`ArrowArrayStreamReader.readSchema` ← `AdbcObjectsReader.readNamespaces`
← `AdbcClient.withConnection` | 0 of 5 |
| Closes completed per run | crashed before finishing | 200 |
| Metadata calls per run | crashed before reporting | 6,019 to 7,238 |
| Calls the close reached while they were using the driver | | 0 |
| Failed closes, failed deferred releases | | 0, 0 (checked with WARN
logging on, where the unit test's deliberate release failure shows up as
one warning) |
The stress runs used the macOS arm64 build of the same SQLite driver, so
the offset differs from the Linux one in CI.
#### Classes, and how they call each other
- `AdbcClient`: `withConnection` brackets the body with `enter` /
`leave`; `close` refuses new calls and releases only when idle;
`release` closes the database, then the allocator; `openDatabase` is the
one place the database is created, so a test can stand a recording
database in for the driver.
- `AdbcClientTest`: five new cases, one against the real driver and four
against a recording database. The recording connection answers
`getObjects` with a result built in memory in the client's allocator:
reading a driver's result into Arrow loads arrow-c-data's JNI shim,
which needs CXXABI_1.3.9 and does not load on the FE UT agent (see
`AdbcNativeTestSupport`, #68510).
- `AdbcObjectsReaderTest`: `response`, `catalog` and `schema` are
package-private, so `AdbcClientTest` builds that result with them.
```
STATS_FETCH: ColumnStatisticsCacheLoader.doLoad
-> PluginDrivenExternalTable.getColumnStatistic
-> AdbcConnectorMetadata.getTableHandle (namespaces, table names)
-> AdbcClient.withConnection
enter() refused once closed; callsInFlight++
-> AdbcConnection.getObjects ->
AdbcObjectsReader.readNamespaces / readTableNames
leave() callsInFlight--; the last one out of a closed client
-> release()
DROP CATALOG / ALTER CATALOG ... SET PROPERTIES
-> ExternalCatalog.onClose / resetToUninitialized ->
ExternalResourceReclaimer
-> AdbcConnector.close -> AdbcClient.close()
closed = true; release() now only if no call is in flight
release(): AdbcDatabase.close (closes its connections and their
results) -> allocator.close
```
Not changed: `AdbcConnector`, the catalog lifecycle and
`ExternalResourceReclaimer`, BE, and the regression suites;
`test_adbc_sqlite_catalog_scan` stays as it is, since what it exposes is
this bug.
---
.../apache/doris/connector/adbc/AdbcClient.java | 131 ++++++++----
.../doris/connector/adbc/AdbcClientTest.java | 229 ++++++++++++++++++++-
.../connector/adbc/AdbcObjectsReaderTest.java | 9 +-
3 files changed, 325 insertions(+), 44 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcClient.java
b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcClient.java
index d19d18b31a4..e5c3f7e7a66 100644
---
a/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcClient.java
+++
b/fe/fe-connector/fe-connector-adbc/src/main/java/org/apache/doris/connector/adbc/AdbcClient.java
@@ -26,6 +26,8 @@ import org.apache.arrow.adbc.core.AdbcException;
import org.apache.arrow.adbc.driver.jni.JniDriver;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import java.io.Closeable;
import java.nio.file.Path;
@@ -48,9 +50,20 @@ import java.util.Map;
* constructor: an FE follower replaying the edit log builds every catalog,
and its filesystem layout need
* not match the leader's, so a missing driver file would otherwise stop FE
from starting instead of
* failing the one catalog that cannot work.
+ *
+ * <p><b>Closing never frees what a call is still using.</b> A catalog is
dropped or altered on one thread
+ * while others may still be inside {@link #withConnection} for it -- a
statistics loader, a statement being
+ * planned. Closing the {@code AdbcDatabase} closes every connection still
open on it as well (the JNI driver
+ * closes its children first), and native driver state freed under a thread
that is inside the driver
+ * crashes the whole FE process rather than failing that one call. So {@link
#close} only refuses new calls,
+ * and the native objects are released by whoever leaves last: {@code close}
itself when no call is in
+ * flight, otherwise the last call to finish. Waiting for those calls inside
{@code close} instead would hold
+ * DROP and ALTER CATALOG for as long as a remote source takes to answer.
*/
public class AdbcClient implements Closeable {
+ private static final Logger LOG = LogManager.getLogger(AdbcClient.class);
+
private final Path driverPath;
private final String driverUrl;
private final String entrypoint;
@@ -59,9 +72,12 @@ public class AdbcClient implements Closeable {
private final String password;
private final Map<String, String> driverOptions;
- private volatile BufferAllocator allocator;
- private volatile AdbcDatabase database;
- private volatile boolean closed;
+ // All guarded by this.
+ private BufferAllocator allocator;
+ private AdbcDatabase database;
+ private boolean closed;
+ /** Calls inside {@link #withConnection}; the last one to leave a closed
client releases the driver. */
+ private int callsInFlight;
public AdbcClient(Path driverPath, String driverUrl, String entrypoint,
String uri,
String user, String password, Map<String, String> driverOptions) {
@@ -83,7 +99,7 @@ public class AdbcClient implements Closeable {
* cost does not show.
*/
public <T> T withConnection(AdbcConnectionCall<T> body) {
- AdbcDatabase db = getOrOpenDatabase();
+ AdbcDatabase db = enter();
ClassLoader previous = Thread.currentThread().getContextClassLoader();
try {
// The plugin loads child-first and the ADBC/Arrow classes live in
it; pin the context
@@ -101,43 +117,60 @@ public class AdbcClient implements Closeable {
throw new DorisConnectorException("ADBC operation failed: " +
e.getMessage(), e);
} finally {
Thread.currentThread().setContextClassLoader(previous);
+ leave();
}
}
- private AdbcDatabase getOrOpenDatabase() {
+ /** Counts the calling thread in, opening the database on first use. */
+ private synchronized AdbcDatabase enter() {
if (closed) {
throw new DorisConnectorException("AdbcClient has been closed");
}
- AdbcDatabase db = database;
- if (db != null) {
- return db;
+ if (database == null) {
+ AdbcDriverPathResolver.checkExists(driverPath, driverUrl);
+ ClassLoader previous =
Thread.currentThread().getContextClassLoader();
+ try {
+
Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
+ if (allocator == null) {
+ allocator = new RootAllocator();
+ }
+ database = openDatabase(allocator, buildParameters());
+ } catch (AdbcException e) {
+ throw translate(e, "Failed to open the ADBC driver " +
driverPath);
+ } catch (UnsatisfiedLinkError e) {
+ throw new DorisConnectorException("Failed to load the ADBC JNI
bridge."
+ + " FE loads it from the directory named by the JVM
property"
+ + " arrow.adbc.driver.jni.library.path (set in
fe.conf, normally"
+ + " ${DORIS_HOME}/lib): " + e.getMessage(), e);
+ } finally {
+ Thread.currentThread().setContextClassLoader(previous);
+ }
}
+ callsInFlight++;
+ return database;
+ }
+
+ /** Counts the calling thread out; the last call to leave a closed client
releases the driver. */
+ private void leave() {
synchronized (this) {
- if (closed) {
- throw new DorisConnectorException("AdbcClient has been
closed");
+ callsInFlight--;
+ if (!closed || callsInFlight > 0) {
+ return;
}
- if (database == null) {
- AdbcDriverPathResolver.checkExists(driverPath, driverUrl);
- ClassLoader previous =
Thread.currentThread().getContextClassLoader();
- try {
-
Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
- if (allocator == null) {
- allocator = new RootAllocator();
- }
- database = new
JniDriver(allocator).open(buildParameters());
- } catch (AdbcException e) {
- throw translate(e, "Failed to open the ADBC driver " +
driverPath);
- } catch (UnsatisfiedLinkError e) {
- throw new DorisConnectorException("Failed to load the ADBC
JNI bridge."
- + " FE loads it from the directory named by the
JVM property"
- + " arrow.adbc.driver.jni.library.path (set in
fe.conf, normally"
- + " ${DORIS_HOME}/lib): " + e.getMessage(), e);
- } finally {
- Thread.currentThread().setContextClassLoader(previous);
- }
- }
- return database;
}
+ // close() found this call still inside the driver and left the
release to it. A failure is logged, as
+ // the catalog logs a failed close
(PluginDrivenExternalCatalog#closeResources), not thrown: the call
+ // itself is done, and its caller is not the one that closed the
catalog.
+ try {
+ release();
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to release the ADBC driver {} of a closed
catalog", driverPath, e);
+ }
+ }
+
+ /** The one place the database is created, so a test can stand a recording
one in for the driver. */
+ AdbcDatabase openDatabase(BufferAllocator allocator, Map<String, Object>
parameters) throws AdbcException {
+ return new JniDriver(allocator).open(parameters);
}
private Map<String, Object> buildParameters() {
@@ -184,20 +217,40 @@ public class AdbcClient implements Closeable {
return new DorisConnectorException(sb.toString(), e);
}
+ /**
+ * Refuses new calls, and releases the driver now unless a call is still
inside it -- then the last such
+ * call releases it as it leaves. Does not wait for those calls; see the
class comment.
+ */
@Override
- public synchronized void close() {
- closed = true;
+ public void close() {
+ synchronized (this) {
+ closed = true;
+ if (callsInFlight > 0) {
+ return;
+ }
+ }
+ release();
+ }
+
+ /** Frees the database and then its allocator; a client never used, or
already released, holds neither. */
+ private void release() {
+ AdbcDatabase db;
+ BufferAllocator alloc;
+ synchronized (this) {
+ db = database;
+ alloc = allocator;
+ database = null;
+ allocator = null;
+ }
try {
- if (database != null) {
- database.close();
+ if (db != null) {
+ db.close();
}
} catch (Exception e) {
throw new DorisConnectorException("Failed to close the ADBC
database: " + e.getMessage(), e);
} finally {
- database = null;
- if (allocator != null) {
- allocator.close();
- allocator = null;
+ if (alloc != null) {
+ alloc.close();
}
}
}
diff --git
a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcClientTest.java
b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcClientTest.java
index 05583d94192..f87978c97c0 100644
---
a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcClientTest.java
+++
b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcClientTest.java
@@ -19,18 +19,35 @@ package org.apache.doris.connector.adbc;
import org.apache.doris.connector.spi.DorisConnectorException;
+import org.apache.arrow.adbc.core.AdbcConnection;
+import org.apache.arrow.adbc.core.AdbcDatabase;
import org.apache.arrow.adbc.core.AdbcException;
+import org.apache.arrow.adbc.core.AdbcStatement;
import org.apache.arrow.adbc.core.AdbcStatusCode;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.ipc.ArrowReader;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Exercises {@link AdbcClient} against the real JNI bridge and the real
SQLite driver from thirdparty.
- * Skips (loudly) when those libraries are absent -- see {@link
AdbcNativeTestSupport}.
+ * Skips (loudly) when those libraries are absent -- see {@link
AdbcNativeTestSupport}. The tests of when the
+ * driver is released stand a {@link RecordingClient} in for it instead, so
they run everywhere.
*/
class AdbcClientTest {
@@ -74,6 +91,135 @@ class AdbcClientTest {
Assertions.assertTrue(e.getMessage().contains("closed"),
e.getMessage());
}
+ @Test
+ void closeLeavesACallThatIsStillUsingTheDriverAlone(@TempDir Path tempDir)
throws Exception {
+ AdbcClient client = sqliteClient(tempDir.resolve("probe.db"));
+ CountDownLatch insideCall = new CountDownLatch(1);
+ CountDownLatch mayContinue = new CountDownLatch(1);
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ try {
+ Future<String> call = caller.submit(() ->
client.withConnection(connection -> {
+ insideCall.countDown();
+ Assertions.assertTrue(mayContinue.await(30, TimeUnit.SECONDS));
+ // Back into the driver after the client was closed, as a
statistics loader is when the catalog
+ // it reads is dropped under it. Closing the database used to
close this connection too; here
+ // the JNI handle check turns that into an error, but a thread
already inside the driver read
+ // freed memory and took FE down with it (SIGSEGV in
sqlite3FindTable).
+ return connection.getCurrentCatalog();
+ }));
+ Assertions.assertTrue(insideCall.await(30, TimeUnit.SECONDS));
+ closeWithoutWaitingForCalls(client);
+ mayContinue.countDown();
+ Assertions.assertEquals("main", call.get(30, TimeUnit.SECONDS));
+ } finally {
+ caller.shutdownNow();
+ }
+ }
+
+ @Test
+ void closeLeavesAResultACallIsStillReadingAlone(@TempDir Path tempDir)
throws Exception {
+ RecordingClient client = new
RecordingClient(Files.createFile(tempDir.resolve("driver.so")));
+ CountDownLatch insideCall = new CountDownLatch(1);
+ CountDownLatch mayContinue = new CountDownLatch(1);
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ try {
+ Future<List<AdbcNamespace>> call = caller.submit(() ->
client.withConnection(connection -> {
+ try (ArrowReader reader = connection.getObjects(
+ AdbcConnection.GetObjectsDepth.DB_SCHEMAS, null, null,
null, null, null)) {
+ insideCall.countDown();
+ Assertions.assertTrue(mayContinue.await(30,
TimeUnit.SECONDS));
+ // The getObjects result behind getTableHandle, read after
the client was closed, as a
+ // statistics load reads it when the catalog is dropped
under it. Closing the client used to
+ // release the database, and with it this connection and
its result, and then the allocator the
+ // result lives in; with the SQLite driver, a read caught
in between took FE down (SIGSEGV in
+ // get_next_wrapper or PrivateArrowSchemaDeepCopy).
+ return AdbcObjectsReader.readNamespaces(reader);
+ }
+ }));
+ Assertions.assertTrue(insideCall.await(30, TimeUnit.SECONDS));
+ closeWithoutWaitingForCalls(client);
+ mayContinue.countDown();
+ Assertions.assertEquals(List.of(new AdbcNamespace("main", "")),
call.get(30, TimeUnit.SECONDS));
+ } finally {
+ caller.shutdownNow();
+ }
+ }
+
+ @Test
+ void theLastCallToLeaveAClosedClientReleasesTheDatabase(@TempDir Path
tempDir) throws Exception {
+ RecordingClient client = new
RecordingClient(Files.createFile(tempDir.resolve("driver.so")));
+ CountDownLatch insideCall = new CountDownLatch(1);
+ CountDownLatch mayLeave = new CountDownLatch(1);
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ try {
+ Future<String> call = caller.submit(() ->
client.withConnection(connection -> {
+ insideCall.countDown();
+ Assertions.assertTrue(mayLeave.await(30, TimeUnit.SECONDS));
+ return "done";
+ }));
+ Assertions.assertTrue(insideCall.await(30, TimeUnit.SECONDS));
+
+ closeWithoutWaitingForCalls(client);
+ // Refused from now on, yet nothing is released while a call is
still inside.
+ DorisConnectorException e =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> client.withConnection(connection -> "late"));
+ Assertions.assertTrue(e.getMessage().contains("closed"),
e.getMessage());
+ Assertions.assertEquals(0, client.database.closes.get());
+
+ mayLeave.countDown();
+ Assertions.assertEquals("done", call.get(30, TimeUnit.SECONDS));
+ Assertions.assertEquals(1, client.database.closes.get());
+ } finally {
+ caller.shutdownNow();
+ }
+ client.close();
+ Assertions.assertEquals(1, client.database.closes.get());
+ }
+
+ @Test
+ void releaseFailingAsTheLastCallLeavesIsNotThatCallsFailure(@TempDir Path
tempDir) throws Exception {
+ RecordingClient client = new
RecordingClient(Files.createFile(tempDir.resolve("driver.so")),
+ new RecordingDatabase(true));
+ CountDownLatch insideCall = new CountDownLatch(1);
+ CountDownLatch mayLeave = new CountDownLatch(1);
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ try {
+ Future<String> call = caller.submit(() ->
client.withConnection(connection -> {
+ insideCall.countDown();
+ Assertions.assertTrue(mayLeave.await(30, TimeUnit.SECONDS));
+ return "done";
+ }));
+ Assertions.assertTrue(insideCall.await(30, TimeUnit.SECONDS));
+ closeWithoutWaitingForCalls(client);
+
+ mayLeave.countDown();
+ // The call did its work; the driver failing to close as it leaves
is logged, not thrown at a caller that
+ // did not close the catalog - a statistics load or a statement
that happened to be the last one in.
+ Assertions.assertEquals("done", call.get(30, TimeUnit.SECONDS));
+ Assertions.assertEquals(1, client.database.closes.get());
+ } finally {
+ caller.shutdownNow();
+ }
+ // Released, if unsuccessfully: nothing is closed again.
+ client.close();
+ Assertions.assertEquals(1, client.database.closes.get());
+ }
+
+ @Test
+ void closingAClientNoCallIsUsingReleasesTheDatabaseAtOnce(@TempDir Path
tempDir) throws Exception {
+ RecordingClient client = new
RecordingClient(Files.createFile(tempDir.resolve("driver.so")));
+ Assertions.assertEquals("done", client.withConnection(connection ->
"done"));
+ Assertions.assertEquals(0, client.database.closes.get());
+
+ client.close();
+ Assertions.assertEquals(1, client.database.closes.get());
+ }
+
+ /** Fails instead of hanging the run if close() ever starts waiting for
the calls still in flight. */
+ private static void closeWithoutWaitingForCalls(AdbcClient client) throws
Exception {
+ CompletableFuture.runAsync(client::close).get(30, TimeUnit.SECONDS);
+ }
+
@Test
void missingDriverFileFailsAtFirstUseNotAtConstruction(@TempDir Path
tempDir) {
// An FE follower replaying the edit log constructs every catalog; if
construction reached the
@@ -130,4 +276,85 @@ class AdbcClientTest {
Assertions.assertTrue(message.contains("42P01"), message);
Assertions.assertTrue(message.contains("7"), message);
}
+
+ /** An {@link AdbcClient} whose database records how it is used instead of
reaching a driver. */
+ private static final class RecordingClient extends AdbcClient {
+
+ private final RecordingDatabase database;
+
+ private RecordingClient(Path driver) {
+ this(driver, new RecordingDatabase(false));
+ }
+
+ private RecordingClient(Path driver, RecordingDatabase database) {
+ super(driver, driver.toString(), null, "file:/unused.db", null,
null, Map.of());
+ this.database = database;
+ }
+
+ @Override
+ AdbcDatabase openDatabase(BufferAllocator allocator, Map<String,
Object> parameters) {
+ database.allocator = allocator;
+ return database;
+ }
+ }
+
+ private static final class RecordingDatabase implements AdbcDatabase {
+
+ private final AtomicInteger openConnections = new AtomicInteger();
+ private final AtomicInteger closes = new AtomicInteger();
+ private final boolean closeFails;
+ /** The client's allocator, handed over when the client opens its
database, as the JNI driver is given it. */
+ private BufferAllocator allocator;
+
+ private RecordingDatabase(boolean closeFails) {
+ this.closeFails = closeFails;
+ }
+
+ @Override
+ public AdbcConnection connect() {
+ openConnections.incrementAndGet();
+ return new AdbcConnection() {
+ @Override
+ public AdbcStatement createStatement() {
+ throw new AssertionError("these tests run no statement");
+ }
+
+ @Override
+ public ArrowReader getInfo(int[] infoCodes) {
+ throw new AssertionError("these tests ask for no info");
+ }
+
+ @Override
+ public ArrowReader getObjects(AdbcConnection.GetObjectsDepth
depth, String catalogPattern,
+ String dbSchemaPattern, String tableNamePattern,
String[] tableTypes,
+ String columnNamePattern) {
+ // SQLite's answer, in the client's allocator, which is
where a driver's result lives. Not read
+ // from the SQLite driver itself: reading a driver's
result into Arrow needs arrow-c-data's JNI
+ // shim, which FE UT cannot load on every host -- see
AdbcNativeTestSupport.
+ try {
+ return AdbcObjectsReaderTest.response(allocator,
+ AdbcObjectsReaderTest.catalog("main",
AdbcObjectsReaderTest.schema("", null)));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public void close() {
+ openConnections.decrementAndGet();
+ }
+ };
+ }
+
+ @Override
+ public void close() throws AdbcException {
+ // The JNI driver closes every connection still open on the
database first, from this thread --
+ // whichever thread is still using one.
+ Assertions.assertEquals(0, openConnections.get(), "the database
was released under an open connection");
+ closes.incrementAndGet();
+ if (closeFails) {
+ throw new AdbcException("the driver failed to close the
database", null, AdbcStatusCode.IO, null, 0);
+ }
+ }
+ }
}
diff --git
a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcObjectsReaderTest.java
b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcObjectsReaderTest.java
index 92da09f5f62..1275c1f597e 100644
---
a/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcObjectsReaderTest.java
+++
b/fe/fe-connector/fe-connector-adbc/src/test/java/org/apache/doris/connector/adbc/AdbcObjectsReaderTest.java
@@ -37,6 +37,7 @@ import org.junit.jupiter.api.Test;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.charset.StandardCharsets;
import java.util.List;
@@ -232,12 +233,12 @@ class AdbcObjectsReaderTest {
// ========= fixture =========
/** One catalog row: the name the source reported, and its schema entries
(null at depth CATALOGS). */
- private static Object[] catalog(String name, Object[]... schemas) {
+ static Object[] catalog(String name, Object[]... schemas) {
return new Object[] {name, schemas == null ? null : List.of(schemas)};
}
/** One schema entry: the name the source reported (SQLite's is the empty
string) and its tables. */
- private static Object[] schema(String name, Object[]... tables) {
+ static Object[] schema(String name, Object[]... tables) {
return new Object[] {name, tables == null ? null : List.of(tables)};
}
@@ -249,9 +250,9 @@ class AdbcObjectsReaderTest {
/**
* Builds the getObjects response these rows describe and returns it as
the reader a driver would have
* handed over -- written to an IPC stream and read back, so the reader
sees real Arrow vectors rather
- * than the fixture's own objects.
+ * than the fixture's own objects. {@link AdbcClientTest} hands it out as
a driver's {@code getObjects} result.
*/
- private static ArrowReader response(BufferAllocator allocator, Object[]...
rows) throws Exception {
+ static ArrowReader response(BufferAllocator allocator, Object[]... rows)
throws IOException {
VectorSchemaRoot root = VectorSchemaRoot.create(GET_OBJECTS_SCHEMA,
allocator);
VarCharVector catalogs = (VarCharVector) root.getVector(CATALOG_NAME);
ListVector schemas = (ListVector) root.getVector(CATALOG_DB_SCHEMAS);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
