You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ja...@apache.org on 2026/08/11 14:25:24 UTC
(camel-quarkus) 01/11: Fixes #8878. Add integration test coverage for new weaviate operations (BATCH_CREATE, HYBRID_QUERY, BM25_QUERY, AGGREGATE)
This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
commit 2d9bf5a5e3343d1e8891d38c7f80ed047d62d308
Author: Jiri Ondrusek <on...@gmail.com>
AuthorDate: Tue Jul 14 11:54:29 2026 +0200
Fixes #8878. Add integration test coverage for new weaviate operations (BATCH_CREATE, HYBRID_QUERY, BM25_QUERY, AGGREGATE)
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---
.../component/weaviate/it/WeaviateResource.java | 95 ++++++++++++----
.../component/weaviate/it/WeaviateTest.java | 122 +++++++++++++++++++++
2 files changed, 194 insertions(+), 23 deletions(-)
diff --git a/integration-tests/weaviate/src/main/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateResource.java b/integration-tests/weaviate/src/main/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateResource.java
index b31e40ab1a..331d1daf75 100644
--- a/integration-tests/weaviate/src/main/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateResource.java
+++ b/integration-tests/weaviate/src/main/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateResource.java
@@ -19,11 +19,14 @@ package org.apache.camel.quarkus.component.weaviate.it;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.stream.Collectors;
-import io.weaviate.client.base.Result;
-import io.weaviate.client.v1.data.model.WeaviateObject;
-import io.weaviate.client.v1.graphql.model.GraphQLResponse;
+import io.weaviate.client6.v1.api.collections.Vectors;
+import io.weaviate.client6.v1.api.collections.WeaviateObject;
+import io.weaviate.client6.v1.api.collections.aggregate.AggregateResponse;
+import io.weaviate.client6.v1.api.collections.data.InsertManyResponse;
+import io.weaviate.client6.v1.api.collections.query.QueryResponse;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.POST;
@@ -33,6 +36,8 @@ import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
+import org.apache.camel.component.weaviate.WeaviateVectorDbAction;
+import org.apache.camel.component.weaviate.WeaviateVectorDbHeaders;
import org.jboss.logging.Logger;
@Path("/weaviate")
@@ -40,6 +45,8 @@ import org.jboss.logging.Logger;
public class WeaviateResource {
public static final String WEAVIATE_CONTAINER_ADDRESS = "cq.weaviate.container.address";
+ public static final String WEAVIATE_CONTAINER_GRPC_HOST = "cq.weaviate.container.grpc.host";
+ public static final String WEAVIATE_CONTAINER_GRPC_PORT = "cq.weaviate.container.grpc.port";
public static final String WEAVIATE_API_KEY_ENV = "WEAVIATE_API_KEY";
public static final String WEAVIATE_HOST_ENV = "WEAVIATE_HOST";
@@ -48,6 +55,7 @@ public class WeaviateResource {
@Inject
CamelContext context;
+ @SuppressWarnings("unchecked")
@Path("/request")
@POST
@Produces(MediaType.APPLICATION_JSON)
@@ -57,10 +65,33 @@ public class WeaviateResource {
Object body = headers.get("body");
headers.remove("body");
- //convert Double to Float in body list (for create)
if (body instanceof List) {
- body = ((List<?>) body).stream().map(o -> o instanceof Double ? ((Double) o).floatValue() : o)
- .collect(Collectors.toList());
+ String action = String.valueOf(headers.get(WeaviateVectorDbHeaders.ACTION));
+ if (WeaviateVectorDbAction.BATCH_CREATE.name().equals(action)) {
+ body = ((List<Map<String, Object>>) body).stream().map(m -> {
+ Map<String, Object> props = (Map<String, Object>) m.get("properties");
+ List<Number> vector = (List<Number>) m.get("vector");
+ float[] floatVector = new float[vector.size()];
+ for (int i = 0; i < vector.size(); i++) {
+ floatVector[i] = vector.get(i).floatValue();
+ }
+ return new WeaviateObject.Builder<Map<String, Object>>()
+ .properties(props)
+ .vectors(Vectors.of(floatVector))
+ .build();
+ }).collect(Collectors.toList());
+ } else {
+ body = ((List<?>) body).stream().map(o -> o instanceof Double ? ((Double) o).floatValue() : o)
+ .collect(Collectors.toList());
+ }
+ }
+
+ Object queryVector = headers.get(WeaviateVectorDbHeaders.QUERY_VECTOR);
+ if (queryVector instanceof List) {
+ headers.put(WeaviateVectorDbHeaders.QUERY_VECTOR,
+ ((List<?>) queryVector).stream()
+ .map(o -> o instanceof Double ? ((Double) o).floatValue() : o)
+ .collect(Collectors.toList()));
}
Exchange response = context.createFluentProducerTemplate()
@@ -69,25 +100,43 @@ public class WeaviateResource {
.withHeaders(headers)
.request(Exchange.class);
- Result<?> result = response.getIn().getBody(Result.class);
- LOG.debugf("Response for collections with headers (%s) is: \"%s\".", headers, result);
+ if (response.getException() != null) {
+ HashMap<String, Object> errorMap = new HashMap<>();
+ errorMap.put("error", response.getException().getMessage());
+ return Response.ok(errorMap).build();
+ }
+
+ Object result = response.getIn().getBody();
+ LOG.debugf("Response for request with headers (%s) is: \"%s\".", headers, result);
if (result != null) {
- HashMap<String, Object> map = new HashMap();
- map.put("error", result.getError() == null ? "" : result.getError());
-
- if (result.getResult() instanceof Boolean) {
- map.put("result", result.getResult());
- } else if (result.getResult() instanceof WeaviateObject) {
- map.put("result", ((WeaviateObject) result.getResult()).getId());
- map.put("resultProperties", ((WeaviateObject) result.getResult()).getProperties());
- } else if (result.getResult() instanceof List) {
- @SuppressWarnings("unchecked")
- List<WeaviateObject> objects = (List<WeaviateObject>) result.getResult();
- map.put("result",
- objects.stream().collect(Collectors.toMap(WeaviateObject::getId, WeaviateObject::getProperties)));
- } else if (result.getResult() instanceof GraphQLResponse) {
- map.put("result", result.getResult());
+ HashMap<String, Object> map = new HashMap<>();
+
+ if (result instanceof Boolean) {
+ map.put("result", result);
+ } else if (result instanceof WeaviateObject) {
+ WeaviateObject<Map<String, Object>> wo = (WeaviateObject<Map<String, Object>>) result;
+ map.put("result", wo.uuid());
+ map.put("resultProperties", wo.properties());
+ } else if (result instanceof Optional) {
+ Optional<WeaviateObject<Map<String, Object>>> opt = (Optional<WeaviateObject<Map<String, Object>>>) result;
+ if (opt.isPresent()) {
+ WeaviateObject<Map<String, Object>> wo = opt.get();
+ map.put("result", Map.of(wo.uuid(), wo.properties()));
+ }
+ } else if (result instanceof InsertManyResponse) {
+ InsertManyResponse imr = (InsertManyResponse) result;
+ map.put("uuids", imr.uuids());
+ map.put("errors", imr.errors());
+ } else if (result instanceof AggregateResponse) {
+ AggregateResponse ar = (AggregateResponse) result;
+ map.put("totalCount", ar.totalCount());
+ } else if (result instanceof QueryResponse) {
+ QueryResponse<Map<String, Object>> qr = (QueryResponse<Map<String, Object>>) result;
+ List<Map<String, Object>> objects = qr.objects().stream()
+ .map(WeaviateObject::properties)
+ .collect(Collectors.toList());
+ map.put("result", objects);
}
return Response.ok(map).build();
diff --git a/integration-tests/weaviate/src/test/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateTest.java b/integration-tests/weaviate/src/test/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateTest.java
index 65b66180d1..331b2d5d62 100644
--- a/integration-tests/weaviate/src/test/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateTest.java
+++ b/integration-tests/weaviate/src/test/java/org/apache/camel/quarkus/component/weaviate/it/WeaviateTest.java
@@ -225,6 +225,128 @@ class WeaviateTest {
.body("result", Matchers.is(true));
}
+ @Test
+ public void newOperations() {
+ String collectionName = "WeaviateCQNewOps" + System.currentTimeMillis();
+
+ boolean collectionCreated = false;
+ try {
+ createCollection(collectionName);
+ LOG.infof("Collection created: %s", collectionName);
+ collectionCreated = true;
+
+ batchCreate(collectionName);
+
+ aggregate(collectionName);
+
+ // BM25 for "10" finds green and yellow (both have age=10) — pure text, no ranking preference
+ bm25Query(collectionName);
+
+ // Hybrid for "10" with vector [7.5, 8.5, 9.5] and alpha=0.75 also matches both,
+ // but ranks yellow first because its vector is closest to the query vector
+ // and the vector-heavy alpha makes that dominate
+ hybridQuery(collectionName);
+
+ } finally {
+ if (collectionCreated) {
+ deleteCollection(collectionName);
+ }
+ }
+ }
+
+ private void batchCreate(String collectionName) {
+ List<Map<String, Object>> objects = List.of(
+ Map.of("properties", Map.of("sky", "green", "age", "10"),
+ "vector", Arrays.asList(4.0, 5.0, 6.0)),
+ Map.of("properties", Map.of("sky", "red", "age", "20"),
+ "vector", Arrays.asList(7.0, 8.0, 9.0)),
+ Map.of("properties", Map.of("sky", "yellow", "age", "10"),
+ "vector", Arrays.asList(7.5, 8.5, 9.5)));
+
+ Map<String, Object> payload = Map.of(
+ "body", objects,
+ WeaviateVectorDbHeaders.ACTION, WeaviateVectorDbAction.BATCH_CREATE,
+ WeaviateVectorDbHeaders.COLLECTION_NAME, collectionName);
+
+ List<String> uuids = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(payload)
+ .post("/weaviate/request")
+ .then()
+ .statusCode(200)
+ .body("uuids", Matchers.hasSize(3))
+ .body("errors", Matchers.empty())
+ .extract().path("uuids");
+
+ queryById(collectionName, uuids.get(0))
+ .body("result", Matchers.aMapWithSize(1))
+ .body("result." + uuids.get(0), Matchers.is(Map.of("sky", "green", "age", "10")));
+
+ queryById(collectionName, uuids.get(1))
+ .body("result", Matchers.aMapWithSize(1))
+ .body("result." + uuids.get(1), Matchers.is(Map.of("sky", "red", "age", "20")));
+
+ queryById(collectionName, uuids.get(2))
+ .body("result", Matchers.aMapWithSize(1))
+ .body("result." + uuids.get(2), Matchers.is(Map.of("sky", "yellow", "age", "10")));
+ }
+
+ private void aggregate(String collectionName) {
+ Map<String, Object> payload = Map.of(
+ WeaviateVectorDbHeaders.ACTION, WeaviateVectorDbAction.AGGREGATE,
+ WeaviateVectorDbHeaders.COLLECTION_NAME, collectionName);
+
+ RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(payload)
+ .post("/weaviate/request")
+ .then()
+ .statusCode(200)
+ .body("totalCount", Matchers.equalTo(3));
+ }
+
+ private void bm25Query(String collectionName) {
+ Map<String, Object> fields = Map.of("sky", "", "age", "");
+
+ Map<String, Object> payload = Map.of(
+ "body", "10",
+ WeaviateVectorDbHeaders.ACTION, WeaviateVectorDbAction.BM25_QUERY,
+ WeaviateVectorDbHeaders.COLLECTION_NAME, collectionName,
+ WeaviateVectorDbHeaders.QUERY_TOP_K, 10,
+ WeaviateVectorDbHeaders.FIELDS, fields);
+
+ RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(payload)
+ .post("/weaviate/request")
+ .then()
+ .statusCode(200)
+ .body("result", Matchers.hasSize(2))
+ .body("result.sky", Matchers.containsInAnyOrder("green", "yellow"));
+ }
+
+ private void hybridQuery(String collectionName) {
+ Map<String, Object> fields = Map.of("sky", "", "age", "");
+
+ Map<String, Object> payload = Map.of(
+ "body", "10",
+ WeaviateVectorDbHeaders.ACTION, WeaviateVectorDbAction.HYBRID_QUERY,
+ WeaviateVectorDbHeaders.COLLECTION_NAME, collectionName,
+ WeaviateVectorDbHeaders.QUERY_TOP_K, 10,
+ WeaviateVectorDbHeaders.HYBRID_ALPHA, 0.75,
+ WeaviateVectorDbHeaders.QUERY_VECTOR, Arrays.asList(7.5, 8.5, 9.5),
+ WeaviateVectorDbHeaders.FIELDS, fields);
+
+ RestAssured.given()
+ .contentType(ContentType.JSON)
+ .body(payload)
+ .post("/weaviate/request")
+ .then()
+ .statusCode(200)
+ .body("result", Matchers.hasSize(Matchers.greaterThanOrEqualTo(2)))
+ .body("result[0].sky", Matchers.equalTo("yellow"));
+ }
+
private ValidatableResponse query(String collectionName, List<Float> values, Map<String, String> fields) {
return query(collectionName, values, fields, false);
}