Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 5 additions & 8 deletions src/main/java/org/duckdb/DuckDBConnection.java
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ public final class DuckDBConnection implements java.sql.Connection {
public static final String DEFAULT_SCHEMA = "main";

ByteBuffer connRef;
final Lock connRefLock = new ReentrantLock();
final ReentrantLock connRefLock = new ReentrantLock();
final LinkedHashSet<DuckDBPreparedStatement> preparedStatements = new LinkedHashSet<>();
volatile boolean closing;

Expand Down Expand Up @@ -488,14 +488,11 @@ void checkOpen() throws SQLException {
* This function calls the underlying C++ interrupt function which aborts the query running on this connection.
*/
void interrupt() throws SQLException {
checkOpen();
connRefLock.lock();
try {
checkOpen();
DuckDBNative.duckdb_jdbc_interrupt(connRef);
} finally {
connRefLock.unlock();
if (!connRefLock.isHeldByCurrentThread()) {
throw new SQLException("Connection lock state error");
}
checkOpen();
DuckDBNative.duckdb_jdbc_interrupt(connRef);
}

QueryProgress queryProgress() throws SQLException {
Expand Down
50 changes: 2 additions & 48 deletions src/main/java/org/duckdb/DuckDBDriver.java
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ public class DuckDBDriver implements java.sql.Driver {
public static final String JDBC_STREAM_RESULTS = "jdbc_stream_results";
public static final String JDBC_AUTO_COMMIT = "jdbc_auto_commit";
public static final String JDBC_PIN_DB = "jdbc_pin_db";
public static final String JDBC_IGNORE_UNSUPPORTED_OPTIONS = "jdbc_ignore_unsupported_options";

static final String DUCKDB_URL_PREFIX = "jdbc:duckdb:";
static final String MEMORY_DB = ":memory:";
Expand All @@ -34,9 +33,6 @@ public class DuckDBDriver implements java.sql.Driver {
private static boolean pinnedDbRefsShutdownHookRegistered = false;
private static boolean pinnedDbRefsShutdownHookRun = false;

private static final Set<String> supportedOptions = new LinkedHashSet<>();
private static final ReentrantLock supportedOptionsLock = new ReentrantLock();

static {
try {
DriverManager.registerDriver(new DuckDBDriver());
Expand All @@ -56,20 +52,13 @@ public Connection connect(String url, Properties info) throws SQLException {
props = (Properties) info.clone();
}

// URL options
ParsedProps pp = parsePropsFromUrl(url);
for (Map.Entry<String, String> en : pp.props.entrySet()) {
props.put(en.getKey(), en.getValue());
}

// Ignore unsupported
removeUnsupportedOptions(props);

// Read-only option
String readOnlyStr = removeOption(props, DUCKDB_READONLY_PROPERTY);
boolean readOnly = isStringTruish(readOnlyStr, false);

// Client name option
props.put("duckdb_api", "jdbc");

// Apache Spark passes this option when SELECT on a JDBC DataSource
Expand All @@ -78,7 +67,6 @@ public Connection connect(String url, Properties info) throws SQLException {
// to be established.
props.remove("path");

// DuckLake options
String ducklake = removeOption(props, DUCKLAKE_OPTION);
String ducklakeAlias = removeOption(props, DUCKLAKE_ALIAS_OPTION, DUCKLAKE_OPTION);
final String shortUrl;
Expand All @@ -95,13 +83,13 @@ public Connection connect(String url, Properties info) throws SQLException {
shortUrl = pp.shortUrl;
}

// Pin DB option
String pinDbOptStr = removeOption(props, JDBC_PIN_DB);
boolean pinDBOpt = isStringTruish(pinDbOptStr, false);

// Create connection
DuckDBConnection conn = DuckDBConnection.newConnection(shortUrl, readOnly, props);

pinDB(pinDBOpt, shortUrl, conn);

initDucklake(conn, shortUrl, ducklake, ducklakeAlias);

return conn;
Expand All @@ -128,8 +116,6 @@ public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) throws
list.add(createDriverPropInfo(JDBC_AUTO_COMMIT, "", "Set default auto-commit mode"));
list.add(createDriverPropInfo(JDBC_PIN_DB, "",
"Do not close the DB instance after all connections to it are closed"));
list.add(createDriverPropInfo(JDBC_IGNORE_UNSUPPORTED_OPTIONS, "",
"Silently discard unsupported connection options"));
list.sort((o1, o2) -> o1.name.compareToIgnoreCase(o2.name));
return list.toArray(new DriverPropertyInfo[0]);
}
Expand Down Expand Up @@ -265,38 +251,6 @@ private static DriverPropertyInfo createDriverPropInfo(String name, String value
return dpi;
}

private static void removeUnsupportedOptions(Properties props) throws SQLException {
String ignoreStr = removeOption(props, JDBC_IGNORE_UNSUPPORTED_OPTIONS);
boolean ignore = isStringTruish(ignoreStr, false);
if (!ignore) {
return;
}
supportedOptionsLock.lock();
try {
if (supportedOptions.isEmpty()) {
Driver driver = DriverManager.getDriver(DUCKDB_URL_PREFIX);
Properties dpiProps = new Properties();
dpiProps.put("threads", 1);
DriverPropertyInfo[] dpis = driver.getPropertyInfo(DUCKDB_URL_PREFIX, dpiProps);
for (DriverPropertyInfo dpi : dpis) {
supportedOptions.add(dpi.name);
}
}
List<String> unsupportedNames = new ArrayList<>();
for (Object nameObj : props.keySet()) {
String name = String.valueOf(nameObj);
if (!supportedOptions.contains(name)) {
unsupportedNames.add(name);
}
}
for (String name : unsupportedNames) {
props.remove(name);
}
} finally {
supportedOptionsLock.unlock();
}
}

private static class ParsedProps {
final String shortUrl;
final LinkedHashMap<String, String> props;
Expand Down
41 changes: 35 additions & 6 deletions src/main/java/org/duckdb/DuckDBPreparedStatement.java
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ public class DuckDBPreparedStatement implements PreparedStatement {
private DuckDBConnection conn;

private ByteBuffer stmtRef = null;
final Lock stmtRefLock = new ReentrantLock();
final ReentrantLock stmtRefLock = new ReentrantLock();
volatile boolean closeOnCompletion = false;

private DuckDBResultSet selectResult = null;
Expand Down Expand Up @@ -159,6 +159,11 @@ private boolean execute(boolean startTransaction) throws SQLException {
checkOpen();
checkPrepared();

// Wait with dispatching a new query if connection is locked by cancel() call
Lock connLock = getConnRefLock();
connLock.lock();
connLock.unlock();

ByteBuffer resultRef = null;

stmtRefLock.lock();
Expand Down Expand Up @@ -442,12 +447,27 @@ public void setQueryTimeout(int seconds) throws SQLException {
@Override
public void cancel() throws SQLException {
checkOpen();
// Only proceed to interrupt call after ensuring that the query on
// this statement is still running.
if (!stmtRefLock.isLocked()) {
return;
}
// Cancel is intended to be called concurrently with execute,
// thus we cannot take the statement lock that is held while
// query is running. NPE may be thrown if connection is closed
// concurrently.
try {
// Cancel is intended to be called concurrently with execute,
// thus we cannot take the statement lock that is held while
// query is running. NPE may be thrown if connection is closed
// concurrently.
conn.interrupt();
// Taking connection lock will prevent new queries to be executed
Lock connLock = getConnRefLock();
connLock.lock();
try {
if (!stmtRefLock.isLocked()) {
return;
}
conn.interrupt();
} finally {
connLock.unlock();
}
} catch (NullPointerException e) {
throw new SQLException(e);
}
Expand Down Expand Up @@ -1215,4 +1235,13 @@ private int[] intArrayFromLong(long[] arr) {
}
return res;
}

private Lock getConnRefLock() throws SQLException {
// NPE can be thrown if statement is closed concurrently.
try {
return conn.connRefLock;
} catch (NullPointerException e) {
throw new SQLException(e);
}
}
}
34 changes: 34 additions & 0 deletions src/test/java/org/duckdb/TestClosure.java
Original file line number Diff line number Diff line change
Expand Up @@ -251,4 +251,38 @@ public static void test_results_fetch_no_hang() throws Exception {
}
}
}

public static void test_stmt_can_only_cancel_self() throws Exception {
try (Connection conn = DriverManager.getConnection(JDBC_URL); Statement stmt1 = conn.createStatement();
Statement stmt2 = conn.createStatement()) {
stmt1.execute("DROP TABLE IF EXISTS test_fib1");
stmt1.execute("CREATE TABLE test_fib1(i bigint, p double, f double)");
stmt1.execute("INSERT INTO test_fib1 values(1, 0, 1)");
long start = System.currentTimeMillis();
Thread th = new Thread(() -> {
try {
Thread.sleep(200);
stmt1.cancel();
} catch (Exception e) {
e.printStackTrace();
}
});
th.start();
try (
ResultSet rs = stmt2.executeQuery(
"WITH RECURSIVE cte AS ("
+
"SELECT * from test_fib1 UNION ALL SELECT cte.i + 1, cte.f, cte.p + cte.f from cte WHERE cte.i < 40000) "
+ "SELECT avg(f) FROM cte")) {
rs.next();
assertTrue(rs.getDouble(1) > 0);
}
th.join();
long elapsed = System.currentTimeMillis() - start;
assertTrue(elapsed > 1000);
assertFalse(conn.isClosed());
assertFalse(stmt1.isClosed());
assertFalse(stmt2.isClosed());
}
}
}
9 changes: 0 additions & 9 deletions src/test/java/org/duckdb/TestDuckDBJDBC.java
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static org.duckdb.DuckDBDriver.DUCKDB_USER_AGENT_PROPERTY;
import static org.duckdb.DuckDBDriver.JDBC_STREAM_RESULTS;
import static org.duckdb.DuckDBTimestamp.localDateTimeFromTimestamp;
import static org.duckdb.test.Assertions.*;
import static org.duckdb.test.Runner.runTests;
Expand Down Expand Up @@ -3657,14 +3656,6 @@ public static void test_driver_property_info() throws Exception {
assertTrue(dpis.length > 0);
}

public static void test_ignore_unsupported_options() throws Exception {
assertThrows(() -> { DriverManager.getConnection("jdbc:duckdb:;foo=bar;"); }, SQLException.class);
Properties config = new Properties();
config.put("boo", "bar");
config.put(JDBC_STREAM_RESULTS, true);
DriverManager.getConnection("jdbc:duckdb:;foo=bar;jdbc_ignore_unsupported_options=yes;", config).close();
}

public static void main(String[] args) throws Exception {
String arg1 = args.length > 0 ? args[0] : "";
final int statusCode;
Expand Down