This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a1b1c8b719b0 CAMEL-26150: camel-clickhouse - return binary query
results as byte[] and fail List inserts with a clear message (#28922)
a1b1c8b719b0 is described below
commit a1b1c8b719b0f1f83e3a2a618c1ae0701675684a
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:47:07 2026 +0100
CAMEL-26150: camel-clickhouse - return binary query results as byte[] and
fail List inserts with a clear message (#28922)
* CAMEL-26150: camel-clickhouse - return binary query results as byte[]…
… and fail List inserts with a clear message
A query with a binary format (RowBinary, Native, Parquet, ...) decoded the
result as UTF-8 text into a String, which replaced the bytes that are not
valid UTF-8. Binary formats now give a byte[] body; text formats keep a
String.
Inserting a List of Maps, or a List of objects when no class is registered
on
the Client for the table, failed with a NullPointerException from client-v2.
Both now fail with a ClickHouseException that says why.
The CamelClickHouseReadRows header is documented as the rows the server
read,
not the rows in the result.
* CAMEL-26150: camel-clickhouse - the insert error does not fail on an …
…empty List
When the client threw a NullPointerException for an empty List body,
building the message with
batch.get(0) threw an IndexOutOfBoundsException and hid the cause. The
message now handles an
empty List (and a null first element).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/catalog/components/clickhouse.json | 4 +-
.../camel/catalog/docs/clickhouse-component.adoc | 13 ++--
.../camel/component/clickhouse/clickhouse.json | 4 +-
.../src/main/docs/clickhouse-component.adoc | 13 ++--
.../component/clickhouse/ClickHouseConstants.java | 4 +-
.../component/clickhouse/ClickHouseEndpoint.java | 13 ++--
.../component/clickhouse/ClickHouseProducer.java | 55 +++++++++++---
.../clickhouse/ClickHouseProducerTest.java | 85 +++++++++++++++++++++-
.../integration/ClickHouseProducerIT.java | 42 +++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 +++
.../dsl/ClickHouseEndpointBuilderFactory.java | 4 +-
11 files changed, 213 insertions(+), 35 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/clickhouse.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/clickhouse.json
index db94b15eabb7..25b52c50cbcc 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/clickhouse.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/clickhouse.json
@@ -39,12 +39,12 @@
"CamelClickHouseTable": { "index": 2, "kind": "header", "displayName": "",
"group": "producer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Overrides the target table configured on the endpoint.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#TABLE" },
"CamelClickHouseFormat": { "index": 3, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Overrides the data format configured on the endpoint.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#FORMAT" },
"CamelClickHouseWrittenRows": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of rows written by an insert
operation.", "constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#WRITTEN_ROWS" },
- "CamelClickHouseReadRows": { "index": 5, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The number of rows read by a query operation.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#READ_ROWS" },
+ "CamelClickHouseReadRows": { "index": 5, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The number of rows the server read (scanned) to run a
query operation. This is not the number of rows in the result: with a WHERE or
a LIMIT it is usually more.", "constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#RE [...]
"CamelClickHousePingOk": { "index": 6, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The boolean result of a ping operation.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#PING_OK" }
},
"bodyTypes": {
"producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"byOption": [ { "option": "operation", "value": "QUERY", "carriers": [ "text",
"bytes", "stream", "file" ] }, { "option": "operation", "value": "PING",
"carriers": [ "any" ] } ], "description": "The rows to insert, in the format
option (JSONEachRow: one JSON object per line), as text, bytes, a stream or a
file; or a List of objects whose class is registered on the Client. A Map or a
List of Maps fails: marshal [...]
- "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "QUERY", "carriers": [ "text" ], "javaType":
"java.lang.String" }, { "option": "operation", "value": "PING", "carriers": [
"value" ], "javaType": "java.lang.Boolean" } ], "description": "Insert leaves
the body unchanged, with the count in the CamelClickHouseWrittenRows header.
Query gives the result as a String in the format option (JSONEachRow: one JSON
object per line); binary formats such [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "QUERY", "carriers": [ "text", "bytes" ] }, { "option":
"operation", "value": "PING", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" } ], "description": "Insert leaves the body unchanged, with
the count in the CamelClickHouseWrittenRows header. Query gives the result in
the format option: a String for text formats (JSONEachRow: one JSON object per
line), a byte[] for binary formats [...]
},
"properties": {
"database": { "index": 0, "kind": "path", "displayName": "Database",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The ClickHouse database. A
table may also be provided using the database.table syntax." },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/clickhouse-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/clickhouse-component.adoc
index add6e2933a59..a2a03d957ed4 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/clickhouse-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/clickhouse-component.adoc
@@ -60,11 +60,14 @@ The component is producer-only and supports the following
operations, selected w
| Operation | Description
| `insert` (default) | Streams the message body into the target table using
the configured `format`. Accepted body
types are `InputStream`, `byte[]`, `String`, `java.io.File` (matched to the
`format`), or a `List` of objects
-whose class has been registered on the shared `Client`. When the body is a
`List` and `batchSize` is greater
-than 0, the list is split into batches of that size, each sent as a separate
insert. The number of written
-rows is set on the `CamelClickHouseWrittenRows` header.
-| `query` | Runs the SQL statement contained in the message body and returns
the result set (in the configured
-`format`) as a `String` body. The number of read rows is set on the
`CamelClickHouseReadRows` header.
+whose class has been registered on the shared `Client` with
`Client.register(Class, TableSchema)`. A `List` of
+`Map` is not supported: marshal the rows to the `format` (such as
`JSONEachRow`) first. When the body is a `List`
+and `batchSize` is greater than 0, the list is split into batches of that
size, each sent as a separate insert.
+The number of written rows is set on the `CamelClickHouseWrittenRows` header.
+| `query` | Runs the SQL statement contained in the message body and returns
the result set in the configured
+`format`: a `String` body for text formats (such as `JSONEachRow`, `CSV` or
`TSV`), and a `byte[]` body for binary
+formats (such as `RowBinary`, `Native`, `Parquet`, `Arrow` or `Avro`). The
`CamelClickHouseReadRows` header is set
+to the number of rows the server read to run the query, which is not the
number of rows in the result.
| `ping` | Pings the ClickHouse server and sets the boolean result on both the
message body and the
`CamelClickHousePingOk` header.
|===
diff --git
a/components/camel-clickhouse/src/generated/resources/META-INF/org/apache/camel/component/clickhouse/clickhouse.json
b/components/camel-clickhouse/src/generated/resources/META-INF/org/apache/camel/component/clickhouse/clickhouse.json
index db94b15eabb7..25b52c50cbcc 100644
---
a/components/camel-clickhouse/src/generated/resources/META-INF/org/apache/camel/component/clickhouse/clickhouse.json
+++
b/components/camel-clickhouse/src/generated/resources/META-INF/org/apache/camel/component/clickhouse/clickhouse.json
@@ -39,12 +39,12 @@
"CamelClickHouseTable": { "index": 2, "kind": "header", "displayName": "",
"group": "producer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Overrides the target table configured on the endpoint.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#TABLE" },
"CamelClickHouseFormat": { "index": 3, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "Overrides the data format configured on the endpoint.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#FORMAT" },
"CamelClickHouseWrittenRows": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The number of rows written by an insert
operation.", "constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#WRITTEN_ROWS" },
- "CamelClickHouseReadRows": { "index": 5, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The number of rows read by a query operation.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#READ_ROWS" },
+ "CamelClickHouseReadRows": { "index": 5, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The number of rows the server read (scanned) to run a
query operation. This is not the number of rows in the result: with a WHERE or
a LIMIT it is usually more.", "constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#RE [...]
"CamelClickHousePingOk": { "index": 6, "kind": "header", "displayName":
"", "group": "producer", "label": "", "required": false, "javaType": "boolean",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The boolean result of a ping operation.",
"constantName":
"org.apache.camel.component.clickhouse.ClickHouseConstants#PING_OK" }
},
"bodyTypes": {
"producerIn": { "carriers": [ "text", "bytes", "stream", "file", "list" ],
"byOption": [ { "option": "operation", "value": "QUERY", "carriers": [ "text",
"bytes", "stream", "file" ] }, { "option": "operation", "value": "PING",
"carriers": [ "any" ] } ], "description": "The rows to insert, in the format
option (JSONEachRow: one JSON object per line), as text, bytes, a stream or a
file; or a List of objects whose class is registered on the Client. A Map or a
List of Maps fails: marshal [...]
- "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "QUERY", "carriers": [ "text" ], "javaType":
"java.lang.String" }, { "option": "operation", "value": "PING", "carriers": [
"value" ], "javaType": "java.lang.Boolean" } ], "description": "Insert leaves
the body unchanged, with the count in the CamelClickHouseWrittenRows header.
Query gives the result as a String in the format option (JSONEachRow: one JSON
object per line); binary formats such [...]
+ "producerOut": { "carriers": [ "unchanged" ], "byOption": [ { "option":
"operation", "value": "QUERY", "carriers": [ "text", "bytes" ] }, { "option":
"operation", "value": "PING", "carriers": [ "value" ], "javaType":
"java.lang.Boolean" } ], "description": "Insert leaves the body unchanged, with
the count in the CamelClickHouseWrittenRows header. Query gives the result in
the format option: a String for text formats (JSONEachRow: one JSON object per
line), a byte[] for binary formats [...]
},
"properties": {
"database": { "index": 0, "kind": "path", "displayName": "Database",
"group": "producer", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The ClickHouse database. A
table may also be provided using the database.table syntax." },
diff --git
a/components/camel-clickhouse/src/main/docs/clickhouse-component.adoc
b/components/camel-clickhouse/src/main/docs/clickhouse-component.adoc
index add6e2933a59..a2a03d957ed4 100644
--- a/components/camel-clickhouse/src/main/docs/clickhouse-component.adoc
+++ b/components/camel-clickhouse/src/main/docs/clickhouse-component.adoc
@@ -60,11 +60,14 @@ The component is producer-only and supports the following
operations, selected w
| Operation | Description
| `insert` (default) | Streams the message body into the target table using
the configured `format`. Accepted body
types are `InputStream`, `byte[]`, `String`, `java.io.File` (matched to the
`format`), or a `List` of objects
-whose class has been registered on the shared `Client`. When the body is a
`List` and `batchSize` is greater
-than 0, the list is split into batches of that size, each sent as a separate
insert. The number of written
-rows is set on the `CamelClickHouseWrittenRows` header.
-| `query` | Runs the SQL statement contained in the message body and returns
the result set (in the configured
-`format`) as a `String` body. The number of read rows is set on the
`CamelClickHouseReadRows` header.
+whose class has been registered on the shared `Client` with
`Client.register(Class, TableSchema)`. A `List` of
+`Map` is not supported: marshal the rows to the `format` (such as
`JSONEachRow`) first. When the body is a `List`
+and `batchSize` is greater than 0, the list is split into batches of that
size, each sent as a separate insert.
+The number of written rows is set on the `CamelClickHouseWrittenRows` header.
+| `query` | Runs the SQL statement contained in the message body and returns
the result set in the configured
+`format`: a `String` body for text formats (such as `JSONEachRow`, `CSV` or
`TSV`), and a `byte[]` body for binary
+formats (such as `RowBinary`, `Native`, `Parquet`, `Arrow` or `Avro`). The
`CamelClickHouseReadRows` header is set
+to the number of rows the server read to run the query, which is not the
number of rows in the result.
| `ping` | Pings the ClickHouse server and sets the boolean result on both the
message body and the
`CamelClickHousePingOk` header.
|===
diff --git
a/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseConstants.java
b/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseConstants.java
index 1bc03b65844c..7cd3dfae9d25 100644
---
a/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseConstants.java
+++
b/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseConstants.java
@@ -31,7 +31,9 @@ public final class ClickHouseConstants {
public static final String FORMAT = "CamelClickHouseFormat";
@Metadata(description = "The number of rows written by an insert
operation.", javaType = "long")
public static final String WRITTEN_ROWS = "CamelClickHouseWrittenRows";
- @Metadata(description = "The number of rows read by a query operation.",
javaType = "long")
+ @Metadata(description = "The number of rows the server read (scanned) to
run a query operation. This is not the"
+ + " number of rows in the result: with a WHERE or
a LIMIT it is usually more.",
+ javaType = "long")
public static final String READ_ROWS = "CamelClickHouseReadRows";
@Metadata(description = "The boolean result of a ping operation.",
javaType = "boolean")
public static final String PING_OK = "CamelClickHousePingOk";
diff --git
a/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseEndpoint.java
b/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseEndpoint.java
index f788c6bfa659..7b4f3230a561 100644
---
a/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseEndpoint.java
+++
b/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseEndpoint.java
@@ -51,15 +51,16 @@ import org.apache.camel.util.ObjectHelper;
+ " it first. Query takes the
SQL statement"),
producerOut = @BodyType(value = Carrier.UNCHANGED,
byOption = {
- @BodyType.When(option =
"operation", value = "QUERY", carriers = Carrier.TEXT,
- javaType =
"java.lang.String"),
+ @BodyType.When(option =
"operation", value = "QUERY",
+ carriers = {
Carrier.TEXT, Carrier.BYTES }),
@BodyType.When(option =
"operation", value = "PING", carriers = Carrier.VALUE,
javaType =
"java.lang.Boolean") },
description = "Insert leaves the body
unchanged, with the count in the"
- + "
CamelClickHouseWrittenRows header. Query gives the result as a String"
- + " in the format option
(JSONEachRow: one JSON object per line); binary"
- + " formats such as RowBinary
do not survive the String. Ping gives a"
- + " Boolean. The
CamelClickHouseOperation header overrides the operation"))
+ + "
CamelClickHouseWrittenRows header. Query gives the result in the"
+ + " format option: a String
for text formats (JSONEachRow: one JSON object"
+ + " per line), a byte[] for
binary formats such as RowBinary, Native or"
+ + " Parquet. Ping gives a
Boolean. The CamelClickHouseOperation header"
+ + " overrides the operation"))
public class ClickHouseEndpoint extends DefaultEndpoint {
private Client client;
diff --git
a/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseProducer.java
b/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseProducer.java
index 823f6a5f7740..1dea7d1c7ac3 100644
---
a/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseProducer.java
+++
b/components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseProducer.java
@@ -22,6 +22,8 @@ import java.io.FileInputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import com.clickhouse.client.api.Client;
import com.clickhouse.client.api.insert.InsertResponse;
@@ -122,24 +124,46 @@ public class ClickHouseProducer extends DefaultProducer {
}
private long insertList(Client client, String target, List<?> data,
InsertSettings settings) throws Exception {
+ if (!data.isEmpty() && data.get(0) instanceof Map) {
+ throw new ClickHouseException(
+ "Cannot insert a List of Maps into " + target + ": a List
body must hold objects whose class is"
+ + " registered on the Client with
Client.register(Class, TableSchema)."
+ + " Marshal the rows to the format
option (such as JSONEachRow) instead.");
+ }
int batchSize = endpoint.getBatchSize();
if (batchSize <= 0 || data.size() <= batchSize) {
- try (InsertResponse response = client.insert(target, data,
settings).get()) {
- return response.getWrittenRows();
- }
+ return insertBatch(client, target, data, settings);
}
// split the list into batches of batchSize, inserting each batch
separately
long writtenRows = 0;
for (int start = 0; start < data.size(); start += batchSize) {
List<?> batch = data.subList(start, Math.min(start + batchSize,
data.size()));
- try (InsertResponse response = client.insert(target, batch,
settings).get()) {
- writtenRows += response.getWrittenRows();
- }
+ writtenRows += insertBatch(client, target, batch, settings);
}
return writtenRows;
}
+ private long insertBatch(Client client, String target, List<?> batch,
InsertSettings settings) throws Exception {
+ CompletableFuture<InsertResponse> future;
+ try {
+ future = client.insert(target, batch, settings);
+ } catch (NullPointerException e) {
+ // client-v2 throws a NullPointerException that does not say why
when no class is registered for the table
+ Object first = batch.isEmpty() ? null : batch.get(0);
+ String what = first != null ? "a List of " +
first.getClass().getName() : "the List";
+ throw new ClickHouseException(
+ "Cannot insert " + what + " into " + target
+ + ": no class is registered on the
Client for this table. Register the class"
+ + " with Client.register(Class,
TableSchema), or send the rows as text, bytes,"
+ + " a stream or a file in the format
option.",
+ e);
+ }
+ try (InsertResponse response = future.get()) {
+ return response.getWrittenRows();
+ }
+ }
+
private void doQuery(Exchange exchange) throws Exception {
Client client = endpoint.getClient();
String sql = exchange.getIn().getBody(String.class);
@@ -147,21 +171,27 @@ public class ClickHouseProducer extends DefaultProducer {
throw new ClickHouseException("The SQL query must be provided in
the message body for the query operation");
}
+ ClickHouseFormat format = resolveFormat(exchange);
QuerySettings settings = new QuerySettings();
- settings.setFormat(resolveFormat(exchange));
+ settings.setFormat(format);
String database = resolveDatabase(exchange);
if (ObjectHelper.isNotEmpty(database)) {
settings.setDatabase(database);
}
try (QueryResponse response = client.query(sql, settings).get()) {
- String result;
+ byte[] result;
try (InputStream is = response.getInputStream()) {
- result = new String(is.readAllBytes(), StandardCharsets.UTF_8);
+ result = is.readAllBytes();
}
Message message = exchange.getMessage();
message.setHeader(ClickHouseConstants.READ_ROWS,
response.getReadRows());
- message.setBody(result);
+ if (isBinary(format)) {
+ // binary formats such as RowBinary, Native or Parquet are not
valid UTF-8 text
+ message.setBody(result);
+ } else {
+ message.setBody(new String(result, StandardCharsets.UTF_8));
+ }
}
}
@@ -188,6 +218,11 @@ public class ClickHouseProducer extends DefaultProducer {
return ClickHouseFormat.valueOf(format);
}
+ private static boolean isBinary(ClickHouseFormat format) {
+ // NDJSON is JSON text (one object per line), but clickhouse-data
flags it as binary
+ return format.isBinary() && format != ClickHouseFormat.NDJSON;
+ }
+
private String resolveDatabase(Exchange exchange) {
String database =
exchange.getIn().getHeader(ClickHouseConstants.DATABASE, String.class);
if (ObjectHelper.isNotEmpty(database)) {
diff --git
a/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/ClickHouseProducerTest.java
b/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/ClickHouseProducerTest.java
index 2b9e08b1e57c..e25747a17896 100644
---
a/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/ClickHouseProducerTest.java
+++
b/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/ClickHouseProducerTest.java
@@ -31,6 +31,7 @@ import com.clickhouse.client.api.query.QueryResponse;
import com.clickhouse.client.api.query.QuerySettings;
import com.clickhouse.data.ClickHouseFormat;
import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
import org.apache.camel.WrappedFile;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
@@ -44,6 +45,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -97,6 +99,12 @@ class ClickHouseProducerTest extends CamelTestSupport {
from("direct:query")
.to("clickhouse://analytics?operation=query&format=CSV")
.to("mock:result");
+ from("direct:queryRowBinary")
+
.to("clickhouse://analytics?operation=query&format=RowBinary")
+ .to("mock:result");
+ from("direct:queryNdjson")
+
.to("clickhouse://analytics?operation=query&format=NDJSON")
+ .to("mock:result");
from("direct:ping")
.to("clickhouse://analytics?operation=ping")
.to("mock:result");
@@ -138,18 +146,60 @@ class ClickHouseProducerTest extends CamelTestSupport {
@Test
void insertUsesNativeListApiForListBodies() throws Exception {
- List<Map<String, Object>> rows = List.of(Map.of("id", 1, "name",
"alice"), Map.of("id", 2, "name", "bob"));
+ List<Event> rows = List.of(new Event(1, "alice"), new Event(2, "bob"));
template.sendBody("direct:insert", rows);
verify(client).insert(anyString(), anyList(),
any(InsertSettings.class));
}
+ @Test
+ void insertListOfMapsFailsWithClearMessage() {
+ List<Map<String, Object>> rows = List.of(Map.of("id", 1, "name",
"alice"));
+
+ Exchange out = template.send("direct:insert", exchange ->
exchange.getIn().setBody(rows));
+
+ assertThat(out.getException())
+ .isInstanceOf(ClickHouseException.class)
+ .hasMessageContaining("List of Maps")
+ .hasMessageContaining("JSONEachRow");
+ verify(client, never()).insert(anyString(), anyList(),
any(InsertSettings.class));
+ }
+
+ @Test
+ void insertListWithoutRegisteredClassFailsWithClearMessage() {
+ // client-v2 throws a NullPointerException when no class was
registered for the table
+ when(client.insert(anyString(), anyList(),
any(InsertSettings.class))).thenThrow(new NullPointerException());
+
+ Exchange out = template.send("direct:insert",
+ exchange -> exchange.getIn().setBody(List.of(new Event(1,
"alice"))));
+
+ assertThat(out.getException())
+ .isInstanceOf(ClickHouseException.class)
+ .hasMessageContaining(Event.class.getName())
+ .hasMessageContaining("Client.register")
+ .hasCauseInstanceOf(NullPointerException.class);
+ }
+
+ @Test
+ void insertEmptyListKeepsTheClientErrorAsCause() {
+ // the message must not fail on the empty batch and hide the
NullPointerException of the client
+ when(client.insert(anyString(), anyList(),
any(InsertSettings.class))).thenThrow(new NullPointerException());
+
+ Exchange out = template.send("direct:insert", exchange ->
exchange.getIn().setBody(List.of()));
+
+ assertThat(out.getException())
+ .isInstanceOf(ClickHouseException.class)
+ .hasMessageContaining("Cannot insert the List")
+ .hasMessageContaining("Client.register")
+ .hasCauseInstanceOf(NullPointerException.class);
+ }
+
@Test
void insertSplitsListIntoBatchesWhenBatchSizeSet() throws Exception {
// batchSize=2 with 5 rows -> 3 insert calls (2 + 2 + 1)
- List<Map<String, Object>> rows = List.of(
- Map.of("id", 1), Map.of("id", 2), Map.of("id", 3),
Map.of("id", 4), Map.of("id", 5));
+ List<Event> rows = List.of(
+ new Event(1, "a"), new Event(2, "b"), new Event(3, "c"), new
Event(4, "d"), new Event(5, "e"));
template.sendBody("direct:insertBatched", rows);
@@ -206,6 +256,32 @@ class ClickHouseProducerTest extends CamelTestSupport {
assertThat(sqlCaptor.getValue()).isEqualTo("SELECT * FROM
analytics.events");
}
+ @Test
+ void queryWithBinaryFormatReturnsBytes() throws Exception {
+ // RowBinary of (UInt32 200, String 'é'): bytes that are not valid
UTF-8 on their own
+ byte[] rowBinary = { (byte) 0xC8, 0x00, 0x00, 0x00, 0x02, (byte) 0xC3,
(byte) 0xA9 };
+ when(queryResponse.getInputStream()).thenReturn(new
ByteArrayInputStream(rowBinary));
+
+ Object result = template.requestBody("direct:queryRowBinary", "SELECT
id, name FROM analytics.events");
+
+ assertThat(result).isInstanceOf(byte[].class);
+ assertThat((byte[]) result).containsExactly(rowBinary);
+
+ ArgumentCaptor<QuerySettings> settingsCaptor =
ArgumentCaptor.forClass(QuerySettings.class);
+ verify(client).query(anyString(), settingsCaptor.capture());
+
assertThat(settingsCaptor.getValue().getFormat()).isEqualTo(ClickHouseFormat.RowBinary);
+ }
+
+ @Test
+ void queryWithNdjsonFormatReturnsString() throws Exception {
+ when(queryResponse.getInputStream())
+ .thenReturn(new
ByteArrayInputStream("{\"id\":1,\"name\":\"alice\"}\n".getBytes(StandardCharsets.UTF_8)));
+
+ Object result = template.requestBody("direct:queryNdjson", "SELECT id,
name FROM analytics.events");
+
+ assertThat(result).isEqualTo("{\"id\":1,\"name\":\"alice\"}\n");
+ }
+
@Test
void pingReturnsBooleanBodyAndHeader() throws Exception {
Object result = template.requestBody("direct:ping", (Object) null);
@@ -249,4 +325,7 @@ class ClickHouseProducerTest extends CamelTestSupport {
.isInstanceOf(ClickHouseException.class)
.hasMessageContaining("SQL query");
}
+
+ public record Event(int id, String name) {
+ }
}
diff --git
a/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/integration/ClickHouseProducerIT.java
b/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/integration/ClickHouseProducerIT.java
index 03a141356a96..68bb594b8b47 100644
---
a/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/integration/ClickHouseProducerIT.java
+++
b/components/camel-clickhouse/src/test/java/org/apache/camel/component/clickhouse/integration/ClickHouseProducerIT.java
@@ -16,11 +16,16 @@
*/
package org.apache.camel.component.clickhouse.integration;
+import java.util.List;
+import java.util.Map;
+
import com.clickhouse.client.api.Client;
import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.clickhouse.ClickHouseComponent;
import org.apache.camel.component.clickhouse.ClickHouseConstants;
+import org.apache.camel.component.clickhouse.ClickHouseException;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.infra.clickhouse.services.ClickHouseService;
import
org.apache.camel.test.infra.clickhouse.services.ClickHouseServiceFactory;
@@ -83,6 +88,9 @@ class ClickHouseProducerIT extends CamelTestSupport {
from("direct:query")
.to("clickhouse://" + service.getDatabaseName() +
"?operation=query&format=CSV")
.to("mock:result");
+ from("direct:queryRowBinary")
+ .to("clickhouse://" + service.getDatabaseName() +
"?operation=query&format=RowBinary")
+ .to("mock:result");
from("direct:ping")
.to("clickhouse://" + service.getDatabaseName() +
"?operation=ping")
.to("mock:result");
@@ -108,9 +116,43 @@ class ClickHouseProducerIT extends CamelTestSupport {
assertThat(result.trim()).isEqualTo("2");
}
+ @Test
+ void queryWithBinaryFormatReturnsBytes() {
+ template.sendBody("direct:insert",
"{\"id\":200,\"name\":\"\u00e9\"}\n");
+
+ Object result = template.requestBody("direct:queryRowBinary",
+ "SELECT id, name FROM " + service.getDatabaseName() +
".camel_events");
+
+ // RowBinary: UInt32 200 little endian, then the String as its length
and its UTF-8 bytes
+ assertThat(result).isInstanceOf(byte[].class);
+ assertThat((byte[]) result).containsExactly((byte) 0xC8, 0x00, 0x00,
0x00, 0x02, (byte) 0xC3, (byte) 0xA9);
+ }
+
+ @Test
+ void insertListOfMapsFailsWithClearMessage() {
+ Exchange out = template.send("direct:insert",
+ exchange -> exchange.getIn().setBody(List.of(Map.of("id", 1,
"name", "alice"))));
+
+
assertThat(out.getException()).isInstanceOf(ClickHouseException.class).hasMessageContaining("List
of Maps");
+ }
+
+ @Test
+ void insertListWithoutRegisteredClassFailsWithClearMessage() {
+ Exchange out = template.send("direct:insert",
+ exchange -> exchange.getIn().setBody(List.of(new Event(1,
"alice"))));
+
+ assertThat(out.getException())
+ .isInstanceOf(ClickHouseException.class)
+ .hasMessageContaining(Event.class.getName())
+ .hasMessageContaining("Client.register");
+ }
+
@Test
void pingReturnsTrue() {
Boolean result = template.requestBody("direct:ping", null,
Boolean.class);
assertThat(result).isTrue();
}
+
+ public record Event(int id, String name) {
+ }
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index e6138b4d018b..31e006dc2fbf 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -653,6 +653,17 @@ message sent to that endpoint returned only the first
page, and after one messag
`CamelOaimphIgnoreSSLWarnings=true` every later message accepted a self-signed
certificate. A message without the
header uses the `onlyFirst` and `ignoreSSLWarnings` options of the endpoint.
+=== camel-clickhouse - a query in a binary format returns a byte[] body
+
+A `query` with a binary `format` (such as `RowBinary`, `Native`, `Parquet`,
`Arrow` or `Avro`) now returns the result
+as a `byte[]` body. Before, the bytes were decoded as UTF-8 text into a
`String`, which replaced the bytes that are not
+valid UTF-8, so the result could not be read. A query with a text format (such
as `JSONEachRow`, `CSV` or `TSV`) still
+returns a `String`.
+
+Inserting a `List` of `Map` now fails with a `ClickHouseException` that says
why, and so does a `List` of objects when no
+class is registered on the `Client` for the table. Before, both failed with a
`NullPointerException` from the ClickHouse
+client.
+
=== camel-cm-sms - the from of an SMSMessage is the sender of the SMS
An `SMSMessage` with a `from` is now sent with that sender. Before, the sender
was set only when `from` was empty (to
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ClickHouseEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ClickHouseEndpointBuilderFactory.java
index 43d70e9e4b28..edb2fcf14583 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ClickHouseEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/ClickHouseEndpointBuilderFactory.java
@@ -511,7 +511,9 @@ public interface ClickHouseEndpointBuilderFactory {
return "CamelClickHouseWrittenRows";
}
/**
- * The number of rows read by a query operation.
+ * The number of rows the server read (scanned) to run a query
+ * operation. This is not the number of rows in the result: with a
WHERE
+ * or a LIMIT it is usually more.
*
* The option is a: {@code long} type.
*
