Skip to content

Commit d7341a4

Browse files
chernserchernser
committed
fixed multiresult logic
1 parent 8ad1aab commit d7341a4

File tree

4 files changed

+88
-60
lines changed

4 files changed

+88
-60
lines changed

‎jdbc-v2/src/main/java/com/clickhouse/jdbc/PreparedStatementImpl.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,7 @@ public ResultSet executeQuery() throws SQLException {
134134
@Override
135135
public int executeUpdate() throws SQLException {
136136
ensureOpen();
137-
return super.executeUpdateImpl(buildSQL(), localSettings);
137+
return (int) super.executeUpdateImpl(buildSQL(), localSettings);
138138
}
139139

140140
@Override
@@ -298,7 +298,7 @@ public int[] executeBatch() throws SQLException {
298298
} else {
299299
List<Integer> results = new ArrayList<>();
300300
for (String sql : batch) {
301-
results.add(executeUpdateImpl(sql, localSettings));
301+
results.add((int) executeUpdateImpl(sql, localSettings));
302302
}
303303
return results.stream().mapToInt(Integer::intValue).toArray();
304304
}
@@ -313,7 +313,7 @@ public long[] executeLargeBatch() throws SQLException {
313313
} else {
314314
List<Integer> results = new ArrayList<>();
315315
for (String sql : batch) {
316-
results.add(executeUpdateImpl(sql, localSettings));
316+
results.add((int) executeUpdateImpl(sql, localSettings));
317317
}
318318
return results.stream().mapToLong(Integer::longValue).toArray();
319319
}
@@ -328,7 +328,7 @@ private List<Integer> executeInsertBatch() throws SQLException {
328328
}
329329
insertSql.setLength(insertSql.length() - 1);
330330

331-
int updateCount = super.executeUpdateImpl(insertSql.toString(), localSettings);
331+
int updateCount = (int) super.executeUpdateImpl(insertSql.toString(), localSettings);
332332
if (updateCount == batchValues.size()) {
333333
return Collections.nCopies(batchValues.size(), 1);
334334
} else {

‎jdbc-v2/src/main/java/com/clickhouse/jdbc/StatementImpl.java

Lines changed: 33 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,6 @@
33
import com.clickhouse.client.api.ClientConfigProperties;
44
import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
55
import com.clickhouse.client.api.internal.ServerSettings;
6-
import com.clickhouse.client.api.metrics.OperationMetrics;
7-
import com.clickhouse.client.api.metrics.ServerMetrics;
86
import com.clickhouse.client.api.query.QueryResponse;
97
import com.clickhouse.client.api.query.QuerySettings;
108
import com.clickhouse.client.api.sql.SQLUtils;
@@ -16,7 +14,6 @@
1614

1715
import java.sql.ResultSet;
1816
import java.sql.SQLException;
19-
import java.sql.SQLFeatureNotSupportedException;
2017
import java.sql.SQLWarning;
2118
import java.sql.Statement;
2219
import java.util.ArrayList;
@@ -35,9 +32,9 @@ public class StatementImpl implements Statement, JdbcV2Wrapper {
3532

3633
// State
3734
private volatile boolean closed;
38-
private final ConcurrentLinkedQueue<ResultSetImpl> resultSets;
35+
private final ConcurrentLinkedQueue<ResultSetImpl> resultSets; // all result sets linked to this statement
3936
protected ResultSetImpl currentResultSet;
40-
protected OperationMetrics metrics;
37+
protected long currentUpdateCount = -1;
4138
protected List<String> batch;
4239
private String lastStatementSql;
4340
private ParsedStatement parsedStatement;
@@ -54,7 +51,6 @@ public StatementImpl(ConnectionImpl connection) throws SQLException {
5451
this.connection = connection;
5552
this.queryTimeout = 0;
5653
this.closed = false;
57-
this.metrics = null;
5854
this.batch = new ArrayList<>();
5955
this.maxRows = 0;
6056
this.localSettings = QuerySettings.merge(connection.getDefaultQuerySettings(), new QuerySettings());
@@ -97,7 +93,9 @@ protected String getLastStatementSql() {
9793
@Override
9894
public ResultSet executeQuery(String sql) throws SQLException {
9995
ensureOpen();
100-
return executeQueryImpl(sql, localSettings);
96+
currentUpdateCount = -1;
97+
currentResultSet = executeQueryImpl(sql, localSettings);
98+
return currentResultSet;
10199
}
102100

103101
private void closeCurrentResultSet() {
@@ -149,9 +147,7 @@ protected ResultSetImpl executeQueryImpl(String sql, QuerySettings settings) thr
149147
if (reader.getSchema() == null) {
150148
throw new SQLException("Called method expects empty or filled result set but query has returned none. Consider using `java.sql.Statement.execute(java.lang.String)`", ExceptionUtils.SQL_STATE_CLIENT_ERROR);
151149
}
152-
metrics = response.getMetrics();
153-
setCurrentResultSet(new ResultSetImpl(this, response, reader));
154-
return currentResultSet;
150+
return new ResultSetImpl(this, response, reader);
155151
} catch (Exception e) {
156152
if (response != null) {
157153
try {
@@ -168,13 +164,10 @@ protected ResultSetImpl executeQueryImpl(String sql, QuerySettings settings) thr
168164
@Override
169165
public int executeUpdate(String sql) throws SQLException {
170166
ensureOpen();
171-
parsedStatement = connection.getSqlParser().parsedStatement(sql);
172-
int updateCount = executeUpdateImpl(sql, localSettings);
173-
postUpdateActions();
174-
return updateCount;
167+
return (int)executeLargeUpdate(sql);
175168
}
176169

177-
protected int executeUpdateImpl(String sql, QuerySettings settings) throws SQLException {
170+
protected long executeUpdateImpl(String sql, QuerySettings settings) throws SQLException {
178171
ensureOpen();
179172

180173
// Closing before trying to do next request. Otherwise, deadlock because previous connection will not be
@@ -196,9 +189,7 @@ protected int executeUpdateImpl(String sql, QuerySettings settings) throws SQLEx
196189
int updateCount = 0;
197190
try (QueryResponse response = queryTimeout == 0 ? connection.client.query(lastStatementSql, mergedSettings).get()
198191
: connection.client.query(lastStatementSql, mergedSettings).get(queryTimeout, TimeUnit.SECONDS)) {
199-
setCurrentResultSet(null);
200192
updateCount = Math.max(0, (int) response.getWrittenRows()); // when statement alters schema no result rows returned.
201-
metrics = response.getMetrics();
202193
lastQueryId = response.getQueryId();
203194
} catch (Exception e) {
204195
throw ExceptionUtils.toSqlState(e);
@@ -308,28 +299,17 @@ public void setCursorName(String name) throws SQLException {
308299
ensureOpen();
309300
}
310301

311-
/**
312-
* Remembers current result set to be able to close it later.
313-
* Sets current resultset to a new value
314-
* @param resultSet new current resultset
315-
*/
316-
protected void setCurrentResultSet(ResultSetImpl resultSet) {
317-
ResultSetImpl tmp = currentResultSet;
318-
currentResultSet = resultSet;
319-
if (tmp != null) {
320-
resultSets.add(tmp);
321-
}
322-
}
323-
324302
@Override
325303
public boolean execute(String sql) throws SQLException {
326304
ensureOpen();
327305
parsedStatement = connection.getSqlParser().parsedStatement(sql);
306+
currentUpdateCount = -1;
307+
currentResultSet = null;
328308
if (parsedStatement.isHasResultSet()) {
329-
executeQueryImpl(sql, localSettings); // keep open to allow getResultSet()
309+
currentResultSet = executeQueryImpl(sql, localSettings);
330310
return true;
331311
} else {
332-
executeUpdateImpl(sql, localSettings);
312+
currentUpdateCount = executeUpdateImpl(sql, localSettings);
333313
postUpdateActions();
334314
return false;
335315
}
@@ -339,9 +319,7 @@ public boolean execute(String sql) throws SQLException {
339319
public ResultSet getResultSet() throws SQLException {
340320
ensureOpen();
341321

342-
ResultSet resultSet = currentResultSet;
343-
setCurrentResultSet(null);
344-
return resultSet;
322+
return currentResultSet;
345323
}
346324

347325
@Override
@@ -353,7 +331,7 @@ public int getUpdateCount() throws SQLException {
353331
@Override
354332
public boolean getMoreResults() throws SQLException {
355333
ensureOpen();
356-
return false;
334+
return getMoreResults(Statement.CLOSE_CURRENT_RESULT);
357335
}
358336

359337
@Override
@@ -435,8 +413,17 @@ public QuerySettings getLocalSettings() {
435413

436414
@Override
437415
public boolean getMoreResults(int current) throws SQLException {
438-
// TODO: implement query batches. When multiple selects in the batch.
439-
return false;
416+
// This method designed to iterate over multiple resultsets after "execute(sql)" method is called
417+
// But we have at most only one always
418+
// Then we should close any existing and return false to indicate that no more result are present
419+
420+
if (currentResultSet != null && current != Statement.KEEP_CURRENT_RESULT) {
421+
currentResultSet.close();
422+
}
423+
424+
currentResultSet = null;
425+
currentUpdateCount = -1;
426+
return false; // false indicates that no more results (or it is an update count)
440427
}
441428

442429
@Override
@@ -547,13 +534,7 @@ public boolean isCloseOnCompletion() throws SQLException {
547534
@Override
548535
public long getLargeUpdateCount() throws SQLException {
549536
ensureOpen();
550-
if (currentResultSet == null && metrics != null) {
551-
long updateCount = metrics.getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong();
552-
metrics = null;// clear metrics
553-
return updateCount;
554-
}
555-
556-
return -1L;
537+
return currentUpdateCount;
557538
}
558539

559540
@Override
@@ -590,22 +571,25 @@ public long[] executeLargeBatch() throws SQLException {
590571

591572
@Override
592573
public long executeLargeUpdate(String sql) throws SQLException {
593-
return executeUpdate(sql);
574+
parsedStatement = connection.getSqlParser().parsedStatement(sql);
575+
long updateCount = executeUpdateImpl(sql, localSettings);
576+
postUpdateActions();
577+
return updateCount;
594578
}
595579

596580
@Override
597581
public long executeLargeUpdate(String sql, int autoGeneratedKeys) throws SQLException {
598-
return executeUpdate(sql, autoGeneratedKeys);
582+
return executeLargeUpdate(sql);
599583
}
600584

601585
@Override
602586
public long executeLargeUpdate(String sql, int[] columnIndexes) throws SQLException {
603-
return executeUpdate(sql, columnIndexes);
587+
return executeLargeUpdate(sql);
604588
}
605589

606590
@Override
607591
public long executeLargeUpdate(String sql, String[] columnNames) throws SQLException {
608-
return executeUpdate(sql, columnNames);
592+
return executeLargeUpdate(sql);
609593
}
610594

611595
/**

‎jdbc-v2/src/main/java/com/clickhouse/jdbc/WriterStatementImpl.java

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -118,9 +118,7 @@ public long executeLargeUpdate() throws SQLException {
118118
try (InsertResponse response = queryTimeout == 0 ?
119119
connection.client.insert(tableSchema.getTableName(),in, writer.getFormat(), settings).get()
120120
: connection.client.insert(tableSchema.getTableName(),in, writer.getFormat(), settings).get(queryTimeout, TimeUnit.SECONDS)) {
121-
setCurrentResultSet(null);
122121
updateCount = Math.max(0, (int) response.getWrittenRows()); // when statement alters schema no result rows returned.
123-
metrics = response.getMetrics();
124122
lastQueryId = response.getQueryId();
125123
} catch (Exception e) {
126124
throw ExceptionUtils.toSqlState(e);

‎jdbc-v2/src/test/java/com/clickhouse/jdbc/StatementTest.java

Lines changed: 51 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,6 @@
33
import com.clickhouse.client.api.ClientConfigProperties;
44
import com.clickhouse.client.api.internal.ServerSettings;
55
import com.clickhouse.client.api.query.GenericRecord;
6-
import com.clickhouse.client.api.sql.SQLUtils;
7-
import com.clickhouse.jdbc.internal.ClickHouseParser;
86
import com.clickhouse.jdbc.internal.DriverProperties;
97
import org.slf4j.Logger;
108
import org.slf4j.LoggerFactory;
@@ -32,9 +30,9 @@
3230
import static org.testng.Assert.assertFalse;
3331
import static org.testng.Assert.assertNotNull;
3432
import static org.testng.Assert.assertNull;
33+
import static org.testng.Assert.assertSame;
3534
import static org.testng.Assert.assertThrows;
3635
import static org.testng.Assert.assertTrue;
37-
import static org.testng.Assert.fail;
3836

3937

4038
@Test(groups = { "integration" })
@@ -170,8 +168,6 @@ public void testExecuteUpdateSimpleFloats() throws Exception {
170168
try (Statement stmt = conn.createStatement()) {
171169
assertEquals(stmt.executeUpdate("CREATE TABLE IF NOT EXISTS " + getDatabase() + ".simpleFloats (num Float32) ENGINE = MergeTree ORDER BY ()"), 0);
172170
assertEquals(stmt.executeUpdate("INSERT INTO " + getDatabase() + ".simpleFloats VALUES (1.1), (2.2), (3.3)"), 3);
173-
assertEquals(stmt.getUpdateCount(), 3);
174-
assertEquals(stmt.getLargeUpdateCount(), -1L);
175171
try (ResultSet rs = stmt.executeQuery("SELECT num FROM " + getDatabase() + ".simpleFloats ORDER BY num")) {
176172
assertTrue(rs.next());
177173
assertEquals(rs.getFloat(1), 1.1f);
@@ -963,4 +959,54 @@ public void testVariousSimpleMethods() throws Exception {
963959
assertTrue(stmt.isPoolable());
964960
}
965961
}
962+
963+
@Test(groups = {"integration"})
964+
public void testExecute() throws Exception {
965+
// This test verifies multi-resultset scenario (we may have only one resultset at a time)
966+
try (Connection conn = getJdbcConnection(); Statement stmt = conn.createStatement()) {
967+
// has result set and no update count
968+
Assert.assertTrue(stmt.execute("SELECT 1"));
969+
ResultSet rs = stmt.getResultSet();
970+
Assert.assertTrue(rs.next());
971+
assertEquals(rs.getInt(1), 1);
972+
ResultSet rs2 = stmt.getResultSet();
973+
assertSame(rs, rs2);
974+
Assert.assertFalse(rs.next());
975+
Assert.assertFalse(rs2.next());
976+
Assert.assertEquals(stmt.getUpdateCount(), -1);
977+
assertFalse(rs.isClosed());
978+
Assert.assertFalse(stmt.getMoreResults());
979+
assertTrue(rs.isClosed());
980+
Assert.assertNull(stmt.getResultSet());
981+
}
982+
983+
try (Connection conn = getJdbcConnection(); Statement stmt = conn.createStatement()) {
984+
// no result set and update count
985+
Assert.assertFalse(stmt.execute("CREATE TABLE test_multi_result (id Int32) Engine MergeTree ORDER BY ()"));
986+
Assert.assertNull(stmt.getResultSet());
987+
Assert.assertEquals(stmt.getUpdateCount(), 0);
988+
Assert.assertFalse(stmt.getMoreResults());
989+
990+
// no result set and has update count
991+
Assert.assertFalse(stmt.execute("INSERT INTO test_multi_result VALUES (1), (2), (3)"));
992+
Assert.assertNull(stmt.getResultSet());
993+
Assert.assertEquals(stmt.getUpdateCount(), 3);
994+
Assert.assertFalse(stmt.getMoreResults());
995+
Assert.assertEquals(stmt.getUpdateCount(), -1);
996+
}
997+
998+
// keep current resultset
999+
try (Connection conn = getJdbcConnection(); Statement stmt = conn.createStatement()) {
1000+
// has result set and no update count
1001+
Assert.assertTrue(stmt.execute("SELECT 1"));
1002+
ResultSet rs = stmt.getResultSet();
1003+
Assert.assertEquals(stmt.getUpdateCount(), -1);
1004+
assertFalse(rs.isClosed());
1005+
Assert.assertFalse(stmt.getMoreResults(Statement.KEEP_CURRENT_RESULT));
1006+
Assert.assertNull(stmt.getResultSet());
1007+
assertFalse(rs.isClosed());
1008+
assertTrue(rs.next());
1009+
assertEquals(rs.getInt(1), 1);
1010+
}
1011+
}
9661012
}

0 commit comments

Comments
 (0)