Skip to content

refactor(jdbc): collapse duplicated connection send paths #227

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
magnusp merged 1 commit into from
Oct 24, 2026
Merged

refactor(jdbc): collapse duplicated connection send paths #227

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
111 changes: 48 additions & 63 deletions libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;
import java.util.function.Supplier;

/**
* A JDBC connection whose state is serialized on one connection-level lock, the way sqlite-jdbc
Expand Down Expand Up @@ -120,65 +121,44 @@ public void checkClosed() throws SQLException {
public StmtResult executeStatement(Stmt stmt) throws SQLException {
synchronized (stateLock) {
checkClosed();
try {
if (autoCommit) {
// One-shot: single-round-trip close, so no idle stream is left for the server to reap.
return client.executeOneShot(stmt);
} else {
ensureTransactionStarted();
return client.executeOnStream(stmt);
}
} catch (LibsqlException e) {
throw LibsqlErrorMapper.toSQLException(e);
if (autoCommit) {
// One-shot: single-round-trip close, so no idle stream is left for the server to reap.
return sending(() -> client.executeOneShot(stmt));
}
ensureTransactionStarted();
return sendOnStream(stmt);
}
}

public int[] executeBatch(List<Stmt> statements) throws SQLException {
List<StmtResult> results = executeBatchResults(statements);
int[] updateCounts = new int[results.size()];
for (int i = 0; i < results.size(); i++) {
updateCounts[i] = (int) results.get(i).affectedRowCount();
}
return updateCounts;
return steppedUpdateCounts(executeBatchResults(statements));
}

/**
* Executes a batch and returns the {@code BatchResult} step results, one per statement. The
* caller derives both the update counts and the generated keys from these results, so the
* driver needs no follow-up query and no extra round trip.
*/
/** Executes a batch and returns one step result per statement, so counts and keys need no follow-up query. */
List<StmtResult> executeBatchResults(List<Stmt> statements) throws SQLException {
synchronized (stateLock) {
checkClosed();
if (statements == null || statements.isEmpty()) {
return List.of();
}

try {
List<BatchStep> steps = new ArrayList<>(statements.size());
for (int i = 0; i < statements.size(); i++) {
BatchCond condition = i == 0 ? null : new BatchCond.StepOk(i - 1);
steps.add(new BatchStep(condition, statements.get(i)));
}
Batch batch = new Batch(steps);

BatchResult result;
if (autoCommit) {
result = client.executeBatchOneShot(batch);
} else {
ensureTransactionStarted();
result = client.executeBatchOnStream(batch);
}
return extractBatchResults(result, statements.size());
} catch (LibsqlException e) {
throw LibsqlErrorMapper.toSQLException(e);
List<BatchStep> steps = new ArrayList<>(statements.size());
for (int i = 0; i < statements.size(); i++) {
BatchCond condition = i == 0 ? null : new BatchCond.StepOk(i - 1);
steps.add(new BatchStep(condition, statements.get(i)));
}
Batch batch = new Batch(steps);
BatchResult result;
if (autoCommit) {
result = sending(() -> client.executeBatchOneShot(batch));
} else {
ensureTransactionStarted();
result = sendBatchOnStream(batch);
}
return extractBatchResults(result, statements.size());
}
}

private List<StmtResult> extractBatchResults(BatchResult batchResult, int expectedSize)
throws SQLException {
private List<StmtResult> extractBatchResults(BatchResult batchResult, int expectedSize) throws SQLException {
List<StmtResult> results = new ArrayList<>(expectedSize);
for (int i = 0; i < expectedSize; i++) {
HranaError error = batchResult.stepErrors().get(i);
Expand Down Expand Up @@ -213,6 +193,24 @@ private static int[] steppedUpdateCounts(List<StmtResult> results) {
return updateCounts;
}

/** Maps a transport failure into the JDBC exception hierarchy. */
private <T> T sending(Supplier<T> call) throws SQLException {
try {
return call.get();
} catch (LibsqlException e) {
throw LibsqlErrorMapper.toSQLException(e);
}
}

/** Stream sends. Callers hold {@link #stateLock}. */
private StmtResult sendOnStream(Stmt stmt) throws SQLException {
return sending(() -> client.executeOnStream(stmt));
}

private BatchResult sendBatchOnStream(Batch batch) throws SQLException {
return sending(() -> client.executeBatchOnStream(batch));
}

private void ensureTransactionStarted() throws SQLException {
if (beginPending()) {
beginTransaction();
Expand All @@ -233,11 +231,11 @@ private boolean beginPending() {
/** Callers hold {@link #stateLock}. */
private void beginTransaction() throws SQLException {
try {
client.executeOnStream(new Stmt(transactionMode.beginSql()));
} catch (LibsqlException e) {
sendOnStream(new Stmt(transactionMode.beginSql()));
} catch (SQLException e) {
// The begin failed, so the fresh stream holds no transaction state.
closeStreamQuietly();
throw LibsqlErrorMapper.toSQLException(e);
throw e;
}
}

Expand Down Expand Up @@ -354,11 +352,7 @@ private void endTransaction(String sql) throws SQLException {
return;
}

try {
client.executeOnStream(new Stmt(sql));
} catch (LibsqlException e) {
throw LibsqlErrorMapper.toSQLException(e);
}
sendOnStream(new Stmt(sql));

// The server committed or rolled back, so the stream has no state left worth holding.
// Closing it lets an idle connection drop its server-side stream. On an error above the
Expand Down Expand Up @@ -552,7 +546,7 @@ private Savepoint setSavepointInternal(String name) throws SQLException {
String sql = "SAVEPOINT " + quoteSavepointName(savepoint.sqlName());
if (autoCommit) {
try {
executeTransactionStatement(sql);
sendOnStream(new Stmt(sql));
} catch (SQLException e) {
// No transaction was started, so the fresh stream holds no state.
closeStreamQuietly();
Expand All @@ -564,7 +558,7 @@ private Savepoint setSavepointInternal(String name) throws SQLException {
savepointNames.add(savepoint.sqlName());
} else {
ensureTransactionStarted();
executeTransactionStatement(sql);
sendOnStream(new Stmt(sql));
savepointNames.add(savepoint.sqlName());
}
return savepoint;
Expand All @@ -580,7 +574,7 @@ public void rollback(Savepoint savepoint) throws SQLException {
throw new SQLException("Cannot rollback to a savepoint when in autoCommit mode");
}
ensureTransactionStarted();
executeTransactionStatement("ROLLBACK TO SAVEPOINT " + quoteSavepointName(sqlName));
sendOnStream(new Stmt("ROLLBACK TO SAVEPOINT " + quoteSavepointName(sqlName)));
int index = savepointNames.lastIndexOf(sqlName);
if (index >= 0) {
// Rolling back cancels the savepoints created after the target; the target
Expand All @@ -599,7 +593,7 @@ public void releaseSavepoint(Savepoint savepoint) throws SQLException {
throw new SQLException("Cannot release a savepoint when in autoCommit mode");
}
ensureTransactionStarted();
executeTransactionStatement("RELEASE SAVEPOINT " + quoteSavepointName(sqlName));
sendOnStream(new Stmt("RELEASE SAVEPOINT " + quoteSavepointName(sqlName)));
int index = savepointNames.lastIndexOf(sqlName);
if (index >= 0) {
boolean outermost = index == 0;
Expand All @@ -616,15 +610,6 @@ public void releaseSavepoint(Savepoint savepoint) throws SQLException {
}
}

/** Callers hold {@link #stateLock}. */
private void executeTransactionStatement(String sql) throws SQLException {
try {
client.executeOnStream(new Stmt(sql));
} catch (LibsqlException e) {
throw LibsqlErrorMapper.toSQLException(e);
}
}

private static String savepointSqlName(Savepoint savepoint) throws SQLException {
if (savepoint == null) {
throw new SQLException("Savepoint cannot be null");
Expand Down
Loading