diff --git a/CHANGELOG.md b/CHANGELOG.md index 9eff8d530..eb870ea1f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -142,6 +142,13 @@ engine-to-table-type mapping, so it fell back to the default `TABLE`, and `getTables(..., types = {"REMOTE TABLE"})` returned no row for such a table. `BigQuery` is now mapped to `REMOTE TABLE`, like the other external-storage engines. (https://github.com/ClickHouse/clickhouse-java/issues/3049) +- **[client-v2, jdbc-v2]** Fixed ClickHouse exceptions appended to an HTTP 200 response body being exposed as result + data while a query was streamed. `client-v2` now authenticates the in-band exception frame with the + `X-ClickHouse-Exception-Tag` response header and throws a `ServerException` when the stream reaches it. When the + server error is `TIMEOUT_EXCEEDED` (code 159), `jdbc-v2` now reports `SQLTimeoutException` with SQLState `HYT00` + from `ResultSet.next()` while preserving the original exception chain. Previously the tagged frame could be read as + row data and the timeout was exposed as a generic `SQLException`. + (https://github.com/ClickHouse/clickhouse-java/issues/2702, https://github.com/ClickHouse/clickhouse-java/issues/3077) - **[jdbc-v2]** Fixed `PreparedStatement.getMetaData()` losing the result-set schema for a statement whose SQL contains a comment. The `DESCRIBE` query used to resolve the metadata was built by re-scanning the SQL with a regex that knew only quoted tokens, so a `?` inside a `--` / `#` / `/* */` comment was rewritten to `NULL` and diff --git a/client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java b/client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java index 860f74ca6..506af93f1 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/http/ClickHouseHttpProto.java @@ -27,6 +27,12 @@ public class ClickHouseHttpProto { */ public static final String HEADER_EXCEPTION_CODE = "X-ClickHouse-Exception-Code"; + /** + * Response only header containing the tag used to identify exception frames in a successful response body. + * Cannot be used in request. + */ + public static final String HEADER_EXCEPTION_TAG = "X-ClickHouse-Exception-Tag"; + /** * Response only header to indicate a query progress. * Cannot be used in request. diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java index 54d53eec9..949ec241f 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java @@ -693,7 +693,11 @@ public void close() throws IOException { @Override public InputStream createDataInputStream() { try { - return delegate.getEntity().getContent(); + InputStream input = delegate.getEntity().getContent(); + Header exceptionTag = delegate.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_TAG); + return exceptionTag == null || exceptionTag.getValue().isEmpty() + ? input + : new HttpExceptionInputStream(input, exceptionTag.getValue(), delegate.getCode(), getQueryId()); } catch (Exception e) { throw new ClientException("Failed to construct input stream", e); } @@ -1055,7 +1059,8 @@ public static int getHeaderInt(Header header, int defaultValue) { ClickHouseHttpProto.HEADER_DB_USER, ClickHouseHttpProto.HEADER_TIMEZONE, ClickHouseHttpProto.HEADER_FORMAT, - ClickHouseHttpProto.HEADER_PROGRESS + ClickHouseHttpProto.HEADER_PROGRESS, + ClickHouseHttpProto.HEADER_EXCEPTION_TAG )); /** diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpExceptionInputStream.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpExceptionInputStream.java new file mode 100644 index 000000000..90534e354 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpExceptionInputStream.java @@ -0,0 +1,253 @@ +package com.clickhouse.client.api.internal; + +import com.clickhouse.client.api.ClientException; +import com.clickhouse.client.api.ServerException; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Hides tagged exception frames appended to successful HTTP response bodies. A possible frame prefix remains buffered + * until it is matched or disproved, so callers never observe part of a marker when it crosses source read boundaries. + */ +final class HttpExceptionInputStream extends InputStream { + + private static final byte[] EXCEPTION_MARKER = "\r\n__exception__\r\n".getBytes(StandardCharsets.UTF_8); + private static final String EXCEPTION_END_MARKER = "\r\n__exception__\r\n"; + private static final int BUFFER_SIZE = 8192; + private static final int MAX_EXCEPTION_SIZE = 32 * 1024; + private static final Pattern ERROR_CODE_PATTERN = Pattern.compile("^Code:\\s*(\\d+)\\."); + + private final InputStream source; + private final String exceptionTag; + private final int transportStatus; + private final String queryId; + private final byte[] exceptionPrefix; + private final byte[] sourceBuffer = new byte[BUFFER_SIZE]; + + private byte[] pending = new byte[BUFFER_SIZE]; + private int pendingStart; + private int pendingEnd; + private int scanOffset; + private boolean sourceDone; + private RuntimeException terminalException; + private IOException terminalIOException; + + HttpExceptionInputStream(InputStream source, String exceptionTag, int transportStatus, String queryId) { + this.source = source; + this.exceptionTag = exceptionTag; + this.transportStatus = transportStatus; + this.queryId = queryId; + byte[] tagBytes = exceptionTag.getBytes(StandardCharsets.UTF_8); + this.exceptionPrefix = Arrays.copyOf(EXCEPTION_MARKER, EXCEPTION_MARKER.length + tagBytes.length + 2); + System.arraycopy(tagBytes, 0, exceptionPrefix, EXCEPTION_MARKER.length, tagBytes.length); + exceptionPrefix[exceptionPrefix.length - 2] = '\r'; + exceptionPrefix[exceptionPrefix.length - 1] = '\n'; + } + + @Override + public int read() throws IOException { + byte[] oneByte = new byte[1]; + int read = read(oneByte, 0, 1); + return read < 0 ? -1 : oneByte[0] & 0xff; + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + if (buffer == null) { + throw new NullPointerException("buffer"); + } + if (offset < 0 || length < 0 || length > buffer.length - offset) { + throw new IndexOutOfBoundsException(); + } + if (length == 0) { + return 0; + } + + while (true) { + int safeLength = safeLength(); + if (safeLength > 0) { + int read = Math.min(length, safeLength); + System.arraycopy(pending, pendingStart, buffer, offset, read); + pendingStart += read; + return read; + } + if (terminalException != null) { + throw terminalException; + } + if (terminalIOException != null) { + throw terminalIOException; + } + if (sourceDone) { + return -1; + } + + fillPending(); + scanPending(); + } + } + + @Override + public int available() { + return safeLength(); + } + + @Override + public void close() throws IOException { + source.close(); + } + + private int safeLength() { + if (sourceDone || terminalException != null || terminalIOException != null) { + return pendingEnd - pendingStart; + } + return Math.max(0, scanOffset - pendingStart); + } + + private void fillPending() { + try { + int read = source.read(sourceBuffer); + if (read < 0) { + sourceDone = true; + scanOffset = pendingEnd; + return; + } + appendPending(sourceBuffer, read); + } catch (IOException e) { + sourceDone = true; + terminalIOException = e; + scanOffset = pendingEnd; + } + } + + private void appendPending(byte[] bytes, int length) { + compactPending(length); + System.arraycopy(bytes, 0, pending, pendingEnd, length); + pendingEnd += length; + } + + private void compactPending(int additionalLength) { + int currentLength = pendingEnd - pendingStart; + if (pending.length - pendingEnd >= additionalLength) { + return; + } + + int newLength = Math.max(pending.length * 2, currentLength + additionalLength); + byte[] compacted = new byte[newLength]; + System.arraycopy(pending, pendingStart, compacted, 0, currentLength); + scanOffset -= pendingStart; + pendingStart = 0; + pendingEnd = currentLength; + pending = compacted; + } + + private void scanPending() { + int exceptionStart = indexOf(pending, scanOffset, pendingEnd, exceptionPrefix); + if (exceptionStart >= 0) { + captureException(exceptionStart); + return; + } + + int suffixLength = matchingSuffixLength(pending, pendingStart, pendingEnd, exceptionPrefix); + scanOffset = pendingEnd - suffixLength; + } + + private void captureException(int exceptionStart) { + ByteArrayOutputStream exceptionBody = new ByteArrayOutputStream(); + int bodyStart = exceptionStart + exceptionPrefix.length; + exceptionBody.write(pending, bodyStart, pendingEnd - bodyStart); + pendingEnd = exceptionStart; + scanOffset = exceptionStart; + + try { + while (exceptionBody.size() <= MAX_EXCEPTION_SIZE) { + int read = source.read(sourceBuffer); + if (read < 0) { + terminalException = parseException(exceptionBody.toByteArray()); + sourceDone = true; + return; + } + int remaining = MAX_EXCEPTION_SIZE + 1 - exceptionBody.size(); + exceptionBody.write(sourceBuffer, 0, Math.min(read, remaining)); + if (read > remaining || exceptionBody.size() > MAX_EXCEPTION_SIZE) { + terminalException = new ClientException("ClickHouse exception frame exceeds " + + MAX_EXCEPTION_SIZE + " bytes"); + sourceDone = true; + return; + } + } + } catch (IOException e) { + ClientException truncatedFrame = new ClientException( + "Failed to finish reading ClickHouse exception frame", parseException(exceptionBody.toByteArray())); + truncatedFrame.addSuppressed(e); + terminalException = truncatedFrame; + sourceDone = true; + } + } + + private ServerException parseException(byte[] body) { + String message = stripTrailer(new String(body, StandardCharsets.UTF_8)).trim(); + Matcher matcher = ERROR_CODE_PATTERN.matcher(message); + int errorCode = matcher.find() ? Integer.parseInt(matcher.group(1)) : ServerException.CODE_UNKNOWN; + return new ServerException(errorCode, message, transportStatus, queryId); + } + + private String stripTrailer(String body) { + int closingMarker = body.lastIndexOf(EXCEPTION_END_MARKER); + if (closingMarker < 0) { + return body; + } + + String beforeMarker = body.substring(0, closingMarker); + int trailerStart = beforeMarker.lastIndexOf("\r\n"); + if (trailerStart < 0) { + return body; + } + + String trailer = beforeMarker.substring(trailerStart + 2); + int separator = trailer.indexOf(' '); + if (separator <= 0 || !trailer.substring(separator + 1).equals(exceptionTag)) { + return body; + } + for (int i = 0; i < separator; i++) { + if (!Character.isDigit(trailer.charAt(i))) { + return body; + } + } + return beforeMarker.substring(0, trailerStart); + } + + private static int indexOf(byte[] data, int from, int to, byte[] pattern) { + int lastStart = to - pattern.length; + for (int i = from; i <= lastStart; i++) { + int j = 0; + while (j < pattern.length && data[i + j] == pattern[j]) { + j++; + } + if (j == pattern.length) { + return i; + } + } + return -1; + } + + private static int matchingSuffixLength(byte[] data, int from, int to, byte[] pattern) { + int maxLength = Math.min(pattern.length - 1, to - from); + for (int length = maxLength; length > 0; length--) { + int suffixStart = to - length; + int i = 0; + while (i < length && data[suffixStart + i] == pattern[i]) { + i++; + } + if (i == length) { + return length; + } + } + return 0; + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/HttpResponseExceptionTest.java b/client-v2/src/test/java/com/clickhouse/client/api/HttpResponseExceptionTest.java new file mode 100644 index 000000000..de4e55dc2 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/HttpResponseExceptionTest.java @@ -0,0 +1,93 @@ +package com.clickhouse.client.api; + +import com.clickhouse.client.api.http.ClickHouseHttpProto; +import com.clickhouse.client.api.internal.ClickHouseLZ4OutputStream; +import com.clickhouse.client.api.query.QueryResponse; +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.client.WireMock; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import net.jpountz.lz4.LZ4Factory; +import org.apache.hc.core5.http.HttpStatus; +import org.testng.Assert; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.TimeUnit; + +public class HttpResponseExceptionTest { + + @DataProvider(name = "responseCompression") + public Object[][] responseCompression() { + return new Object[][] {{false}, {true}}; + } + + @Test(dataProvider = "responseCompression") + public void shouldThrowServerExceptionWhileReadingSuccessfulResponse(boolean compressedResponse) throws Exception { + String exceptionTag = "0123456789abcdef"; + String queryId = "mid-stream-timeout"; + byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8); + String errorMessage = "Code: 159. DB::Exception: Timeout exceeded. (TIMEOUT_EXCEEDED)"; + String exceptionFrame = "\r\n__exception__\r\n" + exceptionTag + "\r\n" + errorMessage + + "\r\n" + errorMessage.getBytes(StandardCharsets.UTF_8).length + " " + exceptionTag + + "\r\n__exception__\r\n"; + byte[] body = responseBody(resultPrefix, exceptionFrame.getBytes(StandardCharsets.UTF_8), compressedResponse); + + WireMockServer mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort()); + mockServer.start(); + + try { + mockServer.stubFor(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader("X-ClickHouse-Exception-Tag", exceptionTag) + .withHeader(ClickHouseHttpProto.HEADER_QUERY_ID, queryId) + .withBody(body))); + + try (Client client = new Client.Builder() + .addEndpoint("http://localhost:" + mockServer.port()) + .setUsername("default") + .setPassword("") + .setDefaultDatabase("default") + .compressServerResponse(compressedResponse) + .useHttpCompression(false) + .build(); + QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS); + InputStream input = response.getInputStream()) { + byte[] actualPrefix = new byte[resultPrefix.length]; + new DataInputStream(input).readFully(actualPrefix); + Assert.assertEquals(actualPrefix, resultPrefix); + + ServerException exception = Assert.expectThrows(ServerException.class, input::read); + Assert.assertEquals(exception.getCode(), 159); + Assert.assertEquals(exception.getTransportProtocolCode(), HttpStatus.SC_OK); + Assert.assertEquals(exception.getQueryId(), queryId); + Assert.assertTrue(exception.getMessage().startsWith(errorMessage), exception.getMessage()); + } + } finally { + mockServer.stop(); + } + } + + private static byte[] responseBody(byte[] resultPrefix, byte[] exceptionFrame, boolean compressed) + throws IOException { + ByteArrayOutputStream body = new ByteArrayOutputStream(); + if (!compressed) { + body.write(resultPrefix); + body.write(exceptionFrame); + return body.toByteArray(); + } + + try (ClickHouseLZ4OutputStream output = new ClickHouseLZ4OutputStream(body, + LZ4Factory.fastestInstance().fastCompressor(), ClickHouseLZ4OutputStream.UNCOMPRESSED_BUFF_SIZE)) { + output.write(resultPrefix); + output.flush(); + output.write(exceptionFrame); + } + return body.toByteArray(); + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpExceptionInputStreamTest.java b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpExceptionInputStreamTest.java new file mode 100644 index 000000000..dcebd709c --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpExceptionInputStreamTest.java @@ -0,0 +1,115 @@ +package com.clickhouse.client.api.internal; + +import com.clickhouse.client.api.ClientException; +import com.clickhouse.client.api.ServerException; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; + +public class HttpExceptionInputStreamTest { + + private static final String EXCEPTION_TAG = "0123456789abcdef"; + private static final String ERROR_MESSAGE = + "Code: 159. DB::Exception: Timeout exceeded. (TIMEOUT_EXCEEDED)"; + + @Test + public void shouldDetectExceptionAcrossReadBoundaries() throws Exception { + byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8); + byte[] body = responseBody(resultPrefix, exceptionFrame(EXCEPTION_TAG)); + InputStream fragmentedSource = new FilterInputStream(new ByteArrayInputStream(body)) { + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + return super.read(buffer, offset, Math.min(length, 1)); + } + }; + + try (InputStream input = new HttpExceptionInputStream(fragmentedSource, EXCEPTION_TAG, 200, "query-id")) { + byte[] actualPrefix = new byte[resultPrefix.length]; + int offset = 0; + while (offset < actualPrefix.length) { + int read = input.read(actualPrefix, offset, actualPrefix.length - offset); + Assert.assertTrue(read > 0); + offset += read; + } + Assert.assertEquals(actualPrefix, resultPrefix); + + ServerException exception = Assert.expectThrows(ServerException.class, input::read); + Assert.assertEquals(exception.getCode(), 159); + Assert.assertEquals(exception.getQueryId(), "query-id"); + } + } + + @Test + public void shouldIgnoreFrameWithMismatchedTag() throws Exception { + byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8); + byte[] body = responseBody(resultPrefix, exceptionFrame("fedcba9876543210")); + + try (InputStream input = new HttpExceptionInputStream( + new ByteArrayInputStream(body), EXCEPTION_TAG, 200, "query-id")) { + Assert.assertEquals(readAll(input), body); + } + } + + @Test + public void shouldPreserveServerExceptionWhenFrameReadFails() throws Exception { + byte[] resultPrefix = "result-data".getBytes(StandardCharsets.UTF_8); + byte[] body = responseBody(resultPrefix, exceptionFrame(EXCEPTION_TAG)); + InputStream failingSource = new FilterInputStream(new ByteArrayInputStream(body)) { + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + int read = super.read(buffer, offset, length); + if (read < 0) { + throw new IOException("truncated response"); + } + return read; + } + }; + + try (InputStream input = new HttpExceptionInputStream(failingSource, EXCEPTION_TAG, 200, "query-id")) { + byte[] actualPrefix = new byte[resultPrefix.length]; + int offset = 0; + while (offset < actualPrefix.length) { + int read = input.read(actualPrefix, offset, actualPrefix.length - offset); + Assert.assertTrue(read > 0); + offset += read; + } + Assert.assertEquals(actualPrefix, resultPrefix); + + ClientException exception = Assert.expectThrows(ClientException.class, input::read); + Assert.assertTrue(exception.getCause() instanceof ServerException); + Assert.assertEquals(((ServerException) exception.getCause()).getCode(), 159); + Assert.assertEquals(exception.getSuppressed().length, 1); + Assert.assertEquals(exception.getSuppressed()[0].getMessage(), "truncated response"); + } + } + + private static byte[] responseBody(byte[] resultPrefix, byte[] exceptionFrame) throws IOException { + ByteArrayOutputStream body = new ByteArrayOutputStream(); + body.write(resultPrefix); + body.write(exceptionFrame); + return body.toByteArray(); + } + + private static byte[] exceptionFrame(String tag) { + String frame = "\r\n__exception__\r\n" + tag + "\r\n" + ERROR_MESSAGE + + "\r\n" + ERROR_MESSAGE.getBytes(StandardCharsets.UTF_8).length + " " + tag + + "\r\n__exception__\r\n"; + return frame.getBytes(StandardCharsets.UTF_8); + } + + private static byte[] readAll(InputStream input) throws IOException { + ByteArrayOutputStream output = new ByteArrayOutputStream(); + byte[] buffer = new byte[32]; + int read; + while ((read = input.read(buffer)) >= 0) { + output.write(buffer, 0, read); + } + return output.toByteArray(); + } +} diff --git a/docs/features.md b/docs/features.md index 308f694de..02d30bebf 100644 --- a/docs/features.md +++ b/docs/features.md @@ -13,7 +13,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t - Runtime credential updates: Existing `Client` instances can update username/password or bearer-token credentials for subsequent requests without rebuilding the client. - Proxy support: Can send requests through configured HTTP proxies, including proxy credentials. - Connection and socket tuning: Exposes pool sizing, keep-alive, reuse strategy, connect/request/socket timeouts, and low-level socket options. -- Query execution: Executes SQL asynchronously and returns streaming query responses with response metadata and metrics. +- Query execution: Executes SQL asynchronously and returns streaming query responses with response metadata and metrics. When a successful HTTP response ends with a tagged ClickHouse exception frame, the stream validates the frame against the `X-ClickHouse-Exception-Tag` response header and throws `ServerException` when the caller reaches it instead of exposing the frame as result data. - Query settings: Supports per-query database selection, output format, execution limits, roles, log comments, headers, reusable `Session` objects, session settings, server settings, and network timeout overrides. Settings explicitly set to `null` will not be sent to the server. The output format must be set through the query settings (`QuerySettings#setFormat`) and not with a `FORMAT` clause in the query: the client always sends the format of the settings in the `X-ClickHouse-Format` header, and a `26.8+` server uses that header in preference to a `FORMAT` clause in the query. - Parameterized SQL: Accepts named query parameters and can send them through supported HTTP request encodings. - Result materialization helpers: Provides streaming `Records`, generic row access, and convenience APIs that materialize all rows into generic records or typed POJOs. @@ -82,7 +82,7 @@ Compatibility-sensitive traits: - Prepared statements: Supports `?` parameters through client-side SQL rendering and validates that all parameters are bound before execution. - SQL parsing and classification: Classifies SQL to distinguish queries, updates, inserts, `USE`, and role-changing statements, with selectable parser backends. - JDBC escape processing: Translates supported JDBC escape syntax for dates, timestamps, and functions before execution. Escape sequences are only recognized outside of quoted text, so string literals and quoted identifiers — including inlined parameter values that contain `{fn `, `{d '...'}`, or `{ts '...'}` — are passed through unchanged. -- Result set streaming: Streams result sets from ClickHouse binary formats and `FORMAT JSONEachRow`, enforces max-row limits, and manages result-set lifecycle correctly. +- Result set streaming: Streams result sets from ClickHouse binary formats and `FORMAT JSONEachRow`, enforces max-row limits, and manages result-set lifecycle correctly. A server-side `TIMEOUT_EXCEEDED` (code 159) encountered while consuming a result set is reported as `SQLTimeoutException` with SQLState `HYT00`, including when the `ServerException` is nested in a client read failure. - Binary string reads: `ResultSet#getBytes(int|String)` and `ResultSet#getBinaryStream(int|String)` return the raw bytes of a `String`/`FixedString` column. Combined with the `binary_string_support` connection property, non-UTF-8/binary content stored in `String` columns round-trips byte-for-byte; `NULL` values report `null` with `wasNull()` set. `ResultSet#getObject(...)` never exposes the internal `StringValue` holder for these columns: `getObject(column, byte[].class)` returns the raw bytes, while `getObject(column, Object.class)` and the no-type `getObject(column)` overloads return a decoded `String`. - Result-set metadata: Exposes JDBC `ResultSetMetaData` backed by ClickHouse column schema. - Database metadata: Implements JDBC `DatabaseMetaData` for ClickHouse catalogs, schemas, tables, columns, and related capability reporting. diff --git a/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java b/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java index d0947d735..6b9620ef9 100644 --- a/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java +++ b/jdbc-v2/src/main/java/com/clickhouse/jdbc/internal/ExceptionUtils.java @@ -8,6 +8,7 @@ import java.net.MalformedURLException; import java.sql.SQLDataException; import java.sql.SQLException; +import java.sql.SQLTimeoutException; /** * Helper class for building {@link SQLException}. @@ -15,6 +16,7 @@ public final class ExceptionUtils { public static final String SQL_STATE_CLIENT_ERROR = "HY000"; public static final String SQL_STATE_OPERATION_CANCELLED = "HY008"; + public static final String SQL_STATE_TIMEOUT = "HYT00"; public static final String SQL_STATE_CONNECTION_EXCEPTION = "08000"; public static final String SQL_STATE_SQL_ERROR = "07000"; public static final String SQL_STATE_NO_DATA = "02000"; @@ -27,6 +29,8 @@ public final class ExceptionUtils { public static final String SQL_STATE_WRONG_OBJECT_TYPE = "42809"; public static final String SQL_STATE_TYPE_MISMATCH = "2200G"; + private static final int CLICKHOUSE_TIMEOUT_EXCEEDED = 159; + private ExceptionUtils() {}//Private constructor // https://en.wikipedia.org/wiki/SQLSTATE @@ -55,6 +59,12 @@ public static SQLException toSqlState(String message, String debugMessage, Excep if (cause instanceof SQLException) { return (SQLException) cause; + } + + ServerException serverException = findServerException(cause); + if (serverException != null && serverException.getCode() == CLICKHOUSE_TIMEOUT_EXCEEDED) { + String timeoutMessage = message == null ? serverException.getMessage() : message; + return new SQLTimeoutException(timeoutMessage, SQL_STATE_TIMEOUT, serverException.getCode(), cause); } else if (cause instanceof ClientMisconfigurationException) { return new SQLException(exceptionMessage, SQL_STATE_CLIENT_ERROR, cause); } else if (cause instanceof ConnectionInitiationException) { @@ -70,6 +80,18 @@ public static SQLException toSqlState(String message, String debugMessage, Excep return new SQLException(exceptionMessage, SQL_STATE_CLIENT_ERROR, cause);//Default } + private static ServerException findServerException(Throwable throwable) { + for (Throwable cause = throwable; cause != null; cause = cause.getCause()) { + if (cause instanceof ServerException) { + return (ServerException) cause; + } + if (cause.getCause() == cause) { + break; + } + } + return null; + } + public static Throwable getRootCause(Throwable throwable) { Throwable cause = throwable; while (cause.getCause() != null) { diff --git a/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetTimeoutTest.java b/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetTimeoutTest.java new file mode 100644 index 000000000..58396b5ab --- /dev/null +++ b/jdbc-v2/src/test/java/com/clickhouse/jdbc/ResultSetTimeoutTest.java @@ -0,0 +1,74 @@ +package com.clickhouse.jdbc; + +import com.clickhouse.client.api.ClientConfigProperties; +import com.clickhouse.client.api.ServerException; +import com.clickhouse.client.api.http.ClickHouseHttpProto; +import com.clickhouse.jdbc.internal.ExceptionUtils; +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.client.WireMock; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import org.apache.hc.core5.http.HttpStatus; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLTimeoutException; +import java.sql.Statement; +import java.util.Properties; + +public class ResultSetTimeoutTest { + + @Test + public void shouldMapServerTimeoutWhileReadingResultSet() throws Exception { + String exceptionTag = "0123456789abcdef"; + String queryId = "result-set-timeout"; + String errorMessage = "Code: 159. DB::Exception: Timeout exceeded. (TIMEOUT_EXCEEDED)"; + String exceptionFrame = "\r\n__exception__\r\n" + exceptionTag + "\r\n" + errorMessage + + "\r\n" + errorMessage.getBytes(StandardCharsets.UTF_8).length + " " + exceptionTag + + "\r\n__exception__\r\n"; + // RowBinaryWithNamesAndTypes schema followed by two rows; next() prefetches one row ahead. + byte[] rowBinaryResultPrefix = { + 0x01, 0x01, 0x31, 0x05, 0x55, 0x49, 0x6e, 0x74, 0x38, 0x01, 0x02 + }; + ByteArrayOutputStream body = new ByteArrayOutputStream(); + body.write(rowBinaryResultPrefix); + body.write(exceptionFrame.getBytes(StandardCharsets.UTF_8)); + + WireMockServer mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort()); + mockServer.start(); + + try { + mockServer.stubFor(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader(ClickHouseHttpProto.HEADER_EXCEPTION_TAG, exceptionTag) + .withHeader(ClickHouseHttpProto.HEADER_QUERY_ID, queryId) + .withBody(body.toByteArray()))); + + Properties properties = new Properties(); + properties.setProperty(ClientConfigProperties.SERVER_TIMEZONE.getKey(), "UTC"); + properties.setProperty(ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getKey(), "false"); + properties.setProperty(ClientConfigProperties.USE_HTTP_COMPRESSION.getKey(), "false"); + + String jdbcUrl = "jdbc:clickhouse://localhost:" + mockServer.port() + "/default"; + try (Connection connection = new ConnectionImpl(jdbcUrl, properties); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("SELECT 1")) { + Assert.assertTrue(resultSet.next()); + Assert.assertEquals(resultSet.getInt(1), 1); + + SQLTimeoutException exception = Assert.expectThrows(SQLTimeoutException.class, resultSet::next); + Assert.assertEquals(exception.getErrorCode(), 159); + Assert.assertEquals(exception.getSQLState(), ExceptionUtils.SQL_STATE_TIMEOUT); + Assert.assertTrue(ExceptionUtils.getRootCause(exception) instanceof ServerException); + Assert.assertEquals(((ServerException) ExceptionUtils.getRootCause(exception)).getCode(), 159); + Assert.assertTrue(exception.getMessage().startsWith(errorMessage), exception.getMessage()); + } + } finally { + mockServer.stop(); + } + } +}